title: How to export flink user custom metric to prometheus & grafana
body: <p>I'm trying to create my custom metric variable according to <a href="https://www.youtube.com/watch?v=vesj-ghLimA" rel="nofollow noreferrer">this tutorial</a></p>
  
  <p>With the sample code it's provided, I can get the events and the Histogram. </p>
  
  <p>I'm confused how the identifier been used by prometheus &amp; grafana. I also trying to modify the sample code little bit but the metric just no longer work.</p>
  
  <p>Also, I'm only able to access the system metric but not my own.</p>
  
  <p>My question is: </p>
  
  <ol>
  <li>how can I access the counter I created? for example counter1 </li>
  <li>What is the metricGroup exactly?  </li>
  <li>For example, I'd like to detect a pattern
  from an input stream, and it's more reasonable to do it in the
  metric or just output the result to a timeseries database like
  influxdb?</li>
  </ol>
  
  <p>thanks in advance.</p>
  
  <p>Here is the map function</p>
  
  <pre class="lang-java prettyprint-override"><code>class FlinkMetricsExposingMapFunction extends RichMapFunction&lt;SensorReading, SensorReading&gt; {
    private static final long serialVersionUID = 1L;
  
    private transient Counter eventCounter;
    private transient Counter customCounter1;
    private transient Counter customCounter2;
  
    @Override
    public void open(Configuration parameters) {
      eventCounter = getRuntimeContext()
          .getMetricGroup().counter("events");
  
      customCounter1 = getRuntimeContext()
          .getMetricGroup()
          .addGroup("customCounterKey", "mod2")
          .counter("counter1");
      customCounter2 = getRuntimeContext()
          .getMetricGroup()
          .addGroup("customCounterKey", "mod5")
          .counter("counter2");
  
   //    meter = getRuntimeContext().getMetricGroup().meter("eventMeter", new DropwizardMeterWrapper(dropwizardMeter));
    }
  
    @Override
    public SensorReading map(SensorReading value) {
      eventCounter.inc();
      if (value.getCurrTimestamp() % 2 == 0)
        customCounter1.inc();
  
      if (value.getCurrTimestamp() % 5 == 0)
        customCounter2.inc();
  
      if (value.getCurrTimestamp() % 2 == 0 &amp;&amp; value.getCurrTimestamp() % 5 == 0)
        customCounter1.dec();
      return value;
    }
  }
  
  </code></pre>
  
  <p>Example Job:</p>
  
  <pre class="lang-java prettyprint-override"><code>env
  .addSource(new SimpleSensorReadingGenerator())
  .name(SimpleSensorReadingGenerator.class.getSimpleName())
  .map(new FlinkMetricsExposingMapFunction())
  .name(FlinkMetricsExposingMapFunction.class.getSimpleName())
  .print()
  .name(DataStreamSink.class.getSimpleName());
  </code></pre>
  
  <hr>
  
  <h1>Update</h1>
  
  <p>Screenshot for access flink metrics from grafana:</p>
  
  <p><a href="https://i.stack.imgur.com/QOhM6.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/QOhM6.png" alt="enter image description here"></a></p>
  
  <p>flink-config.yaml </p>
  
  <pre><code>FROM flink:1.9.0
  RUN echo "metrics.reporters: prom" &gt;&gt; "$FLINK_HOME/conf/flink-conf.yaml"; \
      echo "metrics.latency.interval: 1000" &gt;&gt; "$FLINK_HOME/conf/flink-conf.yaml"; \
      echo "metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter" &gt;&gt; "$FLINK_HOME/conf/flink-conf.yaml"; \
      mv $FLINK_HOME/opt/flink-metrics-prometheus-*.jar $FLINK_HOME/lib
  COPY --from=builder /home/gradle/build/libs/*.jar $FLINK_HOME/lib/
  
  </code></pre>
  
  <p>default map function from tutorial:</p>
  
  <pre class="lang-java prettyprint-override"><code> @Override
   public void open(Configuration parameters) {
      eventCounter = getRuntimeContext().getMetricGroup().counter("events");
      valueHistogram = getRuntimeContext()
              .getMetricGroup()
              .histogram("value_histogram", new DescriptiveStatisticsHistogram(10_000_000));
   }
  
  </code></pre>