/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.ambari.metrics.spark


import java.io.{FileInputStream, IOException, InputStream}
import java.util
import java.util.Properties
import java.util.logging.LogManager

import com.fasterxml.jackson.databind.ObjectMapper
import org.apache.ambari.metrics.alertservice.prototype.core.MetricsCollectorInterface
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._
import org.apache.ambari.metrics.alertservice.prototype.methods.{AnomalyDetectionTechnique, MetricAnomaly}
import org.apache.ambari.metrics.alertservice.prototype.methods.ema.{EmaModelLoader, EmaTechnique}
import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics
import org.apache.log4j.Logger
import org.apache.spark.storage.StorageLevel

object MetricAnomalyDetector {

  /*
    Load current EMA model
    Filter step - Check if anomaly
    Collect / Write to AMS / Print.
   */

//  var brokers = "avijayan-ams-1.openstacklocal:2181,avijayan-ams-2.openstacklocal:2181,avijayan-ams-3.openstacklocal:2181"
//  var groupId = "ambari-metrics-group"
//  var topicName = "ambari-metrics-topic"
//  var numThreads = 1
//  val anomalyDetectionModels: Array[AnomalyDetectionTechnique] = Array[AnomalyDetectionTechnique]()
//
//  def readProperties(propertiesFile: String): Properties = try {
//    val properties = new Properties
//    var inputStream = ClassLoader.getSystemResourceAsStream(propertiesFile)
//    if (inputStream == null) inputStream = new FileInputStream(propertiesFile)
//    properties.load(inputStream)
//    properties
//  } catch {
//    case ioEx: IOException =>
//      null
//  }
//
//  def main(args: Array[String]): Unit = {
//
//    @transient
//    lazy val log = org.apache.log4j.LogManager.getLogger("MetricAnomalyDetectorLogger")
//
//    if (args.length < 1) {
//      System.err.println("Usage: MetricSparkConsumer <input-config-file>")
//      System.exit(1)
//    }
//
//    //Read properties
//    val properties = readProperties(propertiesFile = args(0))
//
//    //Load EMA parameters - w, n
//    val emaW = properties.getProperty("emaW").toDouble
//    val emaN = properties.getProperty("emaN").toDouble
//
//    //collector info
//    val collectorHost: String = properties.getProperty("collectorHost")
//    val collectorPort: String = properties.getProperty("collectorPort")
//    val collectorProtocol: String = properties.getProperty("collectorProtocol")
//    val anomalyMetricPublisher = new MetricsCollectorInterface(collectorHost, collectorProtocol, collectorPort)
//
//    //Instantiate Kafka stream reader
//    val sparkConf = new SparkConf().setAppName("AmbariMetricsAnomalyDetector")
//    val streamingContext = new StreamingContext(sparkConf, Duration(10000))
//
//    val topicsSet = topicName.toSet
//    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
////    val stream = KafkaUtils.createDirectStream()
//
//    val kafkaStream = KafkaUtils.createStream(streamingContext, zkQuorum, groupId, Map(topicName -> numThreads), StorageLevel.MEMORY_AND_DISK_SER_2)
//    kafkaStream.print()
//
//    var timelineMetricsStream = kafkaStream.map( message => {
//      val mapper = new ObjectMapper
//      val metrics = mapper.readValue(message._2, classOf[TimelineMetrics])
//      metrics
//    })
//    timelineMetricsStream.print()
//
//    var appMetricStream = timelineMetricsStream.map( timelineMetrics => {
//      (timelineMetrics.getMetrics.get(0).getAppId, timelineMetrics)
//    })
//    appMetricStream.print()
//
//    var filteredAppMetricStream = appMetricStream.filter( appMetricTuple => {
//      appIds.contains(appMetricTuple._1)
//    } )
//    filteredAppMetricStream.print()
//
//    filteredAppMetricStream.foreachRDD( rdd => {
//      rdd.foreach( appMetricTuple => {
//        val timelineMetrics = appMetricTuple._2
//        logger.info("Received Metric (1): " + timelineMetrics.getMetrics.get(0).getMetricName)
//        log.info("Received Metric (2): " + timelineMetrics.getMetrics.get(0).getMetricName)
//        for (timelineMetric <- timelineMetrics.getMetrics) {
//          var anomalies = emaModel.test(timelineMetric)
//          anomalyMetricPublisher.publish(anomalies)
//        }
//      })
//    })
//
//    streamingContext.start()
//    streamingContext.awaitTermination()
//  }
  }