/**
 * 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 com.alibaba.jstorm.metric;

import backtype.storm.generated.MetricInfo;
import backtype.storm.generated.MetricSnapshot;
import backtype.storm.generated.TopologyMetric;
import backtype.storm.utils.Utils;
import com.alibaba.jstorm.client.ConfigExtension;
import com.alibaba.jstorm.common.metric.*;
import com.alibaba.jstorm.common.metric.codahale.JAverageReservoir;
import com.alibaba.jstorm.common.metric.codahale.JAverageSnapshot;
import com.alibaba.jstorm.common.metric.snapshot.*;
import com.alibaba.jstorm.utils.JStormUtils;
import com.alibaba.jstorm.utils.TimeUtils;
import com.codahale.metrics.*;
import com.google.common.collect.Maps;
import com.google.common.primitives.Longs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * @author Cody (weiyue.wy@alibaba-inc.com)
 * @since 2.0.5
 */
public class MetricUtils {
    private static final Logger LOG = LoggerFactory.getLogger(MetricUtils.class);

    public static final char AT = '@';
    public static final String DELIM = AT + "";
    public static final String EMPTY = "";
    public static final String DEFAULT_GROUP = "sys";
    public static final int NETTY_METRIC_PAGE_SIZE = 200;

    public static volatile boolean metricAccurateCal = false;

    public static boolean isValidId(long metricId) {
        return metricId != 0;
    }

    public static MetricInfo mkMetricInfo() {
        MetricInfo ret = new MetricInfo();
        ret.set_metrics(new HashMap<String, Map<Integer, MetricSnapshot>>());

        return ret;
    }

    public static TopologyMetric mkTopologyMetric() {
        TopologyMetric emptyTopologyMetric = new TopologyMetric();

        emptyTopologyMetric.set_topologyMetric(new MetricInfo());
        emptyTopologyMetric.set_componentMetric(new MetricInfo());
        emptyTopologyMetric.set_workerMetric(new MetricInfo());
        emptyTopologyMetric.set_taskMetric(new MetricInfo());
        emptyTopologyMetric.set_streamMetric(new MetricInfo());
        emptyTopologyMetric.set_nettyMetric(new MetricInfo());

        return emptyTopologyMetric;
    }

    public static boolean isEnableNettyMetrics(Map stormConf) {
        boolean enableMetrics = ConfigExtension.isEnableMetrics(stormConf);
        if (!enableMetrics) {
            return false;
        }

        int maxWorkerNumForNetty = ConfigExtension.getTopologyMaxWorkerNumForNettyMetrics(stormConf);
        int workerNum = JStormUtils.parseInt(stormConf.get("topology.workers"), 1);
        return workerNum < maxWorkerNumForNetty;
    }

    /**
     * a metric name composites of: type@topologyId@componentId@taskId@streamId@group@name for non-worker metrics
     * OR type@topologyId@host@port@group@name for worker metrics
     */
    public static String metricName(String type, String topologyId, String componentId, int taskId, String streamId, String group, String name) {
        return concat(type, topologyId, componentId, taskId, streamId, group, name);
    }

    public static String streamMetricName(String topologyId, String componentId, int taskId, String streamId, String name, MetricType type) {
        return concat(MetaType.STREAM.getV() + type.getV(), topologyId, componentId, taskId, streamId, DEFAULT_GROUP, name);
    }

    public static String workerMetricName(String topologyId, String host, int port, String name, MetricType type) {
        return concat(MetaType.WORKER.getV() + type.getV(), topologyId, host, port, DEFAULT_GROUP, name);
    }

    public static String workerMetricName(String name, MetricType type) {
        return concat(MetaType.WORKER.getV() + type.getV(), EMPTY, EMPTY, 0, DEFAULT_GROUP, name);
    }

    public static String nettyMetricName(String name, MetricType type) {
        return concat(MetaType.NETTY.getV() + type.getV(), EMPTY, EMPTY, 0, JStormMetrics.NETTY_GROUP, name);
    }

    public static String workerMetricPrefix(String topologyId, String host, int port) {
        return concat(MetaType.WORKER.getV(), topologyId, host, port);
    }

    public static String taskMetricName(String topologyId, String componentId, int taskId, String name, MetricType type) {
        return concat(MetaType.TASK.getV() + type.getV(), topologyId, componentId, taskId, EMPTY, DEFAULT_GROUP, name);
    }

    public static String taskMetricName(String topologyId, String componentId, int taskId, String group, String name, MetricType type) {
        return concat(MetaType.TASK.getV() + type.getV(), topologyId, componentId, taskId, EMPTY, group, name);
    }

    public static String compMetricName(String topologyId, String componentId, String name, MetricType type) {
        return concat(MetaType.COMPONENT.getV() + type.getV(), topologyId, componentId, 0, EMPTY, DEFAULT_GROUP, name);
    }

    public static String removeDelimIfPossible(String name) {
        if (name.contains(DELIM)) {
            return name.replace(DELIM, EMPTY);
        }
        return name;
    }

    public static MetaType metaType(String name) {
        return MetaType.parse(name.charAt(0) + EMPTY);
    }

    public static MetricType metricType(String name) {
        return MetricType.parse(name.charAt(1) + EMPTY);
    }

    public static String metricName(String name) {
        String[] parts = name.split(DELIM);
        return parts[parts.length - 1];
    }

    /**
     * make streamId empty, remain other parts the same
     */
    public static String stream2taskName(String old) {
        String[] parts = old.split(DELIM);
        if (parts.length >= 7) {
            parts[0] = MetaType.TASK.getV() + parts[0].charAt(1);
            parts[parts.length - 3] = EMPTY;

            String metricName = getMergeMetricName(parts[parts.length - 1]);
            parts[parts.length - 1] = metricName;
        }
        return concat(parts);
    }

    /**
     * make taskId=0 and streamId empty.
     */
    public static String task2compName(String old) {
        String[] parts = old.split(DELIM);
        if (parts.length >= 7) {
            parts[0] = MetaType.COMPONENT.getV() + parts[0].charAt(1);
            parts[parts.length - 3] = EMPTY;
            parts[parts.length - 4] = "0";
        }
        return concat(parts);
    }

    /**
     * converts a task metric name to a component-level stream metric name
     */
    public static String stream2compStreamName(String old) {
        String[] parts = old.split(DELIM);
        if (parts.length >= 7) {
            parts[0] = MetaType.COMPONENT_STREAM.getV() + parts[0].charAt(1);
            //parts[parts.length - 3] = EMPTY; retain stream name
            parts[parts.length - 4] = "0"; // task
            parts[parts.length - 1] = getMergeMetricName(parts[parts.length - 1]);
        }
        return concat(parts);
    }

    /**
     * make taskId=0 and streamId empty and metricName remain the string after `.`.
     */
    public static String task2MergeCompName(String old) {
        String[] parts = old.split(DELIM);
        if (parts.length >= 7) {
            parts[0] = MetaType.COMPONENT.getV() + parts[0].charAt(1);
            parts[parts.length - 3] = EMPTY;
            parts[parts.length - 4] = "0";

            String metricName = getMergeMetricName(parts[parts.length - 1]);
            parts[parts.length - 1] = metricName;
        }
        return concat(parts);
    }

    private static String getMergeMetricName(String name) {
        int dotIndex = name.lastIndexOf(".");
        if (dotIndex != -1) {
            return name.substring(dotIndex + 1);
        }
        return name;
    }

    /**
     * change component metric name to topology metric name
     */
    public static String comp2topologyName(String old) {
        String[] parts = old.split(DELIM);
        parts[0] = MetaType.TOPOLOGY.getV() + parts[0].charAt(1);
        // type + topologyId + comp + task + group + name
        return concat(parts[0], parts[1], EMPTY, "0", parts[5], parts[6]);
    }

    public static String worker2topologyName(String old) {
        String[] parts = old.split(DELIM);
        if (parts.length >= 5) {
            parts[0] = MetaType.TOPOLOGY.getV() + parts[0].charAt(1);
            parts[2] = EMPTY;   // host
            parts[3] = "0";     // port
        }
        return concat(parts);
    }

    public static String topo2clusterName(String old) {
        String[] parts = old.split(DELIM);
        parts[1] = JStormMetrics.CLUSTER_METRIC_KEY;
        return concat(parts);
    }

    public static String concat(Object... args) {
        StringBuilder sb = new StringBuilder(50);
        int last = args.length - 1;
        if (args[last] instanceof String) {
            args[last] = removeDelimIfPossible((String) args[last]);
        }
        for (Object arg : args) {
            sb.append(arg).append(DELIM);
        }
        sb.deleteCharAt(sb.length() - 1);
        return sb.toString();
    }

    public static String concat2(Object... args) {
        StringBuilder sb = new StringBuilder(50);
        for (Object arg : args) {
            sb.append(arg).append(DELIM);
        }
        sb.deleteCharAt(sb.length() - 1);
        return sb.toString();
    }

    public static Histogram metricSnapshot2Histogram(MetricSnapshot snapshot) {
        Histogram histogram;
        if (metricAccurateCal) {
            histogram = new Histogram(new ExponentiallyDecayingReservoir());
            byte[] points = snapshot.get_points();
            int len = snapshot.get_pointSize();
            updateHistogramPoints(histogram, points, len);
        } else {
            histogram = new Histogram(new JAverageReservoir());
            JAverageSnapshot averageSnapshot = (JAverageSnapshot) histogram.getSnapshot();
            averageSnapshot.setMetricSnapshot(snapshot.deepCopy());
        }
        return histogram;
    }

    public static void updateHistogramPoints(Histogram histogram, byte[] points, int len) {
        if (points != null && len > 0) {
            for (int i = 0; i < len; i++) {
                long pt = Bytes.toLong(points, i * Longs.BYTES, Longs.BYTES);
                histogram.update(pt);
            }
        }
    }

    public static Map<Integer, MetricSnapshot> toThriftCounterSnapshots(Map<Integer, AsmSnapshot> snapshots) {
        Map<Integer, MetricSnapshot> ret = Maps.newHashMapWithExpectedSize(snapshots.size());
        for (Map.Entry<Integer, AsmSnapshot> entry : snapshots.entrySet()) {
            ret.put(entry.getKey(), convert((AsmCounterSnapshot) entry.getValue()));
        }
        return ret;
    }

    public static Map<Integer, MetricSnapshot> toThriftGaugeSnapshots(Map<Integer, AsmSnapshot> snapshots) {
        Map<Integer, MetricSnapshot> ret = Maps.newHashMapWithExpectedSize(snapshots.size());
        for (Map.Entry<Integer, AsmSnapshot> entry : snapshots.entrySet()) {
            ret.put(entry.getKey(), convert((AsmGaugeSnapshot) entry.getValue()));
        }
        return ret;
    }

    public static Map<Integer, MetricSnapshot> toThriftMeterSnapshots(Map<Integer, AsmSnapshot> snapshots) {
        Map<Integer, MetricSnapshot> ret = Maps.newHashMapWithExpectedSize(snapshots.size());
        for (Map.Entry<Integer, AsmSnapshot> entry : snapshots.entrySet()) {
            ret.put(entry.getKey(), convert((AsmMeterSnapshot) entry.getValue()));
        }
        return ret;
    }

    public static Map<Integer, MetricSnapshot> toThriftHistoSnapshots(Map<Integer, AsmSnapshot> snapshots) {
        Map<Integer, MetricSnapshot> ret = Maps.newHashMapWithExpectedSize(snapshots.size());
        for (Map.Entry<Integer, AsmSnapshot> entry : snapshots.entrySet()) {
            MetricSnapshot histogramSnapshot = convert((AsmHistogramSnapshot) entry.getValue());
            if (histogramSnapshot != null) {
                ret.put(entry.getKey(), histogramSnapshot);
            }
        }
        return ret;
    }

    public static Map<Integer, MetricSnapshot> toThriftHistoSnapshots(MetaType metaType, Map<Integer, AsmSnapshot> snapshots) {
        Map<Integer, MetricSnapshot> ret = Maps.newHashMapWithExpectedSize(snapshots.size());
        for (Map.Entry<Integer, AsmSnapshot> entry : snapshots.entrySet()) {
            MetricSnapshot histogramSnapshot = convert(metaType, (AsmHistogramSnapshot) entry.getValue());
            if (histogramSnapshot != null) {
                ret.put(entry.getKey(), histogramSnapshot);
            }
        }
        return ret;
    }

    public static Map<Integer, MetricSnapshot> toThriftSnapshots(Map<Integer, AsmSnapshot> snapshots, MetricType metricType) {
        if (metricType == MetricType.COUNTER) {
            return MetricUtils.toThriftCounterSnapshots(snapshots);
        } else if (metricType == MetricType.GAUGE) {
            return MetricUtils.toThriftGaugeSnapshots(snapshots);
        } else if (metricType == MetricType.METER) {
            return MetricUtils.toThriftMeterSnapshots(snapshots);
        } else if (metricType == MetricType.HISTOGRAM) {
            return MetricUtils.toThriftHistoSnapshots(snapshots);
        } else {
            throw new RuntimeException("invalid metricType " + metricType);
        }
    }

    public static MetricSnapshot convert(AsmCounterSnapshot snapshot) {
        MetricSnapshot ret = new MetricSnapshot();
        ret.set_metricId(snapshot.getMetricId());
        ret.set_ts(TimeUtils.alignTimeToMin(snapshot.getTs()));
        ret.set_metricType(MetricType.COUNTER.getT());
        ret.set_longValue(snapshot.getV());

        return ret;
    }

    public static MetricSnapshot convert(AsmGaugeSnapshot snapshot) {
        MetricSnapshot ret = new MetricSnapshot();
        ret.set_metricId(snapshot.getMetricId());
        ret.set_ts(TimeUtils.alignTimeToMin(snapshot.getTs()));
        ret.set_metricType(MetricType.GAUGE.getT());
        ret.set_doubleValue(snapshot.getV());

        return ret;
    }

    public static MetricSnapshot convert(AsmMeterSnapshot snapshot) {
        MetricSnapshot ret = new MetricSnapshot();
        ret.set_metricId(snapshot.getMetricId());
        ret.set_ts(TimeUtils.alignTimeToMin(snapshot.getTs()));
        ret.set_metricType(MetricType.METER.getT());

        ret.set_m1(snapshot.getM1());
        ret.set_m5(snapshot.getM5());
        ret.set_m15(snapshot.getM15());
        ret.set_mean(snapshot.getMean());

        return ret;
    }

    public static MetricSnapshot convert(AsmHistogramSnapshot snapshot) {
        // some histograms are never updated, skip such metrics
        //if (snapshot.getSnapshot().getValues().length == 0) {
        //    return null;
        //}

        MetricSnapshot ret = new MetricSnapshot();
        ret.set_metricId(snapshot.getMetricId());
        ret.set_ts(TimeUtils.alignTimeToMin(snapshot.getTs()));
        ret.set_metricType(MetricType.HISTOGRAM.getT());

        Snapshot ws = snapshot.getSnapshot();
        ret.set_min(ws.getMin());
        ret.set_max(ws.getMax());
        ret.set_p50(ws.getMedian());
        ret.set_p75(ws.get75thPercentile());
        ret.set_p95(ws.get95thPercentile());
        ret.set_p98(ws.get98thPercentile());
        ret.set_p99(ws.get99thPercentile());
        ret.set_p999(ws.get999thPercentile());
        ret.set_mean(ws.getMean());
        ret.set_stddev(ws.getStdDev());
        ret.set_points(new byte[0]);
        ret.set_pointSize(0);

        return ret;
    }


    public static MetricSnapshot convert(MetaType metaType, AsmHistogramSnapshot snapshot) {
        // some histograms are never updated, skip such metrics
        //if (snapshot.getSnapshot().getValues().length == 0) {
        //    return null;
        //}

        MetricSnapshot ret = new MetricSnapshot();
        ret.set_metricId(snapshot.getMetricId());
        ret.set_ts(TimeUtils.alignTimeToMin(snapshot.getTs()));
        ret.set_metricType(MetricType.HISTOGRAM.getT());

        Snapshot ws = snapshot.getSnapshot();
        ret.set_min(ws.getMin());
        ret.set_max(ws.getMax());
        ret.set_p50(ws.getMedian());
        ret.set_p75(ws.get75thPercentile());
        ret.set_p95(ws.get95thPercentile());
        ret.set_p98(ws.get98thPercentile());
        ret.set_p99(ws.get99thPercentile());
        ret.set_p999(ws.get999thPercentile());
        ret.set_mean(ws.getMean());
        ret.set_stddev(ws.getStdDev());

        // only upload points for component metrics
        if (metricAccurateCal && (metaType == MetaType.COMPONENT || metaType == MetaType.TOPOLOGY)) {
            byte[] points = longs2bytes(ws.getValues());

            ret.set_points(points);
            ret.set_pointSize(ws.getValues().length);
        } else {
            ret.set_points(new byte[0]);
            ret.set_pointSize(0);
        }

        return ret;
    }

    public static byte[] longs2bytes(long[] points) {
        int len = points.length;
        byte[] bytePts = new byte[len * Longs.BYTES];
        for (int i = 0; i < len; i++) {
            Bytes.putLong(bytePts, i * Longs.BYTES, points[i]);
        }

        return bytePts;
    }

    public static String getMetricName(String fullName) {
        String[] parts = fullName.split(DELIM);
        return parts[parts.length - 1];
    }

    public static String str(Object obj) {
        if (obj instanceof MetricSnapshot) {
            MetricSnapshot snapshot = (MetricSnapshot) obj;
            MetricType type = MetricType.parse(snapshot.get_metricType());
            if (type == MetricType.COUNTER) {
                return counterStr(snapshot);
            } else if (type == MetricType.GAUGE) {
                return gaugeStr(snapshot);
            } else if (type == MetricType.METER) {
                return meterStr(snapshot);
            } else if (type == MetricType.HISTOGRAM) {
                return histogramStr(snapshot);
            }
        } else if (obj instanceof AsmSnapshot) {
            if (obj instanceof AsmCounterSnapshot) {
                return counterStr((AsmCounterSnapshot) obj);
            } else if (obj instanceof AsmGaugeSnapshot) {
                return gaugeStr((AsmGaugeSnapshot) obj);
            } else if (obj instanceof AsmMeterSnapshot) {
                return meterStr((AsmMeterSnapshot) obj);
            } else if (obj instanceof AsmHistogramSnapshot) {
                return histogramStr((AsmHistogramSnapshot) obj);
            }
        }
        return obj.toString();
    }

    public static String counterStr(MetricSnapshot snapshot) {
        StringBuilder sb = new StringBuilder(32);
        sb.append("id:").append(snapshot.get_metricId()).append(",v:").append(snapshot.get_longValue());

        return sb.toString();
    }

    public static String counterStr(AsmCounterSnapshot snapshot) {
        StringBuilder sb = new StringBuilder(32);
        sb.append("id:").append(snapshot.getMetricId()).append(",v:").append(snapshot.getV());

        return sb.toString();
    }

    public static String gaugeStr(MetricSnapshot snapshot) {
        StringBuilder sb = new StringBuilder(32);
        sb.append("id:").append(snapshot.get_metricId()).append(",v:").append(snapshot.get_doubleValue());

        return sb.toString();
    }

    public static String gaugeStr(AsmGaugeSnapshot snapshot) {
        StringBuilder sb = new StringBuilder(32);
        sb.append("id:").append(snapshot.getMetricId()).append(",v:").append(snapshot.getV());

        return sb.toString();
    }

    public static String meterStr(MetricSnapshot snapshot) {
        StringBuilder sb = new StringBuilder(50);
        sb.append("id:").append(snapshot.get_metricId());
        sb.append(",m1:").append(snapshot.get_m1()).append(",m5:").append(snapshot.get_m5())
                .append(",m15:").append(snapshot.get_m15()).append(",mean:").append(snapshot.get_mean());
        return sb.toString();
    }

    public static String meterStr(AsmMeterSnapshot snapshot) {
        StringBuilder sb = new StringBuilder(50);
        sb.append("id:").append(snapshot.getMetricId());
        sb.append(",m1:").append(snapshot.getM1()).append(",m5:").append(snapshot.getM5())
                .append(",m15:").append(snapshot.getM15()).append(",mean:").append(snapshot.getMean());
        return sb.toString();
    }

    public static String histogramStr(MetricSnapshot snapshot) {
        StringBuilder sb = new StringBuilder(128);
        sb.append("histogram");
        sb.append("(").append("id:").append(snapshot.get_metricId()).append(",min:").append(snapshot.get_min()).append(",max:")
                .append(snapshot.get_max()).append(",mean:").append(snapshot.get_mean()).append(",p50:").append(snapshot.get_p50())
                .append(",p75:").append(snapshot.get_p75()).append(",p95:").append(snapshot.get_p95()).append(",p98:")
                .append(snapshot.get_p98()).append(",p99:").append(snapshot.get_p99()).append(",pts:").append(snapshot.get_pointSize())
                .append(")");
        return sb.toString();
    }

    public static String histogramStr(AsmHistogramSnapshot snapshot) {
        StringBuilder sb = new StringBuilder(128);
        sb.append("histogram");
        Snapshot s = snapshot.getSnapshot();
        sb.append("(").append("id:").append(snapshot.getMetricId()).append(",min:").append(s.getMin()).append(",max:")
                .append(s.getMax()).append(",mean:").append(s.getMean()).append(",p50:").append(s.getMedian())
                .append(",p75:").append(s.get75thPercentile()).append(",p95:").append(s.get95thPercentile()).append(",p98:")
                .append(s.get98thPercentile()).append(",p99:").append(s.get99thPercentile()).append(")");
        return sb.toString();
    }

    public static void printMetricSnapshot(AsmMetric metric, Map<Integer, AsmSnapshot> snapshots) {
        StringBuilder sb = new StringBuilder(128);
        sb.append("metric:").append(metric.getMetricName()).append(", ");
        for (Map.Entry<Integer, AsmSnapshot> entry : snapshots.entrySet()) {
            sb.append("win:").append(entry.getKey()).append(", v:")
                    .append(getSnapshotDefaultValue(entry.getValue())).append("; ");
        }

        LOG.info(sb.toString());
    }

    public static double getSnapshotDefaultValue(AsmSnapshot snapshot) {
        if (snapshot instanceof AsmCounterSnapshot) {
            return ((AsmCounterSnapshot) snapshot).getV();
        } else if (snapshot instanceof AsmGaugeSnapshot) {
            return ((AsmGaugeSnapshot) snapshot).getV();
        } else if (snapshot instanceof AsmMeterSnapshot) {
            return ((AsmMeterSnapshot) snapshot).getM1();
        } else if (snapshot instanceof AsmHistogramSnapshot) {
            return ((AsmHistogramSnapshot) snapshot).getSnapshot().getMean();
        }
        return 0;
    }

    public static void printMetricInfo(MetricInfo metricInfo) {
        iterateMap(metricInfo.get_metrics());
    }

    public static void printMetricInfo(MetricInfo metricInfo, Set<String> metrics) {
        iterateMap(metricInfo.get_metrics(), metrics);
    }

    public static <T> void iterateMap(Map<String, Map<Integer, T>> map) {
        iterateMap(map, null);
    }

    public static <T> void iterateMap(Map<String, Map<Integer, T>> map, Set<String> metrics) {
        for (Map.Entry<String, Map<Integer, T>> entry : map.entrySet()) {
            String name = entry.getKey();
            boolean print = false;
            if (metrics == null) {
                print = true;
            } else {
                for (String metric : metrics) {
                    if (name.contains(metric)) {
                        print = true;
                        break;
                    }
                }
            }
            if (print) {
                Map<Integer, T> winData = entry.getValue();
                for (Map.Entry<Integer, T> win : winData.entrySet()) {
                    T v = win.getValue();
                    String str;
                    if (v instanceof MetricSnapshot) {
                        str = MetricUtils.str(v);
                    } else {
                        str = v.toString();
                    }
                    LOG.info("metric:{}, win:{}, data:{}", name, win.getKey(), str);
                }
            }
        }
    }

    private static <T> void iter(Map<String, T> map, Func func, Object... args) {
        for (Map.Entry<String, T> entry : map.entrySet()) {
            func.exec(entry, args);
        }
    }

    public interface Func {
        void exec(Object... args);
    }

    /**
     * print default value for all metrics, in the format of: name|type|value
     */
    public static void logMetrics(MetricInfo metricInfo) {
        Map<String, Map<Integer, MetricSnapshot>> metrics = metricInfo.get_metrics();
        if (metrics != null) {
            LOG.info("\nprint metrics:");
            for (Map.Entry<String, Map<Integer, MetricSnapshot>> entry : metrics.entrySet()) {
                String name = entry.getKey();
                MetricSnapshot metricSnapshot = entry.getValue().get(AsmWindow.M1_WINDOW);
                if (metricSnapshot != null) {
                    MetricType metricType = MetricType.parse(metricSnapshot.get_metricType());
                    double v;
                    if (metricType == MetricType.COUNTER) {
                        v = metricSnapshot.get_longValue();
                    } else if (metricType == MetricType.GAUGE) {
                        v = metricSnapshot.get_doubleValue();
                    } else if (metricType == MetricType.METER) {
                        v = metricSnapshot.get_m1();
                    } else if (metricType == MetricType.HISTOGRAM) {
                        v = metricSnapshot.get_mean();
                    } else {
                        v = 0;
                    }
                    LOG.info("{}|{}|{}", metricType, v, name);
                }
            }
            LOG.info("\n");
        }
    }

    public static void mergeMetricSnapshotMap(Map<Integer, MetricSnapshot> returnSnapshotMaps,
                                              Map<Integer, MetricSnapshot> rootSnapshotMap,
                                              MetricType metricType) {
        for (Map.Entry<Integer, MetricSnapshot> entry : rootSnapshotMap.entrySet()) {
            Integer key = entry.getKey();
            MetricSnapshot snapshot = entry.getValue();
            MetricSnapshot old = returnSnapshotMaps.get(key);
            if (old == null) {
                old = new MetricSnapshot(snapshot);
                returnSnapshotMaps.put(key, old);
            } else {
                if (metricType == MetricType.COUNTER) {
                    old.set_longValue(old.get_longValue() + snapshot.get_longValue());
                } else if (metricType == MetricType.GAUGE) {
                    old.set_doubleValue(old.get_doubleValue() + snapshot.get_doubleValue());
                } else if (metricType == MetricType.METER) {
                    old.set_mean(old.get_mean() + snapshot.get_mean());
                    old.set_m1(old.get_m1() + snapshot.get_m1());
                    old.set_m5(old.get_m5() + snapshot.get_m5());
                    old.set_m15(old.get_m15() + snapshot.get_m15());
                } else if (metricType == MetricType.HISTOGRAM) {
                    old.set_min(old.get_min() + snapshot.get_min());
                    old.set_max(old.get_max() + snapshot.get_max());
                    old.set_p50(old.get_p50() + snapshot.get_p50());
                    old.set_p75(old.get_p75() + snapshot.get_p75());
                    old.set_p95(old.get_p95() + snapshot.get_p95());
                    old.set_p98(old.get_p98() + snapshot.get_p98());
                    old.set_p99(old.get_p99() + snapshot.get_p99());
                    old.set_p999(old.get_p999() + snapshot.get_p999());
                    old.set_mean(old.get_mean() + snapshot.get_mean());
                    old.set_stddev(old.get_stddev() + snapshot.get_stddev());
                }
            }
        }
    }

    public static void main(String[] args) {
        String streamName = "SM@b2b_crm_order_PaymentSpout-1-1461743650@__topology_master" +
                "@2@__master_task_heartbeat@sys@com.alibaba.crm.searchincr.jstorm.martini.ordcash.PaymentToplogy.RecvTps";
        String taskName = stream2taskName(streamName);
        System.out.println(taskName);
        String compName = task2compName(taskName);
        System.out.println(compName);
        System.out.println(task2MergeCompName(taskName));
        String tpName = comp2topologyName(compName);
        System.out.println(tpName);
    }
}
