'use strict';

const constants = require('.././constants_local.json');
const secrets = require('./secrets.json')
const fetch = require('node-fetch');
const util = require('util')
const prom = require('prom-client');
const sharedMeta = require('./shared_meta');

const Registry = prom.Registry;
const register = new Registry();

const alpha = constants.metrics.alpha
let log_channel = constants.topics.log_channel,
    metrics = {  }


const intervalCollector = prom.collectDefaultMetrics({ prefix: 'xanadu', timeout: 5000, register });
const workerCountMetric = new prom.Gauge({ name: "worker_count", help: "worker count" });
const warmstartMetric = new prom.Histogram({ name: "warmstart", help: "warm start latency" });
const coldstartMetric = new prom.Histogram({ name: "coldstart", help: "cold start latency"});
const starttimeMetric = new prom.Histogram({ name: "starttime", help: "worker start times" });
const requestMetric = new prom.Summary({ name: "requests", help: "request RTT times",
    percentiles: [0.01, 0.05, 0.5, 0.9, 0.95, 0.99, 0.999]
});

register.registerMetric(workerCountMetric);
register.registerMetric(warmstartMetric);
register.registerMetric(coldstartMetric);
register.registerMetric(starttimeMetric);
register.registerMetric(requestMetric);

let metricsDB = sharedMeta.metricsDB
let kafka = require('kafka-node'),
    Producer = kafka.Producer,
    client = new kafka.KafkaClient({
        kafkaHost: constants.network.external.kafka_host,
        autoConnect: true
    }),
    producer = new Producer(client)

/**
 * Function called to report metric data related to functions
 * @param {JSON} metric 
 */
function collectMetrics(metric) {
    /**
     * If metrics for a new function comes in, 
     * provision required structure for the function
     */
    if (!(metric.functionHash in metrics)) {
        metrics[metric.functionHash] = {}
    } 
    if (!(metric.runtime in metrics[metric.functionHash])) {
        metrics[metric.functionHash][metric.runtime] = {
            shortterm: {
                coldstart: 0,
                coldstart_total_request: 0,
                warm_total_request: 0,
                scale_count: 0,
                warmstart: 0,
                worker_count: 0,
                starttime: 0
            }
        }
    }
    if (metric.type === 'coldstart') {
        metrics[metric.functionHash][metric.runtime].shortterm.coldstart += metric.value
        metrics[metric.functionHash][metric.runtime].shortterm.coldstart_total_request += 1
        coldstartMetric.observe(metric.value)
        requestMetric.observe(metric.value)
    } else if (metric.type === 'warmstart') {
        metrics[metric.functionHash][metric.runtime].shortterm.warmstart += metric.value
        metrics[metric.functionHash][metric.runtime].shortterm.warm_total_request += 1
        warmstartMetric.observe(metric.value)
        requestMetric.observe(metric.value)
    } else if (metric.type === 'scale') {
        metrics[metric.functionHash][metric.runtime].shortterm.worker_count = metric.value
        workerCountMetric.set(metric.value)
        if (metric.starttime !== undefined) {
            metrics[metric.functionHash][metric.runtime].shortterm.starttime += metric.starttime
            metrics[metric.functionHash][metric.runtime].shortterm.scale_count += 1
            starttimeMetric.observe(metric.starttime)
        }
    }
    
}


/**
 * Run periodically to calculate average runtime metrics like coldstart and
 * warmstart latencies.
 * The module provides two granularities for metrics - shortterm and longterm
 * shortterm - realtime data at a granularity of 5s (set in dispatch_manager/lib.js)
 *             shortterm data is calculated using Simple Moving Average (SMA)
 * longterm - longterm data is held and averaged out over a period of time.
 *            longterm data is calculated using Expontential Moving Average (EMA)
 */
async function broadcastMetrics() {
    if (Object.keys(metrics).length !== 0) {
        for (let [functionHash, data] of Object.entries(metrics)) {
            for (let [runtime, metricData] of Object.entries(data)) {
                if (metricData.shortterm.coldstart != 0 || metricData.shortterm.longterm != 0) {
                    let { metric, dbData } = await fetchData(functionHash, metricData, runtime)
                    /**
                    * Shortterm moving average
                    */
                    metric.shortterm.coldstart /= (metric.shortterm.coldstart_total_request != 0) ?
                        metric.shortterm.coldstart_total_request : 1
                    metric.shortterm.starttime /= (metric.shortterm.scale_count != 0) ?
                        metric.shortterm.scale_count : 1
                    metric.shortterm.warmstart /= (metric.shortterm.warm_total_request != 0) ?
                        metric.shortterm.warm_total_request : 1
                    
                    /**
                     * Longterm exponential moving average
                     */
                    if (metric.shortterm.coldstart != 0)
                        metric.longterm.coldstart = (metric.longterm.coldstart != 0) ? metric.longterm.coldstart * alpha
                            + metric.shortterm.coldstart * (1 - alpha) : metric.shortterm.coldstart
                    if (metric.shortterm.starttime && metric.shortterm.starttime != 0)
                        metric.longterm.starttime = (metric.longterm.starttime != 0) ? metric.longterm.starttime * alpha
                            + metric.shortterm.starttime * (1 - alpha) : metric.shortterm.starttime
                    if (metric.shortterm.warmstart != 0)
                        metric.longterm.warmstart = (metric.longterm.warmstart != 0) ? metric.longterm.warmstart * alpha
                            + metric.shortterm.warmstart * (1 - alpha) : metric.shortterm.warmstart
                    dbData[runtime] = {
                        coldstart: metric.longterm.coldstart,
                        warmstart: metric.longterm.warmstart,
                        starttime: metric.longterm.starttime
                    }
                    
                    let payload = {
                        method: 'put',
                        body: JSON.stringify(dbData),
                        headers: { 'Content-Type': 'application/json' }
                    }
                    await fetch(metricsDB + functionHash, payload)
                    
                    
                    metric.timestamp = Date.now()
                }
            }
        }
        
        let log = [{
            topic: log_channel,
            messages: JSON.stringify({
                metrics
            }),
            partition: 0
        }]
        producer.send(log, () => { })
        
        for (let [functionHash, data] of Object.entries(metrics)) {
            for (let [runtime, metric] of Object.entries(data)) {
                metric.shortterm = {
                    coldstart: 0,
                    coldstart_total_request: 0,
                    warm_total_request: 0,
                    warmstart: 0,
                    worker_count: 0,
                    starttime: 0,
                    scale_count: 0
                }
            }
        }
    }
}

/**
 * Function to fetch the latest data from metric DB
 * @param {String} functionHash 
 * @param {JSON} metric 
 */
async function fetchData(functionHash, metric, runtime) {
    let res = await fetch(metricsDB + functionHash)
    let json = await res.json()
    
    if (json.error === "not_found" || json[runtime] === undefined) {
        metric.longterm = {
            coldstart: 0,
            warmstart: 0,
            starttime: 0
        }
    } else {
        metric.longterm = {
            coldstart: json[runtime].coldstart,
            warmstart: json[runtime].warmstart,
            starttime: (json[runtime].starttime) ? json[runtime].starttime: 0
        }
    }
    return {
        metric,
        dbData: (json.error === "not_found")? {}: json
    }
}

module.exports = {
    collectMetrics, broadcastMetrics, register
}
