Adding TestMetricsRegistry for UTs

Signed-off-by: Harsh Garg <[email protected]>
opensearch-project · Mar 26, 2024 · 48e38b0 · 48e38b0
1 parent 63eba93
commit 48e38b0
Show file tree

Hide file tree

Showing 15 changed files with 308 additions and 105 deletions.
diff --git a/server/src/main/java/org/opensearch/cluster/coordination/ClusterManagerMetrics.java b/server/src/main/java/org/opensearch/cluster/coordination/ClusterManagerMetrics.java
@@ -0,0 +1,52 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.cluster.coordination;
+
+import org.opensearch.telemetry.metrics.Counter;
+import org.opensearch.telemetry.metrics.MetricsRegistry;
+import org.opensearch.telemetry.metrics.tags.Tags;
+
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Class containing metrics (counters/latency) specific to ClusterManager.
+ */
+public final class ClusterManagerMetrics {
+
+ private static final String COUNTER_METRICS_UNIT = "1";
+
+ public final Counter leaderCheckFailureCounter;
+ public final Counter followerChecksFailureCounter;
+
+ public ClusterManagerMetrics(MetricsRegistry metricsRegistry) {
+ this.followerChecksFailureCounter = metricsRegistry.createCounter(
+ "followers.checker.failure.count",
+ "Counter for number of failed follower checks",
+ COUNTER_METRICS_UNIT
+ );
+ this.leaderCheckFailureCounter = metricsRegistry.createCounter(
+ "leader.checker.failure.count",
+ "Counter for number of failed leader checks",
+ COUNTER_METRICS_UNIT
+ );
+ }
+
+ public void incrementCounter(Counter counter, Double value) {
+ incrementCounter(counter, value, Optional.empty());
+ }
+
+ public void incrementCounter(Counter counter, Double value, Optional<Tags> tags) {
+ if (Objects.isNull(tags) || tags.isEmpty()) {
+ counter.add(value);
+ return;
+ }
+ counter.add(value, tags.get());
+ }
+}
diff --git a/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java b/server/src/main/java/org/opensearch/cluster/coordination/Coordinator.java
@@ -87,7 +87,6 @@
 import org.opensearch.monitor.NodeHealthService;
 import org.opensearch.monitor.StatusInfo;
 import org.opensearch.node.remotestore.RemoteStoreNodeService;
-import org.opensearch.telemetry.metrics.MetricsRegistry;
 import org.opensearch.threadpool.Scheduler;
 import org.opensearch.threadpool.ThreadPool.Names;
 import org.opensearch.transport.TransportService;
@@ -209,7 +208,7 @@ public Coordinator(
  NodeHealthService nodeHealthService,
  PersistedStateRegistry persistedStateRegistry,
  RemoteStoreNodeService remoteStoreNodeService,
- MetricsRegistry metricsRegistry
+ ClusterManagerMetrics clusterManagerMetrics
  ) {
  this.settings = settings;
  this.transportService = transportService;
@@ -269,7 +268,7 @@ public Coordinator(
  transportService,
  this::onLeaderFailure,
  nodeHealthService,
- metricsRegistry
+ clusterManagerMetrics
  );
  this.followersChecker = new FollowersChecker(
  settings,
@@ -278,7 +277,7 @@ public Coordinator(
  this::onFollowerCheckRequest,
  this::removeNode,
  nodeHealthService,
- metricsRegistry
+ clusterManagerMetrics
  );
  this.nodeRemovalExecutor = new NodeRemovalClusterStateTaskExecutor(allocationService, logger);
  this.clusterApplier = clusterApplier;

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/FollowersChecker.java b/server/src/main/java/org/opensearch/cluster/coordination/FollowersChecker.java
@@ -48,8 +48,6 @@
 import org.opensearch.core.transport.TransportResponse.Empty;
 import org.opensearch.monitor.NodeHealthService;
 import org.opensearch.monitor.StatusInfo;
-import org.opensearch.telemetry.metrics.Counter;
-import org.opensearch.telemetry.metrics.MetricsRegistry;
 import org.opensearch.threadpool.ThreadPool.Names;
 import org.opensearch.transport.ConnectTransportException;
 import org.opensearch.transport.Transport;
@@ -114,8 +112,6 @@ public class FollowersChecker {
  Setting.Property.NodeScope
  );
 
- private static final String UNIT = "1";
-
  private final Settings settings;
 
  private final TimeValue followerCheckInterval;
@@ -131,7 +127,7 @@ public class FollowersChecker {
  private final TransportService transportService;
  private final NodeHealthService nodeHealthService;
  private volatile FastResponseState fastResponseState;
- private Counter followerChecksFailureCounter;
+ private ClusterManagerMetrics clusterManagerMetrics;
 
  public FollowersChecker(
  Settings settings,
@@ -140,7 +136,7 @@ public FollowersChecker(
  Consumer<FollowerCheckRequest> handleRequestAndUpdateState,
  BiConsumer<DiscoveryNode, String> onNodeFailure,
  NodeHealthService nodeHealthService,
- MetricsRegistry metricsRegistry
+ ClusterManagerMetrics clusterManagerMetrics
  ) {
  this.settings = settings;
  this.transportService = transportService;
@@ -167,15 +163,7 @@ public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connecti
  handleDisconnectedNode(node);
  }
  });
- initializeMetrics(metricsRegistry);
- }
-
- private void initializeMetrics(MetricsRegistry metricsRegistry) {
- this.followerChecksFailureCounter = metricsRegistry.createCounter(
- "followers.checker.failure.count",
- "Counter for number of failed follower checks",
- UNIT
- );
+ this.clusterManagerMetrics = clusterManagerMetrics;
  }
 
  private void setFollowerCheckTimeout(TimeValue followerCheckTimeout) {
@@ -416,7 +404,6 @@ public void handleException(TransportException exp) {
  return;
  }
 
- followerChecksFailureCounter.add(1);
  failNode(reason);
  }
 
@@ -442,6 +429,7 @@ public void run() {
  followerCheckers.remove(discoveryNode);
  }
  onNodeFailure.accept(discoveryNode, reason);
+ clusterManagerMetrics.incrementCounter(clusterManagerMetrics.followerChecksFailureCounter, 1.0);
  }
 
  @Override

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/LeaderChecker.java b/server/src/main/java/org/opensearch/cluster/coordination/LeaderChecker.java
@@ -50,8 +50,6 @@
 import org.opensearch.core.transport.TransportResponse.Empty;
 import org.opensearch.monitor.NodeHealthService;
 import org.opensearch.monitor.StatusInfo;
-import org.opensearch.telemetry.metrics.Counter;
-import org.opensearch.telemetry.metrics.MetricsRegistry;
 import org.opensearch.threadpool.ThreadPool.Names;
 import org.opensearch.transport.ConnectTransportException;
 import org.opensearch.transport.NodeDisconnectedException;
@@ -113,8 +111,6 @@ public class LeaderChecker {
  Setting.Property.NodeScope
  );
 
- private static final String UNIT = "1";
-
  private final Settings settings;
 
  private final TimeValue leaderCheckInterval;
@@ -125,15 +121,15 @@ public class LeaderChecker {
  private final NodeHealthService nodeHealthService;
  private AtomicReference<CheckScheduler> currentChecker = new AtomicReference<>();
  private volatile DiscoveryNodes discoveryNodes;
- private Counter leaderCheckFailureCounter;
+ private final ClusterManagerMetrics clusterManagerMetrics;
 
  LeaderChecker(
  final Settings settings,
  final ClusterSettings clusterSettings,
  final TransportService transportService,
  final Consumer<Exception> onLeaderFailure,
  NodeHealthService nodeHealthService,
- final MetricsRegistry metricsRegistry
+ final ClusterManagerMetrics clusterManagerMetrics
  ) {
  this.settings = settings;
  leaderCheckInterval = LEADER_CHECK_INTERVAL_SETTING.get(settings);
@@ -142,6 +138,7 @@ public class LeaderChecker {
  this.transportService = transportService;
  this.onLeaderFailure = onLeaderFailure;
  this.nodeHealthService = nodeHealthService;
+ this.clusterManagerMetrics = clusterManagerMetrics;
  clusterSettings.addSettingsUpdateConsumer(LEADER_CHECK_TIMEOUT_SETTING, this::setLeaderCheckTimeout);
 
  transportService.registerRequestHandler(
@@ -162,15 +159,6 @@ public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connecti
  handleDisconnectedNode(node);
  }
  });
- initializeMetrics(metricsRegistry);
- }
-
- private void initializeMetrics(MetricsRegistry metricsRegistry) {
- this.leaderCheckFailureCounter = metricsRegistry.createCounter(
- "leader.checker.failure.count",
- "Counter for number of failed leader checks",
- UNIT
- );
  }
 
  private void setLeaderCheckTimeout(TimeValue leaderCheckTimeout) {
@@ -319,12 +307,10 @@ public void handleException(TransportException exp) {
  if (exp instanceof ConnectTransportException || exp.getCause() instanceof ConnectTransportException) {
  logger.debug(new ParameterizedMessage("leader [{}] disconnected during check", leader), exp);
  leaderFailed(new ConnectTransportException(leader, "disconnected during check", exp));
- leaderCheckFailureCounter.add(1);
  return;
  } else if (exp.getCause() instanceof NodeHealthCheckFailureException) {
  logger.debug(new ParameterizedMessage("leader [{}] health check failed", leader), exp);
  leaderFailed(new NodeHealthCheckFailureException("node [" + leader + "] failed health checks", exp));
- leaderCheckFailureCounter.add(1);
  return;
  }
  long failureCount = failureCountSinceLastSuccess.incrementAndGet();
@@ -342,7 +328,6 @@ public void handleException(TransportException exp) {
  leaderFailed(
  new OpenSearchException("node [" + leader + "] failed [" + failureCount + "] consecutive checks", exp)
  );
- leaderCheckFailureCounter.add(1);
  return;
  }
 
@@ -373,6 +358,7 @@ void leaderFailed(Exception e) {
  @Override
  public void run() {
  onLeaderFailure.accept(e);
+ clusterManagerMetrics.incrementCounter(clusterManagerMetrics.leaderCheckFailureCounter, 1.0);
  }
 
  @Override

diff --git a/server/src/main/java/org/opensearch/discovery/DiscoveryModule.java b/server/src/main/java/org/opensearch/discovery/DiscoveryModule.java
@@ -35,6 +35,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.opensearch.cluster.ClusterState;
+import org.opensearch.cluster.coordination.ClusterManagerMetrics;
 import org.opensearch.cluster.coordination.Coordinator;
 import org.opensearch.cluster.coordination.ElectionStrategy;
 import org.opensearch.cluster.coordination.PersistedStateRegistry;
@@ -55,7 +56,6 @@
 import org.opensearch.monitor.NodeHealthService;
 import org.opensearch.node.remotestore.RemoteStoreNodeService;
 import org.opensearch.plugins.DiscoveryPlugin;
-import org.opensearch.telemetry.metrics.MetricsRegistry;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.TransportService;
 
@@ -135,7 +135,7 @@ public DiscoveryModule(
  NodeHealthService nodeHealthService,
  PersistedStateRegistry persistedStateRegistry,
  RemoteStoreNodeService remoteStoreNodeService,
- MetricsRegistry metricsRegistry
+ ClusterManagerMetrics clusterManagerMetrics
  ) {
  final Collection<BiConsumer<DiscoveryNode, ClusterState>> joinValidators = new ArrayList<>();
  final Map<String, Supplier<SeedHostsProvider>> hostProviders = new HashMap<>();
@@ -214,7 +214,7 @@ public DiscoveryModule(
  nodeHealthService,
  persistedStateRegistry,
  remoteStoreNodeService,
- metricsRegistry
+ clusterManagerMetrics
  );
  } else {
  throw new IllegalArgumentException("Unknown discovery type [" + discoveryType + "]");

diff --git a/server/src/main/java/org/opensearch/node/Node.java b/server/src/main/java/org/opensearch/node/Node.java
@@ -66,6 +66,7 @@
 import org.opensearch.cluster.InternalClusterInfoService;
 import org.opensearch.cluster.NodeConnectionsService;
 import org.opensearch.cluster.action.index.MappingUpdatedAction;
+import org.opensearch.cluster.coordination.ClusterManagerMetrics;
 import org.opensearch.cluster.coordination.PersistedStateRegistry;
 import org.opensearch.cluster.metadata.AliasValidator;
 import org.opensearch.cluster.metadata.IndexTemplateMetadata;
@@ -644,6 +645,7 @@ protected Node(
 
  final ClusterInfoService clusterInfoService = newClusterInfoService(settings, clusterService, threadPool, client);
  final UsageService usageService = new UsageService();
+ final ClusterManagerMetrics clusterManagerMetrics = new ClusterManagerMetrics(metricsRegistry);
 
  ModulesBuilder modules = new ModulesBuilder();
  // plugin modules must be added here, before others or we can get crazy injection errors...
@@ -1130,7 +1132,7 @@ protected Node(
  fsHealthService,
  persistedStateRegistry,
  remoteStoreNodeService,
- metricsRegistry
+ clusterManagerMetrics
  );
  final SearchPipelineService searchPipelineService = new SearchPipelineService(
  clusterService,