diff --git a/docs/layouts/shortcodes/generated/history_server_configuration.html b/docs/layouts/shortcodes/generated/history_server_configuration.html index ef521d5f999b58..c83079272c3d03 100644 --- a/docs/layouts/shortcodes/generated/history_server_configuration.html +++ b/docs/layouts/shortcodes/generated/history_server_configuration.html @@ -116,5 +116,11 @@ String Local directory that is used by the history server REST API for temporary files. + +
historyserver.yarn.log.enable-custom-url-handlers
+ (none) + Boolean + Enable custom URL handlers for JobManager and TaskManager logs. The HistoryServer will generate a URL based off of these custom TaskManagerLogUrlHandler and JobManagerLogUrlHandler all custom logic should be handled in the above classes. When this configuration is set, the historyserver.log.jobmanager.url-pattern and historyserver.log.taskmanager.url-pattern configurations will be ignored. + diff --git a/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java index 05de6dee6ba37a..c7da3e14fe0432 100644 --- a/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java +++ b/flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java @@ -75,6 +75,18 @@ public class HistoryServerOptions { + " with replacing the special placeholders, ``, to the id of job." + " Only http / https schemes are supported."); + public static final ConfigOption + HISTORY_SERVER_JOBMANAGER_TASKMANAGER_LOG_ENABLE_CUSTOM_HANDLERS = + key("historyserver.yarn.log.enable-custom-url-handlers") + .booleanType() + .noDefaultValue() + .withDescription( + "Enable custom URL handlers for JobManager and TaskManager logs. The HistoryServer will generate a URL based" + + " off of these custom TaskManagerLogUrlHandler and JobManagerLogUrlHandler" + + " all custom logic should be handled in the above classes. When this configuration is set, the" + + " historyserver.log.jobmanager.url-pattern and historyserver.log.taskmanager.url-pattern configurations" + + " will be ignored."); + /** The local directory used by the HistoryServer web-frontend. */ public static final ConfigOption HISTORY_SERVER_WEB_DIR = key("historyserver.web.tmpdir") diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerLogUrlHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerLogUrlHandler.java new file mode 100644 index 00000000000000..300280f3a516af --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerLogUrlHandler.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.runtime.rest.handler.cluster; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.rest.handler.AbstractRestHandler; +import org.apache.flink.runtime.rest.handler.HandlerRequest; +import org.apache.flink.runtime.rest.handler.RestHandlerException; +import org.apache.flink.runtime.rest.messages.EmptyRequestBody; +import org.apache.flink.runtime.rest.messages.JobIDPathParameter; +import org.apache.flink.runtime.rest.messages.JobManagerLogUrlHeaders; +import org.apache.flink.runtime.rest.messages.JobMessageParameters; +import org.apache.flink.runtime.rest.messages.LogUrlResponse; +import org.apache.flink.runtime.rest.messages.MessageHeaders; +import org.apache.flink.runtime.rest.messages.ResponseBody; +import org.apache.flink.runtime.rest.util.EnvironmentInfoUtils; +import org.apache.flink.runtime.scheduler.ExecutionGraphInfo; +import org.apache.flink.runtime.webmonitor.RestfulGateway; +import org.apache.flink.runtime.webmonitor.history.ArchivedJson; +import org.apache.flink.runtime.webmonitor.history.JsonArchivist; +import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever; + +import javax.annotation.Nonnull; + +import java.io.IOException; +import java.time.Duration; +import java.util.Collection; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +/** Request handler for retrieving the job manager log url. */ +public class JobManagerLogUrlHandler + extends AbstractRestHandler< + RestfulGateway, EmptyRequestBody, LogUrlResponse, JobMessageParameters> + implements JsonArchivist { + public static final String JOB_MANAGER_LOG_URL_FORMAT = "http://%s:%s/node/containerlogs/%s/%s"; + + private final Configuration config; + + public JobManagerLogUrlHandler( + GatewayRetriever leaderRetriever, + Duration timeout, + Map responseHeaders, + MessageHeaders messageHeaders, + Configuration configuration) { + super(leaderRetriever, timeout, responseHeaders, messageHeaders); + this.config = configuration; + } + + @Override + protected CompletableFuture handleRequest( + @Nonnull HandlerRequest request, @Nonnull RestfulGateway gateway) + throws RestHandlerException { + final EnvironmentInfoUtils.EnvironmentContext environmentContext = + EnvironmentInfoUtils.getEnvironmentContext(); + return CompletableFuture.completedFuture(createJobManagerURL(environmentContext)); + } + + @Override + public Collection archiveJsonWithPath(ExecutionGraphInfo executionGraphInfo) + throws IOException { + final EnvironmentInfoUtils.EnvironmentContext environmentContext = + EnvironmentInfoUtils.getEnvironmentContext(); + ResponseBody json = createJobManagerURL(environmentContext); + String path = + JobManagerLogUrlHeaders.getInstance() + .getTargetRestEndpointURL() + .replace( + ':' + JobIDPathParameter.KEY, + executionGraphInfo.getJobId().toString()); + return Collections.singletonList(new ArchivedJson(path, json)); + } + + /** + * Generates a URL that will link to the location of the job manager logs. The format of the URL + * will be: CONTAINER-NM-HOST:CONTAINER-NM-HTTP-PORT/node/containerlogs/CONTAINER-ID/USER/ + */ + @VisibleForTesting + public LogUrlResponse createJobManagerURL( + EnvironmentInfoUtils.EnvironmentContext environmentContext) { + return new LogUrlResponse( + String.format( + JOB_MANAGER_LOG_URL_FORMAT, + environmentContext.nodeManagerHostName, + environmentContext.nodeManagerHttpPort, + environmentContext.containerId, + environmentContext.user)); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerLogUrlHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerLogUrlHandler.java new file mode 100644 index 00000000000000..ff7aab3b5be87f --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerLogUrlHandler.java @@ -0,0 +1,244 @@ +/* + * 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.runtime.rest.handler.taskmanager; + +import org.apache.flink.annotation.VisibleForTesting; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.clusterframework.types.ResourceID; +import org.apache.flink.runtime.executiongraph.AccessExecution; +import org.apache.flink.runtime.executiongraph.AccessExecutionGraph; +import org.apache.flink.runtime.executiongraph.AccessExecutionJobVertex; +import org.apache.flink.runtime.executiongraph.AccessExecutionVertex; +import org.apache.flink.runtime.executiongraph.ArchivedExecution; +import org.apache.flink.runtime.executiongraph.ExecutionHistory; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.rest.handler.HandlerRequest; +import org.apache.flink.runtime.rest.handler.RestHandlerException; +import org.apache.flink.runtime.rest.handler.job.AbstractAccessExecutionGraphHandler; +import org.apache.flink.runtime.rest.handler.legacy.ExecutionGraphCache; +import org.apache.flink.runtime.rest.messages.EmptyRequestBody; +import org.apache.flink.runtime.rest.messages.JobIDPathParameter; +import org.apache.flink.runtime.rest.messages.JobTaskManagerMessageParameters; +import org.apache.flink.runtime.rest.messages.LogUrlResponse; +import org.apache.flink.runtime.rest.messages.MessageHeaders; +import org.apache.flink.runtime.rest.messages.ResponseBody; +import org.apache.flink.runtime.rest.messages.TaskManagerLogUrlHeaders; +import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerIdPathParameter; +import org.apache.flink.runtime.rest.util.EnvironmentInfoUtils; +import org.apache.flink.runtime.scheduler.ExecutionGraphInfo; +import org.apache.flink.runtime.taskmanager.TaskManagerLocation; +import org.apache.flink.runtime.webmonitor.RestfulGateway; +import org.apache.flink.runtime.webmonitor.history.ArchivedJson; +import org.apache.flink.runtime.webmonitor.history.JsonArchivist; +import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever; + +import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nonnull; + +import java.io.IOException; +import java.time.Duration; +import java.util.Collection; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Executor; + +/** Request handler for retrieving the task manager log url. */ +public class TaskManagerLogUrlHandler + extends AbstractAccessExecutionGraphHandler + implements JsonArchivist { + + private static final Logger LOG = LoggerFactory.getLogger(TaskManagerLogUrlHandler.class); + public static final String TASK_MANAGER_LOG_URL_FORMAT = + "http://%s:%s/node/containerlogs/%s/%s"; + + private final Configuration config; + + public TaskManagerLogUrlHandler( + GatewayRetriever leaderRetriever, + Duration timeout, + Map responseHeaders, + MessageHeaders + messageHeaders, + ExecutionGraphCache executionGraphCache, + Executor executor, + Configuration configuration) { + super( + leaderRetriever, + timeout, + responseHeaders, + messageHeaders, + executionGraphCache, + executor); + this.config = configuration; + } + + /** + * When the endpoint is hit for a live application, it may not return a log-url if the container + * has not yet been allocated. + */ + @Override + protected LogUrlResponse handleRequest( + @Nonnull HandlerRequest request, AccessExecutionGraph executionGraph) + throws RestHandlerException { + EnvironmentInfoUtils.EnvironmentContext environmentContext = + EnvironmentInfoUtils.getEnvironmentContext(); + ResourceID containerId = request.getPathParameter(TaskManagerIdPathParameter.class); + Map jobVertices = + executionGraph.getAllVertices(); + + for (AccessExecutionJobVertex jobVertex : jobVertices.values()) { + for (AccessExecutionVertex task : jobVertex.getTaskVertices()) { + + TaskManagerLocation currentLocation = task.getCurrentAssignedResourceLocation(); + + if (currentLocation != null + && currentLocation.getResourceID().equals(containerId)) { + return createTaskManagerUrl( + environmentContext, + containerId.toString(), + currentLocation.getFQDNHostname()); + } else { + ExecutionHistory executionHistory = task.getExecutionHistory(); + for (ArchivedExecution execution : executionHistory.getHistoricalExecutions()) { + TaskManagerLocation location = execution.getAssignedResourceLocation(); + if (location != null && location.getResourceID().equals(containerId)) { + return createTaskManagerUrl( + environmentContext, + containerId.toString(), + location.getFQDNHostname()); + } + } + } + } + } + + // If we are unable to find the hostname of the task manager for the container in the + // execution graph, throw an error. + throw new RestHandlerException( + "Unable to find hostname for containerId: " + + containerId + + " and job id: " + + executionGraph.getJobID(), + HttpResponseStatus.NOT_FOUND); + } + + /** + * Generates a URL that will link to the location of the task manager logs. The format of the + * URL will be: + * Job-History-base-URL/CONTAINER-NM-HOST:CONTAINER-NM-PORT/CONTAINER-ID/CONTAINER-ID/USER/ + */ + @Override + public Collection archiveJsonWithPath(ExecutionGraphInfo executionGraphInfo) + throws IOException { + EnvironmentInfoUtils.EnvironmentContext environmentContext = + EnvironmentInfoUtils.getEnvironmentContext(); + Collection vertices = + executionGraphInfo.getArchivedExecutionGraph().getAllVertices().values(); + Set archive = new HashSet<>(); + for (AccessExecutionJobVertex jobVertex : vertices) { + for (AccessExecutionVertex task : jobVertex.getTaskVertices()) { + Set vertexArchivedJson = + processAllTaskManagerExecutionsPerVertex( + executionGraphInfo.getJobId().toString(), task, environmentContext); + archive.addAll(vertexArchivedJson); + } + } + return archive; + } + + private Set processAllTaskManagerExecutionsPerVertex( + String jobId, + AccessExecutionVertex vertex, + EnvironmentInfoUtils.EnvironmentContext environmentContext) + throws IOException { + Set archivedJsons = new HashSet<>(); + + // The current execution is not included in the execution history so it is + // fetched separately + AccessExecution currentExecution = vertex.getCurrentExecutionAttempt(); + ArchivedJson currentJson = null; + if (isValidExecutionResourceLocation(currentExecution)) { + currentJson = + buildTaskManagerJson( + currentExecution.getAssignedResourceLocation(), + jobId, + environmentContext); + archivedJsons.add(currentJson); + } + ExecutionHistory executionHistory = vertex.getExecutionHistory(); + for (ArchivedExecution execution : executionHistory.getHistoricalExecutions()) { + // In case the task manager location is not available, skip the execution + if (!isValidExecutionResourceLocation(execution)) { + LOG.warn( + "Archived execution with jobId: {} has no assigned resource location, skipping building json", + jobId); + continue; + } + TaskManagerLocation location = execution.getAssignedResourceLocation(); + ArchivedJson json = buildTaskManagerJson(location, jobId, environmentContext); + archivedJsons.add(json); + } + return archivedJsons; + } + + private boolean isValidExecutionResourceLocation(AccessExecution execution) { + return execution != null + && execution.getAssignedResourceLocation() != null + && execution.getAssignedResourceLocation().getResourceID() != null; + } + + private ArchivedJson buildTaskManagerJson( + TaskManagerLocation location, + String jobId, + EnvironmentInfoUtils.EnvironmentContext environmentContext) + throws IOException { + String containerId = location.getResourceID().toString(); + String path = + TaskManagerLogUrlHeaders.getInstance() + .getTargetRestEndpointURL() + .replace(':' + JobIDPathParameter.KEY, jobId) + .replace(':' + TaskManagerIdPathParameter.KEY, containerId); + ResponseBody taskManagerUrl = + createTaskManagerUrl(environmentContext, containerId, location.getFQDNHostname()); + return new ArchivedJson(path, taskManagerUrl); + } + + /** + * Generates a URL that will link to the location of the task manager logs. The format of the + * URL will be: CONTAINER-NM-HOST:CONTAINER-NM-HTTP-PORT/node/containerlogs/CONTAINER-ID/USER/ + */ + @VisibleForTesting + public LogUrlResponse createTaskManagerUrl( + EnvironmentInfoUtils.EnvironmentContext environmentContext, + String containerId, + String nodeManagerHttpHostname) { + return new LogUrlResponse( + String.format( + TASK_MANAGER_LOG_URL_FORMAT, + nodeManagerHttpHostname, + environmentContext.nodeManagerHttpPort, + containerId, + environmentContext.user)); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/util/EnvironmentInfoUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/util/EnvironmentInfoUtils.java new file mode 100644 index 00000000000000..1c51b5a92c6ac8 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/util/EnvironmentInfoUtils.java @@ -0,0 +1,79 @@ +/* + * 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.runtime.rest.util; + +/** + * Helper utility class to retrieve the YARN container environment context. This relies on the + * standard environment variables that YARN's NodeManager exports into every launched container, so + * it intentionally avoids adding a dependency on the YARN client libraries. + */ +public class EnvironmentInfoUtils { + + private static final String ENV_CONTAINER_ID = "CONTAINER_ID"; + private static final String ENV_NM_HOST = "NM_HOST"; + private static final String ENV_NM_HTTP_PORT = "NM_HTTP_PORT"; + private static final String ENV_USER = "USER"; + + /** Class that holds the application environment context. */ + public static class EnvironmentContext { + + public final String containerId; + + public final String nodeManagerHostName; + + public final String nodeManagerHttpPort; + + public final String user; + + public EnvironmentContext( + String containerId, + String nodeManagerHostName, + String nodeManagerHttpPort, + String user) { + this.containerId = containerId; + this.nodeManagerHostName = nodeManagerHostName; + this.nodeManagerHttpPort = nodeManagerHttpPort; + this.user = user; + } + } + + public static EnvironmentContext getEnvironmentContext() { + return new EnvironmentContext( + getContainerId(), + getNodeManagerHostName(), + getNodeManagerHttpPort(), + getUserInfo()); + } + + private static String getContainerId() { + return System.getenv(ENV_CONTAINER_ID); + } + + private static String getNodeManagerHostName() { + return System.getenv(ENV_NM_HOST); + } + + private static String getNodeManagerHttpPort() { + return System.getenv(ENV_NM_HTTP_PORT); + } + + private static String getUserInfo() { + return System.getenv(ENV_USER); + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java index 16cf06d4dd9666..2f920f42730cbc 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java @@ -22,6 +22,7 @@ import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.CheckpointingOptions; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.HistoryServerOptions; import org.apache.flink.configuration.RestOptions; import org.apache.flink.configuration.RpcOptions; import org.apache.flink.configuration.SecurityOptions; @@ -47,6 +48,7 @@ import org.apache.flink.runtime.rest.handler.cluster.JobManagerEnvironmentHandler; import org.apache.flink.runtime.rest.handler.cluster.JobManagerLogFileHandler; import org.apache.flink.runtime.rest.handler.cluster.JobManagerLogListHandler; +import org.apache.flink.runtime.rest.handler.cluster.JobManagerLogUrlHandler; import org.apache.flink.runtime.rest.handler.cluster.JobManagerProfilingFileHandler; import org.apache.flink.runtime.rest.handler.cluster.JobManagerProfilingHandler; import org.apache.flink.runtime.rest.handler.cluster.JobManagerProfilingListHandler; @@ -113,6 +115,7 @@ import org.apache.flink.runtime.rest.handler.taskmanager.TaskManagerDetailsHandler; import org.apache.flink.runtime.rest.handler.taskmanager.TaskManagerLogFileHandler; import org.apache.flink.runtime.rest.handler.taskmanager.TaskManagerLogListHandler; +import org.apache.flink.runtime.rest.handler.taskmanager.TaskManagerLogUrlHandler; import org.apache.flink.runtime.rest.handler.taskmanager.TaskManagerProfilingFileHandler; import org.apache.flink.runtime.rest.handler.taskmanager.TaskManagerProfilingHandler; import org.apache.flink.runtime.rest.handler.taskmanager.TaskManagerProfilingListHandler; @@ -728,14 +731,6 @@ protected List> initiali executor, metricFetcher); - final GeneratedLogUrlHandler jobManagerLogUrlHandler = - new GeneratedLogUrlHandler( - localAddressFuture.thenApply(url -> url + "/#/job-manager/logs")); - - final GeneratedLogUrlHandler taskManagerLogUrlHandler = - new GeneratedLogUrlHandler( - localAddressFuture.thenApply(url -> url + "/#/task-manager//logs")); - final SavepointDisposalHandlers savepointDisposalHandlers = new SavepointDisposalHandlers(asyncOperationStoreDuration); @@ -928,8 +923,52 @@ protected List> initiali Tuple2.of( jobManagerJobConfigurationHandler.getMessageHeaders(), jobManagerJobConfigurationHandler)); - handlers.add(Tuple2.of(JobManagerLogUrlHeaders.getInstance(), jobManagerLogUrlHandler)); - handlers.add(Tuple2.of(TaskManagerLogUrlHeaders.getInstance(), taskManagerLogUrlHandler)); + + if (clusterConfiguration.get( + HistoryServerOptions + .HISTORY_SERVER_JOBMANAGER_TASKMANAGER_LOG_ENABLE_CUSTOM_HANDLERS, + false)) { + JobManagerLogUrlHandler jobManagerLogUrlHandler = + new JobManagerLogUrlHandler( + leaderRetriever, + timeout, + responseHeaders, + JobManagerLogUrlHeaders.getInstance(), + clusterConfiguration); + TaskManagerLogUrlHandler taskManagerLogUrlHandler = + new TaskManagerLogUrlHandler( + leaderRetriever, + timeout, + responseHeaders, + TaskManagerLogUrlHeaders.getInstance(), + executionGraphCache, + executor, + clusterConfiguration); + handlers.add( + Tuple2.of( + jobManagerLogUrlHandler.getMessageHeaders(), jobManagerLogUrlHandler)); + handlers.add( + Tuple2.of( + taskManagerLogUrlHandler.getMessageHeaders(), + taskManagerLogUrlHandler)); + } else { + final GeneratedLogUrlHandler jobManagerGeneratedLogUrlHandler = + new GeneratedLogUrlHandler( + localAddressFuture.thenApply(url -> url + "/#/job-manager/logs")); + + final GeneratedLogUrlHandler taskManagerGeneratedLogUrlHandler = + new GeneratedLogUrlHandler( + localAddressFuture.thenApply( + url -> url + "/#/task-manager//logs")); + handlers.add( + Tuple2.of( + JobManagerLogUrlHeaders.getInstance(), + jobManagerGeneratedLogUrlHandler)); + handlers.add( + Tuple2.of( + TaskManagerLogUrlHeaders.getInstance(), + taskManagerGeneratedLogUrlHandler)); + } final AbstractRestHandler jobVertexFlameGraphHandler; if (clusterConfiguration.get(RestOptions.ENABLE_FLAMEGRAPH)) { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerLogUrlHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerLogUrlHandlerTest.java new file mode 100644 index 00000000000000..9701c8953dce62 --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/cluster/JobManagerLogUrlHandlerTest.java @@ -0,0 +1,85 @@ +/* + * 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.runtime.rest.handler.cluster; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.HistoryServerOptions; +import org.apache.flink.runtime.rest.messages.JobManagerLogUrlHeaders; +import org.apache.flink.runtime.rest.messages.LogUrlResponse; +import org.apache.flink.runtime.rest.util.EnvironmentInfoUtils; +import org.apache.flink.testutils.TestingUtils; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.Collections; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** Test for the {@link org.apache.flink.runtime.rest.handler.cluster.JobManagerLogUrlHandler}. */ +public class JobManagerLogUrlHandlerTest { + + private JobManagerLogUrlHandler testInstance; + + private static final String CONTAINER_ID = "container_abc"; + + private static final String NM_HOST = "foo.bar"; + + private static final String NM_PORT = "0000"; + + private static final String USER = "user"; + + private Configuration configuration; + + @BeforeEach + void setup() { + configuration = new Configuration(); + configuration.set( + HistoryServerOptions + .HISTORY_SERVER_JOBMANAGER_TASKMANAGER_LOG_ENABLE_CUSTOM_HANDLERS, + true); + + testInstance = + new JobManagerLogUrlHandler( + () -> null, + TestingUtils.TIMEOUT, + Collections.emptyMap(), + JobManagerLogUrlHeaders.getInstance(), + this.configuration); + } + + @Test + public void testGenerateJobManagerLogUrl() { + EnvironmentInfoUtils.EnvironmentContext environmentContext = + new EnvironmentInfoUtils.EnvironmentContext(CONTAINER_ID, NM_HOST, NM_PORT, USER); + + LogUrlResponse actual = testInstance.createJobManagerURL(environmentContext); + + LogUrlResponse expected = + new LogUrlResponse( + String.format( + JobManagerLogUrlHandler.JOB_MANAGER_LOG_URL_FORMAT, + environmentContext.nodeManagerHostName, + environmentContext.nodeManagerHttpPort, + environmentContext.containerId, + environmentContext.user)); + + assertEquals(actual, expected); + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerLogUrlHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerLogUrlHandlerTest.java new file mode 100644 index 00000000000000..fbddd8b47c9c3b --- /dev/null +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/taskmanager/TaskManagerLogUrlHandlerTest.java @@ -0,0 +1,212 @@ +/* + * 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.runtime.rest.handler.taskmanager; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.HistoryServerOptions; +import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult; +import org.apache.flink.runtime.clusterframework.types.AllocationID; +import org.apache.flink.runtime.clusterframework.types.ResourceProfile; +import org.apache.flink.runtime.execution.ExecutionState; +import org.apache.flink.runtime.executiongraph.ArchivedExecution; +import org.apache.flink.runtime.executiongraph.ArchivedExecutionJobVertex; +import org.apache.flink.runtime.executiongraph.ArchivedExecutionVertex; +import org.apache.flink.runtime.executiongraph.ExecutionHistory; +import org.apache.flink.runtime.jobgraph.JobVertexID; +import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup; +import org.apache.flink.runtime.rest.handler.legacy.DefaultExecutionGraphCache; +import org.apache.flink.runtime.rest.handler.legacy.utils.ArchivedExecutionGraphBuilder; +import org.apache.flink.runtime.rest.messages.LogUrlResponse; +import org.apache.flink.runtime.rest.messages.TaskManagerLogUrlHeaders; +import org.apache.flink.runtime.rest.util.EnvironmentInfoUtils; +import org.apache.flink.runtime.scheduler.ExecutionGraphInfo; +import org.apache.flink.runtime.taskmanager.LocalTaskManagerLocation; +import org.apache.flink.runtime.webmonitor.history.ArchivedJson; +import org.apache.flink.testutils.TestingUtils; +import org.apache.flink.util.concurrent.Executors; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId; +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** Test for the {@link TaskManagerLogUrlHandler}. */ +public class TaskManagerLogUrlHandlerTest { + + private TaskManagerLogUrlHandler testInstance; + + private static final String CONTAINER_ID = "container_abc"; + + private static final String NM_HOST = "foo.bar"; + + private static final String NM_PORT = "0000"; + + private static final String USER = "user"; + + private Configuration configuration; + + @BeforeEach + void setup() { + configuration = new Configuration(); + configuration.set( + HistoryServerOptions + .HISTORY_SERVER_JOBMANAGER_TASKMANAGER_LOG_ENABLE_CUSTOM_HANDLERS, + true); + + testInstance = + new TaskManagerLogUrlHandler( + () -> null, + TestingUtils.TIMEOUT, + Collections.emptyMap(), + TaskManagerLogUrlHeaders.getInstance(), + new DefaultExecutionGraphCache(TestingUtils.TIMEOUT, TestingUtils.TIMEOUT), + Executors.directExecutor(), + this.configuration); + } + + @Test + public void testGenerateMultipleTaskManagerLogUrls() throws IOException { + int numTasks = 5; + + Collection archivedTaskManagerUrls = + testInstance.archiveJsonWithPath(createAccessExecutionGraph(numTasks, 0)); + + assertEquals(archivedTaskManagerUrls.size(), numTasks); + } + + @Test + public void testNoDuplicateTaskManagerLogUrls() throws IOException { + int numTasks = 5; + int numTasksWithSameLocation = 3; + Collection archivedTaskManagerUrls = + testInstance.archiveJsonWithPath( + createAccessExecutionGraph(numTasks, numTasksWithSameLocation)); + + assertEquals(archivedTaskManagerUrls.size(), numTasks - numTasksWithSameLocation + 1); + } + + @Test + public void testGenerateTaskManagerLogUrl() { + EnvironmentInfoUtils.EnvironmentContext environmentContext = + new EnvironmentInfoUtils.EnvironmentContext(CONTAINER_ID, NM_HOST, NM_PORT, USER); + + LogUrlResponse actual = + testInstance.createTaskManagerUrl(environmentContext, CONTAINER_ID, NM_HOST); + + LogUrlResponse expected = + new LogUrlResponse( + String.format( + TaskManagerLogUrlHandler.TASK_MANAGER_LOG_URL_FORMAT, + NM_HOST, + NM_PORT, + CONTAINER_ID, + environmentContext.user)); + + assertEquals(actual, expected); + } + + private static ExecutionGraphInfo createAccessExecutionGraph( + int numTasks, int numTasksWithSameLocation) { + Map tasks = new HashMap<>(); + for (int i = 0; i < numTasks - numTasksWithSameLocation; i++) { + final JobVertexID jobVertexId = new JobVertexID(); + final LocalTaskManagerLocation assignedResourceLocation = + new LocalTaskManagerLocation(); + tasks.put( + jobVertexId, + createArchivedExecutionJobVertexWithLocation( + jobVertexId, assignedResourceLocation)); + } + + final LocalTaskManagerLocation fixedResourceLocation = new LocalTaskManagerLocation(); + for (int i = 0; i < numTasksWithSameLocation; i++) { + final JobVertexID jobVertexId = new JobVertexID(); + tasks.put( + jobVertexId, + createArchivedExecutionJobVertexWithLocation( + jobVertexId, fixedResourceLocation)); + } + + return new ExecutionGraphInfo(new ArchivedExecutionGraphBuilder().setTasks(tasks).build()); + } + + private static ArchivedExecutionJobVertex createArchivedExecutionJobVertexWithLocation( + JobVertexID jobVertexID, LocalTaskManagerLocation assignedResourceLocation) { + final StringifiedAccumulatorResult[] emptyAccumulators = + new StringifiedAccumulatorResult[0]; + final long[] timestamps = new long[ExecutionState.values().length]; + final long[] endTimestamps = new long[ExecutionState.values().length]; + final ExecutionState expectedState = ExecutionState.FINISHED; + + return new ArchivedExecutionJobVertex( + createArchiveExecutionVertices( + 3, + jobVertexID, + expectedState, + assignedResourceLocation, + timestamps, + endTimestamps), + jobVertexID, + jobVertexID.toString(), + 1, + 1, + new SlotSharingGroup(), + ResourceProfile.UNKNOWN, + emptyAccumulators); + } + + private static ArchivedExecutionVertex[] createArchiveExecutionVertices( + int numSubtasks, + JobVertexID jobVertexID, + ExecutionState expectedState, + LocalTaskManagerLocation location, + long[] timestamps, + long[] endTimestamps) { + + ArchivedExecutionVertex[] vertices = new ArchivedExecutionVertex[numSubtasks]; + + for (int i = 0; i < numSubtasks; i++) { + ArchivedExecutionVertex vertex = + new ArchivedExecutionVertex( + i, + "test task", + new ArchivedExecution( + new StringifiedAccumulatorResult[0], + null, + createExecutionAttemptId(jobVertexID, i, 1), + expectedState, + null, + location, + new AllocationID(), + timestamps, + endTimestamps), + new ExecutionHistory(0)); + + vertices[i] = vertex; + } + return vertices; + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpointTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpointTest.java index 3ad06659d8264e..a925e6cc03bbb5 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpointTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpointTest.java @@ -18,26 +18,39 @@ package org.apache.flink.runtime.webmonitor; +import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; +import org.apache.flink.configuration.HistoryServerOptions; import org.apache.flink.configuration.RestOptions; import org.apache.flink.configuration.WebOptions; import org.apache.flink.core.testutils.OneShotLatch; import org.apache.flink.runtime.blob.NoOpTransientBlobService; import org.apache.flink.runtime.leaderelection.StandaloneLeaderElection; import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration; +import org.apache.flink.runtime.rest.handler.RestHandlerSpecification; +import org.apache.flink.runtime.rest.handler.cluster.JobManagerLogUrlHandler; +import org.apache.flink.runtime.rest.handler.job.GeneratedLogUrlHandler; import org.apache.flink.runtime.rest.handler.legacy.metrics.VoidMetricFetcher; +import org.apache.flink.runtime.rest.handler.taskmanager.TaskManagerLogUrlHandler; +import org.apache.flink.runtime.rest.messages.JobManagerLogUrlHeaders; +import org.apache.flink.runtime.rest.messages.TaskManagerLogUrlHeaders; import org.apache.flink.runtime.util.TestingFatalErrorHandler; import org.apache.flink.util.ExecutorUtils; +import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandler; + import org.junit.jupiter.api.Test; import java.time.Duration; +import java.util.List; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import static org.assertj.core.api.Assertions.assertThat; + /** Tests for the {@link WebMonitorEndpoint}. */ class WebMonitorEndpointTest { @@ -75,4 +88,81 @@ void cleansUpExpiredExecutionGraphs() throws Exception { ExecutorUtils.gracefulShutdown(timeout, TimeUnit.MILLISECONDS, executor); } } + + @Test + void usesGeneratedLogUrlHandlersByDefault() throws Exception { + final Tuple2 jobManagerLogUrlHandler = + findHandler(new Configuration(), JobManagerLogUrlHeaders.getInstance()); + final Tuple2 taskManagerLogUrlHandler = + findHandler(new Configuration(), TaskManagerLogUrlHeaders.getInstance()); + + assertThat(jobManagerLogUrlHandler.f1).isInstanceOf(GeneratedLogUrlHandler.class); + assertThat(taskManagerLogUrlHandler.f1).isInstanceOf(GeneratedLogUrlHandler.class); + } + + @Test + void usesGeneratedLogUrlHandlersWhenCustomHandlersAreExplicitlyDisabled() throws Exception { + final Configuration configuration = new Configuration(); + configuration.set( + HistoryServerOptions + .HISTORY_SERVER_JOBMANAGER_TASKMANAGER_LOG_ENABLE_CUSTOM_HANDLERS, + false); + + final Tuple2 jobManagerLogUrlHandler = + findHandler(configuration, JobManagerLogUrlHeaders.getInstance()); + final Tuple2 taskManagerLogUrlHandler = + findHandler(configuration, TaskManagerLogUrlHeaders.getInstance()); + + assertThat(jobManagerLogUrlHandler.f1).isInstanceOf(GeneratedLogUrlHandler.class); + assertThat(taskManagerLogUrlHandler.f1).isInstanceOf(GeneratedLogUrlHandler.class); + } + + @Test + void usesCustomLogUrlHandlersWhenEnabled() throws Exception { + final Configuration configuration = new Configuration(); + configuration.set( + HistoryServerOptions + .HISTORY_SERVER_JOBMANAGER_TASKMANAGER_LOG_ENABLE_CUSTOM_HANDLERS, + true); + + final Tuple2 jobManagerLogUrlHandler = + findHandler(configuration, JobManagerLogUrlHeaders.getInstance()); + final Tuple2 taskManagerLogUrlHandler = + findHandler(configuration, TaskManagerLogUrlHeaders.getInstance()); + + assertThat(jobManagerLogUrlHandler.f1).isInstanceOf(JobManagerLogUrlHandler.class); + assertThat(taskManagerLogUrlHandler.f1).isInstanceOf(TaskManagerLogUrlHandler.class); + } + + private static Tuple2 findHandler( + Configuration configuration, RestHandlerSpecification headers) throws Exception { + configuration.set(RestOptions.ADDRESS, "localhost"); + final ScheduledExecutorService executor = Executors.newScheduledThreadPool(1); + try (final WebMonitorEndpoint webMonitorEndpoint = + new WebMonitorEndpoint<>( + CompletableFuture::new, + configuration, + RestHandlerConfiguration.fromConfiguration(configuration), + CompletableFuture::new, + NoOpTransientBlobService.INSTANCE, + executor, + VoidMetricFetcher.INSTANCE, + new StandaloneLeaderElection(UUID.randomUUID()), + TestingExecutionGraphCache.newBuilder().build(), + new TestingFatalErrorHandler())) { + + final List> handlers = + webMonitorEndpoint.initializeHandlers(CompletableFuture.completedFuture("")); + + return handlers.stream() + .filter(handler -> handler.f0 == headers) + .findFirst() + .orElseThrow( + () -> + new AssertionError( + "No handler registered for " + headers.getClass())); + } finally { + ExecutorUtils.gracefulShutdown(10000L, TimeUnit.MILLISECONDS, executor); + } + } }