Page MenuHomePhabricator
Create Task
Maniphest T418996

Audit and fix observability (logging and metrics) for pyflink jobs
Open, Needs TriagePublic
Actions

Description

We have some doubts about how the logs and metrics produced by our Flink pipelines are reaching Grafana and OpenSearch.

We need to ensure that both the logs produced by Flink itself and by our Python code appear in Opensearch.

We also need to ensure that all metrics are shown in Grafana, where we have a dashboard for Flink, but there are other dashboards that show interesting metrics related to our pipelines, like the number of calls the HTTP APIs.

Task is done if:

  • All logs appear in Opensearch and we have documented it.
  • We have documented where the metrics appear in Grafana.

Ideally, we could improve this to:

  • Ensure we use a common format on logs, like JSON, and the fields produced are parsed by logstach/opensearch
  • Ensure there is an Index Pattern that can understand our logs and extract valuable variables.
  • We build an Opensearch dashboard based on the HTML (or general Flink?) pipeline, with important counts and grouped metrics, like:
    • How many messages were rejected/failed by reason.
    • How many messages were rejected/failed by "change_type_kind"
    • ... TBD
  • We build a Grafana dashboard that shows all the important metrics for our HTML (or all Flink?) pipelines together.
    • General Flink status information
    • Number of calls to the HTTP APIs, with codes returned
    • Latency calling the HTTP APIs
    • Avg latency of messages (from source to sink)
      • Avg latency split by "change_type_kind" (to know if there are issues calling the APIs for specific change types, maybe not hitting cache?)
      • Avg latency split by "Flink process step" (to know if a process is slower than others)

Maybe some of this "ideas" can be split into other tickets, some of them might require more custom logs or metrics to be produced.

Details

Related Changes in GitLab:
TitleReferenceAuthorSource BranchDest Branch
Warnings when exception caughtrepos/data-engineering/eventutilities-python!98javiermontonfeature/warning-errorsmain
Customize query in GitLab

Related Objects
Search...

Event Timeline

JMonton-WMF created this task.Mar 4 2026, 11:14 AM
Ottomata added a comment.Mar 4 2026, 3:18 PM

Nice!

We build a Grafana dashboard that shows all the important metrics for our HTML (or all Flink?) pipelines together.

The Flink App dashboard should be the one. We can surely add more metrics there.

Avg latency split by "change_type_kind" (to know if there are issues calling the APIs for specific change types, maybe not hitting cache?)

This kind of stuff is pretty app specific. If we do this, we should either add it to a dedicated page_change enrichment related section of the dashboard, or a separate dashboard altogether.

Ottomata added a comment.Mar 4 2026, 3:20 PM

Ensure there is an Index Pattern that can understand our logs and extract valuable variables.

This should be ECS. T234565: Standardize the logstash logging format - ECS

JMonton-WMF added a comment.Mar 6 2026, 10:47 AM

These are logs taken from a PyFlink application. They seem to mix JSON with plain text, and many errors are reported as INFO. At the end of these logs are real ERROR logs being produced, it looks like Flink is reporting al logs coming from the Python code (or Java?) as INFOs, until it finally breaks.

{"@timestamp":"2026-03-06T10:23:57.726Z","log.level": "INFO","message":"           ^^^^^^^^^^^", "ecs.version": "1.2.0","process.thread.name":"flink-pekko.actor.default-dispatcher-13","log.logger":"org.apache.flink.client.python.PythonDriver"}
  File "/usr/local/lib/python3.11/dist-packages/py4j/protocol.py", line 326, in get_return_value
{"@timestamp":"2026-03-06T10:23:57.726Z","log.level": "INFO","message":"  File \"/usr/local/lib/python3.11/dist-packages/py4j/protocol.py\", line 326, in get_return_value", "ecs.version": "1.2.0","process.thread.name":"flink-pekko.actor.default-dispatcher-13","log.logger":"org.apache.flink.client.python.PythonDriver"}
    raise Py4JJavaError(
{"@timestamp":"2026-03-06T10:23:57.726Z","log.level": "INFO","message":"    raise Py4JJavaError(", "ecs.version": "1.2.0","process.thread.name":"flink-pekko.actor.default-dispatcher-13","log.logger":"org.apache.flink.client.python.PythonDriver"}
py4j.protocol.Py4JJavaError: An error occurred while calling o0.execute.
{"@timestamp":"2026-03-06T10:23:57.726Z","log.level": "INFO","message":"py4j.protocol.Py4JJavaError: An error occurred while calling o0.execute.", "ecs.version": "1.2.0","process.thread.name":"flink-pekko.actor.default-dispatcher-13","log.logger":"org.apache.flink.client.python.PythonDriver"}
: java.net.MalformedURLException: no protocol: ['file:/usr/local/lib/python3.11/dist-packages/pyflink/opt/flink-python-1.20.2.jar', 'file:/usr/local/lib/python3.11/dist-packages/pyflink/opt/flink-python-1.20.2.jar', 'file:///opt/lib/venv/lib/python3.11/site-packages/eventutilities_python/lib/flink-connector-kafka-3.3.0-1.20.jar', 'file:///opt/lib/venv/lib/python3.11/site-packages/eventutilities_python/lib/eventutilities-flink-1.4.0-jar-with-dependencies.jar', 'file:///opt/lib/venv/lib/python3.11/site-packages/eventutilities_python/lib/flink-python-1.20.0-tests.jar', 'file:///opt/lib/venv/lib/python3.11/site-packages/eventutilities_python/lib/kafka-clients-3.4.0.jar']
{"@timestamp":"2026-03-06T10:23:57.726Z","log.level": "INFO","message":": java.net.MalformedURLException: no protocol: ['file:/usr/local/lib/python3.11/dist-packages/pyflink/opt/flink-python-1.20.2.jar', 'file:/usr/local/lib/python3.11/dist-packages/pyflink/opt/flink-python-1.20.2.jar', 'file:///opt/lib/venv/lib/python3.11/site-packages/eventutilities_python/lib/flink-connector-kafka-3.3.0-1.20.jar', 'file:///opt/lib/venv/lib/python3.11/site-packages/eventutilities_python/lib/eventutilities-flink-1.4.0-jar-with-dependencies.jar', 'file:///opt/lib/venv/lib/python3.11/site-packages/eventutilities_python/lib/flink-python-1.20.0-tests.jar', 'file:///opt/lib/venv/lib/python3.11/site-packages/eventutilities_python/lib/kafka-clients-3.4.0.jar']", "ecs.version": "1.2.0","process.thread.name":"flink-pekko.actor.default-dispatcher-13","log.logger":"org.apache.flink.client.python.PythonDriver"}
	at java.base/java.net.URL.<init>(URL.java:674)
JMonton-WMF moved this task from Next Up to In progress on the Data-Engineering (Q3 FY25/26 January 1st - March 31th) board.Mar 12 2026, 3:35 PM
JMonton-WMF added a comment.Mar 12 2026, 4:18 PM

Another interesting example, it looks like logs defined in Python are picked by Flink and reported as INFO, not matter if they are DEBUG, ERROR or anything else:

Debug as Info:

{"@timestamp":"2026-03-12T15:48:57.088Z","log.level": "INFO","message":"DEBUG:root:No need to resolve https://meta.wikimedia.org/w/api.php, it is already absolute", "ecs.version": "1.2.0","process.thread.name":"flink-pekko.actor.default-dispatcher-11","log.logger":"org.apache.flink.client.python.PythonDriver"}

Error as Info:

{"@timestamp":"2026-03-12T15:48:57.093Z","log.level": "INFO","message":"error: Option 'enrich.mediawiki_api_endpoint_template' is not accepted", "ecs.version": "1.2.0","process.thread.name":"flink-pekko.actor.default-dispatcher-11","log.logger":"org.apache.flink.client.python.PythonDriver"}
AKhatun_WMF subscribed.Mar 13 2026, 12:46 AM
JMonton-WMF created subtask T419997: Fix PyFlink log levels.Mar 13 2026, 2:19 PM
Ottomata renamed this task from Logs and Monitoring for the HTML pipeline to Audit and fix observability (logging and metrics) for pyflink jobs.Mar 16 2026, 1:08 PM
Ottomata mentioned this in T351225: Productionized Edit Types.
Ottomata moved this task from Backlog to Stream Processing on the Event-Platform board.Mar 16 2026, 2:09 PM
Ottomata added a subtask: T373112: [NEEDS INVESTIGATION][BUG] eventutilities_python operator metrics.Mar 16 2026, 2:13 PM
AKhatun_WMF added a comment.Tue, Mar 24, 1:10 AM

Wanted to bring up if we need to consider alerting for these pipelines and want to add it to this ticket. I am not sure what we have for other streaming apps. Defining and alerting on SLA misses, data loss etc. I assume this data would also be needed later for reconcile purposes.

Ottomata added a comment.Tue, Mar 24, 1:49 AM

Ya! We've done it once before!
https://wikitech.wikimedia.org/wiki/MediaWiki_Event_Enrichment/SLO/Mediawiki_Page_Content_Change_Enrichment

Ottomata added a comment.Tue, Mar 24, 1:50 AM

E.g. https://gerrit.wikimedia.org/r/plugins/gitiles/operations/alerts/+/refs/heads/master/team-data-engineering/mw-page-content-change-enrich.yaml

CodeReviewBot added a comment.Thu, Mar 26, 10:57 AM

javiermonton updated https://gitlab.wikimedia.org/repos/data-engineering/eventutilities-python/-/merge_requests/98

Warnings when exception caught

CodeReviewBot added a comment.Thu, Mar 26, 4:12 PM

javiermonton merged https://gitlab.wikimedia.org/repos/data-engineering/eventutilities-python/-/merge_requests/98

Warnings when exception caught

JMonton-WMF added a comment.Tue, Mar 31, 8:42 AM

There is a possible improvement here, not related to logs or monitoring, but I found it while exploring logs:

Config uses deprecated configuration key 'state.backend' instead of proper key 'state.backend.type'
JMonton-WMF added a comment.Tue, Mar 31, 8:44 AM

This warning I'm not sure if it appears because the application was restarting and the metric name already existed, or because we are conflicting with something else:

Name collision: Group already contains a Metric with the name 'pendingCommittables'. Metric will not be reported.[10.67.25.169, taskmanager, flink-app-production-taskmanager-1-4, mw-page-html-content-change-enrich, kafka__mediawiki.page_html_content_change.dev5: Committer, 1]
JMonton-WMF added a comment.EditedTue, Mar 31, 8:47 AM

More warnings to investigate:

The operator name process_error_enrich_with_parent_html, convert_python_to_pyflink__$null__stream.error exceeded the 80 characters length limit and was truncated.
The operator name kafka__mediawiki.page_change.v1__convert_flink_to_python, filter_remove_canary_events, _stream_key_by_map_operator exceeded the 80 characters length limit and was truncated.
Cannot create recoverable writer due to This s3 file system implementation does not support recoverable writers., will use the ordinary writer.
JMonton-WMF added a comment.Tue, Mar 31, 2:24 PM

Possible issues with metrics being reported:

Invalid type for Gauge org.apache.flink.runtime.checkpoint.CheckpointStatsTracker$LatestCompletedCheckpointExternalPathGauge@3230b364: java.lang.String, only number types and booleans are supported by this reporter.
│ {"@timestamp":"2026-03-31T13:48:18.638Z","log.level":"DEBUG","message":"Invalid type for Gauge org.apache.flink.connector.kafka.source.metrics.KafkaSourceReaderMetrics$$Lambda$1569/0x00007f3b0fedb530@1cae3576 │
│ : java.lang.String, only number types and booleans are supported by this reporter.", "ecs.version": "1.2.0","process.thread.name":"prometheus-http-1-2","log.logger":"org.apache.flink.metrics.prometheus.Promet │
│ heusReporter"}
Ottomata added a comment.Tue, Mar 31, 3:03 PM

Config uses deprecated configuration key 'state.backend' instead of proper key 'state.backend.type'

Is this possibly related to https://nightlies.apache.org/flink/flink-docs-stable/docs/deployment/config/#migrate-from-flink-confyaml-to-configyaml ? I don't recall where we are with that related to recent Flink upgrades.

Ahoelzl edited projects, added Data-Engineering (Q4 FS25/26 April 1st - June 30st); removed Data-Engineering (Q3 FY25/26 January 1st - March 31th).Fri, Apr 3, 10:24 PM
Ahoelzl moved this task from Next Up to In progress on the Data-Engineering (Q4 FS25/26 April 1st - June 30st) board.
CodeReviewBot added a project: Patch-For-Review.Thu, Apr 16, 5:44 PM

otto opened https://gitlab.wikimedia.org/repos/data-engineering/eventutilities-python/-/merge_requests/103

Add event process function batch_size_total metric

Content licensed under Creative Commons Attribution-ShareAlike (CC BY-SA) 4.0 unless otherwise noted; code licensed under GNU General Public License (GPL) 2.0 or later and other open source licenses. By using this site, you agree to the Terms of Use, Privacy Policy, and Code of Conduct. · Wikimedia Foundation · Privacy Policy · Code of Conduct · Terms of Use · Disclaimer · CC-BY-SA · GPL · Credits