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 2022/04/29 08:46:11 UTC
[flink] branch release-1.15 updated: [FLINK-27420] Recreate metric groups for each new RM to avoid metric loss
This is an automated email from the ASF dual-hosted git repository.
xtsong pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.15 by this push:
new e0c82d6d528 [FLINK-27420] Recreate metric groups for each new RM to avoid metric loss
e0c82d6d528 is described below
commit e0c82d6d52871dbbea70c9b41384d2d33179bec0
Author: Ben Augarten <ba...@gmail.com>
AuthorDate: Wed Apr 27 13:45:12 2022 -0700
[FLINK-27420] Recreate metric groups for each new RM to avoid metric loss
This closes #19607
---
.../resourcemanager/ResourceManagerFactory.java | 15 ++---
.../ResourceManagerProcessContext.java | 23 ++++---
.../ResourceManagerServiceImplTest.java | 70 ++++++++++++++++++++++
3 files changed, 87 insertions(+), 21 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerFactory.java
index 893ad9b33e3..e661816d8ce 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerFactory.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerFactory.java
@@ -62,11 +62,6 @@ public abstract class ResourceManagerFactory<T extends ResourceIDRetrievable> {
Executor ioExecutor)
throws ConfigurationException {
- final ResourceManagerMetricGroup resourceManagerMetricGroup =
- ResourceManagerMetricGroup.create(metricRegistry, hostname);
- final SlotManagerMetricGroup slotManagerMetricGroup =
- SlotManagerMetricGroup.create(metricRegistry, hostname);
-
final Configuration runtimeServicesAndRmConfig =
getEffectiveConfigurationForResourceManagerAndRuntimeServices(configuration);
@@ -86,8 +81,8 @@ public abstract class ResourceManagerFactory<T extends ResourceIDRetrievable> {
fatalErrorHandler,
clusterInformation,
webInterfaceUrl,
- resourceManagerMetricGroup,
- slotManagerMetricGroup,
+ metricRegistry,
+ hostname,
ioExecutor);
}
@@ -99,7 +94,8 @@ public abstract class ResourceManagerFactory<T extends ResourceIDRetrievable> {
context.getRmRuntimeServicesConfig(),
context.getRpcService(),
context.getHighAvailabilityServices(),
- context.getSlotManagerMetricGroup());
+ SlotManagerMetricGroup.create(
+ context.getMetricRegistry(), context.getHostname()));
return createResourceManager(
context.getRmConfig(),
@@ -110,7 +106,8 @@ public abstract class ResourceManagerFactory<T extends ResourceIDRetrievable> {
context.getFatalErrorHandler(),
context.getClusterInformation(),
context.getWebInterfaceUrl(),
- context.getResourceManagerMetricGroup(),
+ ResourceManagerMetricGroup.create(
+ context.getMetricRegistry(), context.getHostname()),
resourceManagerRuntimeServices,
context.getIoExecutor());
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerProcessContext.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerProcessContext.java
index 994947de407..e1e751fda1b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerProcessContext.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerProcessContext.java
@@ -23,8 +23,7 @@ import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.entrypoint.ClusterInformation;
import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
-import org.apache.flink.runtime.metrics.groups.ResourceManagerMetricGroup;
-import org.apache.flink.runtime.metrics.groups.SlotManagerMetricGroup;
+import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcService;
@@ -50,8 +49,8 @@ public class ResourceManagerProcessContext {
private final FatalErrorHandler fatalErrorHandler;
private final ClusterInformation clusterInformation;
@Nullable private final String webInterfaceUrl;
- private final ResourceManagerMetricGroup resourceManagerMetricGroup;
- private final SlotManagerMetricGroup slotManagerMetricGroup;
+ private final MetricRegistry metricRegistry;
+ private final String hostname;
private final Executor ioExecutor;
public ResourceManagerProcessContext(
@@ -64,8 +63,8 @@ public class ResourceManagerProcessContext {
FatalErrorHandler fatalErrorHandler,
ClusterInformation clusterInformation,
@Nullable String webInterfaceUrl,
- ResourceManagerMetricGroup resourceManagerMetricGroup,
- SlotManagerMetricGroup slotManagerMetricGroup,
+ MetricRegistry metricRegistry,
+ String hostname,
Executor ioExecutor) {
this.rmConfig = checkNotNull(rmConfig);
this.resourceId = checkNotNull(resourceId);
@@ -75,8 +74,8 @@ public class ResourceManagerProcessContext {
this.heartbeatServices = checkNotNull(heartbeatServices);
this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
this.clusterInformation = checkNotNull(clusterInformation);
- this.resourceManagerMetricGroup = checkNotNull(resourceManagerMetricGroup);
- this.slotManagerMetricGroup = checkNotNull(slotManagerMetricGroup);
+ this.metricRegistry = checkNotNull(metricRegistry);
+ this.hostname = checkNotNull(hostname);
this.ioExecutor = checkNotNull(ioExecutor);
this.webInterfaceUrl = webInterfaceUrl;
@@ -119,12 +118,12 @@ public class ResourceManagerProcessContext {
return webInterfaceUrl;
}
- public ResourceManagerMetricGroup getResourceManagerMetricGroup() {
- return resourceManagerMetricGroup;
+ public MetricRegistry getMetricRegistry() {
+ return metricRegistry;
}
- public SlotManagerMetricGroup getSlotManagerMetricGroup() {
- return slotManagerMetricGroup;
+ public String getHostname() {
+ return hostname;
}
public Executor getIoExecutor() {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImplTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImplTest.java
index 3d90b203296..1f5bef00718 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImplTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServiceImplTest.java
@@ -27,6 +27,7 @@ import org.apache.flink.runtime.heartbeat.HeartbeatServices;
import org.apache.flink.runtime.heartbeat.TestingHeartbeatServices;
import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
+import org.apache.flink.runtime.metrics.MetricNames;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
import org.apache.flink.runtime.rpc.RpcUtils;
@@ -34,14 +35,20 @@ import org.apache.flink.runtime.rpc.TestingRpcService;
import org.apache.flink.runtime.util.TestingFatalErrorHandler;
import org.apache.flink.util.TestLogger;
+import org.assertj.core.util.Sets;
import org.junit.After;
import org.junit.AfterClass;
+import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeoutException;
@@ -492,6 +499,62 @@ public class ResourceManagerServiceImplTest extends TestLogger {
deregisterApplicationFuture.get(TIMEOUT.getSize(), TIMEOUT.getUnit());
}
+ @Test
+ public void grantAndRevokeLeadership_verifyMetrics() throws Exception {
+ final Set<String> registeredMetrics = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ TestingMetricRegistry metricRegistry =
+ TestingMetricRegistry.builder()
+ .setRegisterConsumer((a, b, c) -> registeredMetrics.add(b))
+ .setUnregisterConsumer((a, b, c) -> registeredMetrics.remove(b))
+ .build();
+
+ final TestingResourceManagerFactory rmFactory = rmFactoryBuilder.build();
+ resourceManagerService =
+ ResourceManagerServiceImpl.create(
+ rmFactory,
+ new Configuration(),
+ ResourceID.generate(),
+ rpcService,
+ haService,
+ heartbeatServices,
+ delegationTokenManager,
+ fatalErrorHandler,
+ clusterInformation,
+ null,
+ metricRegistry,
+ "localhost",
+ ForkJoinPool.commonPool());
+ resourceManagerService.start();
+
+ Assert.assertEquals(0, registeredMetrics.size());
+ // grant leadership
+ leaderElectionService.isLeader(UUID.randomUUID());
+
+ assertRmStarted();
+ Set<String> expectedMetrics =
+ Sets.set(
+ MetricNames.NUM_REGISTERED_TASK_MANAGERS,
+ MetricNames.TASK_SLOTS_TOTAL,
+ MetricNames.TASK_SLOTS_AVAILABLE);
+ Assert.assertTrue(
+ "Expected RM to register leader metrics",
+ registeredMetrics.containsAll(expectedMetrics));
+
+ // revoke leadership, block until old rm is terminated
+ revokeLeadership();
+
+ Set<String> intersection = new HashSet<>(registeredMetrics);
+ intersection.retainAll(expectedMetrics);
+ Assert.assertTrue("Expected RM to unregister leader metrics", intersection.isEmpty());
+
+ leaderElectionService.isLeader(UUID.randomUUID());
+
+ assertRmStarted();
+ Assert.assertTrue(
+ "Expected RM to re-register leader metrics",
+ registeredMetrics.containsAll(expectedMetrics));
+ }
+
private static void blockOnFuture(CompletableFuture<?> future) {
try {
future.get();
@@ -513,4 +576,11 @@ public class ResourceManagerServiceImplTest extends TestLogger {
private void assertRmStarted() throws Exception {
leaderElectionService.getConfirmationFuture().get(TIMEOUT.getSize(), TIMEOUT.getUnit());
}
+
+ private void revokeLeadership() {
+ ResourceManager<?> leaderResourceManager =
+ resourceManagerService.getLeaderResourceManager();
+ leaderElectionService.notLeader();
+ blockOnFuture(leaderResourceManager.getTerminationFuture());
+ }
}