spuru9 commented on code in PR #29157: URL: https://github.com/apache/flink/pull/29157#discussion_r4370230346
########## 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() { Review Comment: critical: if any of these env vars is unset (flag on outside YARN, or `USER` missing) we serve and archive `http://null:null/node/containerlogs/null/null` with a 200, and the history server then serves that dead link forever. Can we validate once when the handlers are built and fail fast, or fall back to `GeneratedLogUrlHandler`? ########## 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 { Review Comment: minor: needs `@Internal`, and can be `final` with a private constructor. Could `getEnvironmentContext` also take a `Map<String, String>` env (no-arg version delegating to `System.getenv()`)? Right now the real env path can't be tested. ########## flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java: ########## @@ -75,6 +75,18 @@ public class HistoryServerOptions { + " with replacing the special placeholders, `<jobid>`, to the id of job." + " Only http / https schemes are supported."); + public static final ConfigOption<Boolean> + HISTORY_SERVER_JOBMANAGER_TASKMANAGER_LOG_ENABLE_CUSTOM_HANDLERS = + key("historyserver.yarn.log.enable-custom-url-handlers") Review Comment: major: only the JobManager reads this (in `WebMonitorEndpoint`, at archive time), so setting it in the history server config does nothing, which a `historyserver.*` key suggests you'd do. Can we move it under a JM/YARN-scoped key, or at least say in the description that it goes on the job cluster? ########## 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<LogUrlResponse, JobTaskManagerMessageParameters> + 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<? extends RestfulGateway> leaderRetriever, + Duration timeout, + Map<String, String> responseHeaders, + MessageHeaders<EmptyRequestBody, LogUrlResponse, JobTaskManagerMessageParameters> + 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<EmptyRequestBody> request, AccessExecutionGraph executionGraph) + throws RestHandlerException { + EnvironmentInfoUtils.EnvironmentContext environmentContext = + EnvironmentInfoUtils.getEnvironmentContext(); + ResourceID containerId = request.getPathParameter(TaskManagerIdPathParameter.class); + Map<JobVertexID, ? extends AccessExecutionJobVertex> 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<ArchivedJson> archiveJsonWithPath(ExecutionGraphInfo executionGraphInfo) + throws IOException { + EnvironmentInfoUtils.EnvironmentContext environmentContext = + EnvironmentInfoUtils.getEnvironmentContext(); + Collection<? extends AccessExecutionJobVertex> vertices = + executionGraphInfo.getArchivedExecutionGraph().getAllVertices().values(); + Set<ArchivedJson> archive = new HashSet<>(); + for (AccessExecutionJobVertex jobVertex : vertices) { + for (AccessExecutionVertex task : jobVertex.getTaskVertices()) { + Set<ArchivedJson> vertexArchivedJson = + processAllTaskManagerExecutionsPerVertex( + executionGraphInfo.getJobId().toString(), task, environmentContext); + archive.addAll(vertexArchivedJson); + } + } + return archive; + } + + private Set<ArchivedJson> processAllTaskManagerExecutionsPerVertex( + String jobId, + AccessExecutionVertex vertex, + EnvironmentInfoUtils.EnvironmentContext environmentContext) + throws IOException { + Set<ArchivedJson> 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()); Review Comment: major: the port and user come from the JM's container env, so this is only right if every NodeManager uses the same HTTP port as the JM's node. Is that safe to assume? On YARN, `ResourceID.getMetadata()` already holds the NM node id, which seems more accurate than `getFQDNHostname()`. ########## 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<LogUrlResponse, JobTaskManagerMessageParameters> + 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<? extends RestfulGateway> leaderRetriever, + Duration timeout, + Map<String, String> responseHeaders, + MessageHeaders<EmptyRequestBody, LogUrlResponse, JobTaskManagerMessageParameters> + 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 Review Comment: nit: this says it "may not return a log-url" but the code throws a 404. Also the Javadoc on `archiveJsonWithPath` (line 149) describes a `Job-History-base-URL/.../CONTAINER-ID/CONTAINER-ID/USER/` format that doesn't match what's built. Please fix or drop both. ########## 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<LogUrlResponse, JobTaskManagerMessageParameters> + 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<? extends RestfulGateway> leaderRetriever, + Duration timeout, + Map<String, String> responseHeaders, + MessageHeaders<EmptyRequestBody, LogUrlResponse, JobTaskManagerMessageParameters> + 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<EmptyRequestBody> request, AccessExecutionGraph executionGraph) + throws RestHandlerException { + EnvironmentInfoUtils.EnvironmentContext environmentContext = + EnvironmentInfoUtils.getEnvironmentContext(); + ResourceID containerId = request.getPathParameter(TaskManagerIdPathParameter.class); + Map<JobVertexID, ? extends AccessExecutionJobVertex> 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<ArchivedJson> archiveJsonWithPath(ExecutionGraphInfo executionGraphInfo) + throws IOException { + EnvironmentInfoUtils.EnvironmentContext environmentContext = + EnvironmentInfoUtils.getEnvironmentContext(); + Collection<? extends AccessExecutionJobVertex> vertices = + executionGraphInfo.getArchivedExecutionGraph().getAllVertices().values(); + Set<ArchivedJson> archive = new HashSet<>(); + for (AccessExecutionJobVertex jobVertex : vertices) { + for (AccessExecutionVertex task : jobVertex.getTaskVertices()) { + Set<ArchivedJson> vertexArchivedJson = + processAllTaskManagerExecutionsPerVertex( + executionGraphInfo.getJobId().toString(), task, environmentContext); + archive.addAll(vertexArchivedJson); + } + } + return archive; + } + + private Set<ArchivedJson> processAllTaskManagerExecutionsPerVertex( + String jobId, + AccessExecutionVertex vertex, + EnvironmentInfoUtils.EnvironmentContext environmentContext) + throws IOException { + Set<ArchivedJson> 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) { Review Comment: nit: `getResourceID() != null` can't be false since `TaskManagerLocation` does a `checkNotNull`. You can also call `getAssignedResourceLocation()` once instead of three times. ########## 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<ArchivedJson> archivedTaskManagerUrls = + testInstance.archiveJsonWithPath(createAccessExecutionGraph(numTasks, 0)); + + assertEquals(archivedTaskManagerUrls.size(), numTasks); + } + + @Test + public void testNoDuplicateTaskManagerLogUrls() throws IOException { + int numTasks = 5; + int numTasksWithSameLocation = 3; + Collection<ArchivedJson> archivedTaskManagerUrls = + testInstance.archiveJsonWithPath( + createAccessExecutionGraph(numTasks, numTasksWithSameLocation)); + + assertEquals(archivedTaskManagerUrls.size(), numTasks - numTasksWithSameLocation + 1); + } + + @Test + public void testGenerateTaskManagerLogUrl() { Review Comment: minor: same circular expected value as the JM test. Assert the literal URL, and use a TM host that differs from `NM_HOST` so a mix-up between the two shows up. ########## flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java: ########## @@ -928,8 +923,52 @@ protected List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> initiali Tuple2.of( jobManagerJobConfigurationHandler.getMessageHeaders(), jobManagerJobConfigurationHandler)); - handlers.add(Tuple2.of(JobManagerLogUrlHeaders.getInstance(), jobManagerLogUrlHandler)); - handlers.add(Tuple2.of(TaskManagerLogUrlHeaders.getInstance(), taskManagerLogUrlHandler)); + + if (clusterConfiguration.get( Review Comment: minor: please add an INFO log with the resolved state when the custom handlers are enabled (AGENTS.md asks for this on flag-gated features). This also swaps in YARN-only handlers with no YARN check; can we use the validated env context and fall back to the generated handlers when it's missing? Keeping the original `GeneratedLogUrlHandler` registration and only replacing the two handlers in the `if` would shrink the diff too. ########## flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpointTest.java: ########## @@ -75,4 +88,81 @@ void cleansUpExpiredExecutionGraphs() throws Exception { ExecutorUtils.gracefulShutdown(timeout, TimeUnit.MILLISECONDS, executor); } } + + @Test + void usesGeneratedLogUrlHandlersByDefault() throws Exception { + final Tuple2<RestHandlerSpecification, ?> jobManagerLogUrlHandler = + findHandler(new Configuration(), JobManagerLogUrlHeaders.getInstance()); + final Tuple2<RestHandlerSpecification, ?> 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<RestHandlerSpecification, ?> jobManagerLogUrlHandler = + findHandler(configuration, JobManagerLogUrlHeaders.getInstance()); + final Tuple2<RestHandlerSpecification, ?> 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<RestHandlerSpecification, ?> jobManagerLogUrlHandler = + findHandler(configuration, JobManagerLogUrlHeaders.getInstance()); + final Tuple2<RestHandlerSpecification, ?> taskManagerLogUrlHandler = + findHandler(configuration, TaskManagerLogUrlHeaders.getInstance()); + + assertThat(jobManagerLogUrlHandler.f1).isInstanceOf(JobManagerLogUrlHandler.class); + assertThat(taskManagerLogUrlHandler.f1).isInstanceOf(TaskManagerLogUrlHandler.class); + } + + private static Tuple2<RestHandlerSpecification, ?> findHandler( Review Comment: nit: this builds a full endpoint on every call (six times) and mutates the caller's `Configuration`. Please match with `.equals` rather than `==` on the headers singletons, and consider parameterizing the first two tests. ########## flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java: ########## @@ -75,6 +75,18 @@ public class HistoryServerOptions { + " with replacing the special placeholders, `<jobid>`, to the id of job." + " Only http / https schemes are supported."); + public static final ConfigOption<Boolean> + 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" Review Comment: major: "url-pattern configurations will be ignored" isn't what happens. `HistoryServer` registers the pattern routes before `/:*`, so a pattern set on the HS wins over the archived JSON. The description also names internal classes and never mentions YARN or the env vars it needs. Something like: "If true, the JobManager serves and archives YARN NodeManager log URLs built from the CONTAINER_ID, NM_HOST, NM_HTTP_PORT and USER environment variables. Only works on YARN and must be set on the job cluster. Log url patterns configured on the history server still take precedence." ########## flink-core/src/main/java/org/apache/flink/configuration/HistoryServerOptions.java: ########## @@ -75,6 +75,18 @@ public class HistoryServerOptions { + " with replacing the special placeholders, `<jobid>`, to the id of job." + " Only http / https schemes are supported."); + public static final ConfigOption<Boolean> + HISTORY_SERVER_JOBMANAGER_TASKMANAGER_LOG_ENABLE_CUSTOM_HANDLERS = + key("historyserver.yarn.log.enable-custom-url-handlers") + .booleanType() + .noDefaultValue() Review Comment: minor: use `.defaultValue(false)` and drop the `, false` at the read site. The generated doc currently shows `(none)`. ```suggestion .defaultValue(false) ``` ########## 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<LogUrlResponse, JobTaskManagerMessageParameters> + 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; Review Comment: minor: same here, `config` is unused. The class also needs `@Internal`, and `createTaskManagerUrl` can be package-private. ########## 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<LogUrlResponse, JobTaskManagerMessageParameters> + implements JsonArchivist { + + private static final Logger LOG = LoggerFactory.getLogger(TaskManagerLogUrlHandler.class); + public static final String TASK_MANAGER_LOG_URL_FORMAT = Review Comment: nit: identical to `JOB_MANAGER_LOG_URL_FORMAT`, so keep one. The hardcoded `http://` also breaks with `yarn.http.policy=HTTPS_ONLY`; if that's out of scope, a note in the option description would do. ########## 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 Review Comment: minor: needs `@Internal`. `createJobManagerURL` can be package-private with `@VisibleForTesting`. ########## 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; Review Comment: minor: `config` is never read, so please drop the field and the constructor arg (and update `WebMonitorEndpoint` and the tests). ########## 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() { Review Comment: minor: `expected` is built from the same format constant the handler uses, so this passes even if the format is wrong. Assert the literal, e.g. `http://foo.bar:8042/node/containerlogs/container_abc/user`. `archiveJsonWithPath` and `handleRequest` aren't tested either, and the `:jobid` path replacement is what the history server relies on. ########## 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<LogUrlResponse, JobTaskManagerMessageParameters> + 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<? extends RestfulGateway> leaderRetriever, + Duration timeout, + Map<String, String> responseHeaders, + MessageHeaders<EmptyRequestBody, LogUrlResponse, JobTaskManagerMessageParameters> + 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<EmptyRequestBody> request, AccessExecutionGraph executionGraph) + throws RestHandlerException { + EnvironmentInfoUtils.EnvironmentContext environmentContext = + EnvironmentInfoUtils.getEnvironmentContext(); + ResourceID containerId = request.getPathParameter(TaskManagerIdPathParameter.class); + Map<JobVertexID, ? extends AccessExecutionJobVertex> 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 Review Comment: nit: these two comments (here and at line 191) just restate the code, please remove them. ########## 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<LogUrlResponse, JobTaskManagerMessageParameters> + 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<? extends RestfulGateway> leaderRetriever, + Duration timeout, + Map<String, String> responseHeaders, + MessageHeaders<EmptyRequestBody, LogUrlResponse, JobTaskManagerMessageParameters> + 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<EmptyRequestBody> request, AccessExecutionGraph executionGraph) + throws RestHandlerException { + EnvironmentInfoUtils.EnvironmentContext environmentContext = + EnvironmentInfoUtils.getEnvironmentContext(); + ResourceID containerId = request.getPathParameter(TaskManagerIdPathParameter.class); + Map<JobVertexID, ? extends AccessExecutionJobVertex> 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<ArchivedJson> archiveJsonWithPath(ExecutionGraphInfo executionGraphInfo) + throws IOException { + EnvironmentInfoUtils.EnvironmentContext environmentContext = + EnvironmentInfoUtils.getEnvironmentContext(); + Collection<? extends AccessExecutionJobVertex> vertices = + executionGraphInfo.getArchivedExecutionGraph().getAllVertices().values(); + Set<ArchivedJson> archive = new HashSet<>(); + for (AccessExecutionJobVertex jobVertex : vertices) { + for (AccessExecutionVertex task : jobVertex.getTaskVertices()) { + Set<ArchivedJson> vertexArchivedJson = + processAllTaskManagerExecutionsPerVertex( + executionGraphInfo.getJobId().toString(), task, environmentContext); + archive.addAll(vertexArchivedJson); + } + } + return archive; + } + + private Set<ArchivedJson> processAllTaskManagerExecutionsPerVertex( + String jobId, + AccessExecutionVertex vertex, + EnvironmentInfoUtils.EnvironmentContext environmentContext) + throws IOException { + Set<ArchivedJson> 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( Review Comment: minor: this only logs the job id, so we can't tell which execution it was; please add the vertex or attempt id. The current-attempt path skips silently while this one warns, which is inconsistent. ########## 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); Review Comment: nit: AGENTS.md wants AssertJ, so `assertThat(actual).isEqualTo(expected)`. The args are also swapped (`actual, expected`). Test class and methods should be package-private, and `"0000"` isn't a realistic port. ########## 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<ArchivedJson> archivedTaskManagerUrls = + testInstance.archiveJsonWithPath(createAccessExecutionGraph(numTasks, 0)); + + assertEquals(archivedTaskManagerUrls.size(), numTasks); Review Comment: minor: asserting only `size()` means a wrong `:jobid` or `:taskmanagerid` substitution would still pass. Please assert the archived paths and URLs. Also same AssertJ, swapped-args and `public` nits as in the JM test (lines 108, 128). ########## 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<ArchivedJson> archivedTaskManagerUrls = + testInstance.archiveJsonWithPath(createAccessExecutionGraph(numTasks, 0)); + + assertEquals(archivedTaskManagerUrls.size(), numTasks); + } + + @Test + public void testNoDuplicateTaskManagerLogUrls() throws IOException { + int numTasks = 5; + int numTasksWithSameLocation = 3; + Collection<ArchivedJson> 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<JobVertexID, ArchivedExecutionJobVertex> 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)); Review Comment: minor: `new ExecutionHistory(0)` means the historical-execution branch never runs. `handleRequest` (current match, history match, 404) isn't covered at all. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
