You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tk...@apache.org on 2024/06/07 10:19:27 UTC
(ignite-3) branch main updated: IGNITE-21466 Add metrics for partition states (#3835)
This is an automated email from the ASF dual-hosted git repository.
tkalkirill pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 27e4c3e6e9 IGNITE-21466 Add metrics for partition states (#3835)
27e4c3e6e9 is described below
commit 27e4c3e6e9de0ae292bb5f009f7bbab239946b76
Author: Kirill Tkalenko <tk...@yandex.ru>
AuthorDate: Fri Jun 7 13:19:22 2024 +0300
IGNITE-21466 Add metrics for partition states (#3835)
---
.../metrics/exporters/jmx/JmxExporter.java | 6 +-
.../org/apache/ignite/internal/app/IgniteImpl.java | 3 +-
modules/table/build.gradle | 1 +
.../disaster/ItDisasterRecoveryMetricTest.java | 119 ++++++++++++++++++
.../disaster/DisasterRecoveryManager.java | 111 ++++++++---------
.../disaster/LocalPartitionStateEnum.java | 30 ++++-
.../LocalPartitionStateEnumWithLogIndex.java | 56 +++++++++
.../disaster/PartitionStatesMetricSource.java | 133 +++++++++++++++++++++
8 files changed, 399 insertions(+), 60 deletions(-)
diff --git a/modules/metrics/src/main/java/org/apache/ignite/internal/metrics/exporters/jmx/JmxExporter.java b/modules/metrics/src/main/java/org/apache/ignite/internal/metrics/exporters/jmx/JmxExporter.java
index aa14e1f957..526afde60a 100644
--- a/modules/metrics/src/main/java/org/apache/ignite/internal/metrics/exporters/jmx/JmxExporter.java
+++ b/modules/metrics/src/main/java/org/apache/ignite/internal/metrics/exporters/jmx/JmxExporter.java
@@ -46,10 +46,8 @@ public class JmxExporter extends BasicMetricExporter<JmxExporterView> {
*/
public static final String JMX_EXPORTER_NAME = "jmx";
- /**
- * Group attribute of {@link ObjectName} shared for all metric MBeans.
- */
- private static final String JMX_METRIC_GROUP = "metrics";
+ /** Group attribute of {@link ObjectName} shared for all metric MBeans. */
+ public static final String JMX_METRIC_GROUP = "metrics";
/**
* Logger.
diff --git a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index 80b612e8c0..39c0f6b338 100644
--- a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++ b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -819,7 +819,8 @@ public class IgniteImpl implements Ignite {
distributionZoneManager,
raftMgr,
clusterSvc.topologyService(),
- distributedTblMgr
+ distributedTblMgr,
+ metricManager
);
systemViewManager.register(disasterRecoveryManager);
diff --git a/modules/table/build.gradle b/modules/table/build.gradle
index b2e1aa24c8..0aaa8182d2 100644
--- a/modules/table/build.gradle
+++ b/modules/table/build.gradle
@@ -48,6 +48,7 @@ dependencies {
implementation project(':ignite-workers')
implementation project(':ignite-low-watermark')
implementation project(':ignite-system-view-api')
+ implementation project(':ignite-metrics')
implementation libs.jetbrains.annotations
implementation libs.fastutil.core
implementation libs.auto.service.annotations
diff --git a/modules/table/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryMetricTest.java b/modules/table/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryMetricTest.java
new file mode 100644
index 0000000000..657dc07b0e
--- /dev/null
+++ b/modules/table/src/integrationTest/java/org/apache/ignite/internal/disaster/ItDisasterRecoveryMetricTest.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.disaster;
+
+import static java.util.stream.Collectors.toList;
+import static org.apache.ignite.internal.metrics.exporters.jmx.JmxExporter.JMX_METRIC_GROUP;
+import static org.apache.ignite.internal.table.TableTestUtils.TABLE_NAME;
+import static org.apache.ignite.internal.table.TableTestUtils.getTableStrict;
+import static org.apache.ignite.internal.util.IgniteUtils.makeMbeanName;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.not;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.lang.management.ManagementFactory;
+import java.util.List;
+import javax.management.DynamicMBean;
+import javax.management.MBeanServer;
+import javax.management.MBeanServerInvocationHandler;
+import javax.management.ObjectName;
+import org.apache.ignite.InitParametersBuilder;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.metrics.MetricSource;
+import org.apache.ignite.internal.sql.BaseSqlIntegrationTest;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+/** For integration testing of disaster recovery metrics. */
+public class ItDisasterRecoveryMetricTest extends BaseSqlIntegrationTest {
+ private static final String ZONE_NAME = "ZONE_" + TABLE_NAME;
+
+ private final MBeanServer mbeanSrv = ManagementFactory.getPlatformMBeanServer();
+
+ @Override
+ protected int initialNodes() {
+ return 1;
+ }
+
+ @Override
+ protected void configureInitParameters(InitParametersBuilder builder) {
+ super.configureInitParameters(builder);
+
+ builder.clusterConfiguration("{metrics.exporters.jmx.exporterName: jmx}");
+ }
+
+ @AfterEach
+ void tearDown() {
+ sql("DROP TABLE IF EXISTS " + TABLE_NAME);
+ sql("DROP ZONE IF EXISTS " + ZONE_NAME);
+ }
+
+ @Test
+ void testPartitionStatesMetricsForTable() throws Exception {
+ createZoneAndTable(ZONE_NAME, TABLE_NAME, initialNodes(), 1);
+
+ IgniteImpl node = CLUSTER.aliveNode();
+
+ String expectedMetricSourceName = partitionStatesMetricSourceName(node, TABLE_NAME);
+
+ // Let's make sure that the metric source is registered.
+ assertThat(registeredMetricSourceNames(node), hasItem(expectedMetricSourceName));
+
+ node.metricManager().enable(expectedMetricSourceName);
+
+ DynamicMBean metricSourceMbean = metricSourceMbean(expectedMetricSourceName);
+
+ assertEquals(0L, metricSourceMbean.getAttribute("UnavailablePartitionCount"));
+ assertEquals(1L, metricSourceMbean.getAttribute("HealthyPartitionCount"));
+ assertEquals(0L, metricSourceMbean.getAttribute("InitializingPartitionCount"));
+ assertEquals(0L, metricSourceMbean.getAttribute("InstallingSnapshotPartitionCount"));
+ assertEquals(0L, metricSourceMbean.getAttribute("BrokenPartitionCount"));
+ }
+
+ @Test
+ void testPartitionStatesMetricsForTableAfterDropTable() {
+ createZoneAndTable(ZONE_NAME, TABLE_NAME, initialNodes(), 1);
+
+ IgniteImpl node = CLUSTER.aliveNode();
+
+ String expectedMetricSourceName = partitionStatesMetricSourceName(node, TABLE_NAME);
+
+ sql("DROP TABLE IF EXISTS " + TABLE_NAME);
+
+ // Let's make sure that the metric source is unregistered.
+ assertThat(registeredMetricSourceNames(node), not(hasItem(expectedMetricSourceName)));
+ }
+
+ private DynamicMBean metricSourceMbean(String metricSourceName) throws Exception {
+ ObjectName mbeanName = makeMbeanName(JMX_METRIC_GROUP, metricSourceName);
+
+ return MBeanServerInvocationHandler.newProxyInstance(mbeanSrv, mbeanName, DynamicMBean.class, false);
+ }
+
+ private static String partitionStatesMetricSourceName(IgniteImpl node, String tableName) {
+ CatalogTableDescriptor tableDescriptor = getTableStrict(node.catalogManager(), tableName, node.clock().nowLong());
+
+ return String.format("partition.states.zone.%s.table.%s", tableDescriptor.zoneId(), tableDescriptor.id());
+ }
+
+ private static List<String> registeredMetricSourceNames(IgniteImpl node) {
+ return node.metricManager().metricSources().stream().map(MetricSource::name).collect(toList());
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java
index f39a8fb138..fbd4a97c2b 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/DisasterRecoveryManager.java
@@ -24,16 +24,16 @@ import static java.util.stream.Collectors.groupingBy;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toMap;
import static java.util.stream.Collectors.toSet;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.TABLE_CREATE;
+import static org.apache.ignite.internal.catalog.events.CatalogEvent.TABLE_DROP;
+import static org.apache.ignite.internal.event.EventListener.fromConsumer;
import static org.apache.ignite.internal.table.distributed.disaster.DisasterRecoverySystemViews.createGlobalPartitionStatesSystemView;
import static org.apache.ignite.internal.table.distributed.disaster.DisasterRecoverySystemViews.createLocalPartitionStatesSystemView;
import static org.apache.ignite.internal.table.distributed.disaster.GlobalPartitionStateEnum.AVAILABLE;
import static org.apache.ignite.internal.table.distributed.disaster.GlobalPartitionStateEnum.DEGRADED;
import static org.apache.ignite.internal.table.distributed.disaster.GlobalPartitionStateEnum.READ_ONLY;
-import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.BROKEN;
import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.CATCHING_UP;
import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.HEALTHY;
-import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.INITIALIZING;
-import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.INSTALLING_SNAPSHOT;
import static org.apache.ignite.internal.util.CompletableFutures.copyStateTo;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.lang.ErrorGroups.DisasterRecovery.PARTITION_STATE_ERR;
@@ -55,6 +55,8 @@ import org.apache.ignite.internal.catalog.CatalogManager;
import org.apache.ignite.internal.catalog.descriptors.CatalogObjectDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
import org.apache.ignite.internal.catalog.descriptors.CatalogZoneDescriptor;
+import org.apache.ignite.internal.catalog.events.CreateTableEventParameters;
+import org.apache.ignite.internal.catalog.events.DropTableEventParameters;
import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
import org.apache.ignite.internal.distributionzones.NodeWithAttributes;
import org.apache.ignite.internal.distributionzones.exception.DistributionZoneNotFoundException;
@@ -68,6 +70,7 @@ import org.apache.ignite.internal.metastorage.Entry;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.WatchEvent;
import org.apache.ignite.internal.metastorage.WatchListener;
+import org.apache.ignite.internal.metrics.MetricManager;
import org.apache.ignite.internal.network.MessagingService;
import org.apache.ignite.internal.network.NetworkMessage;
import org.apache.ignite.internal.raft.Loza;
@@ -89,8 +92,6 @@ import org.apache.ignite.internal.util.CollectionUtils;
import org.apache.ignite.lang.TableNotFoundException;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.TopologyService;
-import org.apache.ignite.raft.jraft.Node;
-import org.apache.ignite.raft.jraft.core.State;
import org.jetbrains.annotations.Nullable;
/**
@@ -144,6 +145,9 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
/** Table manager. */
final TableManager tableManager;
+ /** Metric manager. */
+ private final MetricManager metricManager;
+
/**
* Map of operations, triggered by local node, that have not yet been processed by {@link #watchListener}. Values in the map are the
* futures, returned from the {@link #processNewRequest(DisasterRecoveryRequest)}, they are completed by
@@ -152,6 +156,8 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
*/
private final Map<UUID, CompletableFuture<Void>> ongoingOperationsById = new ConcurrentHashMap<>();
+ private final Map<Integer, PartitionStatesMetricSource> metricSourceByTableId = new ConcurrentHashMap<>();
+
/** Constructor. */
public DisasterRecoveryManager(
ExecutorService threadPool,
@@ -161,7 +167,8 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
DistributionZoneManager dzManager,
Loza raftManager,
TopologyService topologyService,
- TableManager tableManager
+ TableManager tableManager,
+ MetricManager metricManager
) {
this.threadPool = threadPool;
this.messagingService = messagingService;
@@ -171,6 +178,7 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
this.raftManager = raftManager;
this.topologyService = topologyService;
this.tableManager = tableManager;
+ this.metricManager = metricManager;
watchListener = new WatchListener() {
@Override
@@ -191,6 +199,12 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
metaStorageManager.registerExactWatch(RECOVERY_TRIGGER_KEY, watchListener);
+ catalogManager.listen(TABLE_CREATE, fromConsumer(this::onTableCreate));
+
+ catalogManager.listen(TABLE_DROP, fromConsumer(this::onTableDrop));
+
+ registerMetricSources();
+
return nullCompletedFuture();
}
@@ -536,35 +550,20 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
CatalogTableDescriptor tableDescriptor = catalogManager.table(tablePartitionId.tableId(), catalogVersion);
// Only tables that belong to a specific catalog version will be returned.
- if (tableDescriptor == null || !containsOrEmpty(tableDescriptor.zoneId(), request.zoneIds())
- ) {
+ if (tableDescriptor == null || !containsOrEmpty(tableDescriptor.zoneId(), request.zoneIds())) {
return;
}
- Node raftNode = raftGroupService.getRaftNode();
- State nodeState = raftNode.getNodeState();
-
- LocalPartitionStateEnum localState = convertState(nodeState);
- long lastLogIndex = raftNode.lastLogIndex();
-
- if (localState == HEALTHY) {
- // Node without log didn't process anything yet, it's not really "healthy" before it accepts leader's configuration.
- if (lastLogIndex == 0) {
- localState = INITIALIZING;
- }
-
- if (raftNode.isInstallingSnapshot()) {
- localState = INSTALLING_SNAPSHOT;
- }
- }
+ LocalPartitionStateEnumWithLogIndex localPartitionStateWithLogIndex =
+ LocalPartitionStateEnumWithLogIndex.of(raftGroupService.getRaftNode());
statesList.add(MSG_FACTORY.localPartitionStateMessage()
.partitionId(MSG_FACTORY.tablePartitionIdMessage()
.tableId(tablePartitionId.tableId())
.partitionId(tablePartitionId.partitionId())
.build())
- .state(localState)
- .logIndex(lastLogIndex)
+ .state(localPartitionStateWithLogIndex.state)
+ .logIndex(localPartitionStateWithLogIndex.logIndex)
.build()
);
}
@@ -580,34 +579,6 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
return collection.isEmpty() || collection.contains(item);
}
- /**
- * Converts internal raft node state into public local partition state.
- */
- private static LocalPartitionStateEnum convertState(State nodeState) {
- switch (nodeState) {
- case STATE_LEADER:
- case STATE_TRANSFERRING:
- case STATE_CANDIDATE:
- case STATE_FOLLOWER:
- return HEALTHY;
-
- case STATE_ERROR:
- return BROKEN;
-
- case STATE_UNINITIALIZED:
- return INITIALIZING;
-
- case STATE_SHUTTING:
- case STATE_SHUTDOWN:
- case STATE_END:
- return LocalPartitionStateEnum.UNAVAILABLE;
-
- default:
- // Unrecognized state, better safe than sorry.
- return BROKEN;
- }
- }
-
/**
* Replaces some healthy states with a {@link LocalPartitionStateEnum#CATCHING_UP}, it can only be done once the state of all peers is
* known.
@@ -783,4 +754,36 @@ public class DisasterRecoveryManager implements IgniteComponent, SystemViewProvi
ClusterNode localNode() {
return topologyService.localMember();
}
+
+ private void onTableCreate(CreateTableEventParameters parameters) {
+ registerPartitionStatesMetricSource(parameters.tableDescriptor());
+ }
+
+ private void onTableDrop(DropTableEventParameters parameters) {
+ unregisterPartitionStatesMetricSource(parameters.tableId());
+ }
+
+ private void registerMetricSources() {
+ int catalogVersion = catalogManager.latestCatalogVersion();
+
+ catalogManager.tables(catalogVersion).forEach(this::registerPartitionStatesMetricSource);
+ }
+
+ private void registerPartitionStatesMetricSource(CatalogTableDescriptor tableDescriptor) {
+ var metricSource = new PartitionStatesMetricSource(tableDescriptor, this);
+
+ PartitionStatesMetricSource previous = metricSourceByTableId.putIfAbsent(tableDescriptor.id(), metricSource);
+
+ assert previous == null : "tableId=" + tableDescriptor.id();
+
+ metricManager.registerSource(metricSource);
+ }
+
+ private void unregisterPartitionStatesMetricSource(int tableId) {
+ PartitionStatesMetricSource metricSource = metricSourceByTableId.get(tableId);
+
+ assert metricSource != null : "tableId=" + tableId;
+
+ metricManager.unregisterSource(metricSource);
+ }
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateEnum.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateEnum.java
index fd9536bf66..41ff9bd75a 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateEnum.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateEnum.java
@@ -17,6 +17,8 @@
package org.apache.ignite.internal.table.distributed.disaster;
+import org.apache.ignite.raft.jraft.core.State;
+
/**
* Enum for states of local partitions.
*/
@@ -37,5 +39,31 @@ public enum LocalPartitionStateEnum {
CATCHING_UP,
/** Partition is in broken state, usually it means that its state machine thrown an exception. */
- BROKEN
+ BROKEN;
+
+ /** Converts internal raft node state into public local partition state. */
+ static LocalPartitionStateEnum convert(State raftNodeState) {
+ switch (raftNodeState) {
+ case STATE_LEADER:
+ case STATE_TRANSFERRING:
+ case STATE_CANDIDATE:
+ case STATE_FOLLOWER:
+ return HEALTHY;
+
+ case STATE_ERROR:
+ return BROKEN;
+
+ case STATE_UNINITIALIZED:
+ return INITIALIZING;
+
+ case STATE_SHUTTING:
+ case STATE_SHUTDOWN:
+ case STATE_END:
+ return UNAVAILABLE;
+
+ default:
+ // Unrecognized state, better safe than sorry.
+ return BROKEN;
+ }
+ }
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateEnumWithLogIndex.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateEnumWithLogIndex.java
new file mode 100644
index 0000000000..bd1b27f668
--- /dev/null
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/LocalPartitionStateEnumWithLogIndex.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed.disaster;
+
+import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.HEALTHY;
+import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.INITIALIZING;
+import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.INSTALLING_SNAPSHOT;
+
+import org.apache.ignite.raft.jraft.Node;
+import org.apache.ignite.raft.jraft.core.State;
+
+class LocalPartitionStateEnumWithLogIndex {
+ final LocalPartitionStateEnum state;
+
+ final long logIndex;
+
+ private LocalPartitionStateEnumWithLogIndex(LocalPartitionStateEnum state, long logIndex) {
+ this.state = state;
+ this.logIndex = logIndex;
+ }
+
+ static LocalPartitionStateEnumWithLogIndex of(Node raftNode) {
+ State nodeState = raftNode.getNodeState();
+
+ LocalPartitionStateEnum localState = LocalPartitionStateEnum.convert(nodeState);
+ long lastLogIndex = raftNode.lastLogIndex();
+
+ if (localState == HEALTHY) {
+ // Node without log didn't process anything yet, it's not really "healthy" before it accepts leader's configuration.
+ if (lastLogIndex == 0) {
+ localState = INITIALIZING;
+ }
+
+ if (raftNode.isInstallingSnapshot()) {
+ localState = INSTALLING_SNAPSHOT;
+ }
+ }
+
+ return new LocalPartitionStateEnumWithLogIndex(localState, lastLogIndex);
+ }
+}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/PartitionStatesMetricSource.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/PartitionStatesMetricSource.java
new file mode 100644
index 0000000000..68a19cce41
--- /dev/null
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/disaster/PartitionStatesMetricSource.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.table.distributed.disaster;
+
+import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.BROKEN;
+import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.HEALTHY;
+import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.INITIALIZING;
+import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.INSTALLING_SNAPSHOT;
+import static org.apache.ignite.internal.table.distributed.disaster.LocalPartitionStateEnum.UNAVAILABLE;
+
+import org.apache.ignite.internal.catalog.descriptors.CatalogTableDescriptor;
+import org.apache.ignite.internal.metrics.MetricSet;
+import org.apache.ignite.internal.metrics.MetricSetBuilder;
+import org.apache.ignite.internal.metrics.MetricSource;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.jetbrains.annotations.Nullable;
+
+/** Source of metrics for table partition statuses. */
+class PartitionStatesMetricSource implements MetricSource {
+ private final String metricSourceName;
+
+ private final int tableId;
+
+ private final DisasterRecoveryManager disasterRecoveryManager;
+
+ /** Enablement status. Accessed from different threads under synchronization on this object. */
+ private boolean enabled;
+
+ PartitionStatesMetricSource(
+ CatalogTableDescriptor tableDescriptor,
+ DisasterRecoveryManager disasterRecoveryManager
+ ) {
+ this.tableId = tableDescriptor.id();
+ this.disasterRecoveryManager = disasterRecoveryManager;
+
+ metricSourceName = String.format("partition.states.zone.%s.table.%s", tableDescriptor.zoneId(), tableDescriptor.id());
+ }
+
+ @Override
+ public String name() {
+ return metricSourceName;
+ }
+
+ @Override
+ public synchronized @Nullable MetricSet enable() {
+ if (enabled) {
+ return null;
+ }
+
+ var builder = new MetricSetBuilder(metricSourceName);
+
+ builder.longGauge(
+ "UnavailablePartitionCount",
+ "Count of partitions not yet started.",
+ () -> calculatePartitionCountByLocalState(UNAVAILABLE)
+ );
+
+ builder.longGauge(
+ "HealthyPartitionCount",
+ "Count of living partitions with a healthy state machine.",
+ () -> calculatePartitionCountByLocalState(HEALTHY)
+ );
+
+ builder.longGauge(
+ "InitializingPartitionCount",
+ "Count of partitions that are starting right now.",
+ () -> calculatePartitionCountByLocalState(INITIALIZING)
+ );
+
+ builder.longGauge(
+ "InstallingSnapshotPartitionCount",
+ "Count of partitions that installing Raft snapshots from the leader.",
+ () -> calculatePartitionCountByLocalState(INSTALLING_SNAPSHOT)
+ );
+
+ builder.longGauge(
+ "BrokenPartitionCount",
+ "Count of broken partitions.",
+ () -> calculatePartitionCountByLocalState(BROKEN)
+ );
+
+ return builder.build();
+ }
+
+ @Override
+ public synchronized void disable() {
+ enabled = false;
+ }
+
+ @Override
+ public synchronized boolean enabled() {
+ return enabled;
+ }
+
+ private long calculatePartitionCountByLocalState(LocalPartitionStateEnum state) {
+ long[] count = {0};
+
+ // When receiving/loading metrics for all tables, the total complexity will be O(N*N) where N is the number of tables (for
+ // simplicity). While this is done intentionally, in the future it will be necessary to optimize this or wait for the collocation
+ // of table partitions within one distribution zone.
+ disasterRecoveryManager.raftManager.forEach((raftNodeId, raftGroupService) -> {
+ if (raftNodeId.groupId() instanceof TablePartitionId) {
+ var tablePartitionId = (TablePartitionId) raftNodeId.groupId();
+
+ if (tablePartitionId.tableId() == tableId) {
+ LocalPartitionStateEnumWithLogIndex localPartitionStateWithLogIndex =
+ LocalPartitionStateEnumWithLogIndex.of(raftGroupService.getRaftNode());
+
+ if (localPartitionStateWithLogIndex.state == state) {
+ count[0]++;
+ }
+ }
+ }
+ });
+
+ return count[0];
+ }
+}