Report outcome of data node's shard snapshot master update (#123652)

Improves the information in the IndexShardSnapshotStatus's
statusDescription field to include the success/failure of the remote
call to the master node to update the shard snapshot state. This
allows us to see if there is a discrepancy between the state of the
data node and the master node.

Closes ES-10991

Author: DiannaHohensee

server/src/internalClusterTest/java/org/elasticsearch/snapshots/SnapshotShutdownIT.java

@@ -597,7 +597,7 @@ public void testSnapshotShutdownProgressTracker() throws Exception {
                 "SnapshotShutdownProgressTracker index shard snapshot messages",
                 SnapshotShutdownProgressTracker.class.getCanonicalName(),
                 Level.INFO,
-                "statusDescription='finished: master notification attempt complete'"
+                "*successfully sent shard snapshot state [PAUSED_FOR_NODE_REMOVAL] update to the master node*"
             )
         );
 

server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java

@@ -61,7 +61,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.function.Supplier;
+import java.util.function.Consumer;
 
 import static java.util.Collections.emptyMap;
 import static org.elasticsearch.core.Strings.format;
@@ -340,7 +340,8 @@ private void handleUpdatedSnapshotsInProgressEntry(String localNodeId, boolean r
                                 ShardState.FAILED,
                                 shard.getValue().reason(),
                                 shard.getValue().generation(),
-                                () -> null
+                                // Shard snapshot never began, so there is no status object to update.
+                                (outcomeInfoString) -> {}
                             );
                         }
                     } else {
@@ -403,7 +404,6 @@ private void pauseShardSnapshotsForNodeRemoval(String localNodeId, SnapshotsInPr
         for (final Map.Entry<ShardId, ShardSnapshotStatus> shardEntry : entry.shards().entrySet()) {
             final ShardId shardId = shardEntry.getKey();
             final ShardSnapshotStatus masterShardSnapshotStatus = shardEntry.getValue();
-            IndexShardSnapshotStatus indexShardSnapshotStatus = localShardSnapshots.get(shardId);
 
             if (masterShardSnapshotStatus.state() != ShardState.INIT) {
                 // shard snapshot not currently scheduled by master
@@ -424,10 +424,8 @@ private void pauseShardSnapshotsForNodeRemoval(String localNodeId, SnapshotsInPr
                     ShardState.PAUSED_FOR_NODE_REMOVAL,
                     "paused",
                     masterShardSnapshotStatus.generation(),
-                    () -> {
-                        indexShardSnapshotStatus.updateStatusDescription("finished: master notification attempt complete");
-                        return null;
-                    }
+                    // Shard snapshot never began, so there is no status object to update
+                    (outcomeInfoString) -> {}
                 );
             } else {
                 // shard snapshot currently running, mark for pause
@@ -444,9 +442,8 @@ private Runnable newShardSnapshotTask(
         final IndexVersion entryVersion,
         final long entryStartTime
     ) {
-        Supplier<Void> postMasterNotificationAction = () -> {
-            snapshotStatus.updateStatusDescription("finished: master notification attempt complete");
-            return null;
+        Consumer<String> postMasterNotificationAction = (outcomeInfoString) -> {
+            snapshotStatus.updateStatusDescription("Data node shard snapshot finished. Remote master update outcome: " + outcomeInfoString);
         };
 
         // Listener that runs on completion of the shard snapshot: it will notify the master node of success or failure.
@@ -687,6 +684,8 @@ private void syncShardStatsOnNewMaster(List<SnapshotsInProgress.Entry> entries)
                     if (masterShard != null && masterShard.state().completed() == false) {
                         final IndexShardSnapshotStatus.Copy indexShardSnapshotStatus = localShard.getValue().asCopy();
                         final Stage stage = indexShardSnapshotStatus.getStage();
+                        final String statusDescription = indexShardSnapshotStatus.getStatusDescription();
+                        final int maxStatusAppend = 1000;
                         // Master knows about the shard and thinks it has not completed
                         if (stage == Stage.DONE) {
                             // but we think the shard is done - we need to make new master know that the shard is done
@@ -700,7 +699,20 @@ private void syncShardStatsOnNewMaster(List<SnapshotsInProgress.Entry> entries)
                                 snapshot.snapshot(),
                                 shardId,
                                 localShard.getValue().getShardSnapshotResult(),
-                                () -> null
+                                (outcomeInfoString) -> localShard.getValue()
+                                    .updateStatusDescription(
+                                        Strings.format(
+                                            """
+                                                Data node already successfully finished shard snapshot, but a new master needed to be
+                                                notified. New remote master notification outcome: [%s]. The prior shard snapshot status
+                                                description was [%s]
+                                                """,
+                                            outcomeInfoString,
+                                            statusDescription.length() < maxStatusAppend
+                                                ? statusDescription
+                                                : statusDescription.substring(0, maxStatusAppend)
+                                        )
+                                    )
                             );
                         } else if (stage == Stage.FAILURE) {
                             // but we think the shard failed - we need to make new master know that the shard failed
@@ -716,7 +728,21 @@ private void syncShardStatsOnNewMaster(List<SnapshotsInProgress.Entry> entries)
                                 ShardState.FAILED,
                                 indexShardSnapshotStatus.getFailure(),
                                 localShard.getValue().generation(),
-                                () -> null
+                                // Update the original statusDescription with the latest remote master call outcome, but include the old
+                                // response. This will allow us to see when/whether the information reached the previous and current master.
+                                (outcomeInfoString) -> localShard.getValue()
+                                    .updateStatusDescription(
+                                        Strings.format(
+                                            """
+                                                Data node already failed shard snapshot, but a new master needed to be notified. New remote
+                                                master notification outcome: [%s]. The prior shard snapshot status description was [%s]
+                                                """,
+                                            outcomeInfoString,
+                                            statusDescription.length() < maxStatusAppend
+                                                ? statusDescription
+                                                : statusDescription.substring(0, maxStatusAppend)
+                                        )
+                                    )
                             );
                         } else if (stage == Stage.PAUSED) {
                             // but we think the shard has paused - we need to make new master know that
@@ -730,7 +756,19 @@ private void syncShardStatsOnNewMaster(List<SnapshotsInProgress.Entry> entries)
                                 ShardState.PAUSED_FOR_NODE_REMOVAL,
                                 indexShardSnapshotStatus.getFailure(),
                                 localShard.getValue().generation(),
-                                () -> null
+                                (outcomeInfoString) -> localShard.getValue()
+                                    .updateStatusDescription(
+                                        Strings.format(
+                                            """
+                                                Data node already paused shard snapshot, but a new master needed to be notified. New remote
+                                                master notification outcome: [%s]. The prior shard snapshot status description was [%s]
+                                                """,
+                                            outcomeInfoString,
+                                            statusDescription.length() < maxStatusAppend
+                                                ? statusDescription
+                                                : statusDescription.substring(0, maxStatusAppend)
+                                        )
+                                    )
                             );
                         }
                     }
@@ -747,7 +785,7 @@ private void notifySuccessfulSnapshotShard(
         final Snapshot snapshot,
         final ShardId shardId,
         ShardSnapshotResult shardSnapshotResult,
-        Supplier<Void> postMasterNotificationAction
+        Consumer<String> postMasterNotificationAction
     ) {
         assert shardSnapshotResult != null;
         assert shardSnapshotResult.getGeneration() != null;
@@ -768,7 +806,7 @@ private void notifyUnsuccessfulSnapshotShard(
         final ShardState shardState,
         final String failure,
         final ShardGeneration generation,
-        Supplier<Void> postMasterNotificationAction
+        Consumer<String> postMasterNotificationAction
     ) {
         assert shardState == ShardState.FAILED || shardState == ShardState.PAUSED_FOR_NODE_REMOVAL : shardState;
         sendSnapshotShardUpdate(
@@ -792,29 +830,32 @@ private void sendSnapshotShardUpdate(
         final Snapshot snapshot,
         final ShardId shardId,
         final ShardSnapshotStatus status,
-        Supplier<Void> postMasterNotificationAction
+        Consumer<String> postMasterNotificationAction
     ) {
+        snapshotShutdownProgressTracker.trackRequestSentToMaster(snapshot, shardId);
         ActionListener<Void> updateResultListener = new ActionListener<>() {
             @Override
             public void onResponse(Void aVoid) {
+                snapshotShutdownProgressTracker.releaseRequestSentToMaster(snapshot, shardId);
+                postMasterNotificationAction.accept(
+                    Strings.format("successfully sent shard snapshot state [%s] update to the master node", status.state())
+                );
                 logger.trace("[{}][{}] updated snapshot state to [{}]", shardId, snapshot, status);
             }
 
             @Override
             public void onFailure(Exception e) {
+                snapshotShutdownProgressTracker.releaseRequestSentToMaster(snapshot, shardId);
+                postMasterNotificationAction.accept(
+                    Strings.format("exception trying to send shard snapshot state [%s] update to the master node [%s]", status.state(), e)
+                );
                 logger.warn(() -> format("[%s][%s] failed to update snapshot state to [%s]", shardId, snapshot, status), e);
             }
         };
 
-        snapshotShutdownProgressTracker.trackRequestSentToMaster(snapshot, shardId);
-        var releaseTrackerRequestRunsBeforeResultListener = ActionListener.runBefore(updateResultListener, () -> {
-            snapshotShutdownProgressTracker.releaseRequestSentToMaster(snapshot, shardId);
-            postMasterNotificationAction.get();
-        });
-
         remoteFailedRequestDeduplicator.executeOnce(
             new UpdateIndexShardSnapshotStatusRequest(snapshot, shardId, status),
-            releaseTrackerRequestRunsBeforeResultListener,
+            updateResultListener,
             (req, reqListener) -> transportService.sendRequest(
                 transportService.getLocalNode(),
                 SnapshotsService.UPDATE_SNAPSHOT_STATUS_ACTION_NAME,