深入理解Flink Metrics的内部结构

从Metrics的使用说起

Flink的Metrics种类有四种Counters, Gauges, Histograms和Meters.

如何使用Metrics呢? 以Counter为例,

public class MyMapper extends RichMapFunction<String, String> {
  private transient Counter counter;

  @Override
  public void open(Configuration config) {
    this.counter = getRuntimeContext()
      .getMetricGroup()
      .counter("myCounter");
  }

  @Override
  public String map(String value) throws Exception {
    this.counter.inc();
    return value;
  }
}

行7 getMetricGroup()获取MetricGroup

行8 从MetricGroup中获取Metric实例

那么,我们来探访一下MetricGroup

Metric容器--MetricGroup

MetricGroup是Metric对象和metric subgroups的容器.

调用以下4个方法可以获得Metric对象并调用addMetric()注册这个Metric.

(AbstractMetricGroup.java)

public <C extends Counter> C counter(String name, C counter)     
{
    addMetric(name, counter);
    return counter;
}

public <T, G extends Gauge<T>> G gauge(String name, G gauge) {
    addMetric(name, gauge);
    return gauge;
}

public <H extends Histogram> H histogram(String name, H histogram) {
    addMetric(name, histogram);
    return histogram;
}

public <M extends Meter> M meter(String name, M meter) {
    addMetric(name, meter);
    return meter;
}

注,MetricGroup接口的另一个实现UnregisteredMetricsGroup仅仅返回Metric实例而不对Metric进行注册

注2,MetricGroup接口的第三个实现ProxyMetricGroup有一个parent MetricGroup,ProxyMetricGroup所有的调用都转发到parentMetricGroup上

(AbstractMetricGroup.java)重要的域

    /** 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<>();/** Flag indicating whether this group has been closed. */
    private volatile boolean closed;

(AbstractMetricGroup.java)

    protected void addMetric(String name, Metric metric) {
        if (metric == null) {
            LOG.warn("Ignoring attempted registration of a metric due to being null for name {}.", name);
            return;
        }
        // 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 collision)
                // 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

                    if (groups.containsKey(name)) {
                        // we warn here, rather than failing, because metrics are tools that should not fail the
                        // program when used incorrectly
                        LOG.warn("Name collision: Adding a metric with the same name as a metric subgroup: '" +
                                name + "'. Metric might not get properly reported. " + Arrays.toString(scopeComponents));
                    }

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

                    // we warn here, rather than failing, because metrics are tools that should not fail the
                    // program when used incorrectly
                    LOG.warn("Name collision: Group already contains a Metric with the name '" +
                            name + "'. Metric will not be reported." + Arrays.toString(scopeComponents));
                }
            }
        }
    }

具体再来看一下addMetric()的代码

行7 获得互斥锁

行8 检测当前group是否close

行11~34 把要注册的Metric对象添加到metrics map中

这里一个小trick是,默认没有key的冲突,直接把这个metric对象添加到map中.再回头检测是否有值被替换出来.这样的做法可以优化性能(若没有key冲突,减少了一次map寻址)

行24 在MetricRegister中注册Metric,这个在下一节详谈

调用addGroup()可以添加subgroup

(AbstractMetricGroup.java)

    private AbstractMetricGroup<?> addGroup(String name, ChildType childType) {
        synchronized (this) {
            if (!closed) {
                // adding a group with the same name as a metric creates problems in many reporters/dashboards
                // we warn here, rather than failing, because metrics are tools that should not fail the
                // program when used incorrectly
                if (metrics.containsKey(name)) {
                    LOG.warn("Name collision: Adding a metric subgroup with the same name as an existing metric: '" +
                            name + "'. Metric might not get properly reported. " + Arrays.toString(scopeComponents));
                }

                AbstractMetricGroup newGroup = createChildGroup(name, childType);
                AbstractMetricGroup prior = groups.put(name, newGroup);
                if (prior == null) {
                    // no prior group with that name
                    return newGroup;
                } else {
                    // had a prior group with that name, add the prior group back
                    groups.put(name, prior);
                    return prior;
                }
            }
            else {
                // return a non-registered group that is immediately closed already
                GenericMetricGroup closedGroup = new GenericMetricGroup(registry, this, name);
                closedGroup.close();
                return closedGroup;
            }
        }
    }

    protected GenericMetricGroup createChildGroup(String name, ChildType childType) {
        switch (childType) {
            case KEY:
                return new GenericKeyMetricGroup(registry, this, name);
            default:
                return new GenericMetricGroup(registry, this, name);
        }
    }

    /**
     * Enum for indicating which child group should be created.
     * `KEY` is used to create {@link GenericKeyMetricGroup}.
     * `VALUE` is used to create {@link GenericValueMetricGroup}.
     * `GENERIC` is used to create {@link GenericMetricGroup}.
     */
    protected enum ChildType {
        KEY,
        VALUE,
        GENERIC
    }

行2 获取互斥锁

行12~21 新建MetricGroup对象

注意,添加的subgroup的name与Metric对象的name相同会造成问题.

行25,35,37 同一个tree里的MetricGroup对象使用同一个MetricRegister

行26 close MetricGroup

    public void close() {
        synchronized (this) {
            if (!closed) {
                closed = true;

                // close all subgroups
                for (AbstractMetricGroup group : groups.values()) {
                    group.close();
                }
                groups.clear();

                // un-register all directly contained metrics
                for (Map.Entry<String, Metric> metric : metrics.entrySet()) {
                    registry.unregister(metric.getValue(), metric.getKey(), this);
                }
                metrics.clear();
            }
        }
    }

行2 获取互斥锁

递归地close所有subgroups, 注销所有metrics

MetricGroup中的addMetric(),addGroup(),close()以及上面未提到的getAllVariables()方法需要获取互斥锁

原因: 防止关闭group的同时添加metrics和subgroups造成的资源泄露.

MetricGroup另一个很重要的方法是public String getMetricIdentifier(String metricName, CharacterFilter filter, int reporterIndex).

作用是获取某个Metric的唯一名作为标志(identifier).

identifier分为3部分:System scope, User scope, Metric name

A.B.C   其中,A为System scope,B为User scope,C为Metric name, '.'是分隔符

System Scope可在conf/flink-conf.yaml中定义.

User Scope就是groups tree, 可调用addGroup(String)来定义 (可定义多层group)

MetricGroup与MetricReporter之间的桥梁 -- MetricRegister

MetricRegistry追踪所有已注册的Metric.它作为MetricGroup和MetricReporter之间的桥梁.

在MetricGroup的addMetric()方法中调用了MetricRegister的register()方法:

registry.register(metric, name, this);

在MetricGroup的close()方法中调用了MetricRegister的unregister()方法:

registry.unregister(metric.getValue(), metric.getKey(), this);

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

    @Override
    public void register(Metric metric, String metricName, AbstractMetricGroup group) {
        synchronized (lock) {
            if (isShutdown()) {
                LOG.warn("Cannot register metric, because the MetricRegistry has already been shut down.");
            } else {
                if (reporters != null) {
                    for (int i = 0; i < reporters.size(); i++) {
                        MetricReporter reporter = reporters.get(i);
                        try {
                            if (reporter != null) {
                                FrontMetricGroup front = new FrontMetricGroup<AbstractMetricGroup<?>>(i, group);
                                reporter.notifyOfAddedMetric(metric, metricName, front);
                            }
                        } catch (Exception e) {
                            LOG.warn("Error while registering metric.", e);
                        }
                    }
                }
                try {
                    if (queryService != null) {
                        MetricQueryService.notifyOfAddedMetric(queryService, metric, metricName, group);
                    }
                } catch (Exception e) {
                    LOG.warn("Error while registering metric.", e);
                }
                try {
                    if (metric instanceof View) {
                        if (viewUpdater == null) {
                            viewUpdater = new ViewUpdater(executor);
                        }
                        viewUpdater.notifyOfAddedView((View) metric);
                    }
                } catch (Exception e) {
                    LOG.warn("Error while registering metric.", e);
                }
            }
        }
    }

    @Override
    public void unregister(Metric metric, String metricName, AbstractMetricGroup group) {
        synchronized (lock) {
            if (isShutdown()) {
                LOG.warn("Cannot unregister metric, because the MetricRegistry has already been shut down.");
            } else {
                if (reporters != null) {
                    for (int i = 0; i < reporters.size(); i++) {
                        try {
                        MetricReporter reporter = reporters.get(i);
                            if (reporter != null) {
                                FrontMetricGroup front = new FrontMetricGroup<AbstractMetricGroup<?>>(i, group);
                                reporter.notifyOfRemovedMetric(metric, metricName, front);
                            }
                        } catch (Exception e) {
                            LOG.warn("Error while registering metric.", e);
                        }
                    }
                }
                try {
                    if (queryService != null) {
                        MetricQueryService.notifyOfRemovedMetric(queryService, metric);
                    }
                } catch (Exception e) {
                    LOG.warn("Error while registering metric.", e);
                }
                try {
                    if (metric instanceof View) {
                        if (viewUpdater != null) {
                            viewUpdater.notifyOfRemovedView((View) metric);
                        }
                    }
                } catch (Exception e) {
                    LOG.warn("Error while registering metric.", e);
                }
            }
        }
    }

register()方法和unregister()方法基本相似

行7 获取同步锁. 锁对象不再是this,而是new Object().这样做,方便拓展第二个锁.

行11~23 向所有下属的MetricReporter添加该Metric

行24~30 向MetricQueryService添加该Metric

MetricQueryService是个actor,它会将Metric序列化,然后写入到output stream

行31~40 如果Metric实现了View接口,那么在viewUpdater中注册这个Metric

Metric类实现View接口后,可以按设定时间间隔来更新这个Metric(由viewUpdater来执行update)

MetricReporter

MetricReporter用于把Metric导出到外部backend.

外部backend的参数可在conf/flink-conf.yaml中设定.

可同时设定多个外部backend.

MetricReporter接口

public interface MetricReporter {

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

    void open(MetricConfig config); //

    void close();

    // ------------------------------------------------------------------------
    //  adding / removing metrics
    // ------------------------------------------------------------------------

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


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

行8 配置这个Reporter.

由于reporter的构造器是无参的,这个方法用于初始化reporter的域.  

这个方法总是在对象构造后调用

行10 关闭这个Reporter.

应该在这个方法中关闭 channels,streams以及释放资源.

行16,19 增删metrics

常规的reporter类还需要实现Scheduled接口用于报告当前的measurements

public interface Scheduled {

    void report();
}

行3 由metric registry定期地调用report()方法,来报告当前的measurements