jstorm-core/src/main/java/com/alibaba/jstorm/metric/JStormMetricsReporter.java - storm - Git at Google

 /**
  * 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.Config;
 import backtype.storm.generated.MetricInfo;
 import backtype.storm.generated.TopologyMetric;
 import backtype.storm.generated.WorkerUploadMetrics;
 import backtype.storm.spout.SpoutOutputCollector;
 import backtype.storm.task.OutputCollector;
 import backtype.storm.tuple.Values;
 import backtype.storm.utils.NimbusClient;
 import com.alibaba.jstorm.callback.AsyncLoopThread;
 import com.alibaba.jstorm.callback.RunnableCallback;
 import com.alibaba.jstorm.client.ConfigExtension;
 import com.alibaba.jstorm.cluster.Common;
 import com.alibaba.jstorm.cluster.StormConfig;
 import com.alibaba.jstorm.common.metric.AsmMetric;
 import com.alibaba.jstorm.daemon.nimbus.NimbusData;
 import com.alibaba.jstorm.daemon.nimbus.TopologyMetricsRunnable.Update;
 import com.alibaba.jstorm.daemon.worker.WorkerData;
 import com.alibaba.jstorm.utils.JStormServerUtils;
 import com.alibaba.jstorm.utils.TimeUtils;
 import com.google.common.annotations.VisibleForTesting;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import java.util.*;

 /**
  * report metrics from worker to nimbus server. this class serves as an object in Worker/Nimbus/Supervisor.
  * when in Worker, it reports data via netty transport; otherwise reports via thrift.
  * <p/>
  * there are 2 threads:
  * 1.flush thread: check every 1 sec, when current time is aligned to 1 min, flush all metrics to snapshots
  * 2.check meta thread: use thrift to get metric id from nimbus server.
  *
  * @author Cody (weiyue.wy@alibaba-inc.com)
  * @since 2.0.5
  */
 public class JStormMetricsReporter {
     private static final Logger LOG = LoggerFactory.getLogger(JStormMetricsReporter.class);

     private Map conf;
     protected String clusterName;
     protected String topologyId;
     protected String host;
     protected int port;

     protected boolean localMode = false;

     private AsyncLoopThread checkMetricMetaThread;
     protected final int checkMetaThreadCycle;

     private AsyncLoopThread flushMetricThread;
     protected final int flushMetricThreadCycle;

     private boolean test = false;

     private boolean isInWorker = false;

     private SpoutOutputCollector spoutOutput;
     private OutputCollector boltOutput;

     private boolean enableMetrics;
     private NimbusClient client = null;

     public JStormMetricsReporter(Object role) {
         LOG.info("starting jstorm metrics reporter");
         if (role instanceof WorkerData) {
             WorkerData workerData = (WorkerData) role;
             this.conf = workerData.getStormConf();
             this.topologyId = (String) conf.get(Config.TOPOLOGY_ID);
             this.port = workerData.getPort();
             this.isInWorker = true;
         } else if (role instanceof NimbusData) {
             NimbusData nimbusData = (NimbusData) role;
             this.conf = nimbusData.getConf();
             this.topologyId = JStormMetrics.NIMBUS_METRIC_KEY;
         }
         this.host = JStormMetrics.getHost();
         this.enableMetrics = JStormMetrics.isEnabled();
         if (!enableMetrics) {
             LOG.warn("***** topology metrics is disabled! *****");
         } else {
             LOG.info("topology metrics is enabled.");
         }

         this.checkMetaThreadCycle = 30;
         // flush metric snapshots when time is aligned, check every sec.
         this.flushMetricThreadCycle = 1;

         LOG.info("check meta thread freq:{}, flush metrics thread freq:{}", checkMetaThreadCycle, flushMetricThreadCycle);

         this.localMode = StormConfig.local_mode(conf);
         this.clusterName = ConfigExtension.getClusterName(conf);
         LOG.info("done.");
     }

     @VisibleForTesting
     JStormMetricsReporter() {
         LOG.info("Successfully started jstorm metrics reporter for test.");
         this.test = true;
         this.flushMetricThreadCycle = 1;
         this.checkMetaThreadCycle = 30;
     }

     public void init() {
         if (!localMode && enableMetrics) {
             this.checkMetricMetaThread = new AsyncLoopThread(new CheckMetricMetaThread());
             this.flushMetricThread = new AsyncLoopThread(new FlushMetricThread());
         }
     }

     private Map<String, Long> registerMetrics(Set<String> names) {
         if (test || !enableMetrics) {
             return new HashMap<>();
         }
         try {
             if (client == null) {
                 client = NimbusClient.getConfiguredClient(conf);
             }

             return client.getClient().registerMetrics(topologyId, names);
         } catch (Exception e) {
             LOG.error("Failed to gen metric ids", e);
             if (client != null) {
                 client.close();
                 client = NimbusClient.getConfiguredClient(conf);
             }
         }

         return null;
     }

     public void shutdown() {
         if (!localMode && enableMetrics) {
             this.checkMetricMetaThread.cleanup();
             this.flushMetricThread.cleanup();
         }
     }

     public void doUpload() {
         if (test) {
             return;
         }
         try {
             long start = System.currentTimeMillis();
             MetricInfo workerMetricInfo = JStormMetrics.computeAllMetrics();

             WorkerUploadMetrics upload = new WorkerUploadMetrics();
             upload.set_topologyId(topologyId);
             upload.set_supervisorId(host);
             upload.set_port(port);
             upload.set_allMetrics(workerMetricInfo);

             if (workerMetricInfo.get_metrics_size() > 0) {
                 uploadMetric(upload);
                 LOG.info("Successfully upload worker metrics, size:{}, cost:{}",
                         workerMetricInfo.get_metrics_size(), System.currentTimeMillis() - start);
             } else {
                 LOG.info("No metrics to upload.");
             }
         } catch (Exception e) {
             LOG.error("Failed to upload worker metrics", e);
         }
     }


     public void uploadMetric(WorkerUploadMetrics metrics) {
         if (isInWorker) {
         //in Worker, we upload data via netty transport
             if (boltOutput != null) {
                 LOG.info("emit metrics through bolt collector.");
                 boltOutput.emit(Common.TOPOLOGY_MASTER_METRICS_STREAM_ID,
                         new Values(JStormServerUtils.getName(host, port), metrics));
             } else if (spoutOutput != null) {
                 LOG.info("emit metrics through spout collector.");
                 spoutOutput.emit(Common.TOPOLOGY_MASTER_METRICS_STREAM_ID,
                         new Values(JStormServerUtils.getName(host, port), metrics));
             }
         }else {
         // in supervisor or nimbus, we upload metric data via thrift
             LOG.info("emit metrics through nimbus client.");
             Update event = new Update();
             TopologyMetric tpMetric = MetricUtils.mkTopologyMetric();
             tpMetric.set_workerMetric(metrics.get_allMetrics());

             event.topologyMetrics = tpMetric;
             event.topologyId = topologyId;

             try {
                 if (client == null) {
                     client = NimbusClient.getConfiguredClient(conf);
                 }
                 client.getClient().uploadTopologyMetrics(topologyId, tpMetric);
             } catch (Exception ex) {
                 LOG.error("upload metric error:", ex);
                 if (client != null) {
                     client.close();
                     client = NimbusClient.getConfiguredClient(conf);
                 }
             }
         }
         //MetricUtils.logMetrics(metrics.get_allMetrics());
     }


     public void setOutputCollector(Object outputCollector) {
         if (outputCollector instanceof OutputCollector) {
             this.boltOutput = (OutputCollector) outputCollector;
         } else if (outputCollector instanceof SpoutOutputCollector) {
             this.spoutOutput = (SpoutOutputCollector) outputCollector;
         }

     }


     class FlushMetricThread extends RunnableCallback {
         @Override
         public void run() {
             if (TimeUtils.isTimeAligned()) {
                 int cnt = 0;
                 try {
                     for (AsmMetricRegistry registry : JStormMetrics.allRegistries) {
                         for (Map.Entry<String, AsmMetric> entry : registry.getMetrics().entrySet()) {
                             entry.getValue().flush();
                             cnt++;
                         }
                     }
                     LOG.info("flush metrics, total:{}.", cnt);

                     doUpload();
                 } catch (Exception ex) {
                     LOG.error("Error", ex);
                 }
             }
         }

         @Override
         public Object getResult() {
             return flushMetricThreadCycle;
         }
     }

     class CheckMetricMetaThread extends RunnableCallback {
         private volatile boolean processing = false;
         private final long start = TimeUtils.current_time_secs();
         private final long initialDelay = 30 + new Random().nextInt(15);

         @Override
         public void run() {
             if (TimeUtils.current_time_secs() - start < initialDelay) {
                 return;
             }

             if (processing) {
                 LOG.info("still processing, skip...");
             } else {
                 processing = true;
                 long start = System.currentTimeMillis();
                 try {
                     Set<String> names = new HashSet<>();
                     for (AsmMetricRegistry registry : JStormMetrics.allRegistries) {
                         Map<String, AsmMetric> metricMap = registry.getMetrics();
                         for (Map.Entry<String, AsmMetric> metricEntry : metricMap.entrySet()) {
                             AsmMetric metric = metricEntry.getValue();
                             if (((metric.getOp() & AsmMetric.MetricOp.REPORT) == AsmMetric.MetricOp.REPORT) &&
                                     metric.getMetricId() == 0L) {
                                 names.add(metricEntry.getKey());
                             }
                         }
                     }

                     if (names.size() > 0) {
                         Map<String, Long> nameIdMap = registerMetrics(names);
                         if (nameIdMap != null) {
                             for (String name : nameIdMap.keySet()) {
                                 AsmMetric metric = JStormMetrics.find(name);
                                 if (metric != null) {
                                     long id = nameIdMap.get(name);
                                     metric.setMetricId(id);
                                     LOG.info("set metric id, {}:{}", name, id);
                                 }
                             }
                         }
                         LOG.info("register metrics, size:{}, cost:{}", names.size(), System.currentTimeMillis() - start);
                     }
                 } catch (Exception ex) {
                     LOG.error("Error", ex);
                 }
                 processing = false;
             }
         }

         @Override
         public Object getResult() {
             return checkMetaThreadCycle;
         }
     }
 }
	/**
	* 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.Config;
	import backtype.storm.generated.MetricInfo;
	import backtype.storm.generated.TopologyMetric;
	import backtype.storm.generated.WorkerUploadMetrics;
	import backtype.storm.spout.SpoutOutputCollector;
	import backtype.storm.task.OutputCollector;
	import backtype.storm.tuple.Values;
	import backtype.storm.utils.NimbusClient;
	import com.alibaba.jstorm.callback.AsyncLoopThread;
	import com.alibaba.jstorm.callback.RunnableCallback;
	import com.alibaba.jstorm.client.ConfigExtension;
	import com.alibaba.jstorm.cluster.Common;
	import com.alibaba.jstorm.cluster.StormConfig;
	import com.alibaba.jstorm.common.metric.AsmMetric;
	import com.alibaba.jstorm.daemon.nimbus.NimbusData;
	import com.alibaba.jstorm.daemon.nimbus.TopologyMetricsRunnable.Update;
	import com.alibaba.jstorm.daemon.worker.WorkerData;
	import com.alibaba.jstorm.utils.JStormServerUtils;
	import com.alibaba.jstorm.utils.TimeUtils;
	import com.google.common.annotations.VisibleForTesting;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import java.util.*;

	/**
	* report metrics from worker to nimbus server. this class serves as an object in Worker/Nimbus/Supervisor.
	* when in Worker, it reports data via netty transport; otherwise reports via thrift.
	* <p/>
	* there are 2 threads:
	* 1.flush thread: check every 1 sec, when current time is aligned to 1 min, flush all metrics to snapshots
	* 2.check meta thread: use thrift to get metric id from nimbus server.
	*
	* @author Cody (weiyue.wy@alibaba-inc.com)
	* @since 2.0.5
	*/
	public class JStormMetricsReporter {
	private static final Logger LOG = LoggerFactory.getLogger(JStormMetricsReporter.class);

	private Map conf;
	protected String clusterName;
	protected String topologyId;
	protected String host;
	protected int port;

	protected boolean localMode = false;

	private AsyncLoopThread checkMetricMetaThread;
	protected final int checkMetaThreadCycle;

	private AsyncLoopThread flushMetricThread;
	protected final int flushMetricThreadCycle;

	private boolean test = false;

	private boolean isInWorker = false;

	private SpoutOutputCollector spoutOutput;
	private OutputCollector boltOutput;

	private boolean enableMetrics;
	private NimbusClient client = null;

	public JStormMetricsReporter(Object role) {
	LOG.info("starting jstorm metrics reporter");
	if (role instanceof WorkerData) {
	WorkerData workerData = (WorkerData) role;
	this.conf = workerData.getStormConf();
	this.topologyId = (String) conf.get(Config.TOPOLOGY_ID);
	this.port = workerData.getPort();
	this.isInWorker = true;
	} else if (role instanceof NimbusData) {
	NimbusData nimbusData = (NimbusData) role;
	this.conf = nimbusData.getConf();
	this.topologyId = JStormMetrics.NIMBUS_METRIC_KEY;
	}
	this.host = JStormMetrics.getHost();
	this.enableMetrics = JStormMetrics.isEnabled();
	if (!enableMetrics) {
	LOG.warn("*** topology metrics is disabled! ***");
	} else {
	LOG.info("topology metrics is enabled.");
	}

	this.checkMetaThreadCycle = 30;
	// flush metric snapshots when time is aligned, check every sec.
	this.flushMetricThreadCycle = 1;

	LOG.info("check meta thread freq:{}, flush metrics thread freq:{}", checkMetaThreadCycle, flushMetricThreadCycle);

	this.localMode = StormConfig.local_mode(conf);
	this.clusterName = ConfigExtension.getClusterName(conf);
	LOG.info("done.");
	}

	@VisibleForTesting
	JStormMetricsReporter() {
	LOG.info("Successfully started jstorm metrics reporter for test.");
	this.test = true;
	this.flushMetricThreadCycle = 1;
	this.checkMetaThreadCycle = 30;
	}

	public void init() {
	if (!localMode && enableMetrics) {
	this.checkMetricMetaThread = new AsyncLoopThread(new CheckMetricMetaThread());
	this.flushMetricThread = new AsyncLoopThread(new FlushMetricThread());
	}
	}

	private Map<String, Long> registerMetrics(Set<String> names) {
	if (test \|\| !enableMetrics) {
	return new HashMap<>();
	}
	try {
	if (client == null) {
	client = NimbusClient.getConfiguredClient(conf);
	}

	return client.getClient().registerMetrics(topologyId, names);
	} catch (Exception e) {
	LOG.error("Failed to gen metric ids", e);
	if (client != null) {
	client.close();
	client = NimbusClient.getConfiguredClient(conf);
	}
	}

	return null;
	}

	public void shutdown() {
	if (!localMode && enableMetrics) {
	this.checkMetricMetaThread.cleanup();
	this.flushMetricThread.cleanup();
	}
	}

	public void doUpload() {
	if (test) {
	return;
	}
	try {
	long start = System.currentTimeMillis();
	MetricInfo workerMetricInfo = JStormMetrics.computeAllMetrics();

	WorkerUploadMetrics upload = new WorkerUploadMetrics();
	upload.set_topologyId(topologyId);
	upload.set_supervisorId(host);
	upload.set_port(port);
	upload.set_allMetrics(workerMetricInfo);

	if (workerMetricInfo.get_metrics_size() > 0) {
	uploadMetric(upload);
	LOG.info("Successfully upload worker metrics, size:{}, cost:{}",
	workerMetricInfo.get_metrics_size(), System.currentTimeMillis() - start);
	} else {
	LOG.info("No metrics to upload.");
	}
	} catch (Exception e) {
	LOG.error("Failed to upload worker metrics", e);
	}
	}


	public void uploadMetric(WorkerUploadMetrics metrics) {
	if (isInWorker) {
	//in Worker, we upload data via netty transport
	if (boltOutput != null) {
	LOG.info("emit metrics through bolt collector.");
	boltOutput.emit(Common.TOPOLOGY_MASTER_METRICS_STREAM_ID,
	new Values(JStormServerUtils.getName(host, port), metrics));
	} else if (spoutOutput != null) {
	LOG.info("emit metrics through spout collector.");
	spoutOutput.emit(Common.TOPOLOGY_MASTER_METRICS_STREAM_ID,
	new Values(JStormServerUtils.getName(host, port), metrics));
	}
	}else {
	// in supervisor or nimbus, we upload metric data via thrift
	LOG.info("emit metrics through nimbus client.");
	Update event = new Update();
	TopologyMetric tpMetric = MetricUtils.mkTopologyMetric();
	tpMetric.set_workerMetric(metrics.get_allMetrics());

	event.topologyMetrics = tpMetric;
	event.topologyId = topologyId;

	try {
	if (client == null) {
	client = NimbusClient.getConfiguredClient(conf);
	}
	client.getClient().uploadTopologyMetrics(topologyId, tpMetric);
	} catch (Exception ex) {
	LOG.error("upload metric error:", ex);
	if (client != null) {
	client.close();
	client = NimbusClient.getConfiguredClient(conf);
	}
	}
	}
	//MetricUtils.logMetrics(metrics.get_allMetrics());
	}


	public void setOutputCollector(Object outputCollector) {
	if (outputCollector instanceof OutputCollector) {
	this.boltOutput = (OutputCollector) outputCollector;
	} else if (outputCollector instanceof SpoutOutputCollector) {
	this.spoutOutput = (SpoutOutputCollector) outputCollector;
	}

	}


	class FlushMetricThread extends RunnableCallback {
	@Override
	public void run() {
	if (TimeUtils.isTimeAligned()) {
	int cnt = 0;
	try {
	for (AsmMetricRegistry registry : JStormMetrics.allRegistries) {
	for (Map.Entry<String, AsmMetric> entry : registry.getMetrics().entrySet()) {
	entry.getValue().flush();
	cnt++;
	}
	}
	LOG.info("flush metrics, total:{}.", cnt);

	doUpload();
	} catch (Exception ex) {
	LOG.error("Error", ex);
	}
	}
	}

	@Override
	public Object getResult() {
	return flushMetricThreadCycle;
	}
	}

	class CheckMetricMetaThread extends RunnableCallback {
	private volatile boolean processing = false;
	private final long start = TimeUtils.current_time_secs();
	private final long initialDelay = 30 + new Random().nextInt(15);

	@Override
	public void run() {
	if (TimeUtils.current_time_secs() - start < initialDelay) {
	return;
	}

	if (processing) {
	LOG.info("still processing, skip...");
	} else {
	processing = true;
	long start = System.currentTimeMillis();
	try {
	Set<String> names = new HashSet<>();
	for (AsmMetricRegistry registry : JStormMetrics.allRegistries) {
	Map<String, AsmMetric> metricMap = registry.getMetrics();
	for (Map.Entry<String, AsmMetric> metricEntry : metricMap.entrySet()) {
	AsmMetric metric = metricEntry.getValue();
	if (((metric.getOp() & AsmMetric.MetricOp.REPORT) == AsmMetric.MetricOp.REPORT) &&
	metric.getMetricId() == 0L) {
	names.add(metricEntry.getKey());
	}
	}
	}

	if (names.size() > 0) {
	Map<String, Long> nameIdMap = registerMetrics(names);
	if (nameIdMap != null) {
	for (String name : nameIdMap.keySet()) {
	AsmMetric metric = JStormMetrics.find(name);
	if (metric != null) {
	long id = nameIdMap.get(name);
	metric.setMetricId(id);
	LOG.info("set metric id, {}:{}", name, id);
	}
	}
	}
	LOG.info("register metrics, size:{}, cost:{}", names.size(), System.currentTimeMillis() - start);
	}
	} catch (Exception ex) {
	LOG.error("Error", ex);
	}
	processing = false;
	}
	}

	@Override
	public Object getResult() {
	return checkMetaThreadCycle;
	}
	}
	}