import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
/**
 * Goes over the specified metrics and if both execute-count and execute-latency are present,
 * computes the capacity metric according to the formula capacity = execute-count * execute-latency / time-window-ms
 *
 * @param component2metrics metrics keyed by component name.
 * @param taskInfo          additional task information pertaining to the reporting task.
 * @return The capacity metrics that were calculated based on the specified input metrics.
 */
public static ImmutableList<Metric> calculateCapacityMetrics(final Map<String, List<Metric>> component2metrics,
                                                             final IMetricsConsumer.TaskInfo taskInfo) {

  final Function<Map.Entry<String, List<Metric>>, Optional<Metric>> toCapacityMetric =
          new Function<Map.Entry<String, List<Metric>>, Optional<Metric>>() {
            @Override
            public Optional<Metric> apply(final Map.Entry<String, List<Metric>> componentMetrics) {

              final String component = componentMetrics.getKey();
              final FluentIterable<Metric> metrics = FluentIterable.from(componentMetrics.getValue());
              final Optional<Metric> count = metrics.firstMatch(isExecuteCountMetric);
              final Optional<Metric> latency = metrics.firstMatch(isExecuteLatencyMetric);

              return calculateCapacityMetric(component, count, latency, taskInfo.updateIntervalSecs);
            }
          };

  return FluentIterable
          .from(component2metrics.entrySet())
          .transform(toCapacityMetric)
          .filter(Metric.Option.isPresent)
          .transform(Metric.Option.getValue)
          .toList();
}
 

import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
@Test
public void testValidJMXObjectName() throws Exception {

  final String topologyName = "someTopology";

  Map config = new HashMap();
  config.put(Config.TOPOLOGY_NAME, topologyName);
  processor = new SimpleJMXStormMetricProcessor(config);

  Metric metric = new Metric("component", "kafkaPartition{host=kafka_9092, partition=0}", 1.9);
  IMetricsConsumer.TaskInfo taskInfo = new IMetricsConsumer.TaskInfo("localhost", 1010, "emitBot", 2, System.currentTimeMillis(), 100);

  String name = processor.mBeanName(metric, taskInfo);
  ObjectName objName = new ObjectName(name);

  assertThat(objName.getCanonicalName(), is("storm:component=component,host-port-task=localhost-1010-2,operation=\"kafkaPartition{host=kafka_9092, partition=0}\",topology=someTopology"));
}
 

import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
private static void registerMetricsConsumer(String className, Config stormConfig, BulletStormConfig bulletStormConfig) {
    try {
        Class<? extends IMetricsConsumer> consumer = (Class<? extends IMetricsConsumer>) Class.forName(className);
        Method method = consumer.getMethod(REGISTER_METHOD, Config.class, BulletStormConfig.class);
        log.info("Calling the IMetricsConsumer register method for class {} using method {}", className, method.toGenericString());
        method.invoke(null, stormConfig, bulletStormConfig);
        log.info("Registered the IMetricsConsumer class {}", className);
    } catch (Exception e) {
        log.info("Could not call the register method for " + className, e);
    }
}
 

import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
public static Map<String, String> convertToTags(String stormId, IMetricsConsumer.TaskInfo taskInfo) {
  Map<String, String> tags = new HashMap<>();
  tags.put("srcComponentId", taskInfo.srcComponentId);
  tags.put("stormId", removeNonce(stormId));
  tags.put("rawStormId", stormId);
  tags.put("srcWorkerHost", taskInfo.srcWorkerHost);
  tags.put("srcWorkerPort", String.valueOf(taskInfo.srcWorkerPort));
  tags.put("srcTaskId", String.valueOf(taskInfo.srcTaskId));
  return tags;
}
 

import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
private Object[] generateTaskInfo(String stormId, String rawStormId) {
  HashMap<String, String> tags = generateTags(stormId, rawStormId);
  return new Object[]{
    rawStormId,
    new IMetricsConsumer.TaskInfo(
      tags.get("srcWorkerHost"),
      Integer.valueOf(tags.get("srcWorkerPort")),
      tags.get("srcComponentId"),
      Integer.valueOf(tags.get("srcTaskId")),
      System.currentTimeMillis(),
      10),
    tags
  };
}
 

import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
@Override
public MetricName name(final Metric metric, final IMetricsConsumer.TaskInfo taskInfo) {
    return new MetricName(Metric.joinNameFragments(taskInfo.srcWorkerHost,
                    taskInfo.srcWorkerPort,
                    metric.getComponent()),
                    Integer.toString(taskInfo.srcTaskId),
                    metric.getOperation());
}
 

import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
@Override
public void process(final Metric metric, final IMetricsConsumer.TaskInfo taskInfo) {

    final MetricName metricName = name(metric, taskInfo);
    try {
        createOrUpdateGauge(metric, metricName);
    } catch (final Exception e) {
        LOG.error(String.format("Unable to process metric %s", metricName.toString()), e);
    }
}
 

import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
String mBeanName(Metric metric, IMetricsConsumer.TaskInfo taskInfo) {
    return "storm"
            + ":topology=" + topologyName
            + ",component=" + metric.getComponent()
            + ",operation=" + ObjectName.quote(metric.getOperation())
            + ",host-port-task=" + String.format("%s-%s-%s", taskInfo.srcWorkerHost
                ,taskInfo.srcWorkerPort
                ,taskInfo.srcTaskId);
}
 

import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
@Override
public MetricName name(Metric metric, IMetricsConsumer.TaskInfo taskInfo) {
    return new MetricName("storm",
            topologyName,
            metric.getComponent(),
            metric.getOperation(),
            mBeanName(metric, taskInfo));
}
 

import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
@Test(dataProvider = "taskInfo")
public void convertToTags(String stormId, IMetricsConsumer.TaskInfo taskInfo, Map<String, String> expectedTags) {
  assertThat(TagsHelper.convertToTags(stormId, taskInfo)).isEqualTo(expectedTags);
}
 

import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
/**
 * Returns the metric name for the storm metric produced by a task.
 *
 * @param metric storm metric object
 * @param taskInfo information about the task that generates the metric
 * @return the name for the yammer metric
 */
MetricName name(final Metric metric, final IMetricsConsumer.TaskInfo taskInfo);
 

import org.apache.storm.metric.api.IMetricsConsumer; //导入依赖的package包/类
/**
 * Processes the storm metric
 *
 * @param metric storm metric object
 * @param taskInfo information about the task that generates the metric
 */
void process(final Metric metric, final IMetricsConsumer.TaskInfo taskInfo);