apache-flink
How to export flink task or Back Pressure related metrics to prometheus?
I followed the instructions Reporter to export flink metrics to prometheus, however it seems by default it only export job-manager related metrics to prometheus, see below:
Open http://localhost:9249/, I just get the following info, no task or task manager related metrics found.
# HELP flink_jobmanager_Status_JVM_Memory_Mapped_MemoryUsed MemoryUsed (scope: jobmanager_Status_JVM_Memory_Mapped)
# TYPE flink_jobmanager_Status_JVM_Memory_Mapped_MemoryUsed gauge
flink_jobmanager_Status_JVM_Memory_Mapped_MemoryUsed{host="localhost",} 0.0
# HELP flink_jobmanager_Status_JVM_ClassLoader_ClassesUnloaded ClassesUnloaded (scope: jobmanager_Status_JVM_ClassLoader)
# TYPE flink_jobmanager_Status_JVM_ClassLoader_ClassesUnloaded gauge
flink_jobmanager_Status_JVM_ClassLoader_ClassesUnloaded{host="localhost",} 0.0
# HELP flink_jobmanager_Status_JVM_GarbageCollector_PS_Scavenge_Time Time (scope: jobmanager_Status_JVM_GarbageCollector_PS_Scavenge)
# TYPE flink_jobmanager_Status_JVM_GarbageCollector_PS_Scavenge_Time gauge
flink_jobmanager_Status_JVM_GarbageCollector_PS_Scavenge_Time{host="localhost",} 273.0
# HELP flink_jobmanager_job_lastCheckpointRestoreTimestamp lastCheckpointRestoreTimestamp (scope: jobmanager_job)
# TYPE flink_jobmanager_job_lastCheckpointRestoreTimestamp gauge
flink_jobmanager_job_lastCheckpointRestoreTimestamp{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} -1.0
# HELP flink_jobmanager_job_lastCheckpointAlignmentBuffered lastCheckpointAlignmentBuffered (scope: jobmanager_job)
# TYPE flink_jobmanager_job_lastCheckpointAlignmentBuffered gauge
flink_jobmanager_job_lastCheckpointAlignmentBuffered{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 0.0
# HELP flink_jobmanager_job_lastCheckpointExternalPath lastCheckpointExternalPath (scope: jobmanager_job)
# TYPE flink_jobmanager_job_lastCheckpointExternalPath gauge
flink_jobmanager_job_lastCheckpointExternalPath{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 0.0
# HELP flink_jobmanager_Status_JVM_Memory_Direct_TotalCapacity TotalCapacity (scope: jobmanager_Status_JVM_Memory_Direct)
# TYPE flink_jobmanager_Status_JVM_Memory_Direct_TotalCapacity gauge
flink_jobmanager_Status_JVM_Memory_Direct_TotalCapacity{host="localhost",} 2374599.0
# HELP flink_jobmanager_Status_JVM_Threads_Count Count (scope: jobmanager_Status_JVM_Threads)
# TYPE flink_jobmanager_Status_JVM_Threads_Count gauge
flink_jobmanager_Status_JVM_Threads_Count{host="localhost",} 47.0
# HELP flink_jobmanager_Status_JVM_Memory_Heap_Committed Committed (scope: jobmanager_Status_JVM_Memory_Heap)
# TYPE flink_jobmanager_Status_JVM_Memory_Heap_Committed gauge
flink_jobmanager_Status_JVM_Memory_Heap_Committed{host="localhost",} 1.058013184E9
# HELP flink_jobmanager_Status_JVM_Memory_NonHeap_Used Used (scope: jobmanager_Status_JVM_Memory_NonHeap)
# TYPE flink_jobmanager_Status_JVM_Memory_NonHeap_Used gauge
flink_jobmanager_Status_JVM_Memory_NonHeap_Used{host="localhost",} 7.02964E7
# HELP flink_jobmanager_job_restartingTime restartingTime (scope: jobmanager_job)
# TYPE flink_jobmanager_job_restartingTime gauge
flink_jobmanager_job_restartingTime{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 0.0
# HELP flink_jobmanager_Status_JVM_GarbageCollector_PS_Scavenge_Count Count (scope: jobmanager_Status_JVM_GarbageCollector_PS_Scavenge)
# TYPE flink_jobmanager_Status_JVM_GarbageCollector_PS_Scavenge_Count gauge
flink_jobmanager_Status_JVM_GarbageCollector_PS_Scavenge_Count{host="localhost",} 24.0
# HELP flink_jobmanager_Status_JVM_Memory_NonHeap_Committed Committed (scope: jobmanager_Status_JVM_Memory_NonHeap)
# TYPE flink_jobmanager_Status_JVM_Memory_NonHeap_Committed gauge
flink_jobmanager_Status_JVM_Memory_NonHeap_Committed{host="localhost",} 7.2876032E7
# HELP flink_jobmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Count Count (scope: jobmanager_Status_JVM_GarbageCollector_PS_MarkSweep)
# TYPE flink_jobmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Count gauge
flink_jobmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Count{host="localhost",} 2.0
# HELP flink_jobmanager_job_downtime downtime (scope: jobmanager_job)
# TYPE flink_jobmanager_job_downtime gauge
flink_jobmanager_job_downtime{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 0.0
# HELP flink_jobmanager_job_numberOfFailedCheckpoints numberOfFailedCheckpoints (scope: jobmanager_job)
# TYPE flink_jobmanager_job_numberOfFailedCheckpoints gauge
flink_jobmanager_job_numberOfFailedCheckpoints{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 0.0
# HELP flink_jobmanager_job_numberOfInProgressCheckpoints numberOfInProgressCheckpoints (scope: jobmanager_job)
# TYPE flink_jobmanager_job_numberOfInProgressCheckpoints gauge
flink_jobmanager_job_numberOfInProgressCheckpoints{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 0.0
# HELP flink_jobmanager_job_numberOfCompletedCheckpoints numberOfCompletedCheckpoints (scope: jobmanager_job)
# TYPE flink_jobmanager_job_numberOfCompletedCheckpoints gauge
flink_jobmanager_job_numberOfCompletedCheckpoints{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 961.0
# HELP flink_jobmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Time Time (scope: jobmanager_Status_JVM_GarbageCollector_PS_MarkSweep)
# TYPE flink_jobmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Time gauge
flink_jobmanager_Status_JVM_GarbageCollector_PS_MarkSweep_Time{host="localhost",} 110.0
# HELP flink_jobmanager_Status_JVM_Memory_Mapped_Count Count (scope: jobmanager_Status_JVM_Memory_Mapped)
# TYPE flink_jobmanager_Status_JVM_Memory_Mapped_Count gauge
flink_jobmanager_Status_JVM_Memory_Mapped_Count{host="localhost",} 0.0
# HELP flink_jobmanager_Status_JVM_CPU_Load Load (scope: jobmanager_Status_JVM_CPU)
# TYPE flink_jobmanager_Status_JVM_CPU_Load gauge
flink_jobmanager_Status_JVM_CPU_Load{host="localhost",} 0.0025814303680169446
# HELP flink_jobmanager_Status_JVM_ClassLoader_ClassesLoaded ClassesLoaded (scope: jobmanager_Status_JVM_ClassLoader)
# TYPE flink_jobmanager_Status_JVM_ClassLoader_ClassesLoaded gauge
flink_jobmanager_Status_JVM_ClassLoader_ClassesLoaded{host="localhost",} 7693.0
# HELP flink_jobmanager_Status_JVM_Memory_Heap_Max Max (scope: jobmanager_Status_JVM_Memory_Heap)
# TYPE flink_jobmanager_Status_JVM_Memory_Heap_Max gauge
flink_jobmanager_Status_JVM_Memory_Heap_Max{host="localhost",} 1.058013184E9
# HELP flink_jobmanager_job_uptime uptime (scope: jobmanager_job)
# TYPE flink_jobmanager_job_uptime gauge
flink_jobmanager_job_uptime{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 4811388.0
# HELP flink_jobmanager_Status_JVM_CPU_Time Time (scope: jobmanager_Status_JVM_CPU)
# TYPE flink_jobmanager_Status_JVM_CPU_Time gauge
flink_jobmanager_Status_JVM_CPU_Time{host="localhost",} 1.044894698E11
# HELP flink_jobmanager_Status_JVM_Memory_Direct_Count Count (scope: jobmanager_Status_JVM_Memory_Direct)
# TYPE flink_jobmanager_Status_JVM_Memory_Direct_Count gauge
flink_jobmanager_Status_JVM_Memory_Direct_Count{host="localhost",} 60.0
# HELP flink_jobmanager_Status_JVM_Memory_Heap_Used Used (scope: jobmanager_Status_JVM_Memory_Heap)
# TYPE flink_jobmanager_Status_JVM_Memory_Heap_Used gauge
flink_jobmanager_Status_JVM_Memory_Heap_Used{host="localhost",} 2.15962464E8
# HELP flink_jobmanager_job_lastCheckpointDuration lastCheckpointDuration (scope: jobmanager_job)
# TYPE flink_jobmanager_job_lastCheckpointDuration gauge
flink_jobmanager_job_lastCheckpointDuration{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 16.0
# HELP flink_jobmanager_Status_JVM_Memory_NonHeap_Max Max (scope: jobmanager_Status_JVM_Memory_NonHeap)
# TYPE flink_jobmanager_Status_JVM_Memory_NonHeap_Max gauge
flink_jobmanager_Status_JVM_Memory_NonHeap_Max{host="localhost",} -1.0
# HELP flink_jobmanager_job_lastCheckpointSize lastCheckpointSize (scope: jobmanager_job)
# TYPE flink_jobmanager_job_lastCheckpointSize gauge
flink_jobmanager_job_lastCheckpointSize{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 271280.0
# HELP flink_jobmanager_job_fullRestarts fullRestarts (scope: jobmanager_job)
# TYPE flink_jobmanager_job_fullRestarts gauge
flink_jobmanager_job_fullRestarts{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 0.0
# HELP flink_jobmanager_Status_JVM_Memory_Direct_MemoryUsed MemoryUsed (scope: jobmanager_Status_JVM_Memory_Direct)
# TYPE flink_jobmanager_Status_JVM_Memory_Direct_MemoryUsed gauge
flink_jobmanager_Status_JVM_Memory_Direct_MemoryUsed{host="localhost",} 2374600.0
# HELP flink_jobmanager_job_totalNumberOfCheckpoints totalNumberOfCheckpoints (scope: jobmanager_job)
# TYPE flink_jobmanager_job_totalNumberOfCheckpoints gauge
flink_jobmanager_job_totalNumberOfCheckpoints{job_id="dfac65e575f318970e0225eab9688a2e",host="localhost",job_name="Popular_Places_to_Elasticsearch",} 961.0
# HELP flink_jobmanager_Status_JVM_Memory_Mapped_TotalCapacity TotalCapacity (scope: jobmanager_Status_JVM_Memory_Mapped)
# TYPE flink_jobmanager_Status_JVM_Memory_Mapped_TotalCapacity gauge
flink_jobmanager_Status_JVM_Memory_Mapped_TotalCapacity{host="localhost",} 0.0
My question is how to export task or Back Pressure related metrics such as numRecordsIn,numRecordsInPerSecond,numRecordsOut to prometheus also? what else I need to configure?
BTW, My test environment is Flink 1.5.2 with job manager and task manager located in the same windows machine, also I found the Flink 1.6.0 has the same issue.
Solution
When you are running a jobmanager and a taskmanager on the same host, then each needs its own port. In flink-conf.yaml you can configure a range of ports like this, for example:
metrics.reporters: prom
metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
metrics.reporter.prom.port: 9250-9260
If you do this, then you'll find the jobmanager's metrics at http://localhost:9250 and the taskmanager's metrics at http://localhost:9251.
You will also need to adjust your prometheus.yml to match:
scrape_configs:
- job_name: 'jobmanager'
static_configs:
- targets: ['localhost:9250']
- job_name: 'taskmanager'
static_configs:
- targets: ['localhost:9251']
- What are the benefits of Apache Beam over Spark/Flink for batch processing?
- What is/are the main difference(s) between Flink and Storm?
- FLINK - will SQL window flush the element on regular interval for processing
- Difference between job, task and subtask in flink
- Flink failed to deserialize JSON produced by Debezium
- Flink serialization of java.util.List and java.util.Map
- Flink webUI - GC time
- Where the Upsert Kafka connector consumer start?
- The implementation of the AbstractRichFunction is not serializable when using JDBC Sink in Flink
- Flink standalone mode takes too long to start
- Limiting the state size in flink
- Immediate CEP Event Trigger Issue with WatermarkStrategy in Flink 1.16.1
- Connect a stream with watermarks with another one without watermarks in Flink
- Read a keyed Kafka Record using apache Flink?
- Error in Flink process Kafka topic:java.net.ConnectException: Connection refused (Connection refused)
- Apache Flink with multiple Kafka sources. Ensure one topic is fully read before consuming data on the other topic
- Flink user defined sink connector can not serialize data into JSON format
- Using Spring with Apache Flink - Command line arguments are not available to Spring
- Is there any chance to limit database sessions using jdbc sinks with apache flink?
- Flink GlobalWindow Trigger only process the trigger event
- Why does Flink Table with Kafka Connector not return results for window-based aggregation operations?
- Dependency management and execution environment in apache flink
- The POJO class passes the test ,but shows invalid during execution
- Flink KeyedProcessFunction Creation Count
- Apache Flink Python Datastream API sink to Parquet
- Unable to use s3-fs-hadoop plugin in Kubernetes
- Build a JSON_Object value in Flink SQL
- Kafka Migration with MM2 and Flink: How to Handle Offset Changes and Savepoints?
- Performance difference between Table- and DataStream-API
- Apache Flink: restoring state from checkpoint with changes Kafka topic