You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "rkhachatryan (via GitHub)" <gi...@apache.org> on 2024/02/08 19:17:40 UTC
[PR] [FLINK-34417] Log Job ID via MDC [flink]
rkhachatryan opened a new pull request, #24292:
URL: https://github.com/apache/flink/pull/24292
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
Add Flink JobID to logging MDC to make analysing the logs easier.
## Verifying this change
- added `JobIdLoggingMainThreadExecutor`
- tested manually
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / no)
- The serializers: (yes / no / don't know)
- The runtime per-record code paths (performance sensitive): (yes / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
- The S3 file system connector: (yes / no / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1513207066
##########
flink-tests/src/test/resources/log4j2-test.properties:
##########
@@ -28,7 +28,7 @@ appender.testlogger.name = TestLogger
appender.testlogger.type = CONSOLE
appender.testlogger.target = SYSTEM_ERR
appender.testlogger.layout.type = PatternLayout
-appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
+appender.testlogger.layout.pattern = [%-32X{flink-job-id}] %c{0} [%t] %-5p %m%n
Review Comment:
do we need this for the test to work? I'd expect that the MDC is populated regardless of what the pattern says, but that may very well not be the case.
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,231 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.testutils.logging.LoggerAuditingExtension;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.core.LogEvent;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.slf4j.event.Level.DEBUG;
+
+/**
+ * Tests adding of {@link JobID} to logs (via {@link org.slf4j.MDC}) in the most important cases.
+ */
+public class JobIDLoggingITCase extends TestLogger {
+ private static final Logger logger = LoggerFactory.getLogger(JobIDLoggingITCase.class);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension checkpointCoordinatorLogging =
+ new LoggerAuditingExtension(CheckpointCoordinator.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension streamTaskLogging =
+ new LoggerAuditingExtension(StreamTask.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension taskExecutorLogging =
+ new LoggerAuditingExtension(TaskExecutor.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension taskLogging =
+ new LoggerAuditingExtension(Task.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension executionGraphLogging =
+ new LoggerAuditingExtension(ExecutionGraph.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension jobMasterLogging =
+ new LoggerAuditingExtension(JobMaster.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension asyncCheckpointRunnableLogging =
+ // this class is private
+ new LoggerAuditingExtension(
+ "org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable", DEBUG);
+
+ @RegisterExtension
+ public static MiniClusterExtension miniClusterResource =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ @Test
+ public void testJobIDLogging(@InjectClusterClient ClusterClient<?> clusterClient)
+ throws Exception {
+ JobID jobID = null;
+ try {
+ jobID = runJob(clusterClient);
+ } finally {
+ if (jobID != null) {
+ clusterClient.cancel(jobID).get();
+ }
+ }
+
+ // NOTE: most of the assertions are empirical, such as
+ // - which classes are important
+ // - how many messages to expect
+ // - which log patterns to ignore
+
+ assertJobIDPresent(jobID, 3, checkpointCoordinatorLogging);
+ assertJobIDPresent(jobID, 6, streamTaskLogging);
+ assertJobIDPresent(
+ jobID,
+ 9,
+ taskExecutorLogging,
+ "Successful registration.*",
+ "Establish JobManager connection.*",
+ "Offer reserved slots.*",
+ ".*ResourceManager.*",
+ "Operator event.*");
+
+ assertJobIDPresent(jobID, 10, taskLogging);
+ assertJobIDPresent(jobID, 10, executionGraphLogging);
+ assertJobIDPresent(
+ jobID,
+ 15,
+ jobMasterLogging,
+ "Registration at ResourceManager.*",
+ "Registration with ResourceManager.*",
+ "Resolved ResourceManager address.*");
+ assertJobIDPresent(jobID, 1, asyncCheckpointRunnableLogging);
+ }
+
+ private static void assertJobIDPresent(
+ JobID jobID,
+ int expectedLogMessages,
+ LoggerAuditingExtension ext,
+ String... ignPatterns) {
+ String loggerName = ext.getLoggerName();
+ checkState(
+ ext.getEvents().size() >= expectedLogMessages,
+ "Too few log events recorded for %s (%s) - this must be a bug in the test code",
+ loggerName,
+ ext.getEvents().size());
+
+ final List<LogEvent> eventsWithMissingJobId = new ArrayList<>();
+ final List<LogEvent> eventsWithWrongJobId = new ArrayList<>();
+ final List<LogEvent> ignoredEvents = new ArrayList<>();
+ final List<Pattern> ignorePatterns =
+ Arrays.stream(ignPatterns).map(Pattern::compile).collect(Collectors.toList());
+
+ for (LogEvent e : ext.getEvents()) {
+ if (e.getContextData().containsKey(MdcUtils.JOB_ID)) {
+ if (!Objects.equals(
+ e.getContextData().getValue(MdcUtils.JOB_ID), jobID.toHexString())) {
+ eventsWithWrongJobId.add(e);
+ }
+ } else if (matchesAny(ignorePatterns, e.getMessage().getFormattedMessage())) {
+ ignoredEvents.add(e);
+ } else {
+ eventsWithMissingJobId.add(e);
+ }
+ }
+ logger.debug(
+ "checked events for {}:\n {};\n ignored: {},\n wrong job id: {},\n missing job id: {}",
+ loggerName,
+ ext.getEvents(),
+ ignoredEvents,
+ eventsWithWrongJobId,
+ eventsWithMissingJobId);
+ assertTrue(
+ eventsWithWrongJobId.isEmpty(),
+ "events with a wrong Job ID recorded for "
+ + loggerName
+ + ": "
+ + eventsWithWrongJobId);
Review Comment:
nit: I think this should be equivalent 🤔
```suggestion
assertThat(eventsWithWrongJobId)
.as("events with a wrong Job ID)
.isEmpty();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on PR #24292:
URL: https://github.com/apache/flink/pull/24292#issuecomment-1936234814
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24292:
URL: https://github.com/apache/flink/pull/24292#issuecomment-1934796687
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "605d23d07cc51916ac746b9760a400dd363c0044",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "605d23d07cc51916ac746b9760a400dd363c0044",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 605d23d07cc51916ac746b9760a400dd363c0044 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1491395312
##########
flink-core/src/main/java/org/apache/flink/util/JobIdLoggingExecutorService.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.api.common.JobID;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static org.apache.flink.util.MdcUtils.wrapCallable;
+import static org.apache.flink.util.MdcUtils.wrapRunnable;
+
+class JobIdLoggingExecutorService<T extends ExecutorService> extends JobIdLoggingExecutor<T>
Review Comment:
Should be resolved by 8bba339.
##########
flink-core/src/main/java/org/apache/flink/util/JobIdLoggingScheduledExecutorService.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.api.common.JobID;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.flink.util.MdcUtils.wrapCallable;
+import static org.apache.flink.util.MdcUtils.wrapRunnable;
+
+// todo: tests
+class JobIdLoggingScheduledExecutorService
Review Comment:
Should be resolved by 8bba339.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on PR #24292:
URL: https://github.com/apache/flink/pull/24292#issuecomment-1936944954
The failure is unrelated: FLINK-34420
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1513218720
##########
flink-tests/src/test/resources/log4j2-test.properties:
##########
@@ -28,7 +28,7 @@ appender.testlogger.name = TestLogger
appender.testlogger.type = CONSOLE
appender.testlogger.target = SYSTEM_ERR
appender.testlogger.layout.type = PatternLayout
-appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
+appender.testlogger.layout.pattern = [%-32X{flink-job-id}] %c{0} [%t] %-5p %m%n
Review Comment:
No, I just wanted to add job ID to log messages in tests.
Just `%x` doesn't work, because it stands for Nested DC, while this PR uses Mapped DC.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1486799769
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java:
##########
@@ -1266,15 +1282,21 @@ public CompletableFuture<Acknowledge> freeSlot(
@Override
public void freeInactiveSlots(JobID jobId, Time timeout) {
- log.debug("Freeing inactive slots for job {}.", jobId);
+ MdcUtils.addJobID(jobId);
Review Comment:
It wouldn't work for all cases because sometimes we need to extract job ID from the argument or TM state.
(And also that would be much less readable IMO)
edit: it also removes the logger after the execution completely which is not what we want for non-test loggers
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1485904612
##########
flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcGateway.java:
##########
@@ -34,4 +34,16 @@ public interface RpcGateway {
* @return Fully qualified hostname under which the associated rpc endpoint is reachable
*/
String getHostname();
+
+ /**
+ * Perform optional steps before handling any messages or performing service actions (e.g.
+ * start/stop).
+ */
+ default void beforeInvocation() {}
Review Comment:
Not a fan of these because it means we couple components more. I'd rather see us add some pre/post-runnable args to wherever the MDC is set up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1486776195
##########
docs/content.zh/docs/deployment/advanced/logging.md:
##########
@@ -40,6 +40,21 @@ Flink 中的日志记录是使用 [SLF4J](http://www.slf4j.org/) 日志接口实
<a name="configuring-log4j-2"></a>
+### Structured logging
+
+Flink adds the following fields to [MDC](https://www.slf4j.org/api/org/slf4j/MDC.html) of most of the relevant log messages (experimental feature):
+- Job ID
+ - key: `flink-job-id`
Review Comment:
My reasoning was that users configure the mappings anyways, so making it configurable isn't strictly necessary.
OTH, adding more configuration to Flink makes it more complex; so chose to have it static.
I can add the configuration though if you think it's necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on PR #24292:
URL: https://github.com/apache/flink/pull/24292#issuecomment-1985585478
Re-ran benchmarks - no regressions found:
[jmh-result-runtime.csv](https://github.com/apache/flink/files/14537980/jmh-result-runtime.csv)
[jmh-result-scheduler.csv](https://github.com/apache/flink/files/14537982/jmh-result-scheduler.csv)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1489259793
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.apache.logging.log4j.test.appender.ListAppender;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests adding of {@link JobID} to logs (via {@link org.slf4j.MDC}) in the most important cases.
+ */
+public class JobIDLoggingITCase extends TestLogger {
+ private static final Logger logger = LoggerFactory.getLogger(JobIDLoggingITCase.class);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ @Test
+ public void testJobIDLogging() throws Exception {
+ LoggerContext ctx = LoggerContext.getContext(false);
Review Comment:
You can use multiple extensions to capture multiple loggers..
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.apache.logging.log4j.test.appender.ListAppender;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests adding of {@link JobID} to logs (via {@link org.slf4j.MDC}) in the most important cases.
+ */
+public class JobIDLoggingITCase extends TestLogger {
+ private static final Logger logger = LoggerFactory.getLogger(JobIDLoggingITCase.class);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ @Test
+ public void testJobIDLogging() throws Exception {
+ LoggerContext ctx = LoggerContext.getContext(false);
Review Comment:
You can use multiple extensions to capture multiple loggers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1491016717
##########
flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcGateway.java:
##########
@@ -34,4 +34,16 @@ public interface RpcGateway {
* @return Fully qualified hostname under which the associated rpc endpoint is reachable
*/
String getHostname();
+
+ /**
+ * Perform optional steps before handling any messages or performing service actions (e.g.
+ * start/stop).
+ */
+ default void beforeInvocation() {}
Review Comment:
Yeah, I'd rather keep it as an explicit method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1513218720
##########
flink-tests/src/test/resources/log4j2-test.properties:
##########
@@ -28,7 +28,7 @@ appender.testlogger.name = TestLogger
appender.testlogger.type = CONSOLE
appender.testlogger.target = SYSTEM_ERR
appender.testlogger.layout.type = PatternLayout
-appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
+appender.testlogger.layout.pattern = [%-32X{flink-job-id}] %c{0} [%t] %-5p %m%n
Review Comment:
No, I just wanted to add job ID to log messages in tests.
Just `%x` doesn't work, because it stands for Nested DC, while this PR uses Mapped DC (which is `%X`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on PR #24292:
URL: https://github.com/apache/flink/pull/24292#issuecomment-1979884146
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1485886746
##########
flink-core/src/main/java/org/apache/flink/util/JobIdLoggingExecutorService.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.api.common.JobID;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static org.apache.flink.util.MdcUtils.wrapCallable;
+import static org.apache.flink.util.MdcUtils.wrapRunnable;
+
+class JobIdLoggingExecutorService<T extends ExecutorService> extends JobIdLoggingExecutor<T>
Review Comment:
same comment as for JobIdLoggingExecutor
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1486799769
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java:
##########
@@ -1266,15 +1282,21 @@ public CompletableFuture<Acknowledge> freeSlot(
@Override
public void freeInactiveSlots(JobID jobId, Time timeout) {
- log.debug("Freeing inactive slots for job {}.", jobId);
+ MdcUtils.addJobID(jobId);
Review Comment:
It wouldn't work for all cases because sometimes we need to extract job ID from the argument or TM state.
(And also that would be much less readable IMO)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on PR #24292:
URL: https://github.com/apache/flink/pull/24292#issuecomment-1935958746
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on PR #24292:
URL: https://github.com/apache/flink/pull/24292#issuecomment-1936575490
@flinkbot run azure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "XComp (via GitHub)" <gi...@apache.org>.
XComp commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1760917678
##########
flink-tests/src/test/resources/log4j2-test.properties:
##########
@@ -28,7 +28,7 @@ appender.testlogger.name = TestLogger
appender.testlogger.type = CONSOLE
appender.testlogger.target = SYSTEM_ERR
appender.testlogger.layout.type = PatternLayout
-appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n
+appender.testlogger.layout.pattern = [%-32X{flink-job-id}] %c{0} [%t] %-5p %m%n
Review Comment:
@rkhachatryan just to clarify: Removing the timestamp (`%-4r`) was done unintentional, correct? I will add it back in a hotfix commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1491387438
##########
flink-core/src/main/java/org/apache/flink/util/JobIdLoggingExecutor.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.api.common.JobID;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+class JobIdLoggingExecutor<T extends Executor> implements Executor {
+ protected final JobID jobID;
Review Comment:
Done in 8bba339.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1485884759
##########
docs/content.zh/docs/deployment/advanced/logging.md:
##########
@@ -40,6 +40,21 @@ Flink 中的日志记录是使用 [SLF4J](http://www.slf4j.org/) 日志接口实
<a name="configuring-log4j-2"></a>
+### Structured logging
+
+Flink adds the following fields to [MDC](https://www.slf4j.org/api/org/slf4j/MDC.html) of most of the relevant log messages (experimental feature):
+- Job ID
+ - key: `flink-job-id`
Review Comment:
Can we make the key configurable? Users may have different requirements; like wanting to follow ECS when using the json layout where the appropriate key would be `flink.job.id`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1485888366
##########
flink-core/src/main/java/org/apache/flink/util/MdcUtils.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.api.common.JobID;
+
+import org.slf4j.MDC;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+// todo: tests
+
+/** Utility class to manage common Flink attributes in {@link MDC} (only {@link JobID} ATM). */
+public class MdcUtils {
+
+ public static final String JOB_ID = "flink-job-id";
+
+ /** Add the given {@link JobID} to {@link MDC}. */
+ public static void addJobID(JobID jobID) {
+ MDC.put(JOB_ID, jobID.toHexString());
+ }
+
+ /** Remove any {@link JobID} from {@link MDC}. */
+ public static void removeJobID() {
+ // todo: restore original jid if present
Review Comment:
This is an important one because it may have performance-ramifications. (especially if we go with the standard approach of copying the old MDC)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1485910818
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.apache.logging.log4j.test.appender.ListAppender;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests adding of {@link JobID} to logs (via {@link org.slf4j.MDC}) in the most important cases.
+ */
+public class JobIDLoggingITCase extends TestLogger {
+ private static final Logger logger = LoggerFactory.getLogger(JobIDLoggingITCase.class);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ @Test
+ public void testJobIDLogging() throws Exception {
+ LoggerContext ctx = LoggerContext.getContext(false);
Review Comment:
please use the `LoggerAuditingExtension` instead and extend it if necessary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1489259263
##########
flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcGateway.java:
##########
@@ -34,4 +34,16 @@ public interface RpcGateway {
* @return Fully qualified hostname under which the associated rpc endpoint is reachable
*/
String getHostname();
+
+ /**
+ * Perform optional steps before handling any messages or performing service actions (e.g.
+ * start/stop).
+ */
+ default void beforeInvocation() {}
Review Comment:
yes, (if this is handled in the PekkoRpcActor). Although maybe not runnables but just a Map<String, String>? Not sure if we _really_ want to open us up to running arbitrary things before/after just yet 🤔
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1489262074
##########
flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/MainThreadValidatorUtil.java:
##########
@@ -45,11 +45,13 @@ public MainThreadValidatorUtil(RpcEndpoint endpoint) {
public void enterMainThread() {
assert (endpoint.currentMainThread.compareAndSet(null, Thread.currentThread()))
: "The RpcEndpoint has concurrent access from " + endpoint.currentMainThread.get();
+ endpoint.beforeInvocation();
Review Comment:
Yes, because the main thread executable covers less; it doesnt capture anything sent directly against the rpc service's executor.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1514881575
##########
flink-core/src/main/java/org/apache/flink/util/MdcUtils.java:
##########
@@ -0,0 +1,125 @@
+/*
+ * 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.api.common.JobID;
+
+import org.slf4j.MDC;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+// todo: tests
Review Comment:
Done: e2e8b8d :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1512543707
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java:
##########
@@ -1266,15 +1282,21 @@ public CompletableFuture<Acknowledge> freeSlot(
@Override
public void freeInactiveSlots(JobID jobId, Time timeout) {
- log.debug("Freeing inactive slots for job {}.", jobId);
+ MdcUtils.addJobID(jobId);
Review Comment:
yes, it was more of an idea on what we could do in the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1514408118
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:
##########
@@ -303,7 +308,8 @@ public void onUnknownDeploymentsOf(
this.highAvailabilityServices = checkNotNull(highAvailabilityService);
this.blobWriter = jobManagerSharedServices.getBlobWriter();
this.futureExecutor = jobManagerSharedServices.getFutureExecutor();
Review Comment:
No, I just overlooked it :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1514408450
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java:
##########
@@ -1662,6 +1667,7 @@ private class TaskCanceler implements Runnable {
@Override
public void run() {
try {
Review Comment:
Replaced all instances.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1489255700
##########
docs/content.zh/docs/deployment/advanced/logging.md:
##########
@@ -40,6 +40,21 @@ Flink 中的日志记录是使用 [SLF4J](http://www.slf4j.org/) 日志接口实
<a name="configuring-log4j-2"></a>
+### Structured logging
+
+Flink adds the following fields to [MDC](https://www.slf4j.org/api/org/slf4j/MDC.html) of most of the relevant log messages (experimental feature):
+- Job ID
+ - key: `flink-job-id`
Review Comment:
With something like this?
```
appender.console.json.eventTemplateAdditionalField[0].type = EventTemplateAdditionalField
appender.console.json.eventTemplateAdditionalField[0].key = flink.job.id
appender.console.json.eventTemplateAdditionalField[0].value = {"$resolver": "mdc", "key": "flink-job-id"}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1485908014
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java:
##########
@@ -1266,15 +1282,21 @@ public CompletableFuture<Acknowledge> freeSlot(
@Override
public void freeInactiveSlots(JobID jobId, Time timeout) {
- log.debug("Freeing inactive slots for job {}.", jobId);
+ MdcUtils.addJobID(jobId);
Review Comment:
Theoretically you could implement this part in a generic way in the InvocationHandler by checking if any of the method arguments is a job ID.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1486809430
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.apache.logging.log4j.test.appender.ListAppender;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests adding of {@link JobID} to logs (via {@link org.slf4j.MDC}) in the most important cases.
+ */
+public class JobIDLoggingITCase extends TestLogger {
+ private static final Logger logger = LoggerFactory.getLogger(JobIDLoggingITCase.class);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ @Test
+ public void testJobIDLogging() throws Exception {
+ LoggerContext ctx = LoggerContext.getContext(false);
Review Comment:
The extension is tailored for a single-logger case, while in this test, there are many loggers and a single run.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1485887114
##########
flink-core/src/main/java/org/apache/flink/util/JobIdLoggingScheduledExecutorService.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.api.common.JobID;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.flink.util.MdcUtils.wrapCallable;
+import static org.apache.flink.util.MdcUtils.wrapRunnable;
+
+// todo: tests
+class JobIdLoggingScheduledExecutorService
Review Comment:
same comment as for JobIdLoggingExecutor
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1486789947
##########
flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/MainThreadValidatorUtil.java:
##########
@@ -45,11 +45,13 @@ public MainThreadValidatorUtil(RpcEndpoint endpoint) {
public void enterMainThread() {
assert (endpoint.currentMainThread.compareAndSet(null, Thread.currentThread()))
: "The RpcEndpoint has concurrent access from " + endpoint.currentMainThread.get();
+ endpoint.beforeInvocation();
Review Comment:
When I remove it, I get log lines like this without a job ID:
```
[ ] ExecutionGraph [flink-pekko.actor.default-dispatcher-6] INFO Source: Custom Source -> Timestamps/Watermarks -> transform-1-forward -> Sink: Unnamed (1/4) (2dae9a492649d761d992295551f74d3c_cbc357ccb763df2852fee8c4fc7d55f2_0_0) switched from SCHEDULED to DEPLOYING.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1489259793
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.apache.logging.log4j.test.appender.ListAppender;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests adding of {@link JobID} to logs (via {@link org.slf4j.MDC}) in the most important cases.
+ */
+public class JobIDLoggingITCase extends TestLogger {
+ private static final Logger logger = LoggerFactory.getLogger(JobIDLoggingITCase.class);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ @Test
+ public void testJobIDLogging() throws Exception {
+ LoggerContext ctx = LoggerContext.getContext(false);
Review Comment:
You can use multiple extensions to capture multiple loggers, see `YARNSessionCapacitySchedulerITCase`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1514170487
##########
flink-tests/pom.xml:
##########
@@ -44,7 +44,14 @@ under the License.
</properties>
<dependencies>
-
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <version>2.17.1</version>
+ <scope>test</scope>
+ <type>test-jar</type>
Review Comment:
I guess that's not needed anymore after switching to Junit extension, I'll remove this dependency.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1514074926
##########
flink-core/src/main/java/org/apache/flink/util/MdcAwareScheduledExecutorService.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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 java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.flink.util.MdcUtils.wrapCallable;
+import static org.apache.flink.util.MdcUtils.wrapRunnable;
+
+// todo: tests
Review Comment:
🙈
##########
flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/concurrent/ComponentMainThreadExecutor.java:
##########
@@ -30,7 +32,7 @@
* Interface for an executor that runs tasks in the main thread of an {@link
* org.apache.flink.runtime.rpc.RpcEndpoint}.
*/
-public interface ComponentMainThreadExecutor extends ScheduledExecutor {
+public interface ComponentMainThreadExecutor extends ScheduledExecutor, Closeable {
Review Comment:
We may not need this change anymore
##########
flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java:
##########
@@ -440,15 +470,16 @@ public void validateRunsInMainThread() {
* @return true if all the resources are closed, otherwise false
*/
boolean validateResourceClosed() {
- return mainThreadExecutor.validateScheduledExecutorClosed() && resourceRegistry.isClosed();
+ return ((MainThreadExecutor) mainThreadExecutor).validateScheduledExecutorClosed()
Review Comment:
this maybe can be reverted
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java:
##########
@@ -1662,6 +1667,7 @@ private class TaskCanceler implements Runnable {
@Override
public void run() {
try {
Review Comment:
This could be a good candidate for MdcUtils#withContext or `MDC.putCloseable()`
(more instances in this file)
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.testutils.logging.LoggerAuditingExtension;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.core.LogEvent;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.slf4j.event.Level.DEBUG;
+
+/**
+ * Tests adding of {@link JobID} to logs (via {@link org.slf4j.MDC}) in the most important cases.
+ */
+public class JobIDLoggingITCase extends TestLogger {
Review Comment:
```suggestion
public class JobIDLoggingITCase {
```
This is both out-dated (JUnit 4 tech) and no longer required in flink-tests.
##########
flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcEndpoint.java:
##########
@@ -134,18 +139,30 @@ public abstract class RpcEndpoint implements RpcGateway, AutoCloseableAsync {
* @param rpcService The RPC server that dispatches calls to this RPC endpoint.
* @param endpointId Unique identifier for this endpoint
*/
- protected RpcEndpoint(final RpcService rpcService, final String endpointId) {
+ protected RpcEndpoint(
+ RpcService rpcService, String endpointId, Map<String, String> loggingContext) {
this.rpcService = checkNotNull(rpcService, "rpcService");
this.endpointId = checkNotNull(endpointId, "endpointId");
- this.rpcServer = rpcService.startServer(this);
+ this.rpcServer = rpcService.startServer(this, loggingContext);
this.resourceRegistry = new CloseableRegistry();
this.mainThreadExecutor =
new MainThreadExecutor(rpcServer, this::validateRunsInMainThread, endpointId);
+
Review Comment:
revert
##########
flink-core/src/main/java/org/apache/flink/util/MdcUtils.java:
##########
@@ -0,0 +1,125 @@
+/*
+ * 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.api.common.JobID;
+
+import org.slf4j.MDC;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+// todo: tests
Review Comment:
🙈
##########
flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java:
##########
@@ -303,7 +308,8 @@ public void onUnknownDeploymentsOf(
this.highAvailabilityServices = checkNotNull(highAvailabilityService);
this.blobWriter = jobManagerSharedServices.getBlobWriter();
this.futureExecutor = jobManagerSharedServices.getFutureExecutor();
Review Comment:
Is there a reason we dont wrap this one as well?
##########
flink-tests/pom.xml:
##########
@@ -44,7 +44,14 @@ under the License.
</properties>
<dependencies>
-
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <version>2.17.1</version>
+ <scope>test</scope>
+ <type>test-jar</type>
Review Comment:
What do we need from the test jar?
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,227 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.testutils.logging.LoggerAuditingExtension;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.core.LogEvent;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.slf4j.event.Level.DEBUG;
+
+/**
+ * Tests adding of {@link JobID} to logs (via {@link org.slf4j.MDC}) in the most important cases.
+ */
+public class JobIDLoggingITCase extends TestLogger {
+ private static final Logger logger = LoggerFactory.getLogger(JobIDLoggingITCase.class);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension checkpointCoordinatorLogging =
+ new LoggerAuditingExtension(CheckpointCoordinator.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension streamTaskLogging =
+ new LoggerAuditingExtension(StreamTask.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension taskExecutorLogging =
+ new LoggerAuditingExtension(TaskExecutor.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension taskLogging =
+ new LoggerAuditingExtension(Task.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension executionGraphLogging =
+ new LoggerAuditingExtension(ExecutionGraph.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension jobMasterLogging =
+ new LoggerAuditingExtension(JobMaster.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension asyncCheckpointRunnableLogging =
+ // this class is private
+ new LoggerAuditingExtension(
+ "org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable", DEBUG);
+
+ @RegisterExtension
+ public static MiniClusterExtension miniClusterResource =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ @Test
+ public void testJobIDLogging(@InjectClusterClient ClusterClient<?> clusterClient)
+ throws Exception {
+ JobID jobID = null;
+ try {
+ jobID = runJob(clusterClient);
+ } finally {
+ if (jobID != null) {
+ clusterClient.cancel(jobID).get();
+ }
+ }
Review Comment:
```suggestion
JobID jobID = runJob(clusterClient);
clusterClient.cancel(jobID).get();
```
Maybe I'm missing something but the finally branch will never do anything in an exceptional case, because then the jobID couldn't have been assigned.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1513528600
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java:
##########
@@ -903,22 +907,28 @@ private Stream<ResultPartitionDeploymentDescriptor> filterPartitionsRequiringRel
public CompletableFuture<Acknowledge> cancelTask(
ExecutionAttemptID executionAttemptID, Time timeout) {
final Task task = taskSlotTable.getTask(executionAttemptID);
+ MdcUtils.addJobID(task.getJobID());
Review Comment:
this can fail if the task is null.
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.testutils.logging.LoggerAuditingExtension;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.core.LogEvent;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.slf4j.event.Level.DEBUG;
+
+/**
+ * Tests adding of {@link JobID} to logs (via {@link org.slf4j.MDC}) in the most important cases.
+ */
+public class JobIDLoggingITCase extends TestLogger {
+ private static final Logger logger = LoggerFactory.getLogger(JobIDLoggingITCase.class);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension checkpointCoordinatorLogging =
+ new LoggerAuditingExtension(CheckpointCoordinator.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension streamTaskLogging =
+ new LoggerAuditingExtension(StreamTask.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension taskExecutorLogging =
+ new LoggerAuditingExtension(TaskExecutor.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension taskLogging =
+ new LoggerAuditingExtension(Task.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension executionGraphLogging =
+ new LoggerAuditingExtension(ExecutionGraph.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension jobMasterLogging =
+ new LoggerAuditingExtension(JobMaster.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension asyncCheckpointRunnableLogging =
+ // this class is private
+ new LoggerAuditingExtension(
+ "org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable", DEBUG);
+
+ @RegisterExtension
+ public static MiniClusterExtension miniClusterResource =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ @Test
+ public void testJobIDLogging(@InjectClusterClient ClusterClient<?> clusterClient)
+ throws Exception {
+ JobID jobID = null;
+ try {
+ jobID = runJob(clusterClient);
+ } finally {
+ if (jobID != null) {
+ clusterClient.cancel(jobID).get();
+ }
+ }
+
+ // NOTE: most of the assertions are empirical, such as
+ // - which classes are important
+ // - how many messages to expect
+ // - which log patterns to ignore
Review Comment:
This test seems a bit unstable.
When I run the test locally it reliably fails because one message doesn't have a job ID, but it's also not supposed to have any according to the current code afaict.
```
Un-registering task and sending final execution state CANCELED to JobManager for
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java:
##########
@@ -1039,28 +1049,32 @@ public CompletableFuture<Acknowledge> triggerCheckpoint(
long checkpointId,
long checkpointTimestamp,
CheckpointOptions checkpointOptions) {
- log.debug(
- "Trigger checkpoint {}@{} for {}.",
- checkpointId,
- checkpointTimestamp,
- executionAttemptID);
-
final Task task = taskSlotTable.getTask(executionAttemptID);
+ MdcUtils.addJobID(task.getJobID());
Review Comment:
this can fail if the task is null.
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.testutils.logging.LoggerAuditingExtension;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.core.LogEvent;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertTrue;
Review Comment:
```suggestion
import static org.assertj.core.api.Assertions.assertThat;
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java:
##########
@@ -1070,31 +1084,36 @@ public CompletableFuture<Acknowledge> confirmCheckpoint(
long completedCheckpointId,
long completedCheckpointTimestamp,
long lastSubsumedCheckpointId) {
+ final Task task = taskSlotTable.getTask(executionAttemptID);
+ MdcUtils.addJobID(task.getJobID());
Review Comment:
this can fail if the task is null.
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,228 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.common.time.Deadline;
+import org.apache.flink.client.program.ClusterClient;
+import org.apache.flink.core.execution.CheckpointType;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskexecutor.TaskExecutor;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.junit5.InjectClusterClient;
+import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.testutils.logging.LoggerAuditingExtension;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.core.LogEvent;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertTrue;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.slf4j.event.Level.DEBUG;
+
+/**
+ * Tests adding of {@link JobID} to logs (via {@link org.slf4j.MDC}) in the most important cases.
+ */
+public class JobIDLoggingITCase extends TestLogger {
+ private static final Logger logger = LoggerFactory.getLogger(JobIDLoggingITCase.class);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension checkpointCoordinatorLogging =
+ new LoggerAuditingExtension(CheckpointCoordinator.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension streamTaskLogging =
+ new LoggerAuditingExtension(StreamTask.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension taskExecutorLogging =
+ new LoggerAuditingExtension(TaskExecutor.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension taskLogging =
+ new LoggerAuditingExtension(Task.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension executionGraphLogging =
+ new LoggerAuditingExtension(ExecutionGraph.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension jobMasterLogging =
+ new LoggerAuditingExtension(JobMaster.class, DEBUG);
+
+ @RegisterExtension
+ public final LoggerAuditingExtension asyncCheckpointRunnableLogging =
+ // this class is private
+ new LoggerAuditingExtension(
+ "org.apache.flink.streaming.runtime.tasks.AsyncCheckpointRunnable", DEBUG);
+
+ @RegisterExtension
+ public static MiniClusterExtension miniClusterResource =
+ new MiniClusterExtension(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ @Test
+ public void testJobIDLogging(@InjectClusterClient ClusterClient<?> clusterClient)
+ throws Exception {
+ JobID jobID = null;
+ try {
+ jobID = runJob(clusterClient);
+ } finally {
+ if (jobID != null) {
+ clusterClient.cancel(jobID).get();
+ }
+ }
+
+ // NOTE: most of the assertions are empirical, such as
+ // - which classes are important
+ // - how many messages to expect
+ // - which log patterns to ignore
+
+ assertJobIDPresent(jobID, 3, checkpointCoordinatorLogging);
+ assertJobIDPresent(jobID, 6, streamTaskLogging);
+ assertJobIDPresent(
+ jobID,
+ 9,
+ taskExecutorLogging,
+ "Successful registration.*",
+ "Establish JobManager connection.*",
+ "Offer reserved slots.*",
+ ".*ResourceManager.*",
+ "Operator event.*");
+
+ assertJobIDPresent(jobID, 10, taskLogging);
+ assertJobIDPresent(jobID, 10, executionGraphLogging);
+ assertJobIDPresent(
+ jobID,
+ 15,
+ jobMasterLogging,
+ "Registration at ResourceManager.*",
+ "Registration with ResourceManager.*",
+ "Resolved ResourceManager address.*");
+ assertJobIDPresent(jobID, 1, asyncCheckpointRunnableLogging);
+ }
+
+ private static void assertJobIDPresent(
+ JobID jobID,
+ int expectedLogMessages,
+ LoggerAuditingExtension ext,
+ String... ignPatterns) {
+ String loggerName = ext.getLoggerName();
+ checkState(
+ ext.getEvents().size() >= expectedLogMessages,
+ "Too few log events recorded for %s (%s) - this must be a bug in the test code",
+ loggerName,
+ ext.getEvents().size());
+
+ final List<LogEvent> eventsWithMissingJobId = new ArrayList<>();
+ final List<LogEvent> eventsWithWrongJobId = new ArrayList<>();
+ final List<LogEvent> ignoredEvents = new ArrayList<>();
+ final List<Pattern> ignorePatterns =
+ Arrays.stream(ignPatterns).map(Pattern::compile).collect(Collectors.toList());
+
+ for (LogEvent e : ext.getEvents()) {
+ if (e.getContextData().containsKey(MdcUtils.JOB_ID)) {
+ if (!Objects.equals(
+ e.getContextData().getValue(MdcUtils.JOB_ID), jobID.toHexString())) {
+ eventsWithWrongJobId.add(e);
+ }
+ } else if (matchesAny(ignorePatterns, e.getMessage().getFormattedMessage())) {
+ ignoredEvents.add(e);
+ } else {
+ eventsWithMissingJobId.add(e);
+ }
+ }
+ logger.debug(
+ "checked events for {}:\n {};\n ignored: {},\n wrong job id: {},\n missing job id: {}",
+ loggerName,
+ ext.getEvents(),
+ ignoredEvents,
+ eventsWithWrongJobId,
+ eventsWithMissingJobId);
+ assertThat(eventsWithWrongJobId)
+ .as("events with a wrong Job ID)
Review Comment:
```suggestion
.as("events with a wrong Job ID")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan merged PR #24292:
URL: https://github.com/apache/flink/pull/24292
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1514899954
##########
flink-core/src/main/java/org/apache/flink/util/MdcAwareScheduledExecutorService.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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 java.util.Map;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.flink.util.MdcUtils.wrapCallable;
+import static org.apache.flink.util.MdcUtils.wrapRunnable;
+
+// todo: tests
Review Comment:
Done: e2e8b8d3cecd22fd80d47a8b11607f307ed02ae4 :)
(I think same MdcUtilsTest is sufficient)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1485902184
##########
flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/MainThreadValidatorUtil.java:
##########
@@ -45,11 +45,13 @@ public MainThreadValidatorUtil(RpcEndpoint endpoint) {
public void enterMainThread() {
assert (endpoint.currentMainThread.compareAndSet(null, Thread.currentThread()))
: "The RpcEndpoint has concurrent access from " + endpoint.currentMainThread.get();
+ endpoint.beforeInvocation();
Review Comment:
We don't need both this and the main thread executor. All the main thread executor does is throw things into the rpc actor.
I also don't like this being coupled into the main thread validation logic; and would rather see this being handled in the PekkoRpcActor or encoded into runnables within the PekkoInvocationHandler.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1485888366
##########
flink-core/src/main/java/org/apache/flink/util/MdcUtils.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.api.common.JobID;
+
+import org.slf4j.MDC;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+// todo: tests
+
+/** Utility class to manage common Flink attributes in {@link MDC} (only {@link JobID} ATM). */
+public class MdcUtils {
+
+ public static final String JOB_ID = "flink-job-id";
+
+ /** Add the given {@link JobID} to {@link MDC}. */
+ public static void addJobID(JobID jobID) {
+ MDC.put(JOB_ID, jobID.toHexString());
+ }
+
+ /** Remove any {@link JobID} from {@link MDC}. */
+ public static void removeJobID() {
+ // todo: restore original jid if present
Review Comment:
This is an important one, especially because it has performance-ramifications.
##########
flink-core/src/main/java/org/apache/flink/util/MdcUtils.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.api.common.JobID;
+
+import org.slf4j.MDC;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+
+// todo: tests
+
+/** Utility class to manage common Flink attributes in {@link MDC} (only {@link JobID} ATM). */
+public class MdcUtils {
+
+ public static final String JOB_ID = "flink-job-id";
+
+ /** Add the given {@link JobID} to {@link MDC}. */
+ public static void addJobID(JobID jobID) {
+ MDC.put(JOB_ID, jobID.toHexString());
+ }
+
+ /** Remove any {@link JobID} from {@link MDC}. */
+ public static void removeJobID() {
+ // todo: restore original jid if present
Review Comment:
This is an important one because it has performance-ramifications.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1485886510
##########
flink-core/src/main/java/org/apache/flink/util/JobIdLoggingExecutor.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.api.common.JobID;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+class JobIdLoggingExecutor<T extends Executor> implements Executor {
+ protected final JobID jobID;
Review Comment:
This seems strikingly purpose-built; can't we generalize it to support additional properties in the future?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1486782437
##########
flink-core/src/main/java/org/apache/flink/util/JobIdLoggingExecutor.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.api.common.JobID;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+class JobIdLoggingExecutor<T extends Executor> implements Executor {
+ protected final JobID jobID;
Review Comment:
Yes, it can be generalised, but what would be the use case?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1486795898
##########
flink-rpc/flink-rpc-core/src/main/java/org/apache/flink/runtime/rpc/RpcGateway.java:
##########
@@ -34,4 +34,16 @@ public interface RpcGateway {
* @return Fully qualified hostname under which the associated rpc endpoint is reachable
*/
String getHostname();
+
+ /**
+ * Perform optional steps before handling any messages or performing service actions (e.g.
+ * start/stop).
+ */
+ default void beforeInvocation() {}
Review Comment:
Do you mean passing these Runnables to `actor.Props.create()`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "zentol (via GitHub)" <gi...@apache.org>.
zentol commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1489257016
##########
flink-core/src/main/java/org/apache/flink/util/JobIdLoggingExecutor.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * 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.api.common.JobID;
+
+import java.util.concurrent.Executor;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+class JobIdLoggingExecutor<T extends Executor> implements Executor {
+ protected final JobID jobID;
Review Comment:
cluster id, job name, task id, task name, ...
There's a lot of stuff we could put in the MDC that'd be useful.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1491013640
##########
docs/content.zh/docs/deployment/advanced/logging.md:
##########
@@ -40,6 +40,21 @@ Flink 中的日志记录是使用 [SLF4J](http://www.slf4j.org/) 日志接口实
<a name="configuring-log4j-2"></a>
+### Structured logging
+
+Flink adds the following fields to [MDC](https://www.slf4j.org/api/org/slf4j/MDC.html) of most of the relevant log messages (experimental feature):
+- Job ID
+ - key: `flink-job-id`
Review Comment:
That should already be doable - without changes in code, no?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-34417] Log Job ID via MDC [flink]
Posted by "rkhachatryan (via GitHub)" <gi...@apache.org>.
rkhachatryan commented on code in PR #24292:
URL: https://github.com/apache/flink/pull/24292#discussion_r1491387018
##########
flink-tests/src/test/java/org/apache/flink/test/misc/JobIDLoggingITCase.java:
##########
@@ -0,0 +1,218 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.runtime.checkpoint.CheckpointCoordinator;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.jobmaster.JobMaster;
+import org.apache.flink.runtime.taskmanager.Task;
+import org.apache.flink.runtime.testutils.CommonTestUtils;
+import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.util.MdcUtils;
+import org.apache.flink.util.TestLogger;
+
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.config.LoggerConfig;
+import org.apache.logging.log4j.test.appender.ListAppender;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests adding of {@link JobID} to logs (via {@link org.slf4j.MDC}) in the most important cases.
+ */
+public class JobIDLoggingITCase extends TestLogger {
+ private static final Logger logger = LoggerFactory.getLogger(JobIDLoggingITCase.class);
+
+ @ClassRule
+ public static MiniClusterWithClientResource miniClusterResource =
+ new MiniClusterWithClientResource(
+ new MiniClusterResourceConfiguration.Builder()
+ .setNumberTaskManagers(1)
+ .setNumberSlotsPerTaskManager(1)
+ .build());
+
+ @Test
+ public void testJobIDLogging() throws Exception {
+ LoggerContext ctx = LoggerContext.getContext(false);
Review Comment:
Done in 7c7a83c.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org