/**
 * 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
 * <p/>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p/>
 * 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.daemon.nimbus;

import backtype.storm.generated.MetricInfo;
import backtype.storm.generated.MetricSnapshot;
import backtype.storm.generated.TopologyMetric;
import backtype.storm.utils.Utils;
import com.alibaba.jstorm.callback.AsyncLoopThread;
import com.alibaba.jstorm.callback.RunnableCallback;
import com.alibaba.jstorm.client.ConfigExtension;
import com.alibaba.jstorm.cluster.Cluster;
import com.alibaba.jstorm.cluster.StormClusterState;
import com.alibaba.jstorm.common.metric.AsmGauge;
import com.alibaba.jstorm.common.metric.MetricMeta;
import com.alibaba.jstorm.daemon.nimbus.metric.uploader.DefaultMetricUploader;
import com.alibaba.jstorm.daemon.nimbus.metric.uploader.MetricUploader;
import com.alibaba.jstorm.metric.*;
import com.alibaba.jstorm.schedule.Assignment;
import com.alibaba.jstorm.schedule.default_assign.ResourceWorkerSlot;
import com.alibaba.jstorm.utils.JStormUtils;
import com.alibaba.jstorm.utils.TimeUtils;
import com.codahale.metrics.Gauge;
import com.google.common.collect.Sets;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.lang.management.ManagementFactory;
import java.lang.management.MemoryMXBean;
import java.lang.management.MemoryUsage;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicIntegerArray;

/**
 * Topology metrics thread which resides in nimbus.
 * This class is responsible for generating metrics IDs and uploading metrics to the underlying storage system.
 *
 * @author Cody (weiyue.wy@alibaba-inc.com)
 * @since 2.0.5
 */
public class TopologyMetricsRunnable extends Thread {
    private static final Logger LOG = LoggerFactory.getLogger(TopologyMetricsRunnable.class);

    protected JStormMetricCache metricCache;

    /**
     * map<topologyId, map<worker, metricInfo>>, local memory cache, keeps only one snapshot of metrics.
     */
    protected final ConcurrentMap<String, TopologyMetricContext> topologyMetricContexts =
            new ConcurrentHashMap<>();

    protected final BlockingDeque<TopologyMetricsRunnable.Event> queue = new LinkedBlockingDeque<>();

    private static final String PENDING_UPLOAD_METRIC_DATA = "__pending.upload.metrics__";
    private static final String PENDING_UPLOAD_METRIC_DATA_INFO = "__pending.upload.metrics.info__";

    // the slot is empty
    private static final int UNSET = 0;
    // the slot is ready for uploading
    private static final int SET = 1;
    // the slot is being uploaded
    private static final int UPLOADING = 2;
    // the slot will be set ready for uploading
    private static final int PRE_SET = 3;

    protected final AtomicIntegerArray metricStat;

    protected StormClusterState stormClusterState;

    protected MetricUploader metricUploader;

    protected AtomicBoolean isShutdown;
    protected String clusterName;
    protected int maxPendingUploadMetrics;

    private final boolean localMode;
    private final NimbusData nimbusData;
    private MetricQueryClient metricQueryClient;

    private ScheduledExecutorService clusterMetricsUpdateExecutor;

    /**
     * refreshes alive topologies every min or on startup.
     */
    protected AsyncLoopThread refreshTopologiesThread;

    /**
     * the thread for metric sending, checks every second.
     */
    private final Thread uploadThread = new MetricsUploadThread();

    /**
     * async flush metric meta
     */
    private final Thread flushMetricMetaThread = new FlushMetricMetaThread();

    /**
     * use default UUID generator
     */
    private final MetricIDGenerator metricIDGenerator = new DefaultMetricIDGenerator();

    public TopologyMetricsRunnable(final NimbusData nimbusData) {
        setName(getClass().getSimpleName());

        this.nimbusData = nimbusData;

        this.localMode = nimbusData.isLocalMode();
        if (localMode) {
            this.metricStat = new AtomicIntegerArray(1);
            return;
        }

        LOG.info("create topology metrics runnable.");
        this.metricCache = nimbusData.getMetricCache();
        this.stormClusterState = nimbusData.getStormClusterState();
        this.isShutdown = nimbusData.getIsShutdown();

        clusterName = ConfigExtension.getClusterName(nimbusData.getConf());
        if (clusterName == null) {
            throw new RuntimeException("cluster.name property must be set in storm.yaml!");
        }

        this.maxPendingUploadMetrics = ConfigExtension.getMaxPendingMetricNum(nimbusData.getConf());
        this.metricStat = new AtomicIntegerArray(this.maxPendingUploadMetrics);

        int cnt = 0;
        for (int i = 0; i < maxPendingUploadMetrics; i++) {
            TopologyMetricDataInfo obj = getMetricDataInfoFromCache(i);
            if (obj != null) {
                this.metricStat.set(i, SET);
                cnt++;
            }
        }
        LOG.info("pending upload metrics: {}", cnt);

        // init alive topologies from zk
        this.refreshTopologies();
        this.refreshTopologiesThread = new AsyncLoopThread(new RefreshTopologiesThread());

        this.clusterMetricsUpdateExecutor = Executors.newSingleThreadScheduledExecutor();
        this.clusterMetricsUpdateExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                int secOffset = TimeUtils.secOffset();
                int offset = 55;
                if (secOffset < offset) {
                    JStormUtils.sleepMs((offset - secOffset) * 1000);
                } else if (secOffset == offset) {
                    // do nothing
                } else {
                    JStormUtils.sleepMs((60 - secOffset + offset) * 1000);
                }

                LOG.info("cluster metrics force upload.");
                mergeAndUploadClusterMetrics();
            }
        }, 5, 60, TimeUnit.SECONDS);

        // track nimbus JVM heap
        JStormMetrics.registerWorkerGauge(JStormMetrics.NIMBUS_METRIC_KEY, MetricDef.MEMORY_USED,
                new AsmGauge(new Gauge<Double>() {
                    @Override
                    public Double getValue() {
                        MemoryMXBean memoryMXBean = ManagementFactory.getMemoryMXBean();
                        MemoryUsage memoryUsage = memoryMXBean.getHeapMemoryUsage();
                        return (double) memoryUsage.getUsed();
                    }
                }));
    }

    /**
     * init metric uploader
     */
    public void init() {
        String metricUploadClass = ConfigExtension.getMetricUploaderClass(nimbusData.getConf());
        if (StringUtils.isBlank(metricUploadClass)) {
            metricUploadClass = DefaultMetricUploader.class.getName();
        }
        // init metric uploader
        LOG.info("metric uploader class:{}", metricUploadClass);
        Object instance = Utils.newInstance(metricUploadClass);
        if (!(instance instanceof MetricUploader)) {
            throw new RuntimeException(metricUploadClass + " isn't MetricUploader class ");
        }
        this.metricUploader = (MetricUploader) instance;
        try {
            metricUploader.init(nimbusData);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        LOG.info("Successfully init {}", metricUploadClass);

        // init metric query client
        String metricQueryClientClass = ConfigExtension.getMetricQueryClientClass(nimbusData.getConf());
        if (!StringUtils.isBlank(metricQueryClientClass)) {
            LOG.info("metric query client class:{}", metricQueryClientClass);
            this.metricQueryClient = (MetricQueryClient) Utils.newInstance(metricQueryClientClass);
        } else {
            LOG.warn("use default metric query client class.");
            this.metricQueryClient = new DefaultMetricQueryClient();
        }
        try {
            metricQueryClient.init(nimbusData.getConf());
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        this.uploadThread.start();
        this.flushMetricMetaThread.start();

        LOG.info("init topology metric runnable done.");
    }

    public void shutdown() {
        LOG.info("Begin to shutdown");
        metricUploader.cleanup();

        LOG.info("Successfully shutdown");
    }

    @Override
    public void run() {
        while (!isShutdown.get()) {
            if (localMode) {
                return;
            }

            try {
                // wait for metricUploader to be ready, for some external plugin like database, it'll take a few seconds
                if (this.metricUploader != null) {
                    Event event = queue.poll();
                    if (event == null) {
                        continue;
                    }

                    if (event instanceof Remove) {
                        handleRemoveEvent((Remove) event);
                    } else if (event instanceof Update) {
                        handleUpdateEvent((Update) event);
                    } else if (event instanceof Refresh) {
                        handleRefreshEvent((Refresh) event);
                    } else if (event instanceof KillTopologyEvent) {
                        handleKillTopologyEvent((KillTopologyEvent) event);
                    } else if (event instanceof StartTopologyEvent) {
                        handleStartTopologyEvent((StartTopologyEvent) event);
                    } else if (event instanceof TaskDeadEvent) {
                        handleTaskDeadEvent((TaskDeadEvent) event);
                    } else if (event instanceof TaskStartEvent) {
                        handleTaskStartEvent((TaskStartEvent) event);
                    } else {
                        LOG.error("Unknown event type:{}", event.getClass());
                    }
                }
            } catch (Exception e) {
                if (!isShutdown.get()) {
                    LOG.error(e.getMessage(), e);
                }
            }
        }
    }


    public boolean isTopologyAlive(String topologyId) {
        return topologyMetricContexts.containsKey(topologyId);
    }

    private int getAndPresetFirstEmptyIndex() {
        for (int i = 0; i < maxPendingUploadMetrics; i++) {
            if (metricStat.get(i) == UNSET) {
                if (metricStat.compareAndSet(i, UNSET, PRE_SET)) {
                    return i;
                }
            }
        }
        return -1;
    }

    private int getFirstPendingUploadIndex() {
        for (int i = 0; i < maxPendingUploadMetrics; i++) {
            if (metricStat.get(i) == SET) {
                return i;
            }
        }
        return -1;
    }

    public void markUploaded(int idx) {
        this.metricCache.remove(PENDING_UPLOAD_METRIC_DATA + idx);
        this.metricCache.remove(PENDING_UPLOAD_METRIC_DATA_INFO + idx);
        this.metricStat.set(idx, UNSET);
    }

    public void markUploading(int idx) {
        this.metricStat.set(idx, UPLOADING);
    }

    public void markSet(int idx) {
        this.metricStat.set(idx, SET);
    }

    public TopologyMetric getMetricDataFromCache(int idx) {
        return (TopologyMetric) metricCache.get(PENDING_UPLOAD_METRIC_DATA + idx);
    }

    public TopologyMetricDataInfo getMetricDataInfoFromCache(int idx) {
        return (TopologyMetricDataInfo) metricCache.get(PENDING_UPLOAD_METRIC_DATA_INFO + idx);
    }

    public void pushEvent(Event cmd) {
        queue.offer(cmd);
    }

    public Map<String, Long> registerMetrics(String topologyId, Set<String> metricNames) {
        TimeTicker ticker = new TimeTicker(TimeUnit.MILLISECONDS, true);

        ConcurrentMap<String, Long> memMeta = topologyMetricContexts.get(topologyId).getMemMeta();
        Map<String, Long> ret = new HashMap<>();
        for (String metricName : metricNames) {
            Long id = memMeta.get(metricName);
            if (id != null && MetricUtils.isValidId(id)) {
                ret.put(metricName, id);
            } else {
                id = metricIDGenerator.genMetricId(metricName);
                Long old = memMeta.putIfAbsent(metricName, id);
                if (old == null) {
                    ret.put(metricName, id);
                } else {
                    ret.put(metricName, old);
                }
            }
        }

        long cost = ticker.stop();
        LOG.info("register metrics, topology:{}, size:{}, cost:{}", topologyId, metricNames.size(), cost);

        return ret;
    }

    public void handleRemoveEvent(Remove event) {
        String topologyId = event.topologyId;
        if (topologyId != null) {
            removeTopology(topologyId);
        }
        LOG.info("remove topology:{}.", topologyId);

    }

    private void removeTopology(String topologyId) {
        metricCache.removeTopology(topologyId);
        metricCache.removeSampleRate(topologyId);

        topologyMetricContexts.remove(topologyId);
    }


    public void refreshTopologies() {
        if (!topologyMetricContexts.containsKey(JStormMetrics.NIMBUS_METRIC_KEY)) {
            LOG.info("adding __nimbus__ to metric context.");
            Set<ResourceWorkerSlot> workerSlot = Sets.newHashSet(new ResourceWorkerSlot());
            TopologyMetricContext metricContext = new TopologyMetricContext(workerSlot);
            topologyMetricContexts.putIfAbsent(JStormMetrics.NIMBUS_METRIC_KEY, metricContext);
            syncMetaFromCache(JStormMetrics.NIMBUS_METRIC_KEY, topologyMetricContexts.get(JStormMetrics.NIMBUS_METRIC_KEY));
        }
        if (!topologyMetricContexts.containsKey(JStormMetrics.CLUSTER_METRIC_KEY)) {
            LOG.info("adding __cluster__ to metric context.");
            Set<ResourceWorkerSlot> workerSlot = Sets.newHashSet(new ResourceWorkerSlot());
            Map conf = new HashMap();
            //there's no need to consider sample rate when cluster metrics merge
            conf.put(ConfigExtension.TOPOLOGY_METRIC_SAMPLE_RATE, 1.0);
            TopologyMetricContext metricContext = new TopologyMetricContext(
                    JStormMetrics.CLUSTER_METRIC_KEY, workerSlot, conf);
            topologyMetricContexts.putIfAbsent(JStormMetrics.CLUSTER_METRIC_KEY, metricContext);
            syncMetaFromCache(JStormMetrics.CLUSTER_METRIC_KEY, topologyMetricContexts.get(JStormMetrics.CLUSTER_METRIC_KEY));
        }

        Map<String, Assignment> assignMap;
        try {
            assignMap = Cluster.get_all_assignment(stormClusterState, null);
            for (String topologyId : assignMap.keySet()) {
                if (!topologyMetricContexts.containsKey(topologyId)) {
                    Assignment assignment = assignMap.get(topologyId);
                    TopologyMetricContext metricContext =
                            new TopologyMetricContext(assignment.getWorkers());
                    metricContext.setTaskNum(NimbusUtils.getTopologyTaskNum(assignment));
                    syncMetaFromCache(topologyId, metricContext);

                    LOG.info("adding {} to metric context.", topologyId);
                    topologyMetricContexts.put(topologyId, metricContext);
                }
            }
        } catch (Exception e1) {
            LOG.warn("failed to get assignments");
            return;
        }

        List<String> removing = new ArrayList<>();
        for (String topologyId : topologyMetricContexts.keySet()) {
            if (!JStormMetrics.NIMBUS_METRIC_KEY.equals(topologyId)
                    && !JStormMetrics.CLUSTER_METRIC_KEY.equals(topologyId)
                    && !assignMap.containsKey(topologyId)) {
                removing.add(topologyId);
            }
        }

        for (String topologyId : removing) {
            LOG.info("removing topology:{}", topologyId);
            removeTopology(topologyId);
        }
    }

    /**
     * sync topology metric meta from external storage like TDDL/OTS.
     * nimbus server will skip syncing, only followers do this
     */
    public void syncTopologyMeta() {
        String nimbus = JStormMetrics.NIMBUS_METRIC_KEY;
        if (topologyMetricContexts.containsKey(nimbus)) {
            syncMetaFromRemote(nimbus, topologyMetricContexts.get(nimbus));
        }
        String cluster = JStormMetrics.CLUSTER_METRIC_KEY;
        if (topologyMetricContexts.containsKey(cluster)) {
            syncMetaFromRemote(cluster, topologyMetricContexts.get(cluster));
        }

        Map<String, Assignment> assignMap;
        try {
            assignMap = Cluster.get_all_assignment(stormClusterState, null);
            for (String topologyId : assignMap.keySet()) {
                if (topologyMetricContexts.containsKey(topologyId)) {
                    Assignment assignment = assignMap.get(topologyId);
                    TopologyMetricContext metricContext =
                            new TopologyMetricContext(assignment.getWorkers());
                    metricContext.setTaskNum(NimbusUtils.getTopologyTaskNum(assignment));

                    syncMetaFromCache(topologyId, metricContext);
                    syncMetaFromRemote(topologyId, metricContext);
                }
            }
        } catch (Exception e1) {
            LOG.warn("failed to get assignments");
        }
    }

    /**
     * sync metric meta from rocks db into mem cache on startup
     */
    private void syncMetaFromCache(String topologyId, TopologyMetricContext context) {
        if (!context.syncMeta()) {
            Map<String, Long> meta = metricCache.getMeta(topologyId);
            if (meta != null) {
                context.getMemMeta().putAll(meta);
            }
            context.setSyncMeta(true);
        }
    }

    private void syncMetaFromRemote(String topologyId, TopologyMetricContext context) {
        try {
            int memSize = context.getMemMeta().size();
            int zkSize = (Integer) stormClusterState.get_topology_metric(topologyId);

            if (memSize != zkSize) {
                ConcurrentMap<String, Long> memMeta = context.getMemMeta();
                for (MetaType metaType : MetaType.values()) {
                    List<MetricMeta> metaList = metricQueryClient.getMetricMeta(clusterName, topologyId, metaType);
                    if (metaList != null) {
                        LOG.info("get remote metric meta, topology:{}, metaType:{}, mem:{}, zk:{}, new size:{}",
                                topologyId, metaType, memSize, zkSize, metaList.size());
                        for (MetricMeta meta : metaList) {
                            memMeta.putIfAbsent(meta.getFQN(), meta.getId());
                        }
                    }
                }
                metricCache.putMeta(topologyId, memMeta);
            }
        } catch (Exception ex) {
            LOG.error("failed to sync remote meta", ex);
        }
    }

    /**
     * send topology track to jstorm monitor
     */
    protected void handleKillTopologyEvent(KillTopologyEvent event) {
        metricUploader.sendEvent(this.clusterName, event);
        removeTopology(event.topologyId);
    }

    private void handleStartTopologyEvent(StartTopologyEvent event) {
        this.metricCache.putSampleRate(event.topologyId, event.sampleRate);
        metricUploader.sendEvent(this.clusterName, event);
        if (!topologyMetricContexts.containsKey(event.topologyId)) {
            TopologyMetricContext metricContext = new TopologyMetricContext();
            // note that workerNum is not set here.
            this.topologyMetricContexts.put(event.topologyId, metricContext);
        }
    }

    private void handleTaskDeadEvent(TaskDeadEvent event) {
        metricUploader.sendEvent(this.clusterName, event);

        // unregister dead workers
        Set<ResourceWorkerSlot> workers = new HashSet<>();
        workers.addAll(event.deadTasks.values());
        for (ResourceWorkerSlot worker : workers) {
            metricCache.unregisterWorker(event.topologyId, worker.getHostname(), worker.getPort());
        }
    }

    private void handleTaskStartEvent(final TaskStartEvent event) {
        Assignment assignment = event.newAssignment;
        TopologyMetricContext metricContext = topologyMetricContexts.get(event.topologyId);
        if (metricContext != null) {
            metricContext.setWorkerSet(assignment.getWorkers());
        } else {
            metricContext = new TopologyMetricContext();
            metricContext.setWorkerSet(assignment.getWorkers());
            topologyMetricContexts.put(event.topologyId, metricContext);
        }
        metricUploader.sendEvent(this.clusterName, event);
    }

    /**
     * merge and send all metric data.
     */
    public void handleRefreshEvent(Refresh dummy) {
        TimeTicker ticker = new TimeTicker(TimeUnit.MILLISECONDS, true);
        try {
            refreshTopologies();
            LOG.info("refresh topologies, cost:{}", ticker.stopAndRestart());
            if (!nimbusData.isLeader()) {
                syncTopologyMeta();
                LOG.info("sync topology meta, cost:{}", ticker.stop());
            }
        } catch (Exception ex) {
            LOG.error("handleRefreshEvent error:", ex);
        }
    }

    private TopologyMetricContext getClusterTopologyMetricContext() {
        return topologyMetricContexts.get(JStormMetrics.CLUSTER_METRIC_KEY);
    }

    private void mergeAndUploadClusterMetrics() {
        TopologyMetricContext context = getClusterTopologyMetricContext();
        TopologyMetric tpMetric = context.mergeMetrics();
        if (tpMetric == null) {
            tpMetric = MetricUtils.mkTopologyMetric();
            tpMetric.set_topologyMetric(MetricUtils.mkMetricInfo());
        }

        //reset snapshots metric id
        MetricInfo clusterMetrics = tpMetric.get_topologyMetric();
        Map<String, Long> metricNames = context.getMemMeta();
        for (Map.Entry<String, Map<Integer, MetricSnapshot>> entry : clusterMetrics.get_metrics().entrySet()) {
            String metricName = entry.getKey();
            MetricType metricType = MetricUtils.metricType(metricName);
            Long metricId = metricNames.get(metricName);
            for (Map.Entry<Integer, MetricSnapshot> metric : entry.getValue().entrySet()) {
                MetricSnapshot snapshot = metric.getValue();
                snapshot.set_metricId(metricId);
                if (metricType == MetricType.HISTOGRAM || metricType == MetricType.TIMER) {
                    snapshot.set_points(new ArrayList<Long>(0));
                }
//                entry.getValue().put(metric.getKey(), snapshot);
            }
        }

        //fill the unacquired metrics with zero
        long ts = System.currentTimeMillis();
        for (Map.Entry<String, Long> entry : metricNames.entrySet()) {
            String name = entry.getKey();
            if (!clusterMetrics.get_metrics().containsKey(name)) {
                Map<Integer, MetricSnapshot> metric = new HashMap<>();
                MetricType type = MetricUtils.metricType(name);
                metric.put(AsmWindow.M1_WINDOW, new MetricSnapshot(entry.getValue(), ts, type.getT()));
                clusterMetrics.put_to_metrics(name, metric);
            }
        }

        //upload to cache
        Update event = new Update();
        event.timestamp = System.currentTimeMillis();
        event.topologyMetrics = tpMetric;
        event.topologyId = JStormMetrics.CLUSTER_METRIC_KEY;
        pushEvent(event);

        LOG.info("send update event for cluster metrics, size : {}", clusterMetrics.get_metrics_size());
    }

    //update cluster metrics local cache
    private void updateClusterMetrics(String topologyId, TopologyMetric tpMetric) {
        if (tpMetric.get_topologyMetric().get_metrics_size() > 0) {
            TopologyMetricContext context = getClusterTopologyMetricContext();
            MetricInfo topologyMetrics = tpMetric.get_topologyMetric();
            // make a new MetricInfo to save the topologyId's metric
            MetricInfo clusterMetrics = MetricUtils.mkMetricInfo();
            Set<String> metricNames = new HashSet<>();
            for (Map.Entry<String, Map<Integer, MetricSnapshot>> entry : topologyMetrics.get_metrics().entrySet()) {
                String metricName = MetricUtils.topo2clusterName(entry.getKey());
                MetricType metricType = MetricUtils.metricType(metricName);
                Map<Integer, MetricSnapshot> winData = new HashMap<>();
                for (Map.Entry<Integer, MetricSnapshot> entryData : entry.getValue().entrySet()) {
                    MetricSnapshot snapshot = entryData.getValue().deepCopy();
                    winData.put(entryData.getKey(), snapshot);
                    if (metricType == MetricType.HISTOGRAM || metricType == MetricType.TIMER) {
                        // reset topology metric points
                        entryData.getValue().set_points(new ArrayList<Long>(0));
                    }
                }
                clusterMetrics.put_to_metrics(metricName, winData);
                metricNames.add(metricName);
            }
            // save to local cache, waiting for merging
            context.addToMemCache(topologyId, clusterMetrics);
            registerMetrics(JStormMetrics.CLUSTER_METRIC_KEY, metricNames);
        }
    }

    /**
     * put metric data to metric cache.
     */
    public void handleUpdateEvent(Update event) {
        TopologyMetric topologyMetrics = event.topologyMetrics;
        final String topologyId = event.topologyId;

        if (this.topologyMetricContexts.containsKey(topologyId)) {
            if (!JStormMetrics.CLUSTER_METRIC_KEY.equals(topologyId)) {
                updateClusterMetrics(topologyId, topologyMetrics);
            }

            // overwrite
            metricCache.putMetricData(topologyId, topologyMetrics);

            // below process is kind of a transaction, first we lock an empty slot, mark it as PRE_SET
            // by this time the slot is not yet ready for uploading as the upload thread looks for SET slots only
            // after all metrics data has been saved, we mark it as SET, then it's ready for uploading.
            int idx = getAndPresetFirstEmptyIndex();
            if (idx >= 0) {
                TopologyMetricDataInfo summary = new TopologyMetricDataInfo();
                summary.topologyId = topologyId;
                summary.timestamp = event.timestamp;
                if (topologyId.equals(JStormMetrics.NIMBUS_METRIC_KEY) ||
                        topologyId.equals(JStormMetrics.CLUSTER_METRIC_KEY)) {
                    summary.type = MetricUploader.METRIC_TYPE_TOPLOGY;
                } else {
                    if (topologyMetrics.get_topologyMetric().get_metrics_size() > 0 ||
                            topologyMetrics.get_componentMetric().get_metrics_size() > 0) {
                        if (topologyMetrics.get_taskMetric().get_metrics_size() +
                                topologyMetrics.get_workerMetric().get_metrics_size() +
                                topologyMetrics.get_nettyMetric().get_metrics_size() +
                                topologyMetrics.get_streamMetric().get_metrics_size() > 0) {
                            summary.type = MetricUploader.METRIC_TYPE_ALL;
                        } else {
                            summary.type = MetricUploader.METRIC_TYPE_TOPLOGY;
                        }
                    } else {
                        summary.type = MetricUploader.METRIC_TYPE_TASK;
                    }
                }

                metricCache.put(PENDING_UPLOAD_METRIC_DATA_INFO + idx, summary);
                metricCache.put(PENDING_UPLOAD_METRIC_DATA + idx, topologyMetrics);
                markSet(idx);
                LOG.info("put metric data to local cache, topology:{}, idx:{}", topologyId, idx);
            } else {
                LOG.error("exceeding maxPendingUploadMetrics, skip metrics data for topology:{}", topologyId);
            }
        } else {
            LOG.warn("topology {} has been killed or has not started, skip update.", topologyId);
        }
    }

    /**
     * get topology metrics, note that only topology & component & worker metrics are returned
     */
    public TopologyMetric getTopologyMetric(String topologyId) {
        long start = System.nanoTime();
        try {
            TopologyMetric ret = new TopologyMetric();
            List<MetricInfo> topologyMetrics = metricCache.getMetricData(topologyId, MetaType.TOPOLOGY);
            List<MetricInfo> componentMetrics = metricCache.getMetricData(topologyId, MetaType.COMPONENT);
            List<MetricInfo> workerMetrics = metricCache.getMetricData(topologyId, MetaType.WORKER);

            MetricInfo dummy = MetricUtils.mkMetricInfo();
            if (topologyMetrics.size() > 0) {
                // get the last min topology metric
                ret.set_topologyMetric(topologyMetrics.get(topologyMetrics.size() - 1));
            } else {
                ret.set_topologyMetric(dummy);
            }
            if (componentMetrics.size() > 0) {
                ret.set_componentMetric(componentMetrics.get(0));
            } else {
                ret.set_componentMetric(dummy);
            }
            if (workerMetrics.size() > 0) {
                ret.set_workerMetric(workerMetrics.get(0));
            } else {
                ret.set_workerMetric(dummy);
            }
            ret.set_taskMetric(dummy);
            ret.set_streamMetric(dummy);
            ret.set_nettyMetric(dummy);

            return ret;
        } finally {
            long end = System.nanoTime();
            SimpleJStormMetric.updateNimbusHistogram("getTopologyMetric", (end - start) / TimeUtils.NS_PER_US);
        }
    }

    public static String getWorkerSlotName(String hostname, Integer port) {
        return hostname + ":" + port;
    }

    class RefreshTopologiesThread extends RunnableCallback {
        @Override
        public void run() {
            if (!isShutdown.get()) {
                pushEvent(new Refresh());
            }
        }

        @Override
        public Object getResult() {
            return TimeUtils.SEC_PER_MIN;
        }

        @Override
        public String getThreadName() {
            return "RefreshThread";
        }
    }

    class MetricsUploadThread extends Thread {
        public MetricsUploadThread() {
            setName("main-upload-thread");
        }

        @Override
        public void run() {
            while (!isShutdown.get()) {
                try {
                    if (metricUploader != null && nimbusData.isLeader()) {
                        final int idx = getFirstPendingUploadIndex();
                        if (idx >= 0) {
                            markUploading(idx);
                            upload(clusterName, idx);
                        }
                    }
                    JStormUtils.sleepMs(5);
                } catch (Exception ex) {
                    LOG.error("Error", ex);
                }
            }
        }

        public boolean upload(final String clusterName, final int idx) {
            final TopologyMetricDataInfo summary = getMetricDataInfoFromCache(idx);
            if (summary == null) {
                LOG.warn("metric summary is null from cache idx:{}", idx);
                markUploaded(idx);
                return true;
            }

            final String topologyId = summary.topologyId;
            if (!isTopologyAlive(topologyId)) {
                LOG.warn("topology {} is not alive, skip sending metrics.", topologyId);
                markUploaded(idx);
                return true;
            }

            return metricUploader.upload(clusterName, topologyId, idx, summary.toMap());
        }
    }

    class FlushMetricMetaThread extends Thread {

        public FlushMetricMetaThread() {
            setName("FlushMetricMetaThread");
        }

        @Override
        public void run() {
            while (!isShutdown.get()) {
                long start = System.currentTimeMillis();
                try {
                    // if metricUploader is not fully initialized, return directly
                    if (nimbusData.isLeader() && metricUploader != null) {
                        for (Map.Entry<String, TopologyMetricContext> entry : topologyMetricContexts.entrySet()) {
                            String topologyId = entry.getKey();
                            TopologyMetricContext metricContext = entry.getValue();

                            Map<String, Long> cachedMeta = metricCache.getMeta(topologyId);
                            if (cachedMeta == null) {
                                cachedMeta = new HashMap<>();
                            }
                            Map<String, Long> memMeta = metricContext.getMemMeta();
                            if (memMeta.size() > cachedMeta.size()) {
                                cachedMeta.putAll(memMeta);
                            }
                            metricCache.putMeta(topologyId, cachedMeta);

                            int curSize = cachedMeta.size();
                            if (curSize != metricContext.getFlushedMetaNum()) {
                                metricContext.setFlushedMetaNum(curSize);

                                metricUploader.registerMetrics(clusterName, topologyId, cachedMeta);
                                LOG.info("flush metric meta, topology:{}, total:{}, cost:{}.",
                                        topologyId, curSize, System.currentTimeMillis() - start);
                            }
                            stormClusterState.set_topology_metric(topologyId, curSize);
                        }
                    }

                    JStormUtils.sleepMs(15000);
                } catch (Exception ex) {
                    LOG.error("Error", ex);
                }
            }
        }
    }

    public static class TopologyMetricDataInfo implements Serializable {
        private static final long serialVersionUID = 1303262512351757610L;

        public String topologyId;
        public String type; // "tp" for tp/comp metrics OR "task" for task/stream/worker/netty metrics
        public long timestamp;   // metrics report time

        public Map<String, Object> toMap() {
            Map<String, Object> ret = new HashMap<String, Object>();
            ret.put(MetricUploader.METRIC_TIME, timestamp);
            ret.put(MetricUploader.METRIC_TYPE, type);

            return ret;
        }

        @Override
        public String toString() {
            return ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE);
        }
    }

    // ==============================================
    // =================== events ===================
    // ==============================================
    public static class Event {
        protected Event() {
        }

        public String clusterName;
        public String topologyId;
        public long timestamp;

        @Override
        public String toString() {
            return ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE);
        }
    }

    public static class Update extends Event {
        public TopologyMetric topologyMetrics;
    }

    public static class Remove extends Event {
    }

    public static class Refresh extends Event {
    }


    public static class KillTopologyEvent extends Event {
    }

    public static class StartTopologyEvent extends Event {
        public double sampleRate;
    }

    public static class TaskDeadEvent extends Event {
        public Map<Integer, ResourceWorkerSlot> deadTasks;
    }

    public static class TaskStartEvent extends Event {
        public Assignment oldAssignment;
        public Assignment newAssignment;
        public Map<Integer, String> task2Component;
    }
}