Flink - metrics

/**
 * A MetricGroup is a named container for {@link Metric Metrics} and further metric subgroups.
 * 
 * <p>Instances of this class can be used to register new metrics with Flink and to create a nested
 * hierarchy based on the group names.
 * 
 * <p>A MetricGroup is uniquely identified by it's place in the hierarchy and name.
 */
public interface MetricGroup {
    <C extends Counter> C counter(int name, C counter);
    <T, G extends Gauge<T>> G gauge(int name, G gauge);
    <H extends Histogram> H histogram(String name, H histogram);
    MetricGroup addGroup(String name);
}

/**
 * Abstract {@link MetricGroup} that contains key functionality for adding metrics and groups.
 * 
 */
 
public abstract class AbstractMetricGroup implements MetricGroup {

    /** The registry that this metrics group belongs to */
    protected final MetricRegistry registry;

    /** All metrics that are directly contained in this group */
    private final Map<String, Metric> metrics = new HashMap<>();

    /** All metric subgroups of this group */
    private final Map<String, AbstractMetricGroup> groups = new HashMap<>();

    /** The metrics scope represented by this group.
     *  For example ["host-7", "taskmanager-2", "window_word_count", "my-mapper" ]. */
    private final String[] scopeComponents;  //命名空间

    /** The metrics scope represented by this group, as a concatenated string, lazily computed.
     * For example: "host-7.taskmanager-2.window_word_count.my-mapper" */
    private String scopeString;

    @Override
    public <C extends Counter> C counter(String name, C counter) {
        addMetric(name, counter);
        return counter;
    }
    
    /**
     * Adds the given metric to the group and registers it at the registry, if the group
     * is not yet closed, and if no metric with the same name has been registered before.
     * 
     * @param name the name to register the metric under
     * @param metric the metric to register
     */
    protected void addMetric(String name, Metric metric) {
        // add the metric only if the group is still open
        synchronized (this) { //加锁
            if (!closed) {
                // immediately put without a 'contains' check to optimize the common case (no collition)
                // collisions are resolved later
                Metric prior = metrics.put(name, metric);

                // check for collisions with other metric names
                if (prior == null) {
                    // no other metric with this name yet

                    registry.register(metric, name, this);
                }
                else {
                    // we had a collision. put back the original value
                    metrics.put(name, prior);
                    
                }
            }
        }
    }
}

/**
 * Reporters are used to export {@link Metric Metrics} to an external backend.
 * 
 * <p>Reporters are instantiated via reflection and must be public, non-abstract, and have a
 * public no-argument constructor.
 */
public interface MetricReporter {

    // ------------------------------------------------------------------------
    //  life cycle
    // ------------------------------------------------------------------------

    /**
     * Configures this reporter. Since reporters are instantiated generically and hence parameter-less,
     * this method is the place where the reporters set their basic fields based on configuration values.
     * 
     * <p>This method is always called first on a newly instantiated reporter.
     *
     * @param config The configuration with all parameters.
     */
    void open(MetricConfig config);

    /**
     * Closes this reporter. Should be used to close channels, streams and release resources.
     */
    void close();

    void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group);
    void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group);
}

/**
 * Base interface for custom metric reporters.
 */
public abstract class AbstractReporter implements MetricReporter, CharacterFilter {
    protected final Logger log = LoggerFactory.getLogger(getClass());

    protected final Map<Gauge<?>, String> gauges = new HashMap<>();
    protected final Map<Counter, String> counters = new HashMap<>();
    protected final Map<Histogram, String> histograms = new HashMap<>();

    @Override
    public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup group) {
        final String name = group.getMetricIdentifier(metricName, this); //group只是用来获取metrics完整的name

        synchronized (this) {
            if (metric instanceof Counter) {
                counters.put((Counter) metric, name);
            } else if (metric instanceof Gauge) {
                gauges.put((Gauge<?>) metric, name);
            } else if (metric instanceof Histogram) {
                histograms.put((Histogram) metric, name);
            } else {
                log.warn("Cannot add unknown metric type {}. This indicates that the reporter " +
                    "does not support this metric type.", metric.getClass().getName());
            }
        }
    }

    @Override
    public void notifyOfRemovedMetric(Metric metric, String metricName, MetricGroup group) {
        synchronized (this) {
            if (metric instanceof Counter) {
                counters.remove(metric);
            } else if (metric instanceof Gauge) {
                gauges.remove(metric);
            } else if (metric instanceof Histogram) {
                histograms.remove(metric);
            } else {
                log.warn("Cannot remove unknown metric type {}. This indicates that the reporter " +
                    "does not support this metric type.", metric.getClass().getName());
            }
        }
    }
}

/**
 * A MetricRegistry keeps track of all registered {@link Metric Metrics}. It serves as the
 * connection between {@link MetricGroup MetricGroups} and {@link MetricReporter MetricReporters}.
 */
public class MetricRegistry {
    
    private List<MetricReporter> reporters;
    private ScheduledExecutorService executor;

    private final ScopeFormats scopeFormats;

    private final char delimiter;

    /**
     * Creates a new MetricRegistry and starts the configured reporter.
     */
    public MetricRegistry(Configuration config) {
        // first parse the scope formats, these are needed for all reporters
        ScopeFormats scopeFormats;
        try {
            scopeFormats = createScopeConfig(config);  //从配置中读到scope的格式，即监控数据的namespace的格式是什么
        }
        catch (Exception e) {
            LOG.warn("Failed to parse scope format, using default scope formats", e);
            scopeFormats = new ScopeFormats();
        }
        this.scopeFormats = scopeFormats;

        char delim;
        try {
            delim = config.getString(ConfigConstants.METRICS_SCOPE_DELIMITER, ".").charAt(0);  //从配置里面读出分隔符
        } catch (Exception e) {
            LOG.warn("Failed to parse delimiter, using default delimiter.", e);
            delim = '.';
        }
        this.delimiter = delim;

        // second, instantiate any custom configured reporters
        this.reporters = new ArrayList<>();

        final String definedReporters = config.getString(ConfigConstants.METRICS_REPORTERS_LIST, null);  //读出配置的Reporters

        if (definedReporters == null) {
            // no reporters defined
            // by default, don't report anything
            LOG.info("No metrics reporter configured, no metrics will be exposed/reported.");
            this.executor = null;
        } else {
            // we have some reporters so
            String[] namedReporters = definedReporters.split("\\s*,\\s*");
            for (String namedReporter : namedReporters) {  //对于配置的每个reporter

                DelegatingConfiguration reporterConfig = new DelegatingConfiguration(config, ConfigConstants.METRICS_REPORTER_PREFIX + namedReporter + ".");
                final String className = reporterConfig.getString(ConfigConstants.METRICS_REPORTER_CLASS_SUFFIX, null);  //reporter class名配置

                try {
                    String configuredPeriod = reporterConfig.getString(ConfigConstants.METRICS_REPORTER_INTERVAL_SUFFIX, null); //report interval配置
                    TimeUnit timeunit = TimeUnit.SECONDS;
                    long period = 10;

                    if (configuredPeriod != null) {
                        try {
                            String[] interval = configuredPeriod.split(" ");
                            period = Long.parseLong(interval[0]);
                            timeunit = TimeUnit.valueOf(interval[1]);
                        }
                        catch (Exception e) {
                            LOG.error("Cannot parse report interval from config: " + configuredPeriod +
                                    " - please use values like '10 SECONDS' or '500 MILLISECONDS'. " +
                                    "Using default reporting interval.");
                        }
                    }

                    Class<?> reporterClass = Class.forName(className);
                    MetricReporter reporterInstance = (MetricReporter) reporterClass.newInstance(); //实例化reporter

                    MetricConfig metricConfig = new MetricConfig();
                    reporterConfig.addAllToProperties(metricConfig);
                    reporterInstance.open(metricConfig);  //open reporter

                    if (reporterInstance instanceof Scheduled) {
                        if (this.executor == null) {
                            executor = Executors.newSingleThreadScheduledExecutor(); //创建Executor
                        }
                        LOG.info("Periodically reporting metrics in intervals of {} {} for reporter {} of type {}.", period, timeunit.name(), namedReporter, className);

                        executor.scheduleWithFixedDelay(
                                new ReporterTask((Scheduled) reporterInstance), period, period, timeunit); //Scheduled report
                    }
                    reporters.add(reporterInstance); //加入reporters列表
                }
                catch (Throwable t) {
                    shutdownExecutor();
                    LOG.error("Could not instantiate metrics reporter" + namedReporter + ". Metrics might not be exposed/reported.", t);
                }
            }
        }
    }

    // ------------------------------------------------------------------------
    //  Metrics (de)registration
    // ------------------------------------------------------------------------

    /**
     * Registers a new {@link Metric} with this registry.
     *
     * @param metric      the metric that was added
     * @param metricName  the name of the metric
     * @param group       the group that contains the metric
     */
    public void register(Metric metric, String metricName, MetricGroup group) { //在AbstractMetricGroup.addMetric中被调用，metric被加到group的同时也会加到reporter中

        try {
            if (reporters != null) {
                for (MetricReporter reporter : reporters) {
                    if (reporter != null) {
                        reporter.notifyOfAddedMetric(metric, metricName, group); //把metric加到每个reporters上面
                    }
                }
            }
        } catch (Exception e) {
            LOG.error("Error while registering metric.", e);
        }
    }

    /**
     * Un-registers the given {@link org.apache.flink.metrics.Metric} with this registry.
     *
     * @param metric      the metric that should be removed
     * @param metricName  the name of the metric
     * @param group       the group that contains the metric
     */
    public void unregister(Metric metric, String metricName, MetricGroup group) {
        try {
            if (reporters != null) {
                for (MetricReporter reporter : reporters) {
                    if (reporter != null) {
                        reporter.notifyOfRemovedMetric(metric, metricName, group);
                    }
                }
            }
        } catch (Exception e) {
            LOG.error("Error while registering metric.", e);
        }
    }

    // ------------------------------------------------------------------------

    /**
     * This task is explicitly a static class, so that it does not hold any references to the enclosing
     * MetricsRegistry instance.
     *
     * This is a subtle difference, but very important: With this static class, the enclosing class instance
     * may become garbage-collectible, whereas with an anonymous inner class, the timer thread
     * (which is a GC root) will hold a reference via the timer task and its enclosing instance pointer.
     * Making the MetricsRegistry garbage collectible makes the java.util.Timer garbage collectible,
     * which acts as a fail-safe to stop the timer thread and prevents resource leaks.
     */
    private static final class ReporterTask extends TimerTask {

        private final Scheduled reporter;

        private ReporterTask(Scheduled reporter) {
            this.reporter = reporter;
        }

        @Override
        public void run() {
            try {
                reporter.report();  //Task的核心就是调用reporter.report
            } catch (Throwable t) {
                LOG.warn("Error while reporting metrics", t);
            }
        }
    }
}

metricsRegistry = new FlinkMetricRegistry(config.configuration)

taskManagerMetricGroup = 
  new TaskManagerMetricGroup(metricsRegistry, this.runtimeInfo.getHostname, id.toString)

TaskManager.instantiateStatusMetrics(taskManagerMetricGroup)

private def instantiateStatusMetrics(taskManagerMetricGroup: MetricGroup) : Unit = {
    val jvm = taskManagerMetricGroup
      .addGroup("Status")
      .addGroup("JVM")

    instantiateClassLoaderMetrics(jvm.addGroup("ClassLoader"))
    instantiateGarbageCollectorMetrics(jvm.addGroup("GarbageCollector"))
    instantiateMemoryMetrics(jvm.addGroup("Memory"))
    instantiateThreadMetrics(jvm.addGroup("Threads"))
    instantiateCPUMetrics(jvm.addGroup("CPU"))
  }

  private def instantiateClassLoaderMetrics(metrics: MetricGroup) {
    val mxBean = ManagementFactory.getClassLoadingMXBean //从ManagementFactory可以取出表示JVM指标的MXBean

    metrics.gauge[Long, FlinkGauge[Long]]("ClassesLoaded", new FlinkGauge[Long] {
      override def getValue: Long = mxBean.getTotalLoadedClassCount
    })
    metrics.gauge[Long, FlinkGauge[Long]]("ClassesUnloaded", new FlinkGauge[Long] {
      override def getValue: Long = mxBean.getUnloadedClassCount
    })
  }

  val taskMetricGroup = taskManagerMetricGroup.addTaskForJob(tdd)

  val task = new Task(
    tdd,
    memoryManager,
    ioManager,
    network,
    bcVarManager,
    selfGateway,
    jobManagerGateway,
    config.timeout,
    libCache,
    fileCache,
    runtimeInfo,
    taskMetricGroup)

Environment env = new RuntimeEnvironment(jobId, vertexId, executionId,
        executionConfig, taskInfo, jobConfiguration, taskConfiguration,
        userCodeClassLoader, memoryManager, ioManager,
        broadcastVariableManager, accumulatorRegistry,
        splitProvider, distributedCacheEntries,
        writers, inputGates, jobManager, taskManagerConfig, metrics, this);

// let the task code create its readers and writers
invokable.setEnvironment(env);

public abstract class AbstractInvokable {

    /** The environment assigned to this invokable. */
    private Environment environment;

    /**
     * Starts the execution.
     *
     * <p>Must be overwritten by the concrete task implementation. This method
     * is called by the task manager when the actual execution of the task
     * starts.
     *
     * <p>All resources should be cleaned up when the method returns. Make sure
     * to guard the code with <code>try-finally</code> blocks where necessary.
     * 
     * @throws Exception
     *         Tasks may forward their exceptions for the TaskManager to handle through failure/recovery.
     */
    public abstract void invoke() throws Exception;

    /**
     * Sets the environment of this task.
     * 
     * @param environment
     *        the environment of this task
     */
    public final void setEnvironment(Environment environment) {
        this.environment = environment;
    }

    /**
     * Returns the environment of this task.
     * 
     * @return The environment of this task.
     */
    public Environment getEnvironment() {
        return this.environment;
    }
}

getEnvironment().getMetricGroup().gauge("lastCheckpointSize", new Gauge<Long>() {
    @Override
    public Long getValue() {
        return StreamTask.this.lastCheckpointSize;
    }
});