diff --git a/docs/_docs/snapshots/snapshots.adoc b/docs/_docs/snapshots/snapshots.adoc index fe61b3c49234a..5afeb689779a3 100644 --- a/docs/_docs/snapshots/snapshots.adoc +++ b/docs/_docs/snapshots/snapshots.adoc @@ -289,35 +289,36 @@ control.(sh|bat) --snapshot restore snapshot_09062021 --increment 1 == Getting Snapshot Operation Status -The status of the current snapshot operation in the cluster can be obtained using the `control.sh|bat` script or JMX interface: +The status of current snapshot operations in the cluster can be obtained using the `control.sh|bat` script. + +Snapshot operations include: │ +* `Create` - Full or incremental snapshot creation +* `Restore` - Cache group restoration from snapshot +* `Check` - Snapshot consistency verification (parallel execution allowed) │ [tabs] -- tab:Unix[] [source,shell] ---- -# Get the status of the snapshot operation. +# Get the status of snapshot operation. control.sh --snapshot status ---- tab:Windows[] [source,shell] ---- -# Get the status of the snapshot operation. +# Get the status of snapshot operation. control.bat --snapshot status ---- - -tab:JMX[] -You can also get the current snapshot status via the `SnapshotMXBean` interface: -[source,java] ----- -SnapshotMXBean mxBean = ...; - -// The status of a current snapshot operation in the cluster. -String status = mxBean.status(); ----- -- + [NOTE] │ +==== │ +Parallel and multiple snapshot check operations can be running concurrently with any other snapshot operation. +But only one check for one snapshot. Create and restore operations cannot run in parallel - only one of each type can +be active at a time. The status is printed for each running snapshot operation (if multiple) separately, one after another. + == Cancelling Snapshot Operation To abort create/restore snapshot operation you need to obtain an `operation request ID`. diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java index b5f2659b1fa0b..b58e2eb756d18 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java @@ -77,6 +77,7 @@ import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInternalFuture; import org.apache.ignite.internal.IgniteInterruptedCheckedException; +import org.apache.ignite.internal.IgniteVersionUtils; import org.apache.ignite.internal.Order; import org.apache.ignite.internal.TestRecordingCommunicationSpi; import org.apache.ignite.internal.dto.IgniteDataTransferObject; @@ -116,11 +117,16 @@ import org.apache.ignite.internal.processors.cluster.GridClusterStateProcessor; import org.apache.ignite.internal.processors.datastreamer.DataStreamerRequest; import org.apache.ignite.internal.processors.metric.MetricRegistryImpl; +import org.apache.ignite.internal.processors.nodevalidation.DiscoveryNodeValidationProcessor; +import org.apache.ignite.internal.processors.rollingupgrade.RollingUpgradeProcessor; +import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteCoreFeatureSet; +import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeatureSet; import org.apache.ignite.internal.util.BasicRateLimiter; import org.apache.ignite.internal.util.GridConcurrentHashSet; import org.apache.ignite.internal.util.distributed.DistributedProcess; import org.apache.ignite.internal.util.distributed.SingleNodeMessage; import org.apache.ignite.internal.util.future.IgniteFinishedFutureImpl; +import org.apache.ignite.internal.util.future.IgniteFutureImpl; import org.apache.ignite.internal.util.lang.GridAbsPredicate; import org.apache.ignite.internal.util.lang.GridFunc; import org.apache.ignite.internal.util.typedef.F; @@ -134,7 +140,11 @@ import org.apache.ignite.lang.IgnitePredicate; import org.apache.ignite.lang.IgniteUuid; import org.apache.ignite.metric.MetricRegistry; +import org.apache.ignite.plugin.AbstractTestPluginProvider; +import org.apache.ignite.plugin.PluginContext; +import org.apache.ignite.plugin.PluginProvider; import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.spi.IgniteNodeValidationResult; import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi; import org.apache.ignite.spi.metric.LongMetric; import org.apache.ignite.spi.metric.Metric; @@ -148,6 +158,7 @@ import org.apache.ignite.transactions.TransactionRollbackException; import org.apache.ignite.transactions.TransactionTimeoutException; import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; import org.junit.Assume; import org.junit.Test; @@ -182,6 +193,8 @@ import static org.apache.ignite.internal.processors.diagnostic.DiagnosticProcessor.DEFAULT_TARGET_FOLDER; import static org.apache.ignite.internal.processors.job.GridJobProcessor.JOBS_VIEW; import static org.apache.ignite.internal.processors.task.GridTaskProcessor.TASKS_VIEW; +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CHECK_SNAPSHOT_PARTS; +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.END_SNAPSHOT; import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE; import static org.apache.ignite.testframework.GridTestUtils.assertContains; import static org.apache.ignite.testframework.GridTestUtils.assertNotContains; @@ -233,6 +246,9 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb /** */ protected ListeningTestLogger listeningLog; + /** */ + protected @Nullable PluginProvider pluginProvider; + /** {@inheritDoc} */ @Override protected void beforeTest() throws Exception { super.beforeTest(); @@ -242,13 +258,6 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb cleanDiagnosticDir(); } - /** {@inheritDoc} */ - @Override protected void afterTest() throws Exception { - super.afterTest(); - - listeningLog = null; - } - /** {@inheritDoc} */ @Override protected void cleanPersistenceDir() throws Exception { super.cleanPersistenceDir(); @@ -263,6 +272,9 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb if (listeningLog != null) cfg.setGridLogger(listeningLog); + if (pluginProvider != null) + cfg.setPluginProviders(pluginProvider); + return cfg; } @@ -3658,6 +3670,335 @@ public void testSnapshotRestoreCancelAndStatus() throws Exception { assertNull(ig.cache(DEFAULT_CACHE_NAME)); } + /** + * Tests that snapshot metrics aren't empty when being restored snapshot waits for the check process. + * All nodes support the check status. + */ + @Test + public void testRestoreSnapshotMetricsAtStartAllNodesSupport() throws Exception { + doTestRestoreSnapshotMetricsAtStart(false); + } + + /** + * Tests that snapshot metrics aren't empty when being restored snapshot waits for the check process. + * One node doesn't support the check status. + */ + @Test + public void testRestoreSnapshotMetricsAtStartOneNodeDoesntSupport() throws Exception { + doTestRestoreSnapshotMetricsAtStart(true); + } + + /** + * @param oneNodeDoesntSupport If {@code true}, one node will simulate not supporting snapshot check status. + */ + private void doTestRestoreSnapshotMetricsAtStart(boolean oneNodeDoesntSupport) throws Exception { + // Creates empty feature set unsupporting the snapshot check ststus if required. + pluginProvider = !oneNodeDoesntSupport ? null : new AbstractTestPluginProvider() { + @Override public String name() { + return "Test Ignite features provider"; + } + + @Override public @Nullable T createComponent(PluginContext ctx, Class cls) { + if (!cls.equals(DiscoveryNodeValidationProcessor.class)) + return null; + + boolean doNotSupport = ctx.igniteConfiguration().getIgniteInstanceName().equals(getTestIgniteInstanceName(2)); + + return (T)new RollingUpgradeProcessor( + ((IgniteEx)ctx.grid()).context(), + doNotSupport ? new IgniteCoreFeatureSet(IgniteVersionUtils.VER, new IgniteFeatureSet()) : IgniteCoreFeatureSet.local() + ) { + @Override public @Nullable IgniteNodeValidationResult validateNode(ClusterNode joiningNode) { + // Simulates started rolling updrade allowing node with other features join cluster. + return null; + } + }; + } + }; + + listeningLog = new ListeningTestLogger(log); + + startGrids(3).cluster().state(ClusterState.ACTIVE); + + createCacheAndPreload(grid(1), 1000); + + IgniteSnapshotManager snpMgr = (IgniteSnapshotManager)grid(0).snapshot(); + + snpMgr.createSnapshot("test_snapshot").get(getTestTimeout()); + + grid(0).destroyCache(DEFAULT_CACHE_NAME); + + awaitPartitionMapExchange(); + + TestRecordingCommunicationSpi cm1 = ((TestRecordingCommunicationSpi)grid(1).configuration().getCommunicationSpi()); + TestRecordingCommunicationSpi cm2 = ((TestRecordingCommunicationSpi)grid(2).configuration().getCommunicationSpi()); + + // Block one of the process' first messages of snapshot restoring or snapshot checking. + F.asList(cm1, cm2).forEach(cm -> cm.blockMessages((node, msg) -> msg instanceof SingleNodeMessage sm + && (sm.type() == CHECK_SNAPSHOT_PARTS.ordinal() || sm.type() == RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE.ordinal()))); + + // Snapshot restoration should get paused at the preceeding snapshot check. + IgniteFutureImpl restoreFut = snpMgr.restoreSnapshot("test_snapshot", null, null, 0, true); + + // Waiting for each node to send snapshot check single result. + for (var cm : F.asList(cm1, cm2)) { + assertTrue(waitForCondition(() -> cm.blockedMessages().stream().anyMatch(m -> + m.ioMessage().message() instanceof SingleNodeMessage sm && sm.type() == CHECK_SNAPSHOT_PARTS.ordinal()), + getTestTimeout())); + } + + injectTestSystemOut(); + + LogListener logLsnr = null; + + if (oneNodeDoesntSupport) { + logLsnr = LogListener.matches("Node %s doesn't support the snapshot check status feature" + .formatted(grid(2).localNode().id())).build(); + + listeningLog.registerListener(logLsnr); + } + + int code = execute("--snapshot", "status"); + + assertEquals("Unexpected exit code", EXIT_CODE_OK, code); + + var out = testOut.toString(); + + if (oneNodeDoesntSupport) { + assertTrue(logLsnr.check(getTestTimeout())); + assertContains(log, out, "There is no create or restore snapshot operation in progress"); + } + else { + assertContains(log, out, "Check snapshot operation is in progress"); + assertContains(log, out, "Snapshot name: test_snapshot"); + } + + // Ensure that no snapshot restore started or finished. + assertFalse("Snapshot future has finished", restoreFut.isDone()); + + F.asList(cm1, cm2).forEach(cm -> { + assertTrue(cm.blockedMessages().stream().noneMatch(m -> + m.ioMessage().message() instanceof SingleNodeMessage sm + && sm.type() == RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE.ordinal())); + + cm.stopBlock(); + }); + + restoreFut.get(getTestTimeout()); + } + + /** */ + @Test + public void testOneSnapshotCheckStatus() throws Exception { + doTestSnapshotsChecksStatus(false, false); + } + + /** */ + @Test + public void testOneIncrementalSnapshotCheckStatus() throws Exception { + doTestSnapshotsChecksStatus(false, true); + } + + /** */ + @Test + public void testTwoSnapshotsChecksStatus() throws Exception { + doTestSnapshotsChecksStatus(true, false); + } + + /** */ + @Test + public void testTwoIncrementalsSnapshotsChecksStatus() throws Exception { + doTestSnapshotsChecksStatus(true, true); + } + + /** */ + private void doTestSnapshotsChecksStatus(boolean twoSnapshots, boolean incremental) throws Exception { + walCompactionEnabled(incremental); + + startGrids(3).cluster().state(ClusterState.ACTIVE); + + IgniteSnapshotManager snpMgr = (IgniteSnapshotManager)grid(0).snapshot(); + + createCacheAndPreload(grid(1), DEFAULT_CACHE_NAME, 1000, 32, null); + snpMgr.createSnapshot("testSnapshot0").get(getTestTimeout()); + + if (incremental) { + try (IgniteDataStreamer streamer = grid(0).dataStreamer(DEFAULT_CACHE_NAME)) { + for (int i = 1000; i < 2000; i++) + streamer.addData(i, i); + } + + snpMgr.createIncrementalSnapshot("testSnapshot0").get(getTestTimeout()); + } + + if (twoSnapshots) { + createCacheAndPreload(grid(1), "cache2", 1000, 32, null); + snpMgr.createSnapshot("testSnapshot1").get(getTestTimeout()); + + if (incremental) { + try (IgniteDataStreamer streamer = grid(0).dataStreamer("cache2")) { + for (int i = 1000; i < 2000; i++) + streamer.addData(i, i); + } + + snpMgr.createIncrementalSnapshot("testSnapshot1").get(getTestTimeout()); + } + } + + TestRecordingCommunicationSpi cm1 = ((TestRecordingCommunicationSpi)grid(1).configuration().getCommunicationSpi()); + TestRecordingCommunicationSpi cm2 = ((TestRecordingCommunicationSpi)grid(2).configuration().getCommunicationSpi()); + + F.asList(cm1, cm2).forEach(cm -> cm.blockMessages((node, msg) -> + msg instanceof SingleNodeMessage sm && (sm.type() == CHECK_SNAPSHOT_PARTS.ordinal()))); + + var checkFut0 = runAsync(() -> incremental + ? execute("--snapshot", "check", "testSnapshot0", "--increment", "1") + : execute("--snapshot", "check", "testSnapshot0")); + + var checkFut1 = twoSnapshots + ? runAsync(() -> incremental + ? execute("--snapshot", "check", "testSnapshot1", "--increment", "1") + : execute("--snapshot", "check", "testSnapshot1")) + : null; + + // Waiting for the nodes each to send snapshot check single result. + for (var cm : F.asList(cm1, cm2)) { + assertTrue(waitForCondition(() -> cm.blockedMessages().stream().filter(m -> + m.ioMessage().message() instanceof SingleNodeMessage sm + && sm.type() == CHECK_SNAPSHOT_PARTS.ordinal()).count() == (twoSnapshots ? 2 : 1), + getTestTimeout() + )); + } + + injectTestSystemOut(); + + assertEquals("Unexpected exit code", EXIT_CODE_OK, execute("--snapshot", "status")); + + var out = testOut.toString(); + + if (log.isInfoEnabled()) + log.info("Test out:" + U.nl() + out); + + String chkSnpE = "Check snapshot operation is in progress"; + var chkSnpESum = Arrays.stream(out.split(U.nl())) + .mapToInt(l -> (l.length() - l.replace(chkSnpE, "").length()) / chkSnpE.length()) + .sum(); + assertEquals(twoSnapshots ? 2 : 1, chkSnpESum); + + assertTrue(out.contains("Snapshot name: testSnapshot0")); + + if (incremental) + assertTrue(out.contains("Increment index: 1")); + + if (twoSnapshots) { + assertTrue(out.contains("Snapshot name: testSnapshot1")); + + if (incremental) { + // Number of 'Increment index: 1' entries. + var sum = Arrays.stream(out.split(U.nl())) + .mapToInt(l -> (l.length() - l.replace("Increment index: 1", "").length()) / "Increment index: 1".length()) + .sum(); + + assertEquals(2, sum); + } + } + + F.asList(cm1, cm2).forEach(TestRecordingCommunicationSpi::stopBlock); + + checkFut0.get(); + + if (twoSnapshots) + checkFut1.get(); + } + + /** */ + @Test + public void testSnapshotCheckStatusWithParallelCreate() throws Exception { + doTestSnapshotsChecksStatusWithParallelCreate(false); + } + + /** */ + @Test + public void testTwoSnapshotsChecksStatusWithParallelCreate() throws Exception { + doTestSnapshotsChecksStatusWithParallelCreate(true); + } + + /** */ + private void doTestSnapshotsChecksStatusWithParallelCreate(boolean twoChecks) throws Exception { + startGrids(3).cluster().state(ClusterState.ACTIVE); + + IgniteSnapshotManager snpMgr = (IgniteSnapshotManager)grid(0).snapshot(); + + createCacheAndPreload(grid(1), DEFAULT_CACHE_NAME, 1000, 32, null); + snpMgr.createSnapshot("testSnapshot0").get(getTestTimeout()); + + if (twoChecks) { + createCacheAndPreload(grid(1), "cache2", 1000, 32, null); + snpMgr.createSnapshot("testSnapshot1").get(getTestTimeout()); + } + + createCacheAndPreload(grid(1), "cache3", 1000, 32, null); + + TestRecordingCommunicationSpi cm1 = ((TestRecordingCommunicationSpi)grid(1).configuration().getCommunicationSpi()); + TestRecordingCommunicationSpi cm2 = ((TestRecordingCommunicationSpi)grid(2).configuration().getCommunicationSpi()); + + F.asList(cm1, cm2).forEach(cm -> cm.blockMessages((node, msg) -> + msg instanceof SingleNodeMessage sm && (sm.type() == CHECK_SNAPSHOT_PARTS.ordinal() + || sm.type() == END_SNAPSHOT.ordinal())) + ); + + var createFut = snpMgr.createSnapshot("testSnapshot2"); + + var checkFut0 = runAsync(() -> execute("--snapshot", "check", "testSnapshot0")); + var checkFut1 = twoChecks ? runAsync(() -> execute("--snapshot", "check", "testSnapshot1")) : null; + + // Waiting for the nodes each to send snapshot create single result. + for (var cm : F.asList(cm1, cm2)) { + assertTrue(waitForCondition(() -> cm.blockedMessages().stream().anyMatch(m -> + m.ioMessage().message() instanceof SingleNodeMessage sm && sm.type() == END_SNAPSHOT.ordinal()), + getTestTimeout() + )); + } + + // Waiting for the nodes each to send snapshot check single result. + for (var cm : F.asList(cm1, cm2)) { + assertTrue(waitForCondition(() -> cm.blockedMessages().stream().filter(m -> + m.ioMessage().message() instanceof SingleNodeMessage sm + && sm.type() == CHECK_SNAPSHOT_PARTS.ordinal()).count() == (twoChecks ? 2 : 1), + getTestTimeout() + )); + } + + injectTestSystemOut(); + + assertEquals("Unexpected exit code", EXIT_CODE_OK, execute("--snapshot", "status")); + + var out = testOut.toString(); + + if (log.isInfoEnabled()) + log.info("Test out:" + U.nl() + out); + + assertTrue(out.contains("Create snapshot operation is in progress")); + assertTrue(out.contains("Snapshot name: testSnapshot2")); + + String chkLogE = "Check snapshot operation is in progress"; + var chkLogSum = Arrays.stream(out.split(U.nl())) + .mapToInt(l -> (l.length() - l.replace(chkLogE, "").length()) / chkLogE.length()) + .sum(); + assertEquals(twoChecks ? 2 : 1, chkLogSum); + + assertTrue(out.contains("Snapshot name: testSnapshot0")); + if (twoChecks) + assertTrue(out.contains("Snapshot name: testSnapshot1")); + + F.asList(cm1, cm2).forEach(TestRecordingCommunicationSpi::stopBlock); + + createFut.get(); + checkFut0.get(); + if (twoChecks) + checkFut1.get(); + } + /** @throws Exception If fails. */ @Test public void testSnapshotStatusInMemory() throws Exception { diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotStatusCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotStatusCommand.java index 18f4fc5b690e6..b2210a7e96975 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotStatusCommand.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotStatusCommand.java @@ -26,6 +26,7 @@ import java.util.UUID; import java.util.function.Consumer; import java.util.stream.Collectors; +import java.util.stream.Stream; import org.apache.ignite.internal.management.SystemViewCommand; import org.apache.ignite.internal.management.SystemViewTask; import org.apache.ignite.internal.management.api.NoArg; @@ -64,48 +65,73 @@ public class SnapshotStatusCommand extends AbstractSnapshotCommand 0; + assert status.checkStatuses() != null || status.operation() != SnapshotStatusTask.SnapshotOperation.CHECK + : "No create or restore snapshot operation found but the check statuses are also empty."; - GridStringBuilder s = new GridStringBuilder(); + // The check operation can be run in parallel for different snapshots. + List multipleOpsView; - if (isCreating) - s.a("Create snapshot operation is in progress.").nl(); - else - s.a("Restore snapshot operation is in progress.").nl(); + if (status.operation() == SnapshotStatusTask.SnapshotOperation.CHECK) { + // Check operation always has itself in its aggregated check statuses. + multipleOpsView = status.checkStatuses(); + } + else { + // If the operation is not check, attach possible parallel checks after. + multipleOpsView = status.checkStatuses() == null + ? Collections.singletonList(status) + : Stream.concat(Stream.of(status), status.checkStatuses().stream()).collect(Collectors.toList()); + } + + boolean first = true; + + for (SnapshotStatus s0 : multipleOpsView) { + if (!first) + printer.accept(U.nl()); - s.a("Snapshot name: ").a(status.name()).nl(); - s.a("Incremental: ").a(isIncremental).nl(); + if (s0.operation() == SnapshotStatusTask.SnapshotOperation.CREATE) + printer.accept("Create snapshot operation is in progress."); + else if (s0.operation() == SnapshotStatusTask.SnapshotOperation.RESTORE) + printer.accept("Restore snapshot operation is in progress."); + else + printer.accept("Check snapshot operation is in progress."); + + printer.accept(""); - if (isIncremental) - s.a("Increment index: ").a(status.incrementIndex()).nl(); + GridStringBuilder s = new GridStringBuilder(); - s.a("Operation request ID: ").a(status.requestId()).nl(); - s.a("Started at: ").a(DateFormat.getDateTimeInstance().format(new Date(status.startTime()))).nl(); - s.a("Duration: ").a(X.timeSpan2DHMSM(System.currentTimeMillis() - status.startTime())).nl() - .nl(); - s.a("Estimated operation progress:").nl(); + boolean incremental = s0.incrementIndex() > 0; - printer.accept(s.toString()); + s.a("Snapshot name: ").a(s0.name()).nl(); + s.a("Incremental: ").a(incremental).nl(); - SnapshotTaskProgressDesc desc; + if (incremental) + s.a("Increment index: ").a(s0.incrementIndex()).nl(); - if (isCreating && isIncremental) - desc = new CreateIncrementalSnapshotTaskProgressDesc(); - else if (isCreating) - desc = new CreateFullSnapshotTaskProgressDesc(); - else if (isIncremental) - desc = new RestoreIncrementalSnapshotTaskProgressDesc(); - else - desc = new RestoreFullSnapshotTaskProgressDesc(); + s.a("Operation request ID: ").a(s0.requestId()).nl(); + s.a("Started at: ").a(DateFormat.getDateTimeInstance().format(new Date(s0.startTime()))).nl(); + s.a("Duration: ").a(X.timeSpan2DHMSM(System.currentTimeMillis() - s0.startTime())).nl() + .nl(); + s.a("Estimated operation progress:").nl(); - List> rows = status.progress().entrySet().stream().sorted(Map.Entry.comparingByKey()) - .map(e -> desc.buildRow(e.getKey(), e.getValue())) - .collect(Collectors.toList()); + printer.accept(s.toString()); - SystemViewCommand.printTable(desc.titles(), desc.types(), rows, printer); + SnapshotTaskProgressDesc desc; + + if (s0.operation() == SnapshotStatusTask.SnapshotOperation.CREATE) + desc = incremental ? new CreateIncrementalSnapshotTaskProgressDesc() : new CreateFullSnapshotTaskProgressDesc(); + else if (s0.operation() == SnapshotStatusTask.SnapshotOperation.RESTORE) + desc = incremental ? new RestoreIncrementalSnapshotTaskProgressDesc() : new RestoreFullSnapshotTaskProgressDesc(); + else + desc = new CheckSnapshotTaskProgressDesc(incremental); - printer.accept(U.nl()); + List> rows = s0.progress().entrySet().stream().sorted(Map.Entry.comparingByKey()) + .map(e -> desc.buildRow(e.getKey(), e.getValue())) + .collect(Collectors.toList()); + + SystemViewCommand.printTable(desc.titles(), desc.types(), rows, printer); + + first = false; + } } /** Describes progress of a snapshot task. */ @@ -246,4 +272,29 @@ private static class RestoreIncrementalSnapshotTaskProgressDesc extends Snapshot return result; } } + + /** */ + private static class CheckSnapshotTaskProgressDesc extends SnapshotTaskProgressDesc { + /** */ + private CheckSnapshotTaskProgressDesc(boolean incremental) { + super(incremental + ? F.asList("Node ID", "Processed WAL segments", "Total WAL segments", "Percent") + : F.asList("Node ID", "Processed partitions", "Total partitions", "Percent") + ); + } + + /** {@inheritDoc} */ + @Override public List buildRow(UUID nodeId, T5 progress) { + long total = progress.get2(); + + if (total <= 0) + return F.asList(nodeId, "unknown", "unknown", "unknown"); + + long processed = progress.get1(); + + String percent = (int)(processed * 100 / total) + "%"; + + return F.asList(nodeId, processed, total, percent); + } + } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotStatusTask.java b/modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotStatusTask.java index 01476d309e71f..d092339d039e9 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotStatusTask.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotStatusTask.java @@ -18,6 +18,7 @@ package org.apache.ignite.internal.management.snapshot; import java.io.Serializable; +import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -25,20 +26,29 @@ import java.util.Map; import java.util.UUID; import org.apache.ignite.IgniteException; +import org.apache.ignite.IgniteLogger; import org.apache.ignite.compute.ComputeJobResult; import org.apache.ignite.internal.management.api.NoArg; +import org.apache.ignite.internal.managers.discovery.IgniteClusterNode; import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckProcess; import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationRequest; +import org.apache.ignite.internal.processors.metric.impl.MetricUtils; +import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteCoreFeature; +import org.apache.ignite.internal.processors.rollingupgrade.feature.SupportedFeatureRegistry; import org.apache.ignite.internal.processors.task.GridInternal; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.T5; import org.apache.ignite.internal.util.typedef.internal.CU; +import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.internal.visor.VisorJob; import org.apache.ignite.internal.visor.VisorMultiNodeTask; import org.apache.ignite.internal.visor.VisorTaskArgument; import org.apache.ignite.metric.MetricRegistry; +import org.apache.ignite.resources.LoggerResource; import org.apache.ignite.spi.metric.IntMetric; import org.apache.ignite.spi.metric.LongMetric; +import org.apache.ignite.spi.metric.Metric; import org.jetbrains.annotations.Nullable; import static org.apache.ignite.internal.management.snapshot.SnapshotStatusTask.SnapshotStatus; @@ -75,23 +85,76 @@ public class SnapshotStatusTask extends VisorMultiNodeTask res = F.viewReadOnly(results, ComputeJobResult::getData, r -> r.getData() != null); + // Try to find check statuses. + Map chkStatuses = null; + + // Not empty results. + Collection statuses = new ArrayList<>(results.size()); + + for (ComputeJobResult jobRes : results) { + if (jobRes.getData() == null) + continue; + + assert jobRes.getData() instanceof SnapshotStatus; + + SnapshotStatus s = jobRes.getData(); + + statuses.add(s); + + if (s.checkStatuses == null) { + assert s.op != SnapshotOperation.CHECK; + + continue; + } + + if (chkStatuses == null) + chkStatuses = U.newHashMap(results.size()); + + for (SnapshotStatus s0 : s.checkStatuses) { + var prev = chkStatuses.putIfAbsent(s0.name(), s0); + + if (prev == null) + continue; + + // Merge nodes progress. + prev.progress.putAll(s0.progress()); + } + } + + // First, find create or restore operation. Check statuses, if are, contains any status. + Collection res = F.viewReadOnly(statuses, s -> s, s -> s.operation() != SnapshotOperation.CHECK); + + if (res.isEmpty()) + res = statuses; // There is no snapshot operation. if (res.isEmpty()) return null; - SnapshotStatus s0 = F.first(res); + SnapshotStatus mergerRes = F.first(res); + + var rqId = mergerRes.requestId; // Filter out differing requests due to concurrent updates on nodes. - res = F.view(res, s -> s.requestId.equals(s0.requestId)); + res = F.view(res, s -> s.requestId.equals(rqId)); + + // Create or restore. + if (mergerRes.operation() != SnapshotOperation.CHECK) { + // Merge nodes progress. + Map> progress = new HashMap<>(); + + res.forEach(s -> progress.putAll(s.progress)); - // Merge nodes progress. - Map> progress = new HashMap<>(); + mergerRes = new SnapshotStatus(mergerRes.op, mergerRes.name, mergerRes.incIdx, mergerRes.requestId, + mergerRes.startTime, progress); + } + + if (chkStatuses != null) + mergerRes.checkStatuses = new ArrayList<>(chkStatuses.values()); - res.forEach(s -> progress.putAll(s.progress)); + assert !F.isEmpty(mergerRes.progress); - return new SnapshotStatus(s0.op, s0.name, s0.incIdx, s0.requestId, s0.startTime, progress); + return mergerRes; } /** */ @@ -99,6 +162,26 @@ private static class SnapshotStatusJob extends SnapshotJob metrics; @@ -130,7 +213,7 @@ protected SnapshotStatusJob(@Nullable NoArg arg, boolean debug) { -1L, -1L, -1L); } - return new SnapshotStatus( + res = new SnapshotStatus( SnapshotOperation.CREATE, req.snapshotName(), req.incrementIndex(), @@ -139,32 +222,91 @@ protected SnapshotStatusJob(@Nullable NoArg arg, boolean debug) { F.asMap(ignite.localNode().id(), metrics) ); } - - MetricRegistry mreg = ignite.context().metric().registry(SNAPSHOT_RESTORE_METRICS); - - long startTime = mreg.findMetric("startTime").value(); - - if (startTime > mreg.findMetric("endTime").value()) { - return new SnapshotStatus( - SnapshotOperation.RESTORE, - mreg.findMetric("snapshotName").getAsString(), - mreg.findMetric("incrementIndex").value(), - mreg.findMetric("requestId").getAsString(), - mreg.findMetric("startTime").value(), - F.asMap( - ignite.localNode().id(), - new T5<>( - (long)mreg.findMetric("processedPartitions").value(), - (long)mreg.findMetric("totalPartitions").value(), - (long)mreg.findMetric("processedWalSegments").value(), - (long)mreg.findMetric("totalWalSegments").value(), - mreg.findMetric("processedWalEntries").value() + else { + MetricRegistry mreg = ignite.context().metric().registry(SNAPSHOT_RESTORE_METRICS); + + long startTime = mreg.findMetric("startTime").value(); + + if (startTime > mreg.findMetric("endTime").value()) { + res = new SnapshotStatus( + SnapshotOperation.RESTORE, + mreg.findMetric("snapshotName").getAsString(), + mreg.findMetric("incrementIndex").value(), + mreg.findMetric("requestId").getAsString(), + mreg.findMetric("startTime").value(), + F.asMap( + ignite.localNode().id(), + new T5<>( + (long)mreg.findMetric("processedPartitions").value(), + (long)mreg.findMetric("totalPartitions").value(), + (long)mreg.findMetric("processedWalSegments").value(), + (long)mreg.findMetric("totalWalSegments").value(), + mreg.findMetric("processedWalEntries").value() + ) ) - ) + ); + } + } + + if (!clusterSupportsSnapshotCheckStatus()) + return res; + + List checkStatuses = null; + + for (var snpCheckMReg : ignite.context().metric()) { + if (!snpCheckMReg.name().startsWith(SnapshotCheckProcess.SNAPSHOT_CHECK_METRIC)) + continue; + + Metric rqIdMetric = snpCheckMReg.findMetric("requestId"); + + // The requestId metric is registered last. + if (rqIdMetric == null) + continue; + + if (checkStatuses == null) + checkStatuses = new ArrayList<>(); + + int incIdx = snpCheckMReg.findMetric("incrementIndex") == null + ? 0 + : snpCheckMReg.findMetric("incrementIndex").value(); + + T5 metrics; + + if (incIdx > 0) { + metrics = new T5<>( + (long)snpCheckMReg.findMetric("processedWalSegments").value(), + (long)snpCheckMReg.findMetric("totalWalSegments").value(), + -1L, -1L, -1L + ); + } + else { + metrics = new T5<>( + (long)snpCheckMReg.findMetric("processedPartitions").value(), + (long)snpCheckMReg.findMetric("totalPartitions").value(), + -1L, -1L, -1L + ); + } + + var status = new SnapshotStatus( + SnapshotOperation.CHECK, + MetricUtils.fromFullName(snpCheckMReg.name()).get2(), + incIdx, + rqIdMetric.getAsString(), + ((LongMetric)snpCheckMReg.findMetric("startTime")).value(), + F.asMap(ignite.localNode().id(), metrics) ); + + checkStatuses.add(status); } - return null; + if (checkStatuses != null) { + if (res == null) + res = new SnapshotStatus(checkStatuses); + else + res.checkStatuses = checkStatuses; + } + + return res; } } @@ -191,8 +333,11 @@ public static class SnapshotStatus implements Serializable { /** Progress of operation on nodes. */ private final Map> progress; + /** Nodes' statuses of all snapshot check operations. */ + private @Nullable List checkStatuses; + /** */ - public SnapshotStatus( + SnapshotStatus( SnapshotOperation op, String name, int incIdx, @@ -205,37 +350,57 @@ public SnapshotStatus( this.incIdx = incIdx; this.requestId = requestId; this.startTime = startTime; - this.progress = Collections.unmodifiableMap(progress); + this.progress = progress; + } + + /** */ + private SnapshotStatus(List checkStatuses) { + // Single, V1 status holds first check status. + this( + SnapshotOperation.CHECK, + checkStatuses.get(0).name(), + checkStatuses.get(0).incrementIndex(), + checkStatuses.get(0).requestId(), + checkStatuses.get(0).startTime(), + checkStatuses.get(0).progress() + ); + + this.checkStatuses = checkStatuses; } /** @return Operation type. */ - public SnapshotOperation operation() { + SnapshotOperation operation() { return op; } /** @return Snapshot name. */ - public String name() { + String name() { return name; } /** @return Incremental snapshot index. */ - public int incrementIndex() { + int incrementIndex() { return incIdx; } /** @return Request ID. */ - public String requestId() { + String requestId() { return requestId; } /** @return Start time. */ - public long startTime() { + long startTime() { return startTime; } /** @return Progress of operation on nodes. */ - public Map> progress() { - return progress; + Map> progress() { + return Collections.unmodifiableMap(progress); + } + + /** @return Statuses of parallel check. */ + @Nullable List checkStatuses() { + return checkStatuses; } } @@ -245,6 +410,9 @@ public enum SnapshotOperation { CREATE, /** Restore snapshot. */ - RESTORE + RESTORE, + + /** Check snapshot. */ + CHECK } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java index 2f19878617a15..88a3f25ae451a 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java @@ -751,6 +751,14 @@ private void registerMetrics(SnapshotCheckContext ctx) { mreg.register("processedSnapshotParts", ctx.checkedSnapshotParts::get, "Number of checked snapshot parts (nodes data) on current node."); } + + // Presence of this metrics also marks that the registry is complete. + mreg.register( + "requestId", + () -> ctx.req.requestId().toString(), + String.class, + "The request ID of the last running cluster snapshot check operation on this node." + ); } /** */ @@ -777,7 +785,7 @@ private static final class SnapshotCheckContext { /** All the snapshot metadatas. */ @Nullable private Map> clusterMetas; - /** Common counter of total work units to process on current node. */ + /** Common counter of total work units to process on current node, of all snapshot parts ({@link #checkedSnapshotParts}). */ @GridToStringExclude private final AtomicInteger totalCounter = new AtomicInteger(-1); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMXBeanImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMXBeanImpl.java index 5cf956155c8b5..bedd6243724ec 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMXBeanImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMXBeanImpl.java @@ -23,11 +23,14 @@ import java.util.stream.Collectors; import org.apache.ignite.internal.GridKernalContext; import org.apache.ignite.internal.processors.metric.GridMetricManager; +import org.apache.ignite.internal.processors.metric.impl.MetricUtils; import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.lang.IgniteFuture; import org.apache.ignite.metric.MetricRegistry; import org.apache.ignite.mxbean.SnapshotMXBean; import org.apache.ignite.spi.metric.IntMetric; +import org.apache.ignite.spi.metric.Metric; import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.DFLT_CHECK_ON_RESTORE; import static org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreProcess.SNAPSHOT_RESTORE_METRICS; @@ -116,16 +119,17 @@ public SnapshotMXBeanImpl(GridKernalContext ctx) { /** {@inheritDoc} */ @Override public String status() { + StringBuilder res = null; + SnapshotOperationRequest req = mgr.currentCreateRequest(); if (req != null) { - return "Create snapshot operation is in progress [name=" + req.snapshotName() + + res = new StringBuilder("Create snapshot operation is in progress [name=" + req.snapshotName() + ", incremental=" + req.incremental() + (req.incremental() ? (", incrementIndex=" + req.incrementIndex()) : "") + - ", id=" + req.requestId() + ']'; + ", id=" + req.requestId() + ']'); } - - if (mgr.isRestoring()) { + else if (mgr.isRestoring()) { MetricRegistry mreg = metricMgr.registry(SNAPSHOT_RESTORE_METRICS); String name = mreg.findMetric("snapshotName").getAsString(); @@ -134,10 +138,53 @@ public SnapshotMXBeanImpl(GridKernalContext ctx) { boolean incremental = incIdx > 0; - return "Restore snapshot operation is in progress [name=" + name + ", incremental=" + incremental + - (incremental ? ", incrementIndex=" + incIdx : "") + ", id=" + id + ']'; + res = new StringBuilder("Restore snapshot operation is in progress [name=" + name + ", incremental=" + incremental + + (incremental ? ", incrementIndex=" + incIdx : "") + ", id=" + id + ']'); + } + + boolean one = true; + + for (var mreg : metricMgr) { + if (!mreg.name().startsWith(SnapshotCheckProcess.SNAPSHOT_CHECK_METRIC)) + continue; + + Metric rqIdMetric = mreg.findMetric("requestId"); + + // The requestId metric is registered last. + if (rqIdMetric == null) + continue; + + String name = MetricUtils.fromFullName(mreg.name()).get2(); + int incIdx = mreg.findMetric("incrementIndex").value(); + String id = rqIdMetric.getAsString(); + + if (one) { + if (res == null) + res = new StringBuilder(); + else + res.append(U.nl()).append(U.nl()); + } + + res.append("Check snapshot operations are in progress ["); + + StringBuilder sb0 = new StringBuilder(); + + if (!one) + sb0.append(", "); + + sb0.append("[name=").append(name).append(", incremental=").append(incIdx > 0) + .append(incIdx > 0 ? ", incrementIndex=" + incIdx : "").append(", id=").append(id).append(']'); + + res.append(sb0); + + one = false; } - return "There is no create or restore snapshot operation in progress."; + if (res == null) + return "No snapshot operation is in progress."; + + res.append(']'); + + return res.toString(); } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java index 7b3e55b85d3c4..fc91f1b838065 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java @@ -93,4 +93,7 @@ public class SupportedFeatureRegistry { /** */ public static final IgniteFeature ROLLING_UPGRADE_FEATURE = new IgniteCoreFeature(0); + + /** */ + public static final IgniteFeature SNAPSHOT_CHECK_STATUS_FEATURE = new IgniteCoreFeature(1); } diff --git a/modules/core/src/main/java/org/apache/ignite/mxbean/SnapshotMXBean.java b/modules/core/src/main/java/org/apache/ignite/mxbean/SnapshotMXBean.java index dee7c59a42e7f..dcd3192d3d353 100644 --- a/modules/core/src/main/java/org/apache/ignite/mxbean/SnapshotMXBean.java +++ b/modules/core/src/main/java/org/apache/ignite/mxbean/SnapshotMXBean.java @@ -129,10 +129,11 @@ public void restoreSnapshot( public void cancelSnapshotRestore(@MXBeanParameter(name = "snpName", description = "Snapshot name.") String name); /** - * Get the status of the current snapshot operation in the cluster. + * Get the status of the current snapshot check or restore operation in the cluster. Doesn't provide status of + * snapshots check operations. * - * @return The status of a current snapshot operation in the cluster. + * @return The status of a current snapshot create or restore operation in the cluster. */ - @MXBeanDescription("The status of a current snapshot operation in the cluster.") + @MXBeanDescription("The status of a current snapshot create or restore operation in the cluster.") public String status(); } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java b/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java index cee31a0224a8f..5ba3cecc3dd29 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/TestRecordingCommunicationSpi.java @@ -59,7 +59,7 @@ public class TestRecordingCommunicationSpi extends TcpCommunicationSpi { private List recordedMsgs = new ArrayList<>(); /** */ - private List blockedMsgs = new ArrayList<>(); + private final List blockedMsgs = new ArrayList<>(); /** */ private Map, Set> blockCls = new HashMap<>(); @@ -248,6 +248,13 @@ public void waitForRecorded() throws InterruptedException { } } + /** */ + public List blockedMessages() { + synchronized (this) { + return new ArrayList<>(blockedMsgs); + } + } + /** * @param cls Message class. * @param nodeName Node name.