IGNITE-27150 Add raft snapshot metrics (#7269)

Author: ivanzlenko

docs/_docs/administrators-guide/metrics/metrics-list.adoc

@@ -296,3 +296,21 @@ Transaction metrics.
 | LocalUnrebalancedPartitionsCount | The number of partitions that should be moved to this node.
 | TotalUnrebalancedPartitionsCount | The total number of partitions that should be moved to a new owner.
 |=======================================================================
+
+== raft.snapshots
+
+Metrics related to Raft snapshots of partition replicas.
+
+[width="100%",cols="20%,80%",opts="header"]
+|=======================================================================
+| Metric name | Description
+
+| IncomingSnapshots | The number of incoming Raft snapshots in progress.
+| IncomingSnapshotsLoadingMeta | The number of incoming Raft snapshots loading metadata.
+| IncomingSnapshotsWaitingCatalog | The number of incoming Raft snapshots waiting for catalog.
+| IncomingSnapshotsPreparingStorages | The number of incoming Raft snapshots preparing storages.
+| IncomingSnapshotsPreparingIndexForBuild | The number of incoming Raft snapshots preparing indexes for build.
+| IncomingSnapshotsLoadingMvData | The number of incoming Raft snapshots loading multi-versioned data.
+| IncomingSnapshotsLoadingTxMeta | The number of incoming Raft snapshots loading transaction metadata.
+| OutgoingSnapshots | The number of outgoing Raft snapshots in progress.
+|=======================================================================

modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/CliIntegrationTest.java

@@ -82,7 +82,8 @@ public abstract class CliIntegrationTest extends ClusterPerClassIntegrationTest
             new MetricSource().name("placement-driver").enabled(true),
             new MetricSource().name("resource.vacuum").enabled(true),
             new MetricSource().name("clock.service").enabled(true),
-            new MetricSource().name("index.builder").enabled(true)
+            new MetricSource().name("index.builder").enabled(true),
+            new MetricSource().name("raft.snapshots").enabled(true)
     };
 
     /** Correct ignite jdbc url. */

modules/distribution-zones/src/integrationTest/java/org/apache/ignite/internal/rebalance/ItRebalanceDistributedTest.java

@@ -1556,7 +1556,8 @@ private class Node {
                     txManager,
                     schemaManager,
                     dataStorageMgr,
-                    outgoingSnapshotManager
+                    outgoingSnapshotManager,
+                    metricManager
             );
 
             tableManager = new TableManager(

modules/partition-replicator/build.gradle

@@ -52,6 +52,7 @@ dependencies {
     testImplementation testFixtures(project(':ignite-core'))
     testImplementation testFixtures(project(':ignite-low-watermark'))
     testImplementation testFixtures(project(':ignite-metastorage'))
+    testImplementation testFixtures(project(':ignite-metrics'))
     testImplementation testFixtures(project(':ignite-schema'))
     testImplementation testFixtures(project(':ignite-storage-api'))
     testImplementation testFixtures(project(':ignite-table'))
@@ -60,6 +61,8 @@ dependencies {
     testImplementation testFixtures(project(':ignite-raft'))
     testImplementation testFixtures(project(':ignite-distribution-zones'))
 
+    testImplementation libs.awaitility
+
     integrationTestImplementation testFixtures(project(':ignite-cluster-management'))
     integrationTestImplementation testFixtures(project(':ignite-core'))
     integrationTestImplementation testFixtures(project(':ignite-configuration'))

modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/fixtures/Node.java

@@ -744,7 +744,8 @@ public CompletableFuture<Boolean> invoke(Condition condition, Operation success,
                 txManager,
                 schemaManager,
                 dataStorageMgr,
-                outgoingSnapshotsManager
+                outgoingSnapshotsManager,
+                metricManager
         );
 
         resourceVacuumManager = new ResourceVacuumManager(

modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/raft/ItZonePartitionRaftListenerRecoveryTest.java

@@ -83,6 +83,7 @@
 import org.apache.ignite.internal.partition.replicator.raft.snapshot.PartitionSnapshotStorageFactory;
 import org.apache.ignite.internal.partition.replicator.raft.snapshot.PartitionTxStateAccessImpl;
 import org.apache.ignite.internal.partition.replicator.raft.snapshot.ZonePartitionKey;
+import org.apache.ignite.internal.partition.replicator.raft.snapshot.metrics.RaftSnapshotsMetricsSource;
 import org.apache.ignite.internal.partition.replicator.raft.snapshot.outgoing.OutgoingSnapshotsManager;
 import org.apache.ignite.internal.placementdriver.LeasePlacementDriver;
 import org.apache.ignite.internal.raft.Loza;
@@ -311,7 +312,8 @@ void setUp(
                 catalogService,
                 failureProcessor,
                 executor,
-                new LogStorageAccessImpl(replicaManager)
+                new LogStorageAccessImpl(replicaManager),
+                new RaftSnapshotsMetricsSource()
         );
     }
 

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/PartitionReplicaLifecycleManager.java

@@ -127,6 +127,7 @@
 import org.apache.ignite.internal.metastorage.WatchListener;
 import org.apache.ignite.internal.metastorage.dsl.Condition;
 import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.metrics.MetricManager;
 import org.apache.ignite.internal.network.InternalClusterNode;
 import org.apache.ignite.internal.network.RecipientLeftException;
 import org.apache.ignite.internal.network.TopologyService;
@@ -265,6 +266,8 @@ public class PartitionReplicaLifecycleManager extends
 
     private final ZoneResourcesManager zoneResourcesManager;
 
+    private final MetricManager metricManager;
+
     private final ReliableCatalogVersions reliableCatalogVersions;
 
     private final EventListener<CreateZoneEventParameters> onCreateZoneListener = this::onCreateZone;
@@ -295,6 +298,7 @@ public class PartitionReplicaLifecycleManager extends
      * @param schemaManager Schema manager.
      * @param dataStorageManager Data storage manager.
      * @param outgoingSnapshotsManager Outgoing snapshots manager.
+     * @param metricManager Metric manager.
      */
     public PartitionReplicaLifecycleManager(
             CatalogService catalogService,
@@ -316,7 +320,8 @@ public PartitionReplicaLifecycleManager(
             TxManager txManager,
             SchemaManager schemaManager,
             DataStorageManager dataStorageManager,
-            OutgoingSnapshotsManager outgoingSnapshotsManager
+            OutgoingSnapshotsManager outgoingSnapshotsManager,
+            MetricManager metricManager
     ) {
         this(
                 catalogService,
@@ -346,7 +351,8 @@ public PartitionReplicaLifecycleManager(
                         failureProcessor,
                         partitionOperationsExecutor,
                         replicaMgr
-                )
+                ),
+                metricManager
         );
     }
 
@@ -370,7 +376,8 @@ public PartitionReplicaLifecycleManager(
             TxManager txManager,
             SchemaManager schemaManager,
             DataStorageManager dataStorageManager,
-            ZoneResourcesManager zoneResourcesManager
+            ZoneResourcesManager zoneResourcesManager,
+            MetricManager metricManager
     ) {
         this.catalogService = catalogService;
         this.replicaMgr = replicaMgr;
@@ -390,6 +397,7 @@ public PartitionReplicaLifecycleManager(
         this.schemaManager = schemaManager;
         this.dataStorageManager = dataStorageManager;
         this.zoneResourcesManager = zoneResourcesManager;
+        this.metricManager = metricManager;
 
         rebalanceRetryDelayConfiguration = new SystemDistributedConfigurationPropertyHolder<>(
                 systemDistributedConfiguration,
@@ -431,6 +439,9 @@ public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
 
         executorInclinedPlacementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_EXPIRED, onPrimaryReplicaExpiredListener);
 
+        metricManager.registerSource(zoneResourcesManager.snapshotsMetricsSource());
+        metricManager.enable(zoneResourcesManager.snapshotsMetricsSource());
+
         return processZonesAndAssignmentsOnStart;
     }
 
@@ -1657,6 +1668,8 @@ public CompletableFuture<Void> stopAsync(ComponentContext componentContext) {
         }
 
         try {
+            metricManager.unregisterSource(zoneResourcesManager.snapshotsMetricsSource());
+
             IgniteUtils.closeAllManually(zoneResourcesManager);
         } catch (Exception e) {
             return failedFuture(e);

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZoneResourcesManager.java

@@ -34,6 +34,7 @@
 import org.apache.ignite.internal.partition.replicator.raft.snapshot.PartitionSnapshotStorage;
 import org.apache.ignite.internal.partition.replicator.raft.snapshot.PartitionTxStateAccessImpl;
 import org.apache.ignite.internal.partition.replicator.raft.snapshot.ZonePartitionKey;
+import org.apache.ignite.internal.partition.replicator.raft.snapshot.metrics.RaftSnapshotsMetricsSource;
 import org.apache.ignite.internal.partition.replicator.raft.snapshot.outgoing.OutgoingSnapshotsManager;
 import org.apache.ignite.internal.replicator.ReplicaManager;
 import org.apache.ignite.internal.replicator.ZonePartitionId;
@@ -70,6 +71,8 @@ public class ZoneResourcesManager implements ManuallyCloseable {
 
     private final ReplicaManager replicaManager;
 
+    private final RaftSnapshotsMetricsSource snapshotsMetricsSource = new RaftSnapshotsMetricsSource();
+
     /** Map from zone IDs to their resource holders. */
     private final Map<Integer, ZoneResources> resourcesByZoneId = new ConcurrentHashMap<>();
 
@@ -128,7 +131,8 @@ ZonePartitionResources allocateZonePartitionResources(
                 catalogService,
                 failureProcessor,
                 partitionOperationsExecutor,
-                new LogStorageAccessImpl(replicaManager)
+                new LogStorageAccessImpl(replicaManager),
+                snapshotsMetricsSource
         );
 
         var zonePartitionResources = new ZonePartitionResources(
@@ -218,6 +222,10 @@ TxStatePartitionStorage txStatePartitionStorage(int zoneId, int partitionId) {
         return resources.txStateStorage.getPartitionStorage(partitionId);
     }
 
+    RaftSnapshotsMetricsSource snapshotsMetricsSource() {
+        return snapshotsMetricsSource;
+    }
+
     private static class ZoneResources {
 
         final TxStateStorage txStateStorage;

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/raft/snapshot/PartitionSnapshotStorage.java

@@ -36,6 +36,7 @@
 import org.apache.ignite.internal.network.MessagingService;
 import org.apache.ignite.internal.network.TopologyService;
 import org.apache.ignite.internal.partition.replicator.raft.snapshot.incoming.IncomingSnapshotCopier;
+import org.apache.ignite.internal.partition.replicator.raft.snapshot.metrics.RaftSnapshotsMetricsSource;
 import org.apache.ignite.internal.partition.replicator.raft.snapshot.outgoing.OutgoingSnapshotReader;
 import org.apache.ignite.internal.partition.replicator.raft.snapshot.outgoing.OutgoingSnapshotsManager;
 import org.apache.ignite.internal.raft.RaftGroupConfiguration;
@@ -100,6 +101,8 @@ public class PartitionSnapshotStorage {
 
     private final LogStorageAccess logStorage;
 
+    private final RaftSnapshotsMetricsSource snapshotsMetricsSource;
+
     /** Constructor. */
     public PartitionSnapshotStorage(
             ZonePartitionKey partitionKey,
@@ -109,7 +112,8 @@ public PartitionSnapshotStorage(
             CatalogService catalogService,
             FailureProcessor failureProcessor,
             Executor incomingSnapshotsExecutor,
-            LogStorageAccess logStorage
+            LogStorageAccess logStorage,
+            RaftSnapshotsMetricsSource snapshotsMetricsSource
     ) {
         this(
                 partitionKey,
@@ -120,7 +124,8 @@ public PartitionSnapshotStorage(
                 failureProcessor,
                 incomingSnapshotsExecutor,
                 DEFAULT_WAIT_FOR_METADATA_CATCHUP_MS,
-                logStorage
+                logStorage,
+                snapshotsMetricsSource
         );
     }
 
@@ -134,7 +139,8 @@ public PartitionSnapshotStorage(
             FailureProcessor failureProcessor,
             Executor incomingSnapshotsExecutor,
             long waitForMetadataCatchupMs,
-            LogStorageAccess logStorage
+            LogStorageAccess logStorage,
+            RaftSnapshotsMetricsSource snapshotsMetricsSource
     ) {
         this.partitionKey = partitionKey;
         this.topologyService = topologyService;
@@ -145,6 +151,7 @@ public PartitionSnapshotStorage(
         this.incomingSnapshotsExecutor = incomingSnapshotsExecutor;
         this.waitForMetadataCatchupMs = waitForMetadataCatchupMs;
         this.logStorage = logStorage;
+        this.snapshotsMetricsSource = snapshotsMetricsSource;
     }
 
     public ZonePartitionKey partitionKey() {
@@ -231,7 +238,13 @@ public SnapshotCopier startIncomingSnapshot(String uri) {
 
         SnapshotUri snapshotUri = SnapshotUri.fromStringUri(uri);
 
-        var copier = new IncomingSnapshotCopier(this, snapshotUri, incomingSnapshotsExecutor, waitForMetadataCatchupMs) {
+        var copier = new IncomingSnapshotCopier(
+                this,
+                snapshotUri,
+                incomingSnapshotsExecutor,
+                waitForMetadataCatchupMs,
+                snapshotsMetricsSource
+        ) {
             @Override
             public void close() {
                 try {
@@ -257,7 +270,7 @@ public SnapshotReader startOutgoingSnapshot() {
 
         startSnapshotOperation(snapshotId);
 
-        return new OutgoingSnapshotReader(snapshotId, this) {
+        return new OutgoingSnapshotReader(snapshotId, this, snapshotsMetricsSource) {
             @Override
             public void close() throws IOException {
                 try {
@@ -282,6 +295,8 @@ private void completeSnapshotOperation(UUID snapshotId) {
         LOG.info("Finishing outgoing snapshot [partitionKey={}, snapshotId={}]", partitionKey, snapshotId);
 
         synchronized (snapshotOperationLock) {
+            LOG.info("Finishing outgoing snapshot [partitionKey={}, snapshotId={}]", partitionKey, snapshotId);
+
             CompletableFuture<Void> operationFuture = ongoingSnapshotOperations.remove(snapshotId);
 
             assert operationFuture != null :