Merge branch 'main' into dynamic-action-retry

Signed-off-by: Gaurav Bafna <[email protected]>
opensearch-project · Jun 6, 2024 · 92efc25 · 92efc25
2 parents 2f29b43 + b9ca5a8
commit 92efc25
Show file tree

Hide file tree

Showing 44 changed files with 1,006 additions and 69 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,6 +5,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ## [Unreleased 2.x]
 ### Added
+- Add leader and follower check failure counter metrics ([#12439](https://github.com/opensearch-project/OpenSearch/pull/12439))
 - Add latency metrics for instrumenting critical clusterManager code paths ([#12333](https://github.com/opensearch-project/OpenSearch/pull/12333))
 - Add support for Azure Managed Identity in repository-azure ([#12423](https://github.com/opensearch-project/OpenSearch/issues/12423))
 - Add useCompoundFile index setting ([#13478](https://github.com/opensearch-project/OpenSearch/pull/13478))
@@ -16,12 +17,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Allow setting query parameters on requests ([#13776](https://github.com/opensearch-project/OpenSearch/issues/13776))
 - Add remote routing table for remote state publication with experimental feature flag ([#13304](https://github.com/opensearch-project/OpenSearch/pull/13304))
 - Add dynamic action retry timeout setting ([#14022](https://github.com/opensearch-project/OpenSearch/issues/14022))
+- [Remote Store] Add support to disable flush based on translog reader count ([#14027](https://github.com/opensearch-project/OpenSearch/pull/14027))
 
 ### Dependencies
 - Bump `com.github.spullara.mustache.java:compiler` from 0.9.10 to 0.9.13 ([#13329](https://github.com/opensearch-project/OpenSearch/pull/13329), [#13559](https://github.com/opensearch-project/OpenSearch/pull/13559))
 - Bump `org.gradle.test-retry` from 1.5.8 to 1.5.9 ([#13442](https://github.com/opensearch-project/OpenSearch/pull/13442))
 - Bump `org.apache.commons:commons-text` from 1.11.0 to 1.12.0 ([#13557](https://github.com/opensearch-project/OpenSearch/pull/13557))
-- Bump `org.hdrhistogram:HdrHistogram` from 2.1.12 to 2.2.1 ([#13556](https://github.com/opensearch-project/OpenSearch/pull/13556))
+- Bump `org.hdrhistogram:HdrHistogram` from 2.1.12 to 2.2.2 ([#13556](https://github.com/opensearch-project/OpenSearch/pull/13556), [#13986](https://github.com/opensearch-project/OpenSearch/pull/13986))
 - Bump `com.gradle.enterprise` from 3.17.2 to 3.17.4 ([#13641](https://github.com/opensearch-project/OpenSearch/pull/13641), [#13753](https://github.com/opensearch-project/OpenSearch/pull/13753))
 - Bump `org.apache.hadoop:hadoop-minicluster` from 3.3.6 to 3.4.0 ([#13642](https://github.com/opensearch-project/OpenSearch/pull/13642))
 - Bump `mockito` from 5.11.0 to 5.12.0 ([#13665](https://github.com/opensearch-project/OpenSearch/pull/13665))

diff --git a/buildSrc/version.properties b/buildSrc/version.properties
@@ -23,6 +23,8 @@ guava             = 32.1.1-jre
 protobuf          = 3.22.3
 jakarta_annotation = 1.3.5
 google_http_client = 1.44.1
+tdigest           = 3.3
+hdrhistogram      = 2.2.2
 
 # when updating the JNA version, also update the version in buildSrc/build.gradle
 jna               = 5.13.0

diff --git a/gradle/ide.gradle b/gradle/ide.gradle
@@ -28,7 +28,7 @@ allprojects {
   apply plugin: 'idea'
 
   tasks.named('idea').configure {
-    doFirst { throw new GradleException("Use of the 'idea' task has been deprecated. For details on importing into IntelliJ see CONTRIBUTING.md.") }
+    doFirst { throw new GradleException("Use of the 'idea' task has been deprecated. For details on importing into IntelliJ see DEVELOPER_GUIDE.md.") }
   }
 }
 

diff --git a/...translog/BufferedChecksumStreamInput.java → ...o/stream/BufferedChecksumStreamInput.java b/...translog/BufferedChecksumStreamInput.java → ...o/stream/BufferedChecksumStreamInput.java
@@ -30,12 +30,10 @@
  * GitHub history for details.
  */
 
-package org.opensearch.index.translog;
+package org.opensearch.core.common.io.stream;
 
 import org.apache.lucene.store.BufferedChecksum;
 import org.apache.lucene.util.BitUtil;
-import org.opensearch.core.common.io.stream.FilterStreamInput;
-import org.opensearch.core.common.io.stream.StreamInput;
 
 import java.io.EOFException;
 import java.io.IOException;

diff --git a/...ranslog/BufferedChecksumStreamOutput.java → .../stream/BufferedChecksumStreamOutput.java b/...ranslog/BufferedChecksumStreamOutput.java → .../stream/BufferedChecksumStreamOutput.java
@@ -30,11 +30,10 @@
  * GitHub history for details.
  */
 
-package org.opensearch.index.translog;
+package org.opensearch.core.common.io.stream;
 
 import org.apache.lucene.store.BufferedChecksum;
 import org.opensearch.common.annotation.PublicApi;
-import org.opensearch.core.common.io.stream.StreamOutput;
 
 import java.io.IOException;
 import java.util.zip.CRC32;

diff --git a/server/build.gradle b/server/build.gradle
@@ -96,9 +96,9 @@ dependencies {
   api "joda-time:joda-time:${versions.joda}"
 
   // percentiles aggregation
-  api 'com.tdunning:t-digest:3.3'
-  // precentil ranks aggregation
-  api 'org.hdrhistogram:HdrHistogram:2.2.1'
+  api "com.tdunning:t-digest:${versions.tdigest}"
+  // percentile ranks aggregation
+  api "org.hdrhistogram:HdrHistogram:${versions.hdrhistogram}"
 
   // lucene spatial
   api "org.locationtech.spatial4j:spatial4j:${versions.spatial4j}", optional

diff --git a/server/licenses/HdrHistogram-2.2.1.jar.sha1 b/server/licenses/HdrHistogram-2.2.1.jar.sha1
diff --git a/server/licenses/HdrHistogram-2.2.2.jar.sha1 b/server/licenses/HdrHistogram-2.2.2.jar.sha1
@@ -0,0 +1 @@
+7959933ebcc0f05b2eaa5af0a0c8689fa257b15c
diff --git a/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java b/server/src/internalClusterTest/java/org/opensearch/remotestore/RemoteStoreIT.java
@@ -852,7 +852,9 @@ public void testFlushOnTooManyRemoteTranslogFiles() throws Exception {
 
         ClusterUpdateSettingsRequest updateSettingsRequest = new ClusterUpdateSettingsRequest();
         updateSettingsRequest.persistentSettings(
-            Settings.builder().put(RemoteStoreSettings.CLUSTER_REMOTE_MAX_TRANSLOG_READERS.getKey(), "100")
+            Settings.builder()
+                .put(RemoteStoreSettings.CLUSTER_REMOTE_MAX_TRANSLOG_READERS.getKey(), "100")
+                .put(CLUSTER_REMOTE_TRANSLOG_BUFFER_INTERVAL_SETTING.getKey(), "0ms")
         );
         assertAcked(client().admin().cluster().updateSettings(updateSettingsRequest).actionGet());
 
@@ -883,5 +885,27 @@ public void testFlushOnTooManyRemoteTranslogFiles() throws Exception {
                 assertEquals(totalFiles, 1L);
             }
         }, 30, TimeUnit.SECONDS);
+
+        // Disabling max translog readers
+        assertAcked(
+            internalCluster().client()
+                .admin()
+                .cluster()
+                .prepareUpdateSettings()
+                .setPersistentSettings(Settings.builder().put(RemoteStoreSettings.CLUSTER_REMOTE_MAX_TRANSLOG_READERS.getKey(), "-1"))
+                .get()
+        );
+
+        // Indexing 500 more docs
+        for (int i = 0; i < 500; i++) {
+            indexBulk(INDEX_NAME, 1);
+        }
+
+        // No flush is triggered since max_translog_readers is set to -1
+        // Total tlog files would be incremented by 500
+        try (Stream<Path> files = Files.list(translogLocation)) {
+            long totalFiles = files.filter(f -> f.getFileName().toString().endsWith(Translog.TRANSLOG_FILE_SUFFIX)).count();
+            assertEquals(totalFiles, 501L);
+        }
     }
 }
diff --git a/server/src/main/java/org/opensearch/cluster/ClusterManagerMetrics.java b/server/src/main/java/org/opensearch/cluster/ClusterManagerMetrics.java
@@ -8,6 +8,7 @@
 
 package org.opensearch.cluster;
 
+import org.opensearch.telemetry.metrics.Counter;
 import org.opensearch.telemetry.metrics.Histogram;
 import org.opensearch.telemetry.metrics.MetricsRegistry;
 import org.opensearch.telemetry.metrics.tags.Tags;
@@ -23,13 +24,17 @@
 public final class ClusterManagerMetrics {
 
     private static final String LATENCY_METRIC_UNIT_MS = "ms";
+    private static final String COUNTER_METRICS_UNIT = "1";
 
     public final Histogram clusterStateAppliersHistogram;
     public final Histogram clusterStateListenersHistogram;
     public final Histogram rerouteHistogram;
     public final Histogram clusterStateComputeHistogram;
     public final Histogram clusterStatePublishHistogram;
 
+    public final Counter leaderCheckFailureCounter;
+    public final Counter followerChecksFailureCounter;
+
     public ClusterManagerMetrics(MetricsRegistry metricsRegistry) {
         clusterStateAppliersHistogram = metricsRegistry.createHistogram(
             "cluster.state.appliers.latency",
@@ -56,6 +61,16 @@ public ClusterManagerMetrics(MetricsRegistry metricsRegistry) {
             "Histogram for recording time taken to publish a new cluster state",
             LATENCY_METRIC_UNIT_MS
         );
+        followerChecksFailureCounter = metricsRegistry.createCounter(
+            "followers.checker.failure.count",
+            "Counter for number of failed follower checks",
+            COUNTER_METRICS_UNIT
+        );
+        leaderCheckFailureCounter = metricsRegistry.createCounter(
+            "leader.checker.failure.count",
+            "Counter for number of failed leader checks",
+            COUNTER_METRICS_UNIT
+        );
     }
 
     public void recordLatency(Histogram histogram, Double value) {
@@ -69,4 +84,16 @@ public void recordLatency(Histogram histogram, Double value, Optional<Tags> tags
         }
         histogram.record(value, tags.get());
     }
+
+    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
@@ -36,6 +36,7 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.opensearch.cluster.ClusterChangedEvent;
+import org.opensearch.cluster.ClusterManagerMetrics;
 import org.opensearch.cluster.ClusterName;
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.ClusterStateTaskConfig;
@@ -207,7 +208,8 @@ public Coordinator(
         ElectionStrategy electionStrategy,
         NodeHealthService nodeHealthService,
         PersistedStateRegistry persistedStateRegistry,
-        RemoteStoreNodeService remoteStoreNodeService
+        RemoteStoreNodeService remoteStoreNodeService,
+        ClusterManagerMetrics clusterManagerMetrics
     ) {
         this.settings = settings;
         this.transportService = transportService;
@@ -261,14 +263,22 @@ public Coordinator(
             this::handlePublishRequest,
             this::handleApplyCommit
         );
-        this.leaderChecker = new LeaderChecker(settings, clusterSettings, transportService, this::onLeaderFailure, nodeHealthService);
+        this.leaderChecker = new LeaderChecker(
+            settings,
+            clusterSettings,
+            transportService,
+            this::onLeaderFailure,
+            nodeHealthService,
+            clusterManagerMetrics
+        );
         this.followersChecker = new FollowersChecker(
             settings,
             clusterSettings,
             transportService,
             this::onFollowerCheckRequest,
             this::removeNode,
-            nodeHealthService
+            nodeHealthService,
+            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
@@ -35,6 +35,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.opensearch.cluster.ClusterManagerMetrics;
 import org.opensearch.cluster.coordination.Coordinator.Mode;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.cluster.node.DiscoveryNodes;
@@ -127,14 +128,16 @@ public class FollowersChecker {
     private final TransportService transportService;
     private final NodeHealthService nodeHealthService;
     private volatile FastResponseState fastResponseState;
+    private ClusterManagerMetrics clusterManagerMetrics;
 
     public FollowersChecker(
         Settings settings,
         ClusterSettings clusterSettings,
         TransportService transportService,
         Consumer<FollowerCheckRequest> handleRequestAndUpdateState,
         BiConsumer<DiscoveryNode, String> onNodeFailure,
-        NodeHealthService nodeHealthService
+        NodeHealthService nodeHealthService,
+        ClusterManagerMetrics clusterManagerMetrics
     ) {
         this.settings = settings;
         this.transportService = transportService;
@@ -161,6 +164,7 @@ public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connecti
                 handleDisconnectedNode(node);
             }
         });
+        this.clusterManagerMetrics = clusterManagerMetrics;
     }
 
     private void setFollowerCheckTimeout(TimeValue followerCheckTimeout) {
@@ -413,6 +417,7 @@ public String executor() {
         }
 
         void failNode(String reason) {
+            clusterManagerMetrics.incrementCounter(clusterManagerMetrics.followerChecksFailureCounter, 1.0);
             transportService.getThreadPool().generic().execute(new Runnable() {
                 @Override
                 public void run() {

diff --git a/server/src/main/java/org/opensearch/cluster/coordination/LeaderChecker.java b/server/src/main/java/org/opensearch/cluster/coordination/LeaderChecker.java
@@ -36,6 +36,7 @@
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.opensearch.OpenSearchException;
+import org.opensearch.cluster.ClusterManagerMetrics;
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.cluster.node.DiscoveryNodes;
 import org.opensearch.common.Nullable;
@@ -119,17 +120,17 @@ public class LeaderChecker {
     private final TransportService transportService;
     private final Consumer<Exception> onLeaderFailure;
     private final NodeHealthService nodeHealthService;
-
     private AtomicReference<CheckScheduler> currentChecker = new AtomicReference<>();
-
     private volatile DiscoveryNodes discoveryNodes;
+    private final ClusterManagerMetrics clusterManagerMetrics;
 
     LeaderChecker(
         final Settings settings,
         final ClusterSettings clusterSettings,
         final TransportService transportService,
         final Consumer<Exception> onLeaderFailure,
-        NodeHealthService nodeHealthService
+        NodeHealthService nodeHealthService,
+        final ClusterManagerMetrics clusterManagerMetrics
     ) {
         this.settings = settings;
         leaderCheckInterval = LEADER_CHECK_INTERVAL_SETTING.get(settings);
@@ -138,6 +139,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(
@@ -293,7 +295,6 @@ public void handleResponse(Empty response) {
                             logger.debug("closed check scheduler received a response, doing nothing");
                             return;
                         }
-
                         failureCountSinceLastSuccess.set(0);
                         scheduleNextWakeUp(); // logs trace message indicating success
                     }
@@ -304,7 +305,6 @@ public void handleException(TransportException exp) {
                             logger.debug("closed check scheduler received a response, doing nothing");
                             return;
                         }
-
                         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));
@@ -355,6 +355,7 @@ public String executor() {
 
         void leaderFailed(Exception e) {
             if (isClosed.compareAndSet(false, true)) {
+                clusterManagerMetrics.incrementCounter(clusterManagerMetrics.leaderCheckFailureCounter, 1.0);
                 transportService.getThreadPool().generic().execute(new Runnable() {
                     @Override
                     public void run() {

diff --git a/server/src/main/java/org/opensearch/discovery/DiscoveryModule.java b/server/src/main/java/org/opensearch/discovery/DiscoveryModule.java
@@ -34,6 +34,7 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.opensearch.cluster.ClusterManagerMetrics;
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.coordination.Coordinator;
 import org.opensearch.cluster.coordination.ElectionStrategy;
@@ -133,7 +134,8 @@ public DiscoveryModule(
         RerouteService rerouteService,
         NodeHealthService nodeHealthService,
         PersistedStateRegistry persistedStateRegistry,
-        RemoteStoreNodeService remoteStoreNodeService
+        RemoteStoreNodeService remoteStoreNodeService,
+        ClusterManagerMetrics clusterManagerMetrics
     ) {
         final Collection<BiConsumer<DiscoveryNode, ClusterState>> joinValidators = new ArrayList<>();
         final Map<String, Supplier<SeedHostsProvider>> hostProviders = new HashMap<>();
@@ -211,7 +213,8 @@ public DiscoveryModule(
                 electionStrategy,
                 nodeHealthService,
                 persistedStateRegistry,
-                remoteStoreNodeService
+                remoteStoreNodeService,
+                clusterManagerMetrics
             );
         } else {
             throw new IllegalArgumentException("Unknown discovery type [" + discoveryType + "]");