Hi guys, Currently metrics are implemented in runners/core as CounterCell, GaugeCell, DistributionCell, etc. If we want to send metrics to external systems via metrics reporter, we would have to define another set of metrics, say, codahale metrics, and update codahale metrics periodically with beam sdk metrics, which is inconvenient and inefficient.
Another problem is that Meter/Histogram cannot be updated directly in this way because their internal data decays after time. My opinion would be bridge beam sdk metrics to underlying runners so that updates would directly apply to underlying runners (Flink, Spark, etc) without conversion. Specifically, currently we already delegate Metrics.counter/gauge/distribution to DelegatingCounter/Gauge/Distribution, which uses MetricsContainer to store the actual metrics with the implementation of MetricsContainerImpl. If we can add an API in MetricsEnvironment to allow runners to override the default implementation, say, for flink, we have FlinkMetricsContainerImpl, then all metric updates will directly apply to metrics in FlinkMetricsContainerImpl without intermediate conversion and updates. And since the metrics are runner-specific, it would be a lot easier to support metrics reporters as well as Meters/Histograms. What do you think?
