Collecting metrics for average time taken by flink flatmap function

Question

I have implemented a flink flatmap function and I want to collect metrics for average time for this function which I plan to monitor via prometheus.

What would be good approach for it? I have added a gauge to the method(extending Gauge interface from flink API).

public class SimpleGauge<T> implements Gauge<T> {

    private T mValue;

    @Override
    public T getValue() {
        return mValue;
    }

    public void setValue(T value){
        mValue = value;
    }
}

And I call setValue from flatmap function:

float endTime = (System.currentTimeMillis() - startTime) / 1000F;
this.gauge.setValue(endTime);

Would it work?

Answer 1

Whether it will work or not depends on how do You create the Gauge. You should probably switch to RichFlatMapFunction if You are using the plain FlatMapFunction and then initialize and register Your gauge getRuntimeContext.

Also, You may want to read a little bit on metric types because Histogram seems to be a better idea than Gauge in this case, as measuring the average latency isn't generally the best idea in most of the cases.