Adding warn logging for timed out remote state read tasks

Signed-off-by: Harsh Garg <gkharsh@amazon.com>

Author: Harsh Garg

server/src/main/java/org/opensearch/cluster/routing/remote/InternalRemoteRoutingTableService.java

@@ -256,7 +256,8 @@ protected void doStart() {
             clusterName,
             threadPool,
             ThreadPool.Names.REMOTE_STATE_READ,
-            RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+            RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+            clusterSettings
         );
     }
 

server/src/main/java/org/opensearch/common/remote/RemoteWriteableEntityBlobStore.java

@@ -8,8 +8,13 @@
 
 package org.opensearch.common.remote;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.opensearch.common.blobstore.BlobPath;
 import org.opensearch.common.blobstore.stream.write.WritePriority;
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.Setting;
+import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.index.translog.transfer.BlobStoreTransferService;
 import org.opensearch.repositories.blobstore.BlobStoreRepository;
@@ -29,25 +34,44 @@
  */
 public class RemoteWriteableEntityBlobStore<T, U extends RemoteWriteableBlobEntity<T>> implements RemoteWritableEntityStore<T, U> {
 
+    private static final Logger logger = LogManager.getLogger(RemoteWriteableEntityBlobStore.class);
     private final BlobStoreTransferService transferService;
     private final BlobStoreRepository blobStoreRepository;
     private final String clusterName;
     private final ExecutorService executorService;
     private final String pathToken;
+    /**
+     * To be used for identifying and logging read tasks/entities which take considerably more time in getting completed.
+     * Threshold corresponds to the total time spent in reading the blob along with deserializing the i/p stream.
+     */
+    public static final Setting<TimeValue> REMOTE_STORE_SLOW_READ_LOGGING_THRESHOLD_SETTING = Setting.positiveTimeSetting(
+        "cluster.remote_store.slow_read_logging_threshold",
+        TimeValue.timeValueSeconds(10),
+        Setting.Property.Dynamic,
+        Setting.Property.NodeScope
+    );
+    private volatile TimeValue slowReadLoggingThreshold;
 
     public RemoteWriteableEntityBlobStore(
         final BlobStoreTransferService blobStoreTransferService,
         final BlobStoreRepository blobStoreRepository,
         final String clusterName,
         final ThreadPool threadPool,
         final String executor,
-        final String pathToken
+        final String pathToken,
+        final ClusterSettings clusterSettings
     ) {
         this.transferService = blobStoreTransferService;
         this.blobStoreRepository = blobStoreRepository;
         this.clusterName = clusterName;
         this.executorService = threadPool.executor(executor);
         this.pathToken = pathToken;
+        this.slowReadLoggingThreshold = clusterSettings.get(REMOTE_STORE_SLOW_READ_LOGGING_THRESHOLD_SETTING);
+        clusterSettings.addSettingsUpdateConsumer(REMOTE_STORE_SLOW_READ_LOGGING_THRESHOLD_SETTING, this::setSlowReadLoggingThreshold);
+    }
+
+    private void setSlowReadLoggingThreshold(TimeValue slowReadLoggingThreshold) {
+        this.slowReadLoggingThreshold = slowReadLoggingThreshold;
     }
 
     @Override
@@ -71,10 +95,13 @@ public void writeAsync(final U entity, final ActionListener<Void> listener) {
 
     @Override
     public T read(final U entity) throws IOException {
-        // TODO Add timing logs and tracing
         assert entity.getFullBlobName() != null;
+        final long readStartTimeNS = System.nanoTime();
         try (InputStream inputStream = transferService.downloadBlob(getBlobPathForDownload(entity), entity.getBlobFileName())) {
-            return entity.deserialize(inputStream);
+            final long deserializeStartTimeNS = System.nanoTime();
+            T deserializedBlobEntity = entity.deserialize(inputStream);
+            warnAboutSlowReadIfNeeded(entity, deserializeStartTimeNS, readStartTimeNS);
+            return deserializedBlobEntity;
         }
     }
 
@@ -122,4 +149,18 @@ private static String encodeString(String content) {
         return Base64.getUrlEncoder().withoutPadding().encodeToString(content.getBytes(StandardCharsets.UTF_8));
     }
 
+    private void warnAboutSlowReadIfNeeded(final U entity, final long deserializeStartTimeNS, final long readStartTimeNS) {
+        long totalReadTimeMS = Math.max(0, TimeValue.nsecToMSec(System.nanoTime() - readStartTimeNS));
+        long serdeTimeMS = Math.max(0, TimeValue.nsecToMSec(System.nanoTime() - deserializeStartTimeNS));
+        if (totalReadTimeMS > slowReadLoggingThreshold.getMillis()) {
+            logger.warn(
+                "entity [{}] for [{}] took [{}] for serde out of total read time [{}] which is above the total warn threshold of [{}]",
+                entity.getClass().getSimpleName(),
+                entity.getBlobFileName(),
+                serdeTimeMS,
+                totalReadTimeMS,
+                slowReadLoggingThreshold
+            );
+        }
+    }
 }

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -89,6 +89,7 @@
 import org.opensearch.common.logging.Loggers;
 import org.opensearch.common.network.NetworkModule;
 import org.opensearch.common.network.NetworkService;
+import org.opensearch.common.remote.RemoteWriteableEntityBlobStore;
 import org.opensearch.common.settings.Setting.Property;
 import org.opensearch.common.util.FeatureFlags;
 import org.opensearch.common.util.PageCacheRecycler;
@@ -759,6 +760,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                 RemoteRoutingTableBlobStore.REMOTE_ROUTING_TABLE_PATH_HASH_ALGO_SETTING,
                 RemoteClusterStateService.REMOTE_CLUSTER_STATE_CHECKSUM_VALIDATION_MODE_SETTING,
                 RemoteRoutingTableBlobStore.CLUSTER_REMOTE_STORE_ROUTING_TABLE_PATH_PREFIX,
+                RemoteWriteableEntityBlobStore.REMOTE_STORE_SLOW_READ_LOGGING_THRESHOLD_SETTING,
 
                 // Admission Control Settings
                 AdmissionControlSettings.ADMISSION_CONTROL_TRANSPORT_LAYER_MODE,

server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateAttributesManager.java

@@ -14,6 +14,7 @@
 import org.opensearch.common.remote.AbstractClusterMetadataWriteableBlobEntity;
 import org.opensearch.common.remote.AbstractRemoteWritableEntityManager;
 import org.opensearch.common.remote.RemoteWriteableEntityBlobStore;
+import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.common.io.stream.NamedWriteableRegistry;
 import org.opensearch.gateway.remote.model.RemoteClusterBlocks;
@@ -43,7 +44,8 @@ public class RemoteClusterStateAttributesManager extends AbstractRemoteWritableE
         BlobStoreRepository blobStoreRepository,
         BlobStoreTransferService blobStoreTransferService,
         NamedWriteableRegistry namedWriteableRegistry,
-        ThreadPool threadpool
+        ThreadPool threadpool,
+        ClusterSettings clusterSettings
     ) {
         this.remoteWritableEntityStores.put(
             RemoteDiscoveryNodes.DISCOVERY_NODES,
@@ -53,7 +55,8 @@ public class RemoteClusterStateAttributesManager extends AbstractRemoteWritableE
                 clusterName,
                 threadpool,
                 ThreadPool.Names.REMOTE_STATE_READ,
-                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+                clusterSettings
             )
         );
         this.remoteWritableEntityStores.put(
@@ -64,7 +67,8 @@ public class RemoteClusterStateAttributesManager extends AbstractRemoteWritableE
                 clusterName,
                 threadpool,
                 ThreadPool.Names.REMOTE_STATE_READ,
-                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+                clusterSettings
             )
         );
         this.remoteWritableEntityStores.put(
@@ -75,7 +79,8 @@ public class RemoteClusterStateAttributesManager extends AbstractRemoteWritableE
                 clusterName,
                 threadpool,
                 ThreadPool.Names.REMOTE_STATE_READ,
-                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+                clusterSettings
             )
         );
     }

server/src/main/java/org/opensearch/gateway/remote/RemoteClusterStateService.java

@@ -1114,7 +1114,8 @@ public void start() {
             blobStoreRepository,
             blobStoreTransferService,
             namedWriteableRegistry,
-            threadpool
+            threadpool,
+            clusterSettings
         );
 
         remoteRoutingTableService.start();
@@ -1394,7 +1395,14 @@ ClusterState readClusterStateInParallel(
         try {
             if (latch.await(this.remoteStateReadTimeout.getMillis(), TimeUnit.MILLISECONDS) == false) {
                 RemoteStateTransferException exception = new RemoteStateTransferException(
-                    "Timed out waiting to read cluster state from remote within timeout " + this.remoteStateReadTimeout
+                    String.format(
+                        Locale.ROOT,
+                        "Timed out waiting to read total [%s] tasks for cluster state from remote within timeout of [%s]. Could only read [%s] tasks while [%s] tasks failed to be read",
+                        totalReadTasks,
+                        this.remoteStateReadTimeout,
+                        latch.getCount(),
+                        exceptionList.size()
+                    )
                 );
                 exceptionList.forEach(exception::addSuppressed);
                 throw exception;

server/src/main/java/org/opensearch/gateway/remote/RemoteGlobalMetadataManager.java

@@ -91,7 +91,8 @@ public class RemoteGlobalMetadataManager extends AbstractRemoteWritableEntityMan
                 clusterName,
                 threadpool,
                 ThreadPool.Names.REMOTE_STATE_READ,
-                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+                clusterSettings
             )
         );
         this.remoteWritableEntityStores.put(
@@ -102,7 +103,8 @@ public class RemoteGlobalMetadataManager extends AbstractRemoteWritableEntityMan
                 clusterName,
                 threadpool,
                 ThreadPool.Names.REMOTE_STATE_READ,
-                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+                clusterSettings
             )
         );
         this.remoteWritableEntityStores.put(
@@ -113,7 +115,8 @@ public class RemoteGlobalMetadataManager extends AbstractRemoteWritableEntityMan
                 clusterName,
                 threadpool,
                 ThreadPool.Names.REMOTE_STATE_READ,
-                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+                clusterSettings
             )
         );
         this.remoteWritableEntityStores.put(
@@ -124,7 +127,8 @@ public class RemoteGlobalMetadataManager extends AbstractRemoteWritableEntityMan
                 clusterName,
                 threadpool,
                 ThreadPool.Names.REMOTE_STATE_READ,
-                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+                clusterSettings
             )
         );
         this.remoteWritableEntityStores.put(
@@ -135,7 +139,8 @@ public class RemoteGlobalMetadataManager extends AbstractRemoteWritableEntityMan
                 clusterName,
                 threadpool,
                 ThreadPool.Names.REMOTE_STATE_READ,
-                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+                clusterSettings
             )
         );
         this.remoteWritableEntityStores.put(
@@ -146,7 +151,8 @@ public class RemoteGlobalMetadataManager extends AbstractRemoteWritableEntityMan
                 clusterName,
                 threadpool,
                 ThreadPool.Names.REMOTE_STATE_READ,
-                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+                clusterSettings
             )
         );
         this.remoteWritableEntityStores.put(
@@ -157,7 +163,8 @@ public class RemoteGlobalMetadataManager extends AbstractRemoteWritableEntityMan
                 clusterName,
                 threadpool,
                 ThreadPool.Names.REMOTE_STATE_READ,
-                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+                clusterSettings
             )
         );
         clusterSettings.addSettingsUpdateConsumer(GLOBAL_METADATA_UPLOAD_TIMEOUT_SETTING, this::setGlobalMetadataUploadTimeout);

server/src/main/java/org/opensearch/gateway/remote/RemoteIndexMetadataManager.java

@@ -92,7 +92,8 @@ public RemoteIndexMetadataManager(
                 clusterName,
                 threadpool,
                 ThreadPool.Names.REMOTE_STATE_READ,
-                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+                RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+                clusterSettings
             )
         );
         this.namedXContentRegistry = blobStoreRepository.getNamedXContentRegistry();

server/src/main/java/org/opensearch/gateway/remote/RemoteManifestManager.java

@@ -85,7 +85,8 @@ public class RemoteManifestManager {
             clusterName,
             threadpool,
             ThreadPool.Names.REMOTE_STATE_READ,
-            RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+            RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+            clusterSettings
         );
         ;
         clusterSettings.addSettingsUpdateConsumer(METADATA_MANIFEST_UPLOAD_TIMEOUT_SETTING, this::setMetadataManifestUploadTimeout);

server/src/main/java/org/opensearch/gateway/remote/model/RemoteRoutingTableBlobStore.java

@@ -87,7 +87,8 @@ public RemoteRoutingTableBlobStore(
             clusterName,
             threadPool,
             executor,
-            RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN
+            RemoteClusterStateUtils.CLUSTER_STATE_PATH_TOKEN,
+            clusterSettings
         );
         this.pathType = clusterSettings.get(REMOTE_ROUTING_TABLE_PATH_TYPE_SETTING);
         this.pathHashAlgo = clusterSettings.get(REMOTE_ROUTING_TABLE_PATH_HASH_ALGO_SETTING);

server/src/test/java/org/opensearch/gateway/remote/RemoteClusterStateAttributesManagerTests.java

@@ -97,7 +97,8 @@ public void setup() throws Exception {
             blobStoreRepository,
             blobStoreTransferService,
             writableRegistry(),
-            threadPool
+            threadPool,
+            clusterSettings
         );
     }