[FLINK-40208] Add JobMdcRegistry for config-driven MDC enrichment #28855
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| --- | ||
| title: "Logging Context (MDC)" | ||
| weight: 7 | ||
| type: docs | ||
| --- | ||
| <!-- | ||
| Licensed to the Apache Software Foundation (ASF) under one | ||
| or more contributor license agreements. See the NOTICE file | ||
| distributed with this work for additional information | ||
| regarding copyright ownership. The ASF licenses this file | ||
| to you under the Apache License, Version 2.0 (the | ||
| "License"); you may not use this file except in compliance | ||
| with the License. You may obtain a copy of the License at | ||
|
|
||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
|
|
||
| Unless required by applicable law or agreed to in writing, | ||
| software distributed under the License is distributed on an | ||
| "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| KIND, either express or implied. See the License for the | ||
| specific language governing permissions and limitations | ||
| under the License. | ||
| --> | ||
|
|
||
| # Logging Context (MDC) | ||
|
|
||
| Flink populates the SLF4J [MDC](https://www.slf4j.org/api/org/slf4j/MDC.html) while handling jobs. Logging backends include MDC entries in log output, enabling log collectors to filter and group Flink logs by job without parsing message text. For details on rendering MDC entries with Log4j 2, see [Structured logging]({{< ref "docs/deployment/advanced/logging" >}}#structured-logging). | ||
|
|
||
| By default, the context holds a single entry: | ||
|
|
||
| | MDC key | Value | | ||
| |----------------|---------------------------------------------| | ||
| | `flink-job-id` | Job ID as a 32 character hexadecimal string | | ||
|
|
||
| Operators typically need more than a job ID to route logs, for example a tenant, a deployment name, or a pipeline name that persists across resubmissions. The `mdc.job-configuration-to-mdc-keys` option publishes job configuration entries to the MDC, so application code does not need to manage MDC entries directly. | ||
|
|
||
| ## Configuration | ||
|
|
||
| {{< generated/mdc_configuration >}} | ||
|
|
||
| The value maps a job configuration key to the MDC key it is published under. Flink resolves the mapping when the job is submitted or recovered on the JobManager, and when a TaskManager accepts a task of that job. A configuration key that is absent from the job configuration, or whose value is blank, is skipped. `flink-job-id` is always present, and a mapping that targets `flink-job-id` is ignored. | ||
|
|
||
| The lookup runs against the job configuration, which is the cluster configuration from `config.yaml` merged with job-level configuration supplied at submission time (for example, `-D` arguments to `flink run`). Any key can be referenced, including keys that are not Flink configuration options. | ||
|
|
||
| ## Example | ||
|
|
||
| Publish the pipeline name and an identifier that the operator injects at submission time: | ||
|
|
||
| ```yaml | ||
| mdc.job-configuration-to-mdc-keys: | ||
| pipeline.name: pipeline-name | ||
| my.company.tenant-id: tenant-id | ||
| ``` | ||
|
|
||
| ```bash | ||
| $ ./bin/flink run \ | ||
| -Dpipeline.name=nightly-aggregation \ | ||
| -Dmy.company.tenant-id=acme \ | ||
| ./examples/streaming/StateMachineExample.jar | ||
| ``` | ||
|
|
||
| Log records emitted for this job then carry three MDC entries: | ||
|
|
||
| ```text | ||
| flink-job-id = 4d1e3fbd4b1e4a4b8f9d0c6e2a7b5c31 | ||
| pipeline-name = nightly-aggregation | ||
| tenant-id = acme | ||
| ``` | ||
|
|
||
| JSON layouts that resolve the whole MDC pick the new fields up without further configuration. To include them in a plain text layout, extend the [Log4j 2 pattern]({{< ref "docs/deployment/advanced/logging" >}}#log4j-2-patternlayout), for example `[%X{flink-job-id}] [%X{tenant-id}] %c{0} %m%n`. | ||
|
|
||
| ## Scope and lifetime | ||
|
|
||
| The enriched context lives in a process-local registry. The JobManager populates it when the Dispatcher submits or recovers the job, and each TaskManager populates it when it accepts a task of that job. Entries are dropped when the job reaches a terminal state on the JobManager, and when a TaskManager releases the resources of the job. | ||
|
|
||
| Before a job's configuration reaches a process, log records contain only `flink-job-id`. Client-side records produced while the job graph is being built are not scoped to any job. | ||
|
|
||
| ## Notes | ||
|
|
||
| Values are read from the configuration that was submitted with the job. Changing `mdc.job-configuration-to-mdc-keys` or any mapped key while the job runs has no effect. Recovery reuses the stored job configuration, so cluster-level changes do not apply retroactively. Resubmit the job to pick up a new mapping. | ||
|
|
||
| Mapped values are written to log records as is. Do not map configuration keys that hold credentials or other secrets. | ||
|
|
||
| Every mapped key adds a field to every log record scoped to the job. Keep the mapping small to maintain predictable log volume and index cardinality. | ||
|
|
||
| {{< top >}} |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,86 @@ | ||
| --- | ||
| title: "Logging Context (MDC)" | ||
| weight: 7 | ||
| type: docs | ||
| --- | ||
| <!-- | ||
| Licensed to the Apache Software Foundation (ASF) under one | ||
| or more contributor license agreements. See the NOTICE file | ||
| distributed with this work for additional information | ||
| regarding copyright ownership. The ASF licenses this file | ||
| to you under the Apache License, Version 2.0 (the | ||
| "License"); you may not use this file except in compliance | ||
| with the License. You may obtain a copy of the License at | ||
|
|
||
| http://www.apache.org/licenses/LICENSE-2.0 | ||
|
|
||
| Unless required by applicable law or agreed to in writing, | ||
| software distributed under the License is distributed on an | ||
| "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| KIND, either express or implied. See the License for the | ||
| specific language governing permissions and limitations | ||
| under the License. | ||
| --> | ||
|
|
||
| # Logging Context (MDC) | ||
|
|
||
| Flink populates the SLF4J [MDC](https://www.slf4j.org/api/org/slf4j/MDC.html) while handling jobs. Logging backends include MDC entries in log output, enabling log collectors to filter and group Flink logs by job without parsing message text. For details on rendering MDC entries with Log4j 2, see [Structured logging]({{< ref "docs/deployment/advanced/logging" >}}#structured-logging). | ||
|
|
||
| By default, the context holds a single entry: | ||
|
|
||
| | MDC key | Value | | ||
| |----------------|---------------------------------------------| | ||
| | `flink-job-id` | Job ID as a 32 character hexadecimal string | | ||
|
|
||
| Operators typically need more than a job ID to route logs, for example a tenant, a deployment name, or a pipeline name that persists across resubmissions. The `mdc.job-configuration-to-mdc-keys` option publishes job configuration entries to the MDC, so application code does not need to manage MDC entries directly. | ||
|
|
||
| ## Configuration | ||
|
|
||
| {{< generated/mdc_configuration >}} | ||
|
|
||
| The value maps a job configuration key to the MDC key it is published under. Flink resolves the mapping when the job is submitted or recovered on the JobManager, and when a TaskManager accepts a task of that job. A configuration key that is absent from the job configuration, or whose value is blank, is skipped. `flink-job-id` is always present, and a mapping that targets `flink-job-id` is ignored. | ||
|
|
||
| The lookup runs against the job configuration, which is the cluster configuration from `config.yaml` merged with job-level configuration supplied at submission time (for example, `-D` arguments to `flink run`). Any key can be referenced, including keys that are not Flink configuration options. | ||
|
|
||
| ## Example | ||
|
|
||
| Publish the pipeline name and an identifier that the operator injects at submission time: | ||
|
|
||
| ```yaml | ||
| mdc.job-configuration-to-mdc-keys: | ||
| pipeline.name: pipeline-name | ||
| my.company.tenant-id: tenant-id | ||
| ``` | ||
|
|
||
| ```bash | ||
| $ ./bin/flink run \ | ||
| -Dpipeline.name=nightly-aggregation \ | ||
| -Dmy.company.tenant-id=acme \ | ||
| ./examples/streaming/StateMachineExample.jar | ||
| ``` | ||
|
|
||
| Log records emitted for this job then carry three MDC entries: | ||
|
|
||
| ```text | ||
| flink-job-id = 4d1e3fbd4b1e4a4b8f9d0c6e2a7b5c31 | ||
| pipeline-name = nightly-aggregation | ||
| tenant-id = acme | ||
| ``` | ||
|
|
||
| JSON layouts that resolve the whole MDC pick the new fields up without further configuration. To include them in a plain text layout, extend the [Log4j 2 pattern]({{< ref "docs/deployment/advanced/logging" >}}#log4j-2-patternlayout), for example `[%X{flink-job-id}] [%X{tenant-id}] %c{0} %m%n`. | ||
|
|
||
| ## Scope and lifetime | ||
|
|
||
| The enriched context lives in a process-local registry. The JobManager populates it when the Dispatcher submits or recovers the job, and each TaskManager populates it when it accepts a task of that job. Entries are dropped when the job reaches a terminal state on the JobManager, and when a TaskManager releases the resources of the job. | ||
|
|
||
| Before a job's configuration reaches a process, log records contain only `flink-job-id`. Client-side records produced while the job graph is being built are not scoped to any job. | ||
|
|
||
| ## Notes | ||
|
|
||
| Values are read from the configuration that was submitted with the job. Changing `mdc.job-configuration-to-mdc-keys` or any mapped key while the job runs has no effect. Recovery reuses the stored job configuration, so cluster-level changes do not apply retroactively. Resubmit the job to pick up a new mapping. | ||
|
|
||
| Mapped values are written to log records as is. Do not map configuration keys that hold credentials or other secrets. | ||
|
|
||
| Every mapped key adds a field to every log record scoped to the job. Keep the mapping small to maintain predictable log volume and index cardinality. | ||
|
|
||
| {{< top >}} | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| <table class="configuration table table-bordered"> | ||
| <thead> | ||
| <tr> | ||
| <th class="text-left" style="width: 20%">Key</th> | ||
| <th class="text-left" style="width: 15%">Default</th> | ||
| <th class="text-left" style="width: 10%">Type</th> | ||
| <th class="text-left" style="width: 55%">Description</th> | ||
| </tr> | ||
| </thead> | ||
| <tbody> | ||
| <tr> | ||
| <td><h5>mdc.job-configuration-to-mdc-keys</h5></td> | ||
| <td style="word-wrap: break-word;"></td> | ||
| <td>Map</td> | ||
| <td>Maps job configuration keys to MDC key names. At job start, each listed configuration key is looked up; if the value is present and non-blank it is emitted into MDC under the mapped name. Keys absent or blank in the job configuration are skipped. The job ID is always added to MDC under the key 'flink-job-id' regardless of this setting. Example: 'pipeline.name:pipeline-name' maps the job configuration key 'pipeline.name' to the MDC key 'pipeline-name'.</td> | ||
| </tr> | ||
| </tbody> | ||
| </table> |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.flink.configuration; | ||
|
|
||
| import org.apache.flink.annotation.PublicEvolving; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.Map; | ||
|
|
||
| import static org.apache.flink.configuration.ConfigOptions.key; | ||
|
|
||
| /** Configuration options for MDC (Mapped Diagnostic Context) enrichment. */ | ||
| @PublicEvolving | ||
| public final class MdcOptions { | ||
|
|
||
| /** | ||
| * Maps job configuration keys to MDC key names. Keys absent or blank in the job configuration | ||
| * are skipped. | ||
| */ | ||
| @PublicEvolving | ||
| public static final ConfigOption<Map<String, String>> JOB_CONFIGURATION_TO_MDC_KEYS = | ||
| key("mdc.job-configuration-to-mdc-keys") | ||
| .mapType() | ||
| .defaultValue(Collections.emptyMap()) | ||
| .withDescription( | ||
| "Maps job configuration keys to MDC key names. " | ||
| + "At job start, each listed configuration key is looked up; " | ||
| + "if the value is present and non-blank it is emitted into MDC under the mapped name. " | ||
| + "Keys absent or blank in the job configuration are skipped. " | ||
| + "The job ID is always added to MDC under the key 'flink-job-id' regardless of this setting. " | ||
| + "Example: 'pipeline.name:pipeline-name' maps the job configuration key " | ||
| + "'pipeline.name' to the MDC key 'pipeline-name'."); | ||
|
|
||
| private MdcOptions() {} | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.flink.util; | ||
|
|
||
| import org.apache.flink.annotation.Internal; | ||
| import org.apache.flink.annotation.VisibleForTesting; | ||
| import org.apache.flink.api.common.JobID; | ||
| import org.apache.flink.configuration.Configuration; | ||
|
|
||
| import javax.annotation.Nullable; | ||
| import javax.annotation.concurrent.ThreadSafe; | ||
|
|
||
| import java.util.Map; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| /** | ||
| * Process-wide registry mapping {@link JobID} to enriched MDC context, populated where the job | ||
| * {@link Configuration} is available and consulted by {@link MdcUtils#asContextData(JobID)}. | ||
| */ | ||
| @Internal | ||
| @ThreadSafe | ||
| public final class JobMdcRegistry { | ||
|
|
||
| private static final Map<JobID, Map<String, String>> REGISTRY = new ConcurrentHashMap<>(); | ||
|
|
||
| private JobMdcRegistry() {} | ||
|
|
||
| /** | ||
| * Registers enriched MDC context if the configuration carries any MDC key mappings; clears any | ||
| * stale entry otherwise. Equivalent to {@link #unregister} when the config is unenriched. | ||
| */ | ||
| public static void registerOrClear(final JobID jobID, final Configuration jobConfiguration) { | ||
| final Map<String, String> context = MdcUtils.asContextData(jobID, jobConfiguration); | ||
| if (context.size() > 1) { | ||
| REGISTRY.put(jobID, context); | ||
| } else { | ||
| unregister(jobID); | ||
| } | ||
|
Comment on lines
+48
to
+54
There was a problem hiding this comment. [Suggestion] No diagnostic logging at any level when MDC enrichment is configured but resolves to zero entries, making misconfiguration invisible at runtime. — Failure scenario: an operator configures Consider adding a debug-level log that reports the configured mapping size and the resolved entry count, e.g.: if (LOG.isDebugEnabled()) {
LOG.debug("MDC enrichment for job {}: {} configured key(s), {} resolved entry(ies).",
jobID, mdcKeyMapping.size(), context.size() - 1);
}— qwen3.8-max-preview via Qwen Code /review (v0.21.2)
Contributor
Author
There was a problem hiding this comment. Debug logs are normally not enabled in prod. Presence of the field is a signal on its own to confirm whether the feature is working. More realistic scenario is an alert setup for "missing" field and oncall engineer checking what has changed that led to field disappearing (and end to end tests, to catch such regressions before prod) |
||
| } | ||
|
|
||
| /** Remove the registered context for the job. */ | ||
| public static void unregister(final JobID jobID) { | ||
| REGISTRY.remove(jobID); | ||
| } | ||
|
|
||
| /** | ||
| * Return the registered context for the job, or {@code null} if none. The returned map is | ||
| * unmodifiable. | ||
| */ | ||
| @Nullable | ||
| public static Map<String, String> lookup(final JobID jobID) { | ||
| return REGISTRY.get(jobID); | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| public static void clear() { | ||
| REGISTRY.clear(); | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
[Suggestion] Hugo
weight: 7collides with siblingdocs/content/docs/ops/events.md(alsoweight: 7), making their nav ordering non-deterministic. — Concrete cost: Hugo falls back to alphabetical tie-breaking between "Events" and "Logging Context (MDC)", so the sidebar position becomes an accident of the title rather than an editorial choice. The same collision exists indocs/content.zh/docs/ops/logging_context.md.— qwen3.8-max-preview via Qwen Code /review (v0.21.2)
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This looks minor, but I might include it if I do any more cleanup