argoyal2212 commented on code in PR #29157: URL: https://github.com/apache/flink/pull/29157#discussion_r4471054129
########## 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: Dropped the redundant null check and call getAssignedResourceLocation once. ########## 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: Added an INFO log, and it now uses the validated env context and falls back to the generated handlers if it's missing. ########## 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: Now asserts the literal URL and covers handleRequest and the archive path. ########## 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: Switched to AssertJ with correct arg order, package-private, and 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); + } + + @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: Uses literal URLs now and the TM host differs from NM_HOST. ########## 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: Now asserts the archived paths and URLs, plus the AssertJ and visibility cleanups. ########## 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: Added current and historical execution cases and handleRequest including the 404. -- 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]
