You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by xt...@apache.org on 2023/01/05 11:24:35 UTC
[flink] branch master updated (d9b8463da41 -> 898fe6ba6c8)
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from d9b8463da41 [FLINK-30544][runtime] Speed up finding minimum watermark across all channels by introducing heap-based algorithm
new b9ffa7ac1ff [FLINK-30185][rest][refactor] Distinguish which ExecutionAttemptID each ThreadInfoSample belongs to
new e07aab13857 [FLINK-30185][rest] Add the subtask index parameter for flame graph handler
new 898fe6ba6c8 [FLINK-30185][ui] Allow choose the subtask index in the flame graph page
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../src/test/resources/rest_api_v1.snapshot | 3 +
.../job-overview-drawer-flamegraph.component.html | 18 ++
.../job-overview-drawer-flamegraph.component.less | 2 +-
.../job-overview-drawer-flamegraph.component.ts | 59 +++++-
.../web-dashboard/src/app/services/job.service.ts | 11 ++
.../runtime/messages/TaskThreadInfoResponse.java | 8 +-
.../flink/runtime/messages/ThreadInfoSample.java | 15 +-
.../handler/job/JobVertexFlameGraphHandler.java | 92 +++++++--
.../messages/FlameGraphTypeQueryParameter.java | 4 +-
.../rest/messages/JobVertexFlameGraphHeaders.java | 8 +-
.../messages/JobVertexFlameGraphParameters.java | 7 +-
...ameter.java => SubtaskIndexQueryParameter.java} | 12 +-
.../flink/runtime/taskexecutor/TaskExecutor.java | 16 +-
.../taskexecutor/ThreadInfoSampleService.java | 49 ++---
.../org/apache/flink/runtime/util/JvmUtils.java | 6 +-
.../runtime/webmonitor/WebMonitorEndpoint.java | 4 +-
.../threadinfo/JobVertexThreadInfoTracker.java | 10 +-
.../JobVertexThreadInfoTrackerBuilder.java | 6 +-
.../threadinfo/ThreadInfoRequestCoordinator.java | 18 +-
...VertexFlameGraph.java => VertexFlameGraph.java} | 18 +-
...phFactory.java => VertexFlameGraphFactory.java} | 26 +--
...adInfoStats.java => VertexThreadInfoStats.java} | 19 +-
.../job/JobVertexFlameGraphHandlerTest.java | 216 +++++++++++++++++++++
.../taskexecutor/ThreadInfoSampleServiceTest.java | 82 ++++++--
.../threadinfo/JobVertexThreadInfoTrackerTest.java | 111 +++++------
.../ThreadInfoRequestCoordinatorTest.java | 50 +++--
26 files changed, 639 insertions(+), 231 deletions(-)
copy flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/{SubtaskIndexPathParameter.java => SubtaskIndexQueryParameter.java} (78%)
rename flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/{JobVertexFlameGraph.java => VertexFlameGraph.java} (90%)
rename flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/{JobVertexFlameGraphFactory.java => VertexFlameGraphFactory.java} (80%)
rename flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/{JobVertexThreadInfoStats.java => VertexThreadInfoStats.java} (84%)
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexFlameGraphHandlerTest.java
[flink] 02/03: [FLINK-30185][rest] Add the subtask index parameter for flame graph handler
Posted by xt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e07aab138578d1f65f2d7260bd33391b15b2012a
Author: 1996fanrui <19...@gmail.com>
AuthorDate: Sat Dec 17 22:26:33 2022 +0800
[FLINK-30185][rest] Add the subtask index parameter for flame graph handler
---
.../src/test/resources/rest_api_v1.snapshot | 3 +
.../handler/job/JobVertexFlameGraphHandler.java | 92 +++++++--
.../messages/FlameGraphTypeQueryParameter.java | 4 +-
.../rest/messages/JobVertexFlameGraphHeaders.java | 8 +-
.../messages/JobVertexFlameGraphParameters.java | 7 +-
...meters.java => SubtaskIndexQueryParameter.java} | 31 ++-
.../runtime/webmonitor/WebMonitorEndpoint.java | 4 +-
.../threadinfo/JobVertexThreadInfoTracker.java | 10 +-
.../JobVertexThreadInfoTrackerBuilder.java | 6 +-
.../threadinfo/ThreadInfoRequestCoordinator.java | 11 +-
...VertexFlameGraph.java => VertexFlameGraph.java} | 18 +-
...phFactory.java => VertexFlameGraphFactory.java} | 26 +--
...adInfoStats.java => VertexThreadInfoStats.java} | 9 +-
.../job/JobVertexFlameGraphHandlerTest.java | 216 +++++++++++++++++++++
.../threadinfo/JobVertexThreadInfoTrackerTest.java | 67 +++----
.../ThreadInfoRequestCoordinatorTest.java | 20 +-
16 files changed, 406 insertions(+), 126 deletions(-)
diff --git a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
index 9f1a3bec092..de462b9f204 100644
--- a/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
+++ b/flink-runtime-web/src/test/resources/rest_api_v1.snapshot
@@ -2745,6 +2745,9 @@
"queryParameters" : [ {
"key" : "type",
"mandatory" : false
+ }, {
+ "key" : "subtaskindex",
+ "mandatory" : false
} ]
},
"request" : {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexFlameGraphHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexFlameGraphHandler.java
index eb23f0f1436..516ab8bce91 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexFlameGraphHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexFlameGraphHandler.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.rest.handler.job;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.executiongraph.AccessExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.AccessExecutionVertex;
import org.apache.flink.runtime.rest.handler.AbstractRestHandler;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.RestHandlerException;
@@ -29,28 +30,32 @@ import org.apache.flink.runtime.rest.messages.FlameGraphTypeQueryParameter;
import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
import org.apache.flink.runtime.rest.messages.JobVertexFlameGraphHeaders;
import org.apache.flink.runtime.rest.messages.JobVertexFlameGraphParameters;
+import org.apache.flink.runtime.rest.messages.SubtaskIndexQueryParameter;
import org.apache.flink.runtime.webmonitor.RestfulGateway;
import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
-import org.apache.flink.runtime.webmonitor.threadinfo.JobVertexFlameGraph;
-import org.apache.flink.runtime.webmonitor.threadinfo.JobVertexFlameGraphFactory;
-import org.apache.flink.runtime.webmonitor.threadinfo.JobVertexThreadInfoStats;
-import org.apache.flink.runtime.webmonitor.threadinfo.JobVertexThreadInfoTracker;
+import org.apache.flink.runtime.webmonitor.stats.JobVertexStatsTracker;
+import org.apache.flink.runtime.webmonitor.threadinfo.VertexFlameGraph;
+import org.apache.flink.runtime.webmonitor.threadinfo.VertexFlameGraphFactory;
+import org.apache.flink.runtime.webmonitor.threadinfo.VertexThreadInfoStats;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
+import java.util.function.Function;
+import java.util.stream.Collectors;
/** Request handler for the job vertex Flame Graph. */
public class JobVertexFlameGraphHandler
- extends AbstractJobVertexHandler<JobVertexFlameGraph, JobVertexFlameGraphParameters> {
+ extends AbstractJobVertexHandler<VertexFlameGraph, JobVertexFlameGraphParameters> {
- private final JobVertexThreadInfoTracker<JobVertexThreadInfoStats> threadInfoOperatorTracker;
+ private final JobVertexStatsTracker<VertexThreadInfoStats> threadInfoOperatorTracker;
public JobVertexFlameGraphHandler(
GatewayRetriever<? extends RestfulGateway> leaderRetriever,
@@ -58,7 +63,7 @@ public class JobVertexFlameGraphHandler
Map<String, String> responseHeaders,
ExecutionGraphCache executionGraphCache,
Executor executor,
- JobVertexThreadInfoTracker<JobVertexThreadInfoStats> threadInfoOperatorTracker) {
+ JobVertexStatsTracker<VertexThreadInfoStats> threadInfoOperatorTracker) {
super(
leaderRetriever,
timeout,
@@ -70,34 +75,40 @@ public class JobVertexFlameGraphHandler
}
@Override
- protected JobVertexFlameGraph handleRequest(
+ protected VertexFlameGraph handleRequest(
HandlerRequest<EmptyRequestBody> request, AccessExecutionJobVertex jobVertex)
throws RestHandlerException {
- if (jobVertex.getAggregateState().isTerminal()) {
- return JobVertexFlameGraph.terminated();
+ @Nullable Integer subtaskIndex = getSubtaskIndex(request, jobVertex);
+ if (isTerminated(jobVertex, subtaskIndex)) {
+ return VertexFlameGraph.terminated();
}
- final Optional<JobVertexThreadInfoStats> threadInfoSample =
+ Optional<VertexThreadInfoStats> threadInfoSample =
threadInfoOperatorTracker.getVertexStats(
request.getPathParameter(JobIDPathParameter.class), jobVertex);
+ if (subtaskIndex != null) {
+ threadInfoSample =
+ threadInfoSample.map(generateThreadInfoStatsForSubtask(subtaskIndex));
+ }
+
final FlameGraphTypeQueryParameter.Type flameGraphType = getFlameGraphType(request);
- final Optional<JobVertexFlameGraph> operatorFlameGraph;
+ final Optional<VertexFlameGraph> operatorFlameGraph;
switch (flameGraphType) {
case FULL:
operatorFlameGraph =
- threadInfoSample.map(JobVertexFlameGraphFactory::createFullFlameGraphFrom);
+ threadInfoSample.map(VertexFlameGraphFactory::createFullFlameGraphFrom);
break;
case ON_CPU:
operatorFlameGraph =
- threadInfoSample.map(JobVertexFlameGraphFactory::createOnCpuFlameGraph);
+ threadInfoSample.map(VertexFlameGraphFactory::createOnCpuFlameGraph);
break;
case OFF_CPU:
operatorFlameGraph =
- threadInfoSample.map(JobVertexFlameGraphFactory::createOffCpuFlameGraph);
+ threadInfoSample.map(VertexFlameGraphFactory::createOffCpuFlameGraph);
break;
default:
throw new RestHandlerException(
@@ -105,7 +116,28 @@ public class JobVertexFlameGraphHandler
HttpResponseStatus.BAD_REQUEST);
}
- return operatorFlameGraph.orElse(JobVertexFlameGraph.waiting());
+ return operatorFlameGraph.orElse(VertexFlameGraph.waiting());
+ }
+
+ private Function<VertexThreadInfoStats, VertexThreadInfoStats>
+ generateThreadInfoStatsForSubtask(Integer subtaskIndex) {
+ return stats ->
+ new VertexThreadInfoStats(
+ stats.getRequestId(),
+ stats.getStartTime(),
+ stats.getEndTime(),
+ stats.getSamplesBySubtask().entrySet().stream()
+ .filter(entry -> entry.getKey().getSubtaskIndex() == subtaskIndex)
+ .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
+ }
+
+ private boolean isTerminated(
+ AccessExecutionJobVertex jobVertex, @Nullable Integer subtaskIndex) {
+ if (subtaskIndex == null) {
+ return jobVertex.getAggregateState().isTerminal();
+ }
+ AccessExecutionVertex executionVertex = jobVertex.getTaskVertices()[subtaskIndex];
+ return executionVertex.getExecutionState().isTerminal();
}
private static FlameGraphTypeQueryParameter.Type getFlameGraphType(HandlerRequest<?> request) {
@@ -114,9 +146,27 @@ public class JobVertexFlameGraphHandler
if (flameGraphTypeParameter.isEmpty()) {
return FlameGraphTypeQueryParameter.Type.FULL;
- } else {
- return flameGraphTypeParameter.get(0);
}
+ return flameGraphTypeParameter.get(0);
+ }
+
+ @Nullable
+ private static Integer getSubtaskIndex(
+ HandlerRequest<?> request, AccessExecutionJobVertex jobVertex)
+ throws RestHandlerException {
+ final List<Integer> subtaskIndexParameter =
+ request.getQueryParameter(SubtaskIndexQueryParameter.class);
+
+ if (subtaskIndexParameter.isEmpty()) {
+ return null;
+ }
+ int subtaskIndex = subtaskIndexParameter.get(0);
+ if (subtaskIndex >= jobVertex.getTaskVertices().length || subtaskIndex < 0) {
+ throw new RestHandlerException(
+ "Invalid subtask index for vertex " + jobVertex.getJobVertexId(),
+ HttpResponseStatus.NOT_FOUND);
+ }
+ return subtaskIndex;
}
@Override
@@ -135,7 +185,7 @@ public class JobVertexFlameGraphHandler
extends AbstractRestHandler<
RestfulGateway,
EmptyRequestBody,
- JobVertexFlameGraph,
+ VertexFlameGraph,
JobVertexFlameGraphParameters> {
protected DisabledJobVertexFlameGraphHandler(
GatewayRetriever<? extends RestfulGateway> leaderRetriever,
@@ -149,10 +199,10 @@ public class JobVertexFlameGraphHandler
}
@Override
- protected CompletableFuture<JobVertexFlameGraph> handleRequest(
+ protected CompletableFuture<VertexFlameGraph> handleRequest(
@Nonnull HandlerRequest<EmptyRequestBody> request, @Nonnull RestfulGateway gateway)
throws RestHandlerException {
- return CompletableFuture.completedFuture(JobVertexFlameGraph.disabled());
+ return CompletableFuture.completedFuture(VertexFlameGraph.disabled());
}
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/FlameGraphTypeQueryParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/FlameGraphTypeQueryParameter.java
index ae918fe0fd4..a0625394469 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/FlameGraphTypeQueryParameter.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/FlameGraphTypeQueryParameter.java
@@ -24,10 +24,10 @@ import java.util.Arrays;
public class FlameGraphTypeQueryParameter
extends MessageQueryParameter<FlameGraphTypeQueryParameter.Type> {
- private static final String key = "type";
+ public static final String KEY = "type";
public FlameGraphTypeQueryParameter() {
- super(key, MessageParameterRequisiteness.OPTIONAL);
+ super(KEY, MessageParameterRequisiteness.OPTIONAL);
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexFlameGraphHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexFlameGraphHeaders.java
index 401c2202a03..e605d2e5f40 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexFlameGraphHeaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexFlameGraphHeaders.java
@@ -20,14 +20,14 @@ package org.apache.flink.runtime.rest.messages;
import org.apache.flink.runtime.rest.HttpMethodWrapper;
import org.apache.flink.runtime.rest.handler.job.JobVertexFlameGraphHandler;
-import org.apache.flink.runtime.webmonitor.threadinfo.JobVertexFlameGraph;
+import org.apache.flink.runtime.webmonitor.threadinfo.VertexFlameGraph;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
/** Message headers for the {@link JobVertexFlameGraphHandler}. */
public class JobVertexFlameGraphHeaders
implements RuntimeMessageHeaders<
- EmptyRequestBody, JobVertexFlameGraph, JobVertexFlameGraphParameters> {
+ EmptyRequestBody, VertexFlameGraph, JobVertexFlameGraphParameters> {
private static final JobVertexFlameGraphHeaders INSTANCE = new JobVertexFlameGraphHeaders();
@@ -44,8 +44,8 @@ public class JobVertexFlameGraphHeaders
}
@Override
- public Class<JobVertexFlameGraph> getResponseClass() {
- return JobVertexFlameGraph.class;
+ public Class<VertexFlameGraph> getResponseClass() {
+ return VertexFlameGraph.class;
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexFlameGraphParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexFlameGraphParameters.java
index 8b5bb945dd0..4fb34f7c3f0 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexFlameGraphParameters.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexFlameGraphParameters.java
@@ -18,8 +18,8 @@
package org.apache.flink.runtime.rest.messages;
+import java.util.Arrays;
import java.util.Collection;
-import java.util.Collections;
/** Message parameters for job vertex Flame Graph REST handler. */
public class JobVertexFlameGraphParameters extends JobVertexMessageParameters {
@@ -27,8 +27,11 @@ public class JobVertexFlameGraphParameters extends JobVertexMessageParameters {
public final FlameGraphTypeQueryParameter flameGraphTypeQueryParameter =
new FlameGraphTypeQueryParameter();
+ public final SubtaskIndexQueryParameter subtaskIndexQueryParameter =
+ new SubtaskIndexQueryParameter();
+
@Override
public Collection<MessageQueryParameter<?>> getQueryParameters() {
- return Collections.singleton(flameGraphTypeQueryParameter);
+ return Arrays.asList(flameGraphTypeQueryParameter, subtaskIndexQueryParameter);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexFlameGraphParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/SubtaskIndexQueryParameter.java
similarity index 50%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexFlameGraphParameters.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/SubtaskIndexQueryParameter.java
index 8b5bb945dd0..08c1d0e3ff7 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexFlameGraphParameters.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/SubtaskIndexQueryParameter.java
@@ -18,17 +18,32 @@
package org.apache.flink.runtime.rest.messages;
-import java.util.Collection;
-import java.util.Collections;
+/** Query parameter specifying the index of a subtask. */
+public class SubtaskIndexQueryParameter extends MessageQueryParameter<Integer> {
-/** Message parameters for job vertex Flame Graph REST handler. */
-public class JobVertexFlameGraphParameters extends JobVertexMessageParameters {
+ public static final String KEY = "subtaskindex";
- public final FlameGraphTypeQueryParameter flameGraphTypeQueryParameter =
- new FlameGraphTypeQueryParameter();
+ public SubtaskIndexQueryParameter() {
+ super(KEY, MessageParameterRequisiteness.OPTIONAL);
+ }
+
+ @Override
+ public Integer convertStringToValue(String value) throws ConversionException {
+ final int subtaskIndex = Integer.parseInt(value);
+ if (subtaskIndex >= 0) {
+ return subtaskIndex;
+ } else {
+ throw new ConversionException("subtaskindex must be positive, was: " + subtaskIndex);
+ }
+ }
+
+ @Override
+ public String convertValueToString(Integer value) {
+ return value.toString();
+ }
@Override
- public Collection<MessageQueryParameter<?>> getQueryParameters() {
- return Collections.singleton(flameGraphTypeQueryParameter);
+ public String getDescription() {
+ return "Positive integer value that identifies a subtask.";
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
index 259495a3f33..1ab048995aa 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/WebMonitorEndpoint.java
@@ -148,10 +148,10 @@ import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
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.runtime.webmonitor.threadinfo.JobVertexThreadInfoStats;
import org.apache.flink.runtime.webmonitor.threadinfo.JobVertexThreadInfoTracker;
import org.apache.flink.runtime.webmonitor.threadinfo.JobVertexThreadInfoTrackerBuilder;
import org.apache.flink.runtime.webmonitor.threadinfo.ThreadInfoRequestCoordinator;
+import org.apache.flink.runtime.webmonitor.threadinfo.VertexThreadInfoStats;
import org.apache.flink.util.ConfigurationException;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ExecutorUtils;
@@ -241,7 +241,7 @@ public class WebMonitorEndpoint<T extends RestfulGateway> extends RestServerEndp
this.fatalErrorHandler = Preconditions.checkNotNull(fatalErrorHandler);
}
- private JobVertexThreadInfoTracker<JobVertexThreadInfoStats> initializeThreadInfoTracker(
+ private JobVertexThreadInfoTracker<VertexThreadInfoStats> initializeThreadInfoTracker(
ScheduledExecutorService executor) {
final Duration akkaTimeout = clusterConfiguration.get(AkkaOptions.ASK_TIMEOUT_DURATION);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java
index 6933655c0b4..95350f72ae9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTracker.java
@@ -77,7 +77,7 @@ public class JobVertexThreadInfoTracker<T extends Statistics> implements JobVert
@GuardedBy("lock")
private final ThreadInfoRequestCoordinator coordinator;
- private final Function<JobVertexThreadInfoStats, T> createStatsFn;
+ private final Function<VertexThreadInfoStats, T> createStatsFn;
private final ExecutorService executor;
@@ -108,7 +108,7 @@ public class JobVertexThreadInfoTracker<T extends Statistics> implements JobVert
JobVertexThreadInfoTracker(
ThreadInfoRequestCoordinator coordinator,
GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
- Function<JobVertexThreadInfoStats, T> createStatsFn,
+ Function<VertexThreadInfoStats, T> createStatsFn,
ScheduledExecutorService executor,
Duration cleanUpInterval,
int numSamples,
@@ -193,7 +193,7 @@ public class JobVertexThreadInfoTracker<T extends Statistics> implements JobVert
final CompletableFuture<ResourceManagerGateway> gatewayFuture =
resourceManagerGatewayRetriever.getFuture();
- CompletableFuture<JobVertexThreadInfoStats> sample =
+ CompletableFuture<VertexThreadInfoStats> sample =
gatewayFuture.thenCompose(
(ResourceManagerGateway resourceManagerGateway) ->
coordinator.triggerThreadInfoRequest(
@@ -334,7 +334,7 @@ public class JobVertexThreadInfoTracker<T extends Statistics> implements JobVert
/** Callback on completed thread info sample. */
private class ThreadInfoSampleCompletionCallback
- implements BiConsumer<JobVertexThreadInfoStats, Throwable> {
+ implements BiConsumer<VertexThreadInfoStats, Throwable> {
private final Key key;
private final AccessExecutionJobVertex vertex;
@@ -345,7 +345,7 @@ public class JobVertexThreadInfoTracker<T extends Statistics> implements JobVert
}
@Override
- public void accept(JobVertexThreadInfoStats threadInfoStats, Throwable throwable) {
+ public void accept(VertexThreadInfoStats threadInfoStats, Throwable throwable) {
synchronized (lock) {
try {
if (shutDown) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerBuilder.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerBuilder.java
index 86e15334a05..530654fbcda 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerBuilder.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerBuilder.java
@@ -42,7 +42,7 @@ import static org.apache.flink.util.Preconditions.checkArgument;
public class JobVertexThreadInfoTrackerBuilder<T extends Statistics> {
private final GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever;
- private final Function<JobVertexThreadInfoStats, T> createStatsFn;
+ private final Function<VertexThreadInfoStats, T> createStatsFn;
private final ScheduledExecutorService executor;
private final Time restTimeout;
@@ -56,7 +56,7 @@ public class JobVertexThreadInfoTrackerBuilder<T extends Statistics> {
JobVertexThreadInfoTrackerBuilder(
GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
- Function<JobVertexThreadInfoStats, T> createStatsFn,
+ Function<VertexThreadInfoStats, T> createStatsFn,
ScheduledExecutorService executor,
Time restTimeout) {
this.resourceManagerGatewayRetriever = resourceManagerGatewayRetriever;
@@ -188,7 +188,7 @@ public class JobVertexThreadInfoTrackerBuilder<T extends Statistics> {
*/
public static <T extends Statistics> JobVertexThreadInfoTrackerBuilder<T> newBuilder(
GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever,
- Function<JobVertexThreadInfoStats, T> createStatsFn,
+ Function<VertexThreadInfoStats, T> createStatsFn,
ScheduledExecutorService executor,
Time restTimeout) {
return new JobVertexThreadInfoTrackerBuilder<>(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinator.java
index eb0929124d5..8f080f41f8c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinator.java
@@ -42,7 +42,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/** A coordinator for triggering and collecting thread info stats of running job vertex subtasks. */
public class ThreadInfoRequestCoordinator
extends TaskStatsRequestCoordinator<
- Map<ExecutionAttemptID, Collection<ThreadInfoSample>>, JobVertexThreadInfoStats> {
+ Map<ExecutionAttemptID, Collection<ThreadInfoSample>>, VertexThreadInfoStats> {
/**
* Creates a new coordinator for the job.
@@ -68,7 +68,7 @@ public class ThreadInfoRequestCoordinator
* samples.
* @return A future of the completed thread info stats.
*/
- public CompletableFuture<JobVertexThreadInfoStats> triggerThreadInfoRequest(
+ public CompletableFuture<VertexThreadInfoStats> triggerThreadInfoRequest(
Map<ImmutableSet<ExecutionAttemptID>, CompletableFuture<TaskExecutorThreadInfoGateway>>
executionsWithGateways,
int numSamples,
@@ -161,8 +161,7 @@ public class ThreadInfoRequestCoordinator
private static class PendingThreadInfoRequest
extends PendingStatsRequest<
- Map<ExecutionAttemptID, Collection<ThreadInfoSample>>,
- JobVertexThreadInfoStats> {
+ Map<ExecutionAttemptID, Collection<ThreadInfoSample>>, VertexThreadInfoStats> {
PendingThreadInfoRequest(
int requestId, Collection<? extends Set<ExecutionAttemptID>> tasksToCollect) {
@@ -170,13 +169,13 @@ public class ThreadInfoRequestCoordinator
}
@Override
- protected JobVertexThreadInfoStats assembleCompleteStats(long endTime) {
+ protected VertexThreadInfoStats assembleCompleteStats(long endTime) {
HashMap<ExecutionAttemptID, Collection<ThreadInfoSample>> samples = new HashMap<>();
for (Map<ExecutionAttemptID, Collection<ThreadInfoSample>> map :
statsResultByTaskGroup.values()) {
samples.putAll(map);
}
- return new JobVertexThreadInfoStats(requestId, startTime, endTime, samples);
+ return new VertexThreadInfoStats(requestId, startTime, endTime, samples);
}
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexFlameGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexFlameGraph.java
similarity index 90%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexFlameGraph.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexFlameGraph.java
index 96fe0a21492..2bd160e186c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexFlameGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexFlameGraph.java
@@ -28,12 +28,12 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonPro
import java.util.List;
/**
- * Flame Graph representation for a job vertex.
+ * Flame Graph representation for a job vertex or an execution vertex.
*
* <p>Statistics are gathered by sampling stack traces of running tasks.
*/
@JsonInclude(JsonInclude.Include.NON_NULL)
-public class JobVertexFlameGraph implements ResponseBody {
+public class VertexFlameGraph implements ResponseBody {
private static final String FIELD_NAME_END_TIMESTAMP = "endTimestamp";
private static final String FIELD_NAME_DATA = "data";
@@ -46,7 +46,7 @@ public class JobVertexFlameGraph implements ResponseBody {
private final Node root;
@JsonCreator
- public JobVertexFlameGraph(
+ public VertexFlameGraph(
@JsonProperty(FIELD_NAME_END_TIMESTAMP) long endTimestamp,
@JsonProperty(FIELD_NAME_DATA) Node root) {
this.endTimestamp = endTimestamp;
@@ -69,18 +69,18 @@ public class JobVertexFlameGraph implements ResponseBody {
}
// Indicates that the task execution has been terminated
- public static JobVertexFlameGraph terminated() {
- return new JobVertexFlameGraph(-1, null);
+ public static VertexFlameGraph terminated() {
+ return new VertexFlameGraph(-1, null);
}
// Indicates that the flame graph feature has been disabled
- public static JobVertexFlameGraph disabled() {
- return new JobVertexFlameGraph(-2, null);
+ public static VertexFlameGraph disabled() {
+ return new VertexFlameGraph(-2, null);
}
// Indicates that it is waiting for the first samples to creating the flame graph
- public static JobVertexFlameGraph waiting() {
- return new JobVertexFlameGraph(-3, null);
+ public static VertexFlameGraph waiting() {
+ return new VertexFlameGraph(-3, null);
}
/** Graph node. */
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexFlameGraphFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexFlameGraphFactory.java
similarity index 80%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexFlameGraphFactory.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexFlameGraphFactory.java
index 07f98fe33b0..35f06663274 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexFlameGraphFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexFlameGraphFactory.java
@@ -30,21 +30,21 @@ import java.util.Map;
import java.util.Set;
/** Factory class for creating Flame Graph representations. */
-public class JobVertexFlameGraphFactory {
+public class VertexFlameGraphFactory {
/**
- * Converts {@link JobVertexThreadInfoStats} into a FlameGraph.
+ * Converts {@link VertexThreadInfoStats} into a FlameGraph.
*
* @param sample Thread details sample containing stack traces.
* @return FlameGraph data structure
*/
- public static JobVertexFlameGraph createFullFlameGraphFrom(JobVertexThreadInfoStats sample) {
+ public static VertexFlameGraph createFullFlameGraphFrom(VertexThreadInfoStats sample) {
EnumSet<Thread.State> included = EnumSet.allOf(Thread.State.class);
return createFlameGraphFromSample(sample, included);
}
/**
- * Converts {@link JobVertexThreadInfoStats} into a FlameGraph representing blocked (Off-CPU)
+ * Converts {@link VertexThreadInfoStats} into a FlameGraph representing blocked (Off-CPU)
* threads.
*
* <p>Includes threads in states Thread.State.[TIMED_WAITING, BLOCKED, WAITING].
@@ -52,14 +52,14 @@ public class JobVertexFlameGraphFactory {
* @param sample Thread details sample containing stack traces.
* @return FlameGraph data structure.
*/
- public static JobVertexFlameGraph createOffCpuFlameGraph(JobVertexThreadInfoStats sample) {
+ public static VertexFlameGraph createOffCpuFlameGraph(VertexThreadInfoStats sample) {
EnumSet<Thread.State> included =
EnumSet.of(Thread.State.TIMED_WAITING, Thread.State.BLOCKED, Thread.State.WAITING);
return createFlameGraphFromSample(sample, included);
}
/**
- * Converts {@link JobVertexThreadInfoStats} into a FlameGraph representing actively running
+ * Converts {@link VertexThreadInfoStats} into a FlameGraph representing actively running
* (On-CPU) threads.
*
* <p>Includes threads in states Thread.State.[RUNNABLE, NEW].
@@ -67,13 +67,13 @@ public class JobVertexFlameGraphFactory {
* @param sample Thread details sample containing stack traces.
* @return FlameGraph data structure
*/
- public static JobVertexFlameGraph createOnCpuFlameGraph(JobVertexThreadInfoStats sample) {
+ public static VertexFlameGraph createOnCpuFlameGraph(VertexThreadInfoStats sample) {
EnumSet<Thread.State> included = EnumSet.of(Thread.State.RUNNABLE, Thread.State.NEW);
return createFlameGraphFromSample(sample, included);
}
- private static JobVertexFlameGraph createFlameGraphFromSample(
- JobVertexThreadInfoStats sample, Set<Thread.State> threadStates) {
+ private static VertexFlameGraph createFlameGraphFromSample(
+ VertexThreadInfoStats sample, Set<Thread.State> threadStates) {
final NodeBuilder root = new NodeBuilder("root");
for (Collection<ThreadInfoSample> threadInfoSubSamples :
sample.getSamplesBySubtask().values()) {
@@ -94,7 +94,7 @@ public class JobVertexFlameGraphFactory {
}
}
}
- return new JobVertexFlameGraph(sample.getEndTime(), root.toNode());
+ return new VertexFlameGraph(sample.getEndTime(), root.toNode());
}
private static class NodeBuilder {
@@ -119,12 +119,12 @@ public class JobVertexFlameGraphFactory {
hitCount++;
}
- private JobVertexFlameGraph.Node toNode() {
- final List<JobVertexFlameGraph.Node> childrenNodes = new ArrayList<>(children.size());
+ private VertexFlameGraph.Node toNode() {
+ final List<VertexFlameGraph.Node> childrenNodes = new ArrayList<>(children.size());
for (NodeBuilder builderChild : children.values()) {
childrenNodes.add(builderChild.toNode());
}
- return new JobVertexFlameGraph.Node(
+ return new VertexFlameGraph.Node(
stackTraceLocation, hitCount, Collections.unmodifiableList(childrenNodes));
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoStats.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoStats.java
similarity index 94%
rename from flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoStats.java
rename to flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoStats.java
index f035bebc67b..d56945888c9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoStats.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/VertexThreadInfoStats.java
@@ -28,11 +28,8 @@ import java.util.Map;
import static org.apache.flink.util.Preconditions.checkArgument;
-/**
- * Thread info statistics of multiple tasks. Each subtask can deliver multiple samples for
- * statistical purposes.
- */
-public class JobVertexThreadInfoStats implements Statistics {
+/** Thread info statistics of single JobVertex or ExecutionVertex. */
+public class VertexThreadInfoStats implements Statistics {
/** ID of the corresponding request. */
private final int requestId;
@@ -54,7 +51,7 @@ public class JobVertexThreadInfoStats implements Statistics {
* @param endTime Timestamp, when all thread info samples were collected.
* @param samplesBySubtask Map of thread info samples by subtask (execution ID).
*/
- public JobVertexThreadInfoStats(
+ public VertexThreadInfoStats(
int requestId,
long startTime,
long endTime,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexFlameGraphHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexFlameGraphHandlerTest.java
new file mode 100644
index 00000000000..20e76022b3a
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/JobVertexFlameGraphHandlerTest.java
@@ -0,0 +1,216 @@
+/*
+ * 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.job;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.accumulators.StringifiedAccumulatorResult;
+import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.AccessExecutionJobVertex;
+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.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.HandlerRequestException;
+import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
+import org.apache.flink.runtime.rest.handler.legacy.DefaultExecutionGraphCache;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.FlameGraphTypeQueryParameter;
+import org.apache.flink.runtime.rest.messages.JobIDPathParameter;
+import org.apache.flink.runtime.rest.messages.JobVertexFlameGraphParameters;
+import org.apache.flink.runtime.rest.messages.JobVertexIdPathParameter;
+import org.apache.flink.runtime.rest.messages.SubtaskIndexQueryParameter;
+import org.apache.flink.runtime.webmonitor.stats.JobVertexStatsTracker;
+import org.apache.flink.runtime.webmonitor.threadinfo.VertexFlameGraph;
+import org.apache.flink.runtime.webmonitor.threadinfo.VertexThreadInfoStats;
+import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.TestLogger;
+import org.apache.flink.util.concurrent.Executors;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests of {@link JobVertexFlameGraphHandler}. */
+public class JobVertexFlameGraphHandlerTest extends TestLogger {
+
+ private static final JobID JOB_ID = new JobID();
+ private static final JobVertexID JOB_VERTEX_ID = new JobVertexID();
+
+ private static VertexThreadInfoStats taskThreadInfoStatsDefaultSample;
+ private static JobVertexFlameGraphHandler handler;
+
+ @BeforeAll
+ public static void setUp() {
+ taskThreadInfoStatsDefaultSample =
+ new VertexThreadInfoStats(
+ 8,
+ System.currentTimeMillis(),
+ System.currentTimeMillis() + 100,
+ Collections.emptyMap());
+
+ final RestHandlerConfiguration restHandlerConfiguration =
+ RestHandlerConfiguration.fromConfiguration(new Configuration());
+ handler =
+ new JobVertexFlameGraphHandler(
+ () -> null,
+ Time.milliseconds(100L),
+ Collections.emptyMap(),
+ new DefaultExecutionGraphCache(
+ restHandlerConfiguration.getTimeout(),
+ Time.milliseconds(restHandlerConfiguration.getRefreshInterval())),
+ Executors.directExecutor(),
+ new TestThreadInfoTracker(taskThreadInfoStatsDefaultSample));
+ }
+
+ /** Some subtasks are finished, some subtasks are running. */
+ @Test
+ void testHandleMixedSubtasks() throws Exception {
+ final ArchivedExecutionJobVertex archivedExecutionJobVertex =
+ new ArchivedExecutionJobVertex(
+ new ArchivedExecutionVertex[] {
+ generateExecutionVertex(0, ExecutionState.FINISHED),
+ generateExecutionVertex(1, ExecutionState.RUNNING)
+ },
+ JOB_VERTEX_ID,
+ "test",
+ 2,
+ 2,
+ ResourceProfile.UNKNOWN,
+ new StringifiedAccumulatorResult[0]);
+
+ // Check the finished subtask
+ HandlerRequest<EmptyRequestBody> request = generateJobVertexFlameGraphParameters(0);
+ VertexFlameGraph jobVertexFlameGraph =
+ handler.handleRequest(request, archivedExecutionJobVertex);
+ assertThat(jobVertexFlameGraph.getEndTime())
+ .isEqualTo(VertexFlameGraph.terminated().getEndTime());
+
+ // Check the running subtask
+ request = generateJobVertexFlameGraphParameters(1);
+ jobVertexFlameGraph = handler.handleRequest(request, archivedExecutionJobVertex);
+ assertThat(jobVertexFlameGraph.getEndTime())
+ .isEqualTo(taskThreadInfoStatsDefaultSample.getEndTime());
+
+ // Check the job vertex
+ request = generateJobVertexFlameGraphParameters(null);
+ jobVertexFlameGraph = handler.handleRequest(request, archivedExecutionJobVertex);
+ assertThat(jobVertexFlameGraph.getEndTime())
+ .isEqualTo(taskThreadInfoStatsDefaultSample.getEndTime());
+ }
+
+ /** All subtasks are finished. */
+ @Test
+ void testHandleFinishedJobVertex() throws Exception {
+ final ArchivedExecutionJobVertex archivedExecutionJobVertex =
+ new ArchivedExecutionJobVertex(
+ new ArchivedExecutionVertex[] {
+ generateExecutionVertex(0, ExecutionState.FINISHED),
+ generateExecutionVertex(1, ExecutionState.FINISHED)
+ },
+ JOB_VERTEX_ID,
+ "test",
+ 2,
+ 2,
+ ResourceProfile.UNKNOWN,
+ new StringifiedAccumulatorResult[0]);
+
+ HandlerRequest<EmptyRequestBody> request = generateJobVertexFlameGraphParameters(null);
+ VertexFlameGraph jobVertexFlameGraph =
+ handler.handleRequest(request, archivedExecutionJobVertex);
+ assertThat(jobVertexFlameGraph.getEndTime())
+ .isEqualTo(VertexFlameGraph.terminated().getEndTime());
+ }
+
+ private HandlerRequest<EmptyRequestBody> generateJobVertexFlameGraphParameters(
+ Integer subtaskIndex) throws HandlerRequestException {
+ final HashMap<String, String> receivedPathParameters = new HashMap<>(2);
+ receivedPathParameters.put(JobIDPathParameter.KEY, JOB_ID.toString());
+ receivedPathParameters.put(JobVertexIdPathParameter.KEY, JOB_VERTEX_ID.toString());
+
+ Map<String, List<String>> queryParams = new HashMap<>(2);
+ queryParams.put(
+ FlameGraphTypeQueryParameter.KEY,
+ Collections.singletonList(FlameGraphTypeQueryParameter.Type.FULL.name()));
+ if (subtaskIndex != null) {
+ queryParams.put(
+ SubtaskIndexQueryParameter.KEY,
+ Collections.singletonList(subtaskIndex.toString()));
+ }
+
+ return HandlerRequest.resolveParametersAndCreate(
+ EmptyRequestBody.getInstance(),
+ new JobVertexFlameGraphParameters(),
+ receivedPathParameters,
+ queryParams,
+ Collections.emptyList());
+ }
+
+ private ArchivedExecutionVertex generateExecutionVertex(
+ int subtaskIndex, ExecutionState executionState) {
+ return new ArchivedExecutionVertex(
+ subtaskIndex,
+ "test task",
+ new ArchivedExecution(
+ new StringifiedAccumulatorResult[0],
+ null,
+ createExecutionAttemptId(JOB_VERTEX_ID, subtaskIndex, 0),
+ executionState,
+ null,
+ null,
+ null,
+ new long[ExecutionState.values().length],
+ new long[ExecutionState.values().length]),
+ new ExecutionHistory(0));
+ }
+
+ /**
+ * A {@link JobVertexStatsTracker} which returns the pre-generated thread info stats directly.
+ */
+ private static class TestThreadInfoTracker
+ implements JobVertexStatsTracker<VertexThreadInfoStats> {
+
+ private final VertexThreadInfoStats stats;
+
+ public TestThreadInfoTracker(VertexThreadInfoStats stats) {
+ this.stats = stats;
+ }
+
+ @Override
+ public Optional<VertexThreadInfoStats> getVertexStats(
+ JobID jobId, AccessExecutionJobVertex vertex) {
+ return Optional.of(stats);
+ }
+
+ @Override
+ public void shutDown() throws FlinkException {}
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java
index d45aff28279..afd51257099 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java
@@ -89,7 +89,7 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
.collect(Collectors.toSet());
private static final JobID JOB_ID = new JobID();
- private static JobVertexThreadInfoStats threadInfoStatsDefaultSample;
+ private static VertexThreadInfoStats threadInfoStatsDefaultSample;
private static final Duration CLEAN_UP_INTERVAL = Duration.ofSeconds(60);
private static final Duration STATS_REFRESH_INTERVAL = Duration.ofSeconds(60);
@@ -128,16 +128,16 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
/** Tests that cached result is reused within refresh interval. */
@Test
public void testCachedStatsNotUpdatedWithinRefreshInterval() throws Exception {
- final JobVertexThreadInfoStats unusedThreadInfoStats = createThreadInfoStats(1, TIME_GAP);
+ final VertexThreadInfoStats unusedThreadInfoStats = createThreadInfoStats(1, TIME_GAP);
- final JobVertexThreadInfoTracker<JobVertexThreadInfoStats> tracker =
+ final JobVertexThreadInfoTracker<VertexThreadInfoStats> tracker =
createThreadInfoTracker(
STATS_REFRESH_INTERVAL,
threadInfoStatsDefaultSample,
unusedThreadInfoStats);
// stores threadInfoStatsDefaultSample in cache
doInitialRequestAndVerifyResult(tracker);
- Optional<JobVertexThreadInfoStats> result =
+ Optional<VertexThreadInfoStats> result =
tracker.getVertexStats(JOB_ID, EXECUTION_JOB_VERTEX);
// cached result is returned instead of unusedThreadInfoStats
assertThat(result).isPresent().hasValue(threadInfoStatsDefaultSample);
@@ -149,19 +149,19 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
final Duration shortRefreshInterval = Duration.ofMillis(1);
// first entry is in the past, so refresh is triggered immediately upon fetching it
- final JobVertexThreadInfoStats initialThreadInfoStats =
+ final VertexThreadInfoStats initialThreadInfoStats =
createThreadInfoStats(
Instant.now().minus(10, ChronoUnit.SECONDS),
REQUEST_ID,
Duration.ofMillis(5));
- final JobVertexThreadInfoStats threadInfoStatsAfterRefresh =
+ final VertexThreadInfoStats threadInfoStatsAfterRefresh =
createThreadInfoStats(1, TIME_GAP);
// register a CountDownLatch with the cache so we can await refresh of the entry
CountDownLatch cacheRefreshed = new CountDownLatch(1);
- Cache<JobVertexThreadInfoTracker.Key, JobVertexThreadInfoStats> vertexStatsCache =
+ Cache<JobVertexThreadInfoTracker.Key, VertexThreadInfoStats> vertexStatsCache =
createCache(CLEAN_UP_INTERVAL, new LatchRemovalListener<>(cacheRefreshed));
- final JobVertexThreadInfoTracker<JobVertexThreadInfoStats> tracker =
+ final JobVertexThreadInfoTracker<VertexThreadInfoStats> tracker =
createThreadInfoTracker(
CLEAN_UP_INTERVAL,
shortRefreshInterval,
@@ -182,7 +182,7 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
cacheRefreshed.await();
// verify that we get the second result on the next request
- Optional<JobVertexThreadInfoStats> result =
+ Optional<VertexThreadInfoStats> result =
tracker.getVertexStats(JOB_ID, EXECUTION_JOB_VERTEX);
assertExpectedEqualsReceived(threadInfoStatsAfterRefresh, result);
}
@@ -194,9 +194,9 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
// register a CountDownLatch with the cache so we can await expiry of the entry
CountDownLatch cacheExpired = new CountDownLatch(1);
- Cache<JobVertexThreadInfoTracker.Key, JobVertexThreadInfoStats> vertexStatsCache =
+ Cache<JobVertexThreadInfoTracker.Key, VertexThreadInfoStats> vertexStatsCache =
createCache(shortCleanUpInterval, new LatchRemovalListener<>(cacheExpired));
- final JobVertexThreadInfoTracker<JobVertexThreadInfoStats> tracker =
+ final JobVertexThreadInfoTracker<VertexThreadInfoStats> tracker =
createThreadInfoTracker(
shortCleanUpInterval,
STATS_REFRESH_INTERVAL,
@@ -214,8 +214,7 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
/** Tests that cached results are NOT removed within the cleanup interval. */
@Test
public void testCachedStatsNotCleanedWithinCleanupInterval() throws Exception {
- final JobVertexThreadInfoTracker<JobVertexThreadInfoStats> tracker =
- createThreadInfoTracker();
+ final JobVertexThreadInfoTracker<VertexThreadInfoStats> tracker = createThreadInfoTracker();
doInitialRequestAndVerifyResult(tracker);
@@ -228,8 +227,7 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
/** Tests that cached results are not served after the shutdown. */
@Test
public void testShutDown() throws Exception {
- final JobVertexThreadInfoTracker<JobVertexThreadInfoStats> tracker =
- createThreadInfoTracker();
+ final JobVertexThreadInfoTracker<VertexThreadInfoStats> tracker = createThreadInfoTracker();
doInitialRequestAndVerifyResult(tracker);
// shutdown directly
@@ -241,9 +239,9 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
assertThat(tracker.getVertexStats(JOB_ID, EXECUTION_JOB_VERTEX)).isNotPresent();
}
- private Cache<JobVertexThreadInfoTracker.Key, JobVertexThreadInfoStats> createCache(
+ private Cache<JobVertexThreadInfoTracker.Key, VertexThreadInfoStats> createCache(
Duration cleanUpInterval,
- RemovalListener<JobVertexThreadInfoTracker.Key, JobVertexThreadInfoStats>
+ RemovalListener<JobVertexThreadInfoTracker.Key, VertexThreadInfoStats>
removalListener) {
return CacheBuilder.newBuilder()
.concurrencyLevel(1)
@@ -253,7 +251,7 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
}
private void doInitialRequestAndVerifyResult(
- JobVertexThreadInfoTracker<JobVertexThreadInfoStats> tracker)
+ JobVertexThreadInfoTracker<VertexThreadInfoStats> tracker)
throws InterruptedException, ExecutionException {
// no stats yet, but the request triggers async collection of stats
assertThat(tracker.getVertexStats(JOB_ID, EXECUTION_JOB_VERTEX)).isNotPresent();
@@ -264,10 +262,9 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
}
private static void assertExpectedEqualsReceived(
- JobVertexThreadInfoStats expected,
- Optional<JobVertexThreadInfoStats> receivedOptional) {
+ VertexThreadInfoStats expected, Optional<VertexThreadInfoStats> receivedOptional) {
assertThat(receivedOptional).isPresent();
- JobVertexThreadInfoStats received = receivedOptional.get();
+ VertexThreadInfoStats received = receivedOptional.get();
assertThat(expected.getRequestId()).isEqualTo(received.getRequestId());
assertThat(expected.getEndTime()).isEqualTo(received.getEndTime());
@@ -279,20 +276,20 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
}
}
- private JobVertexThreadInfoTracker<JobVertexThreadInfoStats> createThreadInfoTracker() {
+ private JobVertexThreadInfoTracker<VertexThreadInfoStats> createThreadInfoTracker() {
return createThreadInfoTracker(STATS_REFRESH_INTERVAL, threadInfoStatsDefaultSample);
}
- private JobVertexThreadInfoTracker<JobVertexThreadInfoStats> createThreadInfoTracker(
- Duration statsRefreshInterval, JobVertexThreadInfoStats... stats) {
+ private JobVertexThreadInfoTracker<VertexThreadInfoStats> createThreadInfoTracker(
+ Duration statsRefreshInterval, VertexThreadInfoStats... stats) {
return createThreadInfoTracker(CLEAN_UP_INTERVAL, statsRefreshInterval, null, stats);
}
- private JobVertexThreadInfoTracker<JobVertexThreadInfoStats> createThreadInfoTracker(
+ private JobVertexThreadInfoTracker<VertexThreadInfoStats> createThreadInfoTracker(
Duration cleanUpInterval,
Duration statsRefreshInterval,
- Cache<JobVertexThreadInfoTracker.Key, JobVertexThreadInfoStats> vertexStatsCache,
- JobVertexThreadInfoStats... stats) {
+ Cache<JobVertexThreadInfoTracker.Key, VertexThreadInfoStats> vertexStatsCache,
+ VertexThreadInfoStats... stats) {
final ThreadInfoRequestCoordinator coordinator =
new TestingThreadInfoRequestCoordinator(Runnable::run, REQUEST_TIMEOUT, stats);
@@ -311,11 +308,11 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
.build();
}
- private static JobVertexThreadInfoStats createThreadInfoStats(int requestId, Duration timeGap) {
+ private static VertexThreadInfoStats createThreadInfoStats(int requestId, Duration timeGap) {
return createThreadInfoStats(Instant.now(), requestId, timeGap);
}
- private static JobVertexThreadInfoStats createThreadInfoStats(
+ private static VertexThreadInfoStats createThreadInfoStats(
Instant startTime, int requestId, Duration timeGap) {
Instant endTime = startTime.plus(timeGap);
@@ -331,7 +328,7 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
Collections.singletonList(threadInfoSample.get()));
}
- return new JobVertexThreadInfoStats(
+ return new VertexThreadInfoStats(
requestId, startTime.toEpochMilli(), endTime.toEpochMilli(), samples);
}
@@ -372,19 +369,19 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
*/
private static class TestingThreadInfoRequestCoordinator extends ThreadInfoRequestCoordinator {
- private final JobVertexThreadInfoStats[] jobVertexThreadInfoStats;
+ private final VertexThreadInfoStats[] vertexThreadInfoStats;
private int counter = 0;
TestingThreadInfoRequestCoordinator(
Executor executor,
Duration requestTimeout,
- JobVertexThreadInfoStats... jobVertexThreadInfoStats) {
+ VertexThreadInfoStats... vertexThreadInfoStats) {
super(executor, requestTimeout);
- this.jobVertexThreadInfoStats = jobVertexThreadInfoStats;
+ this.vertexThreadInfoStats = vertexThreadInfoStats;
}
@Override
- public CompletableFuture<JobVertexThreadInfoStats> triggerThreadInfoRequest(
+ public CompletableFuture<VertexThreadInfoStats> triggerThreadInfoRequest(
Map<
ImmutableSet<ExecutionAttemptID>,
CompletableFuture<TaskExecutorThreadInfoGateway>>
@@ -396,7 +393,7 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
assertThat(executionsWithGateways.keySet().iterator().next()).isEqualTo(ATTEMPT_IDS);
return CompletableFuture.completedFuture(
- jobVertexThreadInfoStats[(counter++) % jobVertexThreadInfoStats.length]);
+ vertexThreadInfoStats[(counter++) % vertexThreadInfoStats.length]);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinatorTest.java
index 147195ea859..f937ccd1c5d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinatorTest.java
@@ -104,14 +104,14 @@ public class ThreadInfoRequestCoordinatorTest extends TestLogger {
createMockSubtaskWithGateways(
CompletionType.SUCCESSFULLY, CompletionType.SUCCESSFULLY);
- CompletableFuture<JobVertexThreadInfoStats> requestFuture =
+ CompletableFuture<VertexThreadInfoStats> requestFuture =
coordinator.triggerThreadInfoRequest(
executionWithGateways,
DEFAULT_NUMBER_OF_SAMPLES,
DEFAULT_DELAY_BETWEEN_SAMPLES,
DEFAULT_MAX_STACK_TRACE_DEPTH);
- JobVertexThreadInfoStats threadInfoStats = requestFuture.get();
+ VertexThreadInfoStats threadInfoStats = requestFuture.get();
// verify the request result
assertThat(threadInfoStats.getRequestId()).isEqualTo(0);
@@ -133,7 +133,7 @@ public class ThreadInfoRequestCoordinatorTest extends TestLogger {
createMockSubtaskWithGateways(
CompletionType.SUCCESSFULLY, CompletionType.EXCEPTIONALLY);
- CompletableFuture<JobVertexThreadInfoStats> requestFuture =
+ CompletableFuture<VertexThreadInfoStats> requestFuture =
coordinator.triggerThreadInfoRequest(
executionWithGateways,
DEFAULT_NUMBER_OF_SAMPLES,
@@ -156,7 +156,7 @@ public class ThreadInfoRequestCoordinatorTest extends TestLogger {
createMockSubtaskWithGateways(
CompletionType.SUCCESSFULLY, CompletionType.TIMEOUT);
- CompletableFuture<JobVertexThreadInfoStats> requestFuture =
+ CompletableFuture<VertexThreadInfoStats> requestFuture =
coordinator.triggerThreadInfoRequest(
executionWithGateways,
DEFAULT_NUMBER_OF_SAMPLES,
@@ -184,16 +184,16 @@ public class ThreadInfoRequestCoordinatorTest extends TestLogger {
createMockSubtaskWithGateways(
CompletionType.SUCCESSFULLY, CompletionType.TIMEOUT);
- List<CompletableFuture<JobVertexThreadInfoStats>> requestFutures = new ArrayList<>();
+ List<CompletableFuture<VertexThreadInfoStats>> requestFutures = new ArrayList<>();
- CompletableFuture<JobVertexThreadInfoStats> requestFuture1 =
+ CompletableFuture<VertexThreadInfoStats> requestFuture1 =
coordinator.triggerThreadInfoRequest(
executionWithGateways,
DEFAULT_NUMBER_OF_SAMPLES,
DEFAULT_DELAY_BETWEEN_SAMPLES,
DEFAULT_MAX_STACK_TRACE_DEPTH);
- CompletableFuture<JobVertexThreadInfoStats> requestFuture2 =
+ CompletableFuture<VertexThreadInfoStats> requestFuture2 =
coordinator.triggerThreadInfoRequest(
executionWithGateways,
DEFAULT_NUMBER_OF_SAMPLES,
@@ -204,7 +204,7 @@ public class ThreadInfoRequestCoordinatorTest extends TestLogger {
requestFutures.add(requestFuture1);
requestFutures.add(requestFuture2);
- for (CompletableFuture<JobVertexThreadInfoStats> future : requestFutures) {
+ for (CompletableFuture<VertexThreadInfoStats> future : requestFutures) {
assertThat(future).isNotDone();
}
@@ -212,12 +212,12 @@ public class ThreadInfoRequestCoordinatorTest extends TestLogger {
coordinator.shutDown();
// verify all completed
- for (CompletableFuture<JobVertexThreadInfoStats> future : requestFutures) {
+ for (CompletableFuture<VertexThreadInfoStats> future : requestFutures) {
assertThat(future).isCompletedExceptionally();
}
// verify new trigger returns failed future
- CompletableFuture<JobVertexThreadInfoStats> future =
+ CompletableFuture<VertexThreadInfoStats> future =
coordinator.triggerThreadInfoRequest(
executionWithGateways,
DEFAULT_NUMBER_OF_SAMPLES,
[flink] 03/03: [FLINK-30185][ui] Allow choose the subtask index in the flame graph page
Posted by xt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 898fe6ba6c882b82a266fb5a415d68e9c9113cb8
Author: 1996fanrui <19...@gmail.com>
AuthorDate: Sat Dec 17 22:41:00 2022 +0800
[FLINK-30185][ui] Allow choose the subtask index in the flame graph page
This closes #21447
---
.../job-overview-drawer-flamegraph.component.html | 18 +++++++
.../job-overview-drawer-flamegraph.component.less | 2 +-
.../job-overview-drawer-flamegraph.component.ts | 59 ++++++++++++++++++++--
.../web-dashboard/src/app/services/job.service.ts | 11 ++++
4 files changed, 86 insertions(+), 4 deletions(-)
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.html b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.html
index 92cd970e53b..3c34fbc40ba 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.html
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.html
@@ -48,6 +48,24 @@ Type:
</label>
</nz-radio-group>
+Subtask:
+<nz-select
+ nzSize="small"
+ [ngModel]="subtaskIndex"
+ nzShowSearch
+ (ngModelChange)="selectSubtask($event)"
+>
+ <nz-option
+ *ngFor="let name of listOfRunningSubtasks"
+ [nzLabel]="name"
+ [nzValue]="name"
+ nzCustomContent
+ >
+ <span [title]="name">{{ name }}</span>
+ </nz-option>
+</nz-select>
+
+
<ng-container [ngSwitch]="flameGraph.endTimestamp">
<span *ngSwitchCase="-1">The task has already been terminated</span>
<span *ngSwitchCase="-2">
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.less b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.less
index 58c534074c4..73716a18ca1 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.less
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.less
@@ -19,7 +19,7 @@
@import "theme";
nz-select {
- width: 300px;
+ width: 100px;
}
.metric-selector {
diff --git a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.ts b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.ts
index acc182e6a6a..b4a7e92b46a 100644
--- a/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.ts
+++ b/flink-runtime-web/web-dashboard/src/app/pages/job/overview/flamegraph/job-overview-drawer-flamegraph.component.ts
@@ -16,7 +16,7 @@
* limitations under the License.
*/
-import { NgIf, NgSwitch, NgSwitchCase, NgSwitchDefault } from '@angular/common';
+import { NgForOf, NgIf, NgSwitch, NgSwitchCase, NgSwitchDefault } from '@angular/common';
import { ChangeDetectionStrategy, ChangeDetectorRef, Component, OnDestroy, OnInit } from '@angular/core';
import { FormsModule } from '@angular/forms';
import { Subject } from 'rxjs';
@@ -26,7 +26,9 @@ import { FlameGraphComponent } from '@flink-runtime-web/components/flame-graph/f
import { HumanizeDurationPipe } from '@flink-runtime-web/components/humanize-duration.pipe';
import { FlameGraphType, JobFlameGraph, NodesItemCorrect } from '@flink-runtime-web/interfaces';
import { JobService } from '@flink-runtime-web/services';
+import { isNil } from '@flink-runtime-web/utils';
import { NzRadioModule } from 'ng-zorro-antd/radio';
+import { NzSelectModule } from 'ng-zorro-antd/select';
import { NzSpinModule } from 'ng-zorro-antd/spin';
import { JobLocalService } from '../../job-local.service';
@@ -38,6 +40,8 @@ import { JobLocalService } from '../../job-local.service';
styleUrls: ['./job-overview-drawer-flamegraph.component.less'],
imports: [
NgIf,
+ NgForOf,
+ NzSelectModule,
NzRadioModule,
FormsModule,
NgSwitch,
@@ -55,8 +59,11 @@ export class JobOverviewDrawerFlameGraphComponent implements OnInit, OnDestroy {
public now = Date.now();
public selectedVertex: NodesItemCorrect | null;
public flameGraph = {} as JobFlameGraph;
+ public allSubtasks = 'all';
+ public listOfRunningSubtasks: string[] = [this.allSubtasks];
public graphType = FlameGraphType.ON_CPU;
+ public subtaskIndex = this.allSubtasks;
private readonly destroy$ = new Subject<void>();
@@ -67,6 +74,7 @@ export class JobOverviewDrawerFlameGraphComponent implements OnInit, OnDestroy {
) {}
public ngOnInit(): void {
+ this.requestRunningSubtasks();
this.requestFlameGraph(this.graphType);
}
@@ -76,11 +84,22 @@ export class JobOverviewDrawerFlameGraphComponent implements OnInit, OnDestroy {
}
private requestFlameGraph(graphType: FlameGraphType): void {
+ this.flameGraph = {} as JobFlameGraph;
this.jobLocalService
.jobWithVertexChanges()
.pipe(
tap(data => (this.selectedVertex = data.vertex)),
- mergeMap(data => this.jobService.loadOperatorFlameGraph(data.job.jid, data.vertex!.id, graphType)),
+ mergeMap(data => {
+ if (this.subtaskIndex === this.allSubtasks) {
+ return this.jobService.loadOperatorFlameGraph(data.job.jid, data.vertex!.id, graphType);
+ }
+ return this.jobService.loadOperatorFlameGraphForSingleSubtask(
+ data.job.jid,
+ data.vertex!.id,
+ graphType,
+ this.subtaskIndex
+ );
+ }),
takeUntil(this.destroy$)
)
.subscribe(
@@ -100,9 +119,43 @@ export class JobOverviewDrawerFlameGraphComponent implements OnInit, OnDestroy {
);
}
+ private requestRunningSubtasks(): void {
+ this.jobLocalService
+ .jobWithVertexChanges()
+ .pipe(
+ tap(data => (this.selectedVertex = data.vertex)),
+ mergeMap(data => {
+ return this.jobService.loadSubTasks(data.job.jid, data.vertex!.id);
+ }),
+ takeUntil(this.destroy$)
+ )
+ .subscribe(
+ data => {
+ const runningSubtasks = data?.subtasks
+ .filter(subtaskInfo => subtaskInfo.status === 'RUNNING')
+ .map(subtaskInfo => subtaskInfo.subtask.toString());
+ if (isNil(runningSubtasks)) {
+ return;
+ }
+ this.listOfRunningSubtasks = [this.allSubtasks, ...runningSubtasks];
+ this.cdr.markForCheck();
+ },
+ () => {
+ this.listOfRunningSubtasks = [this.allSubtasks];
+ this.cdr.markForCheck();
+ }
+ );
+ }
+
+ public selectSubtask(subtaskIndex: string): void {
+ this.destroy$.next();
+ this.subtaskIndex = subtaskIndex;
+ this.cdr.markForCheck();
+ this.requestFlameGraph(this.graphType);
+ }
+
public selectFrameGraphType(graphType: FlameGraphType): void {
this.destroy$.next();
- this.flameGraph = {} as JobFlameGraph;
this.requestFlameGraph(graphType);
}
}
diff --git a/flink-runtime-web/web-dashboard/src/app/services/job.service.ts b/flink-runtime-web/web-dashboard/src/app/services/job.service.ts
index c3c92a25e76..21eb73bc3f4 100644
--- a/flink-runtime-web/web-dashboard/src/app/services/job.service.ts
+++ b/flink-runtime-web/web-dashboard/src/app/services/job.service.ts
@@ -122,6 +122,17 @@ export class JobService {
);
}
+ public loadOperatorFlameGraphForSingleSubtask(
+ jobId: string,
+ vertexId: string,
+ type: string,
+ subtaskIndex: string
+ ): Observable<JobFlameGraph> {
+ return this.httpClient.get<JobFlameGraph>(
+ `${this.configService.BASE_URL}/jobs/${jobId}/vertices/${vertexId}/flamegraph?type=${type}&subtaskindex=${subtaskIndex}`
+ );
+ }
+
public loadSubTasks(jobId: string, vertexId: string): Observable<JobVertexSubTaskDetail> {
return this.httpClient.get<JobVertexSubTaskDetail>(
`${this.configService.BASE_URL}/jobs/${jobId}/vertices/${vertexId}`
[flink] 01/03: [FLINK-30185][rest][refactor] Distinguish which ExecutionAttemptID each ThreadInfoSample belongs to
Posted by xt...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b9ffa7ac1ff5db3e9e9189d242149a01e543256d
Author: 1996fanrui <19...@gmail.com>
AuthorDate: Wed Dec 21 17:05:08 2022 +0800
[FLINK-30185][rest][refactor] Distinguish which ExecutionAttemptID each ThreadInfoSample belongs to
---
.../runtime/messages/TaskThreadInfoResponse.java | 8 ++-
.../flink/runtime/messages/ThreadInfoSample.java | 15 ++--
.../flink/runtime/taskexecutor/TaskExecutor.java | 16 +++--
.../taskexecutor/ThreadInfoSampleService.java | 49 +++++++------
.../org/apache/flink/runtime/util/JvmUtils.java | 6 +-
.../threadinfo/JobVertexThreadInfoStats.java | 10 +--
.../threadinfo/ThreadInfoRequestCoordinator.java | 15 ++--
.../taskexecutor/ThreadInfoSampleServiceTest.java | 82 ++++++++++++++++------
.../threadinfo/JobVertexThreadInfoTrackerTest.java | 48 +++++--------
.../ThreadInfoRequestCoordinatorTest.java | 30 +++++---
10 files changed, 170 insertions(+), 109 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/TaskThreadInfoResponse.java b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/TaskThreadInfoResponse.java
index fd2987c1edc..525131f9d0c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/TaskThreadInfoResponse.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/TaskThreadInfoResponse.java
@@ -18,24 +18,26 @@
package org.apache.flink.runtime.messages;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.util.Preconditions;
import java.io.Serializable;
import java.util.Collection;
+import java.util.Map;
/** Response to the request to collect thread details samples. */
public class TaskThreadInfoResponse implements Serializable {
private static final long serialVersionUID = -4786454630050578031L;
- private final Collection<ThreadInfoSample> samples;
+ private final Map<ExecutionAttemptID, Collection<ThreadInfoSample>> samples;
/**
* Creates a response to the request to collect thread details samples.
*
* @param samples Thread info samples.
*/
- public TaskThreadInfoResponse(Collection<ThreadInfoSample> samples) {
+ public TaskThreadInfoResponse(Map<ExecutionAttemptID, Collection<ThreadInfoSample>> samples) {
this.samples = Preconditions.checkNotNull(samples);
}
@@ -44,7 +46,7 @@ public class TaskThreadInfoResponse implements Serializable {
*
* @return A collection of thread info samples for a particular execution attempt (Task)
*/
- public Collection<ThreadInfoSample> getSamples() {
+ public Map<ExecutionAttemptID, Collection<ThreadInfoSample>> getSamples() {
return samples;
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/ThreadInfoSample.java b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/ThreadInfoSample.java
index 2de541a10b9..e13d948c0e1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/messages/ThreadInfoSample.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/messages/ThreadInfoSample.java
@@ -23,6 +23,7 @@ import javax.annotation.Nullable;
import java.io.Serializable;
import java.lang.management.ThreadInfo;
import java.util.Collection;
+import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -63,13 +64,15 @@ public class ThreadInfoSample implements Serializable {
* @param threadInfos the collection of {@link ThreadInfo}.
* @return the collection of the corresponding {@link ThreadInfoSample}s.
*/
- public static Collection<ThreadInfoSample> from(Collection<ThreadInfo> threadInfos) {
+ public static Map<Long, ThreadInfoSample> from(Collection<ThreadInfo> threadInfos) {
return threadInfos.stream()
- .map(
- threadInfo ->
- new ThreadInfoSample(
- threadInfo.getThreadState(), threadInfo.getStackTrace()))
- .collect(Collectors.toList());
+ .collect(
+ Collectors.toMap(
+ ThreadInfo::getThreadId,
+ threadInfo ->
+ new ThreadInfoSample(
+ threadInfo.getThreadState(),
+ threadInfo.getStackTrace())));
}
public Thread.State getThreadState() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index bd8941622f0..218ecdaab46 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -571,11 +571,17 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
}
}
- Collection<SampleableTask> sampleableTasks =
- tasks.stream().map(SampleableTaskAdapter::fromTask).collect(Collectors.toList());
-
- final CompletableFuture<Collection<ThreadInfoSample>> stackTracesFuture =
- threadInfoSampleService.requestThreadInfoSamples(sampleableTasks, requestParams);
+ Map<Long, ExecutionAttemptID> sampleableTasks =
+ tasks.stream()
+ .collect(
+ Collectors.toMap(
+ task -> task.getExecutingThread().getId(),
+ Task::getExecutionId));
+
+ final CompletableFuture<Map<ExecutionAttemptID, Collection<ThreadInfoSample>>>
+ stackTracesFuture =
+ threadInfoSampleService.requestThreadInfoSamples(
+ sampleableTasks, requestParams);
return stackTracesFuture.thenApply(TaskThreadInfoResponse::new);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/ThreadInfoSampleService.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/ThreadInfoSampleService.java
index 69c7861d76c..7e29ba394e4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/ThreadInfoSampleService.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/ThreadInfoSampleService.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.taskexecutor;
+import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.messages.ThreadInfoSample;
import org.apache.flink.runtime.util.JvmUtils;
import org.apache.flink.runtime.webmonitor.threadinfo.ThreadInfoSamplesRequest;
@@ -28,6 +29,8 @@ import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@@ -49,52 +52,55 @@ class ThreadInfoSampleService implements Closeable {
* Returns a future that completes with a given number of thread info samples for a set of task
* threads.
*
- * @param tasks The tasks to be sampled.
+ * @param threads the map key is thread id, the map value is the ExecutionAttemptID.
* @param requestParams Parameters of the sampling request.
* @return A future containing the stack trace samples.
*/
- public CompletableFuture<Collection<ThreadInfoSample>> requestThreadInfoSamples(
- final Collection<? extends SampleableTask> tasks,
- final ThreadInfoSamplesRequest requestParams) {
- checkNotNull(tasks, "task must not be null");
+ public CompletableFuture<Map<ExecutionAttemptID, Collection<ThreadInfoSample>>>
+ requestThreadInfoSamples(
+ Map<Long, ExecutionAttemptID> threads,
+ final ThreadInfoSamplesRequest requestParams) {
+ checkNotNull(threads, "threads must not be null");
checkNotNull(requestParams, "requestParams must not be null");
- CompletableFuture<Collection<ThreadInfoSample>> resultFuture = new CompletableFuture<>();
+ CompletableFuture<Map<ExecutionAttemptID, Collection<ThreadInfoSample>>> resultFuture =
+ new CompletableFuture<>();
scheduledExecutor.execute(
() ->
requestThreadInfoSamples(
- tasks,
+ threads,
requestParams.getNumSamples(),
requestParams.getDelayBetweenSamples(),
requestParams.getMaxStackTraceDepth(),
- new ArrayList<>(requestParams.getNumSamples()),
+ new HashMap<>(threads.size()),
resultFuture));
return resultFuture;
}
private void requestThreadInfoSamples(
- final Collection<? extends SampleableTask> tasks,
+ Map<Long, ExecutionAttemptID> threads,
final int numSamples,
final Duration delayBetweenSamples,
final int maxStackTraceDepth,
- final Collection<ThreadInfoSample> currentTraces,
- final CompletableFuture<Collection<ThreadInfoSample>> resultFuture) {
+ final Map<ExecutionAttemptID, Collection<ThreadInfoSample>> currentTraces,
+ final CompletableFuture<Map<ExecutionAttemptID, Collection<ThreadInfoSample>>>
+ resultFuture) {
- final Collection<Long> threadIds =
- tasks.stream()
- .map(t -> t.getExecutingThread().getId())
- .collect(Collectors.toList());
-
- final Collection<ThreadInfoSample> threadInfoSample =
- JvmUtils.createThreadInfoSample(threadIds, maxStackTraceDepth);
+ final Map<Long, ThreadInfoSample> threadInfoSample =
+ JvmUtils.createThreadInfoSample(threads.keySet(), maxStackTraceDepth);
if (!threadInfoSample.isEmpty()) {
- currentTraces.addAll(threadInfoSample);
+ for (Map.Entry<Long, ThreadInfoSample> entry : threadInfoSample.entrySet()) {
+ ExecutionAttemptID executionAttemptID = threads.get(entry.getKey());
+ Collection<ThreadInfoSample> threadInfoSamples =
+ currentTraces.computeIfAbsent(executionAttemptID, key -> new ArrayList<>());
+ threadInfoSamples.add(entry.getValue());
+ }
if (numSamples > 1) {
scheduledExecutor.schedule(
() ->
requestThreadInfoSamples(
- tasks,
+ threads,
numSamples - 1,
delayBetweenSamples,
maxStackTraceDepth,
@@ -111,8 +117,7 @@ class ThreadInfoSampleService implements Closeable {
resultFuture.complete(currentTraces);
} else {
final String ids =
- tasks.stream()
- .map(SampleableTask::getExecutionId)
+ threads.values().stream()
.map(e -> e == null ? "unknown" : e.toString())
.collect(Collectors.joining(", ", "[", "]"));
resultFuture.completeExceptionally(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/JvmUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/JvmUtils.java
index 128fed7f8bb..89a8f8fe0cc 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/util/JvmUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/JvmUtils.java
@@ -29,6 +29,7 @@ import java.lang.management.ThreadMXBean;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -70,9 +71,10 @@ public final class JvmUtils {
*
* @param threadIds The IDs of the threads to create the thread dump for.
* @param maxStackTraceDepth The maximum number of entries in the stack trace to be collected.
- * @return The thread information for the requested thread IDs.
+ * @return The map key is the thread id, the map value is the thread information for the
+ * requested thread IDs.
*/
- public static Collection<ThreadInfoSample> createThreadInfoSample(
+ public static Map<Long, ThreadInfoSample> createThreadInfoSample(
Collection<Long> threadIds, int maxStackTraceDepth) {
ThreadMXBean threadMxBean = ManagementFactory.getThreadMXBean();
long[] threadIdsArray = threadIds.stream().mapToLong(l -> l).toArray();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoStats.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoStats.java
index 3299bdc9df5..f035bebc67b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoStats.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoStats.java
@@ -22,8 +22,6 @@ import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.messages.ThreadInfoSample;
import org.apache.flink.runtime.webmonitor.stats.Statistics;
-import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableSet;
-
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
@@ -46,8 +44,7 @@ public class JobVertexThreadInfoStats implements Statistics {
private final long endTime;
/** Map of thread info samples by execution ID. */
- private final Map<ImmutableSet<ExecutionAttemptID>, Collection<ThreadInfoSample>>
- samplesBySubtask;
+ private final Map<ExecutionAttemptID, Collection<ThreadInfoSample>> samplesBySubtask;
/**
* Creates a thread details sample.
@@ -61,7 +58,7 @@ public class JobVertexThreadInfoStats implements Statistics {
int requestId,
long startTime,
long endTime,
- Map<ImmutableSet<ExecutionAttemptID>, Collection<ThreadInfoSample>> samplesBySubtask) {
+ Map<ExecutionAttemptID, Collection<ThreadInfoSample>> samplesBySubtask) {
checkArgument(requestId >= 0, "Negative request ID");
checkArgument(startTime >= 0, "Negative start time");
@@ -106,8 +103,7 @@ public class JobVertexThreadInfoStats implements Statistics {
*
* @return Map of thread info samples by task (execution ID)
*/
- public Map<ImmutableSet<ExecutionAttemptID>, Collection<ThreadInfoSample>>
- getSamplesBySubtask() {
+ public Map<ExecutionAttemptID, Collection<ThreadInfoSample>> getSamplesBySubtask() {
return samplesBySubtask;
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinator.java
index 60550e5ed44..eb0929124d5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinator.java
@@ -30,6 +30,7 @@ import org.apache.flink.shaded.guava30.com.google.common.collect.ImmutableSet;
import java.time.Duration;
import java.util.Collection;
+import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
@@ -41,7 +42,7 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
/** A coordinator for triggering and collecting thread info stats of running job vertex subtasks. */
public class ThreadInfoRequestCoordinator
extends TaskStatsRequestCoordinator<
- Collection<ThreadInfoSample>, JobVertexThreadInfoStats> {
+ Map<ExecutionAttemptID, Collection<ThreadInfoSample>>, JobVertexThreadInfoStats> {
/**
* Creates a new coordinator for the job.
@@ -159,7 +160,9 @@ public class ThreadInfoRequestCoordinator
// ------------------------------------------------------------------------
private static class PendingThreadInfoRequest
- extends PendingStatsRequest<Collection<ThreadInfoSample>, JobVertexThreadInfoStats> {
+ extends PendingStatsRequest<
+ Map<ExecutionAttemptID, Collection<ThreadInfoSample>>,
+ JobVertexThreadInfoStats> {
PendingThreadInfoRequest(
int requestId, Collection<? extends Set<ExecutionAttemptID>> tasksToCollect) {
@@ -168,8 +171,12 @@ public class ThreadInfoRequestCoordinator
@Override
protected JobVertexThreadInfoStats assembleCompleteStats(long endTime) {
- return new JobVertexThreadInfoStats(
- requestId, startTime, endTime, statsResultByTaskGroup);
+ HashMap<ExecutionAttemptID, Collection<ThreadInfoSample>> samples = new HashMap<>();
+ for (Map<ExecutionAttemptID, Collection<ThreadInfoSample>> map :
+ statsResultByTaskGroup.values()) {
+ samples.putAll(map);
+ }
+ return new JobVertexThreadInfoStats(requestId, startTime, endTime, samples);
}
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/ThreadInfoSampleServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/ThreadInfoSampleServiceTest.java
index 8928ba25812..6647f46e716 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/ThreadInfoSampleServiceTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/ThreadInfoSampleServiceTest.java
@@ -31,10 +31,12 @@ import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.Collection;
import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
+import java.util.stream.Collectors;
import static org.apache.flink.runtime.taskexecutor.IdleTestTask.executeWithTerminationGuarantee;
import static org.assertj.core.api.Assertions.assertThat;
@@ -75,17 +77,27 @@ public class ThreadInfoSampleServiceTest extends TestLogger {
tasks.add(new IdleTestTask());
tasks.add(new IdleTestTask());
Thread.sleep(2000);
- final Collection<ThreadInfoSample> threadInfoSamples =
+
+ Map<Long, ExecutionAttemptID> threads =
+ tasks.stream()
+ .collect(
+ Collectors.toMap(
+ task -> task.getExecutingThread().getId(),
+ IdleTestTask::getExecutionId));
+ final Map<ExecutionAttemptID, Collection<ThreadInfoSample>> threadInfoSamples =
threadInfoSampleService
- .requestThreadInfoSamples(tasks, requestParams)
+ .requestThreadInfoSamples(threads, requestParams)
.get();
- assertThat(threadInfoSamples).hasSize(NUMBER_OF_SAMPLES * 2);
-
- for (ThreadInfoSample sample : threadInfoSamples) {
- StackTraceElement[] traces = sample.getStackTrace();
- assertThat(traces).hasSizeLessThanOrEqualTo(MAX_STACK_TRACK_DEPTH);
+ int count = 0;
+ for (Collection<ThreadInfoSample> samples : threadInfoSamples.values()) {
+ for (ThreadInfoSample sample : samples) {
+ count++;
+ StackTraceElement[] traces = sample.getStackTrace();
+ assertThat(traces).hasSizeLessThanOrEqualTo(MAX_STACK_TRACK_DEPTH);
+ }
}
+ assertThat(count).isEqualTo(NUMBER_OF_SAMPLES * 2);
},
tasks);
}
@@ -97,15 +109,21 @@ public class ThreadInfoSampleServiceTest extends TestLogger {
executeWithTerminationGuarantee(
() -> {
tasks.add(new IdleTestTask());
- final Collection<ThreadInfoSample> threadInfoSamples1 =
+ Map<Long, ExecutionAttemptID> threads =
+ tasks.stream()
+ .collect(
+ Collectors.toMap(
+ task -> task.getExecutingThread().getId(),
+ IdleTestTask::getExecutionId));
+ final Map<ExecutionAttemptID, Collection<ThreadInfoSample>> threadInfoSamples1 =
threadInfoSampleService
- .requestThreadInfoSamples(tasks, requestParams)
+ .requestThreadInfoSamples(threads, requestParams)
.get();
- final Collection<ThreadInfoSample> threadInfoSamples2 =
+ final Map<ExecutionAttemptID, Collection<ThreadInfoSample>> threadInfoSamples2 =
threadInfoSampleService
.requestThreadInfoSamples(
- tasks,
+ threads,
new ThreadInfoSamplesRequest(
1,
NUMBER_OF_SAMPLES,
@@ -113,13 +131,17 @@ public class ThreadInfoSampleServiceTest extends TestLogger {
MAX_STACK_TRACK_DEPTH - 6))
.get();
- for (ThreadInfoSample sample : threadInfoSamples1) {
- assertThat(sample.getStackTrace())
- .hasSizeLessThanOrEqualTo(MAX_STACK_TRACK_DEPTH);
+ for (Collection<ThreadInfoSample> samples : threadInfoSamples1.values()) {
+ for (ThreadInfoSample sample : samples) {
+ assertThat(sample.getStackTrace())
+ .hasSizeLessThanOrEqualTo(MAX_STACK_TRACK_DEPTH);
+ }
}
- for (ThreadInfoSample sample : threadInfoSamples2) {
- assertThat(sample.getStackTrace()).hasSize(MAX_STACK_TRACK_DEPTH - 6);
+ for (Collection<ThreadInfoSample> samples : threadInfoSamples2.values()) {
+ for (ThreadInfoSample sample : samples) {
+ assertThat(sample.getStackTrace()).hasSize(MAX_STACK_TRACK_DEPTH - 6);
+ }
}
},
tasks);
@@ -134,8 +156,18 @@ public class ThreadInfoSampleServiceTest extends TestLogger {
executeWithTerminationGuarantee(
() -> {
tasks.add(new IdleTestTask());
+
+ Map<Long, ExecutionAttemptID> threads =
+ tasks.stream()
+ .collect(
+ Collectors.toMap(
+ task ->
+ task.getExecutingThread()
+ .getId(),
+ IdleTestTask
+ ::getExecutionId));
threadInfoSampleService.requestThreadInfoSamples(
- tasks,
+ threads,
new ThreadInfoSamplesRequest(
1,
-1,
@@ -153,8 +185,16 @@ public class ThreadInfoSampleServiceTest extends TestLogger {
throws ExecutionException, InterruptedException {
Set<SampleableTask> tasks = new HashSet<>();
tasks.add(new NotRunningTask());
- final CompletableFuture<Collection<ThreadInfoSample>> sampleFuture =
- threadInfoSampleService.requestThreadInfoSamples(tasks, requestParams);
+
+ Map<Long, ExecutionAttemptID> threads =
+ tasks.stream()
+ .collect(
+ Collectors.toMap(
+ task -> task.getExecutingThread().getId(),
+ SampleableTask::getExecutionId));
+ final CompletableFuture<Map<ExecutionAttemptID, Collection<ThreadInfoSample>>>
+ sampleFuture =
+ threadInfoSampleService.requestThreadInfoSamples(threads, requestParams);
assertThat(sampleFuture).failsWithin(Duration.ofSeconds(10));
assertThat(sampleFuture.handle((ignored, e) -> e).get())
@@ -163,12 +203,14 @@ public class ThreadInfoSampleServiceTest extends TestLogger {
private static class NotRunningTask implements SampleableTask {
+ private final ExecutionAttemptID executionId = ExecutionAttemptID.randomId();
+
public Thread getExecutingThread() {
return new Thread();
}
public ExecutionAttemptID getExecutionId() {
- return null;
+ return executionId;
}
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java
index f7d4ebfd288..d45aff28279 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/JobVertexThreadInfoTrackerTest.java
@@ -58,8 +58,6 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -74,6 +72,7 @@ import java.util.function.Function;
import java.util.stream.Collectors;
import static org.apache.flink.runtime.scheduler.SchedulerTestingUtils.createScheduler;
+import static org.apache.flink.util.Preconditions.checkState;
import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
/** Tests for the {@link JobVertexThreadInfoTracker}. */
@@ -90,7 +89,6 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
.collect(Collectors.toSet());
private static final JobID JOB_ID = new JobID();
- private static ThreadInfoSample threadInfoSample;
private static JobVertexThreadInfoStats threadInfoStatsDefaultSample;
private static final Duration CLEAN_UP_INTERVAL = Duration.ofSeconds(60);
@@ -110,13 +108,7 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
// Time gap determines endTime of stats, which controls if the "refresh" is triggered:
// now >= stats.getEndTime() + statsRefreshInterval
// Using a small gap to be able to test cache updates without much delay.
- threadInfoSample =
- JvmUtils.createThreadInfoSample(
- Thread.currentThread().getId(), MAX_STACK_TRACE_DEPTH)
- .get();
- threadInfoStatsDefaultSample =
- createThreadInfoStats(
- REQUEST_ID, SMALL_TIME_GAP, Collections.singletonList(threadInfoSample));
+ threadInfoStatsDefaultSample = createThreadInfoStats(REQUEST_ID, SMALL_TIME_GAP);
executor = Executors.newScheduledThreadPool(1);
}
@@ -136,8 +128,7 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
/** Tests that cached result is reused within refresh interval. */
@Test
public void testCachedStatsNotUpdatedWithinRefreshInterval() throws Exception {
- final JobVertexThreadInfoStats unusedThreadInfoStats =
- createThreadInfoStats(1, TIME_GAP, null);
+ final JobVertexThreadInfoStats unusedThreadInfoStats = createThreadInfoStats(1, TIME_GAP);
final JobVertexThreadInfoTracker<JobVertexThreadInfoStats> tracker =
createThreadInfoTracker(
@@ -162,10 +153,9 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
createThreadInfoStats(
Instant.now().minus(10, ChronoUnit.SECONDS),
REQUEST_ID,
- Duration.ofMillis(5),
- Collections.singletonList(threadInfoSample));
+ Duration.ofMillis(5));
final JobVertexThreadInfoStats threadInfoStatsAfterRefresh =
- createThreadInfoStats(1, TIME_GAP, Collections.singletonList(threadInfoSample));
+ createThreadInfoStats(1, TIME_GAP);
// register a CountDownLatch with the cache so we can await refresh of the entry
CountDownLatch cacheRefreshed = new CountDownLatch(1);
@@ -321,32 +311,28 @@ public class JobVertexThreadInfoTrackerTest extends TestLogger {
.build();
}
- private static JobVertexThreadInfoStats createThreadInfoStats(
- int requestId, Duration timeGap, List<ThreadInfoSample> threadInfoSamples) {
- return createThreadInfoStats(Instant.now(), requestId, timeGap, threadInfoSamples);
+ private static JobVertexThreadInfoStats createThreadInfoStats(int requestId, Duration timeGap) {
+ return createThreadInfoStats(Instant.now(), requestId, timeGap);
}
private static JobVertexThreadInfoStats createThreadInfoStats(
- Instant startTime,
- int requestId,
- Duration timeGap,
- List<ThreadInfoSample> threadInfoSamples) {
+ Instant startTime, int requestId, Duration timeGap) {
Instant endTime = startTime.plus(timeGap);
- final Map<ImmutableSet<ExecutionAttemptID>, Collection<ThreadInfoSample>>
- threadInfoRatiosByTask = new HashMap<>();
+ final Map<ExecutionAttemptID, Collection<ThreadInfoSample>> samples = new HashMap<>();
for (ExecutionVertex vertex : TASK_VERTICES) {
- Set<ExecutionAttemptID> attemptIds = new HashSet<>();
- attemptIds.add(vertex.getCurrentExecutionAttempt().getAttemptId());
- threadInfoRatiosByTask.put(ImmutableSet.copyOf(attemptIds), threadInfoSamples);
+ Optional<ThreadInfoSample> threadInfoSample =
+ JvmUtils.createThreadInfoSample(
+ Thread.currentThread().getId(), MAX_STACK_TRACE_DEPTH);
+ checkState(threadInfoSample.isPresent(), "The threadInfoSample should be empty.");
+ samples.put(
+ vertex.getCurrentExecutionAttempt().getAttemptId(),
+ Collections.singletonList(threadInfoSample.get()));
}
return new JobVertexThreadInfoStats(
- requestId,
- startTime.toEpochMilli(),
- endTime.toEpochMilli(),
- threadInfoRatiosByTask);
+ requestId, startTime.toEpochMilli(), endTime.toEpochMilli(), samples);
}
private static ExecutionJobVertex createExecutionJobVertex() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinatorTest.java
index caaf52bbf39..147195ea859 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/webmonitor/threadinfo/ThreadInfoRequestCoordinatorTest.java
@@ -38,6 +38,7 @@ import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -115,7 +116,7 @@ public class ThreadInfoRequestCoordinatorTest extends TestLogger {
// verify the request result
assertThat(threadInfoStats.getRequestId()).isEqualTo(0);
- Map<ImmutableSet<ExecutionAttemptID>, Collection<ThreadInfoSample>> samplesBySubtask =
+ Map<ExecutionAttemptID, Collection<ThreadInfoSample>> samplesBySubtask =
threadInfoStats.getSamplesBySubtask();
for (Collection<ThreadInfoSample> result : samplesBySubtask.values()) {
@@ -237,15 +238,26 @@ public class ThreadInfoRequestCoordinatorTest extends TestLogger {
() -> {
tasks.add(new IdleTestTask());
tasks.add(new IdleTestTask());
- // Thread.sleep(100);
- List<Long> threadIds =
+ Map<Long, ExecutionAttemptID> threads =
tasks.stream()
- .map(t -> t.getExecutingThread().getId())
- .collect(Collectors.toList());
- Collection<ThreadInfoSample> threadInfoSample =
- JvmUtils.createThreadInfoSample(threadIds, 100);
- responseFuture.complete(
- new TaskThreadInfoResponse(new ArrayList<>(threadInfoSample)));
+ .collect(
+ Collectors.toMap(
+ task ->
+ task.getExecutingThread()
+ .getId(),
+ IdleTestTask::getExecutionId));
+
+ Map<ExecutionAttemptID, Collection<ThreadInfoSample>> threadInfoSample =
+ JvmUtils.createThreadInfoSample(threads.keySet(), 100)
+ .entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ entry -> threads.get(entry.getKey()),
+ entry ->
+ Collections.singletonList(
+ entry.getValue())));
+
+ responseFuture.complete(new TaskThreadInfoResponse(threadInfoSample));
},
tasks);