[FLINK-9016] [flip6] Properly unregister jobs from JobMetricGroup

This commit properly removes jobs from the JobMetricGroup once a job has reached a terminal state.
apache · Mar 18, 2018 · c4a1d09 · c4a1d09
1 parent 91d346e
commit c4a1d09
Show file tree

Hide file tree

Showing 13 changed files with 122 additions and 91 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -49,7 +49,8 @@
 import org.apache.flink.runtime.messages.webmonitor.JobDetails;
 import org.apache.flink.runtime.messages.webmonitor.JobsOverview;
 import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
-import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
+import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
 import org.apache.flink.runtime.resourcemanager.ResourceOverview;
 import org.apache.flink.runtime.rest.handler.legacy.backpressure.OperatorBackPressureStatsResponse;
@@ -97,7 +98,6 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
 	private final JobManagerSharedServices jobManagerSharedServices;
 	private final HeartbeatServices heartbeatServices;
 	private final BlobServer blobServer;
-	private final MetricRegistry metricRegistry;
 
 	private final FatalErrorHandler fatalErrorHandler;
 
@@ -109,6 +109,11 @@ public abstract class Dispatcher extends FencedRpcEndpoint<DispatcherId> impleme
 
 	private final JobManagerRunnerFactory jobManagerRunnerFactory;
 
+	private final JobManagerMetricGroup jobManagerMetricGroup;
+
+	@Nullable
+	private final String metricQueryServicePath;
+
 	@Nullable
 	protected final String restAddress;
 
@@ -123,7 +128,8 @@ public Dispatcher(
 			ResourceManagerGateway resourceManagerGateway,
 			BlobServer blobServer,
 			HeartbeatServices heartbeatServices,
-			MetricRegistry metricRegistry,
+			JobManagerMetricGroup jobManagerMetricGroup,
+			@Nullable String metricServiceQueryPath,
 			ArchivedExecutionGraphStore archivedExecutionGraphStore,
 			JobManagerRunnerFactory jobManagerRunnerFactory,
 			FatalErrorHandler fatalErrorHandler,
@@ -135,9 +141,10 @@ public Dispatcher(
 		this.resourceManagerGateway = Preconditions.checkNotNull(resourceManagerGateway);
 		this.heartbeatServices = Preconditions.checkNotNull(heartbeatServices);
 		this.blobServer = Preconditions.checkNotNull(blobServer);
-		this.metricRegistry = Preconditions.checkNotNull(metricRegistry);
 		this.fatalErrorHandler = Preconditions.checkNotNull(fatalErrorHandler);
 		this.submittedJobGraphStore = Preconditions.checkNotNull(submittedJobGraphStore);
+		this.jobManagerMetricGroup = Preconditions.checkNotNull(jobManagerMetricGroup);
+		this.metricQueryServicePath = metricServiceQueryPath;
 
 		this.jobManagerSharedServices = JobManagerSharedServices.fromConfiguration(
 			configuration,
@@ -192,6 +199,8 @@ public CompletableFuture<Void> postStop() {
 					exception = ExceptionUtils.firstOrSuppressed(e, exception);
 				}
 
+				jobManagerMetricGroup.close();
+
 				if (exception != null) {
 					throw exception;
 				} else {
@@ -251,7 +260,8 @@ public CompletableFuture<Acknowledge> submitJob(JobGraph jobGraph, Time timeout)
 					heartbeatServices,
 					blobServer,
 					jobManagerSharedServices,
-					metricRegistry,
+					jobManagerMetricGroup.addJob(jobGraph),
+					metricQueryServicePath,
 					restAddress);
 
 				jobManagerRunner.getResultFuture().whenCompleteAsync(
@@ -464,8 +474,6 @@ public CompletableFuture<JobResult> requestJobResult(JobID jobId, Time timeout)
 
 	@Override
 	public CompletableFuture<Collection<String>> requestMetricQueryServicePaths(Time timeout) {
-		final String metricQueryServicePath = metricRegistry.getMetricQueryServicePath();
-
 		if (metricQueryServicePath != null) {
 			return CompletableFuture.completedFuture(Collections.singleton(metricQueryServicePath));
 		} else {
@@ -513,6 +521,8 @@ private void removeJob(JobID jobId, boolean cleanupHA) throws Exception {
 			registerOrphanedJobManagerTerminationFuture(jobManagerRunnerTerminationFuture);
 		}
 
+		jobManagerMetricGroup.removeJob(jobId);
+
 		if (cleanupHA) {
 			submittedJobGraphStore.removeJobGraph(jobId);
 		}
@@ -725,7 +735,8 @@ JobManagerRunner createJobManagerRunner(
 			HeartbeatServices heartbeatServices,
 			BlobServer blobServer,
 			JobManagerSharedServices jobManagerServices,
-			MetricRegistry metricRegistry,
+			JobManagerJobMetricGroup jobManagerJobMetricGroup,
+			@Nullable String metricQueryServicePath,
 			@Nullable String restAddress) throws Exception;
 	}
 
@@ -745,7 +756,8 @@ public JobManagerRunner createJobManagerRunner(
 				HeartbeatServices heartbeatServices,
 				BlobServer blobServer,
 				JobManagerSharedServices jobManagerServices,
-				MetricRegistry metricRegistry,
+				JobManagerJobMetricGroup jobManagerJobMetricGroup,
+				@Nullable String metricQueryServicePath,
 				@Nullable String restAddress) throws Exception {
 			return new JobManagerRunner(
 				resourceId,
@@ -756,7 +768,8 @@ public JobManagerRunner createJobManagerRunner(
 				heartbeatServices,
 				blobServer,
 				jobManagerServices,
-				metricRegistry,
+				jobManagerJobMetricGroup,
+				metricQueryServicePath,
 				restAddress);
 		}
 	}

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/MiniDispatcher.java
@@ -30,7 +30,7 @@
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobmaster.JobResult;
 import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.RpcService;
@@ -60,7 +60,8 @@ public MiniDispatcher(
 			ResourceManagerGateway resourceManagerGateway,
 			BlobServer blobServer,
 			HeartbeatServices heartbeatServices,
-			MetricRegistry metricRegistry,
+			JobManagerMetricGroup jobManagerMetricGroup,
+			@Nullable String metricQueryServicePath,
 			ArchivedExecutionGraphStore archivedExecutionGraphStore,
 			JobManagerRunnerFactory jobManagerRunnerFactory,
 			FatalErrorHandler fatalErrorHandler,
@@ -76,7 +77,8 @@ public MiniDispatcher(
 			resourceManagerGateway,
 			blobServer,
 			heartbeatServices,
-			metricRegistry,
+			jobManagerMetricGroup,
+			metricQueryServicePath,
 			archivedExecutionGraphStore,
 			jobManagerRunnerFactory,
 			fatalErrorHandler,

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/StandaloneDispatcher.java
@@ -24,7 +24,7 @@
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobmaster.JobMaster;
-import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
 import org.apache.flink.runtime.rpc.FatalErrorHandler;
 import org.apache.flink.runtime.rpc.RpcService;
@@ -45,7 +45,8 @@ public StandaloneDispatcher(
 			ResourceManagerGateway resourceManagerGateway,
 			BlobServer blobServer,
 			HeartbeatServices heartbeatServices,
-			MetricRegistry metricRegistry,
+			JobManagerMetricGroup jobManagerMetricGroup,
+			@Nullable String metricQueryServicePath,
 			ArchivedExecutionGraphStore archivedExecutionGraphStore,
 			JobManagerRunnerFactory jobManagerRunnerFactory,
 			FatalErrorHandler fatalErrorHandler,
@@ -59,7 +60,8 @@ public StandaloneDispatcher(
 			resourceManagerGateway,
 			blobServer,
 			heartbeatServices,
-			metricRegistry,
+			jobManagerMetricGroup,
+			metricQueryServicePath,
 			archivedExecutionGraphStore,
 			jobManagerRunnerFactory,
 			fatalErrorHandler,

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -51,6 +51,8 @@
 import org.apache.flink.runtime.metrics.MetricRegistry;
 import org.apache.flink.runtime.metrics.MetricRegistryConfiguration;
 import org.apache.flink.runtime.metrics.MetricRegistryImpl;
+import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
+import org.apache.flink.runtime.metrics.util.MetricUtils;
 import org.apache.flink.runtime.resourcemanager.ResourceManager;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
@@ -154,6 +156,9 @@ public abstract class ClusterEntrypoint implements FatalErrorHandler {
 	@GuardedBy("lock")
 	private ClusterInformation clusterInformation;
 
+	@GuardedBy("lock")
+	private JobManagerMetricGroup jobManagerMetricGroup;
+
 	protected ClusterEntrypoint(Configuration configuration) {
 		this.configuration = Preconditions.checkNotNull(configuration);
 		this.terminationFuture = new CompletableFuture<>();
@@ -327,14 +332,17 @@ protected void startClusterComponents(
 				clusterInformation,
 				webMonitorEndpoint.getRestAddress());
 
+			jobManagerMetricGroup = MetricUtils.instantiateJobManagerMetricGroup(metricRegistry, rpcService.getAddress());
+
 			dispatcher = createDispatcher(
 				configuration,
 				rpcService,
 				highAvailabilityServices,
 				resourceManager.getSelfGateway(ResourceManagerGateway.class),
 				blobServer,
 				heartbeatServices,
-				metricRegistry,
+				jobManagerMetricGroup,
+				metricRegistry.getMetricQueryServicePath(),
 				archivedExecutionGraphStore,
 				this,
 				webMonitorEndpoint.getRestAddress());
@@ -488,7 +496,19 @@ protected CompletableFuture<Void> stopClusterComponents() {
 				terminationFutures.add(FutureUtils.completedExceptionally(exception));
 			}
 
-			return FutureUtils.completeAll(terminationFutures);
+			final CompletableFuture<Void> componentTerminationFuture = FutureUtils.completeAll(terminationFutures);
+
+			if (jobManagerMetricGroup != null) {
+				return FutureUtils.runAfterwards(
+					componentTerminationFuture,
+					() -> {
+						synchronized (lock) {
+							jobManagerMetricGroup.close();
+						}
+					});
+			} else {
+				return componentTerminationFuture;
+			}
 		}
 	}
 
@@ -567,7 +587,8 @@ protected abstract Dispatcher createDispatcher(
 		ResourceManagerGateway resourceManagerGateway,
 		BlobServer blobServer,
 		HeartbeatServices heartbeatServices,
-		MetricRegistry metricRegistry,
+		JobManagerMetricGroup jobManagerMetricGroup,
+		@Nullable String metricQueryServicePath,
 		ArchivedExecutionGraphStore archivedExecutionGraphStore,
 		FatalErrorHandler fatalErrorHandler,
 		@Nullable String restAddress) throws Exception;

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/JobClusterEntrypoint.java
@@ -32,7 +32,7 @@
 import org.apache.flink.runtime.jobgraph.JobGraph;
 import org.apache.flink.runtime.jobmaster.MiniDispatcherRestEndpoint;
 import org.apache.flink.runtime.leaderelection.LeaderElectionService;
-import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
 import org.apache.flink.runtime.rest.RestServerEndpointConfiguration;
 import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
@@ -95,7 +95,8 @@ protected Dispatcher createDispatcher(
 			ResourceManagerGateway resourceManagerGateway,
 			BlobServer blobServer,
 			HeartbeatServices heartbeatServices,
-			MetricRegistry metricRegistry,
+			JobManagerMetricGroup jobManagerMetricGroup,
+			@Nullable String metricQueryServicePath,
 			ArchivedExecutionGraphStore archivedExecutionGraphStore,
 			FatalErrorHandler fatalErrorHandler,
 			@Nullable String restAddress) throws Exception {
@@ -114,7 +115,8 @@ protected Dispatcher createDispatcher(
 			resourceManagerGateway,
 			blobServer,
 			heartbeatServices,
-			metricRegistry,
+			jobManagerMetricGroup,
+			metricQueryServicePath,
 			archivedExecutionGraphStore,
 			Dispatcher.DefaultJobManagerRunnerFactory.INSTANCE,
 			fatalErrorHandler,

diff --git a/...k-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java b/...k-runtime/src/main/java/org/apache/flink/runtime/entrypoint/SessionClusterEntrypoint.java
@@ -34,7 +34,7 @@
 import org.apache.flink.runtime.heartbeat.HeartbeatServices;
 import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
 import org.apache.flink.runtime.leaderelection.LeaderElectionService;
-import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.groups.JobManagerMetricGroup;
 import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
 import org.apache.flink.runtime.rest.RestServerEndpointConfiguration;
 import org.apache.flink.runtime.rest.handler.RestHandlerConfiguration;
@@ -104,16 +104,17 @@ protected DispatcherRestEndpoint createRestEndpoint(
 
 	@Override
 	protected Dispatcher createDispatcher(
-		Configuration configuration,
+			Configuration configuration,
-		RpcService rpcService,
+			RpcService rpcService,
-		HighAvailabilityServices highAvailabilityServices,
+			HighAvailabilityServices highAvailabilityServices,
-		ResourceManagerGateway resourceManagerGateway,
+			ResourceManagerGateway resourceManagerGateway,
-		BlobServer blobServer,
+			BlobServer blobServer,
-		HeartbeatServices heartbeatServices,
+			HeartbeatServices heartbeatServices,
-		MetricRegistry metricRegistry,
+			JobManagerMetricGroup jobManagerMetricGroup,
-		ArchivedExecutionGraphStore archivedExecutionGraphStore,
+			@Nullable String metricQueryServicePath,
-		FatalErrorHandler fatalErrorHandler,
+			ArchivedExecutionGraphStore archivedExecutionGraphStore,
-		@Nullable String restAddress) throws Exception {
+			FatalErrorHandler fatalErrorHandler,
+			@Nullable String restAddress) throws Exception {
 
 		// create the default dispatcher
 		return new StandaloneDispatcher(
@@ -124,7 +125,8 @@ protected Dispatcher createDispatcher(
 			resourceManagerGateway,
 			blobServer,
 			heartbeatServices,
-			metricRegistry,
+			jobManagerMetricGroup,
+			metricQueryServicePath,
 			archivedExecutionGraphStore,
 			Dispatcher.DefaultJobManagerRunnerFactory.INSTANCE,
 			fatalErrorHandler,