diff --git a/docs/_docs/monitoring-metrics/new-metrics.adoc b/docs/_docs/monitoring-metrics/new-metrics.adoc index b055dc4069544..acefac9d4fc4d 100644 --- a/docs/_docs/monitoring-metrics/new-metrics.adoc +++ b/docs/_docs/monitoring-metrics/new-metrics.adoc @@ -684,13 +684,13 @@ Register name: `snapshot-restore` [cols="2,1,3",opts="header"] |=== |Name | Type | Description +|requestId | string | The request ID of the last running cluster snapshot restore operation on this node. |endTime | long | The system time when the restore operation of a cluster snapshot on this node ended. |error | string | Error message of the last running cluster snapshot restore operation on this node. |incrementIndex | integer | The index of incremental snapshot of the last snapshot restore operation on this node. |processedPartitions | integer | The number of processed partitions on this node. |processedWalEntries | long | The number of processed entries from incremental snapshot on this node. |processedWalSegments | integer | The number of processed WAL segments in the incremental snapshot on this node. -|requestId | string | The request ID of the last running cluster snapshot restore operation on this node. |snapshotName | string | The snapshot name of the last running cluster snapshot restore operation on this node. |startTime | long | The system time of the start of the cluster snapshot restore operation on this node. |totalPartitions | integer | The total number of partitions to be restored on this node. @@ -706,6 +706,8 @@ Register name: `snapshot-check.{snapshot_name}` [cols="2,1,3",opts="header"] |=== |Name | Type | Description +|requestId | string | The request ID of the last running cluster snapshot check operation on this node. +If snapshot is checked as part of the restoration operation, it has the same request ID. |checkPartitions | boolean | Shows whether full validation of snapshot partitions is enabled. |incrementIndex | integer | The index of incremental snapshot of the snapshot check operation. |processedPartitions | integer | The number of checked partitions on current node. diff --git a/docs/_docs/snapshots/snapshots.adoc b/docs/_docs/snapshots/snapshots.adoc index 8a18cc5cbdbe4..ccb6f664443a3 100644 --- a/docs/_docs/snapshots/snapshots.adoc +++ b/docs/_docs/snapshots/snapshots.adoc @@ -371,35 +371,33 @@ If the metadata isn't found or cannot be read, snapshot isn't deleted. == Getting Snapshot Operation Status -The status of the create/restore snapshot operations in the cluster can be obtained using the `control.sh|bat` script or JMX interface: +The status of the create/restore snapshot operations in the cluster can be obtained using the `control.sh|bat` script. [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] +==== +Multiple snapshot check operations can run concurrently with any other snapshot operation of another snapshot. +Only one check can be running for the same snapshot at a time. Create and restore operations cannot run in parallel — +only one of them 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 2630f58ee4304..199e714ff1336 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 @@ -85,6 +85,7 @@ import org.apache.ignite.internal.management.cache.FindAndDeleteGarbageInPersistenceTaskResult; import org.apache.ignite.internal.management.cache.IdleVerifyDumpTask; import org.apache.ignite.internal.management.cache.VerifyBackupPartitionsTask; +import org.apache.ignite.internal.management.snapshot.SnapshotStatusCommand; import org.apache.ignite.internal.management.tx.TxInfo; import org.apache.ignite.internal.management.tx.TxTaskResult; import org.apache.ignite.internal.managers.communication.GridIoMessage; @@ -121,6 +122,7 @@ 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; @@ -182,6 +184,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; @@ -3653,11 +3657,257 @@ public void testSnapshotRestoreCancelAndStatus() throws Exception { assertTrue(ctxDisposed); assertEquals(EXIT_CODE_OK, execute("--snapshot", "status")); - assertContains(log, testOut.toString(), "There is no create or restore snapshot operation in progress."); + assertContains(log, testOut.toString(), SnapshotStatusCommand.NO_SNP_OP_IN_PROGRESS_MSG); assertNull(ig.cache(DEFAULT_CACHE_NAME)); } + /** Tests that snapshot status isn't empty when being restored snapshot waits for the check process. */ + @Test + public void testRestoreSnapshotStatusAtStart() throws Exception { + startGrids(3).cluster().state(ClusterState.ACTIVE); + + createCacheAndPreload(grid(1), 100); + + IgniteSnapshotManager snpMgr = (IgniteSnapshotManager)grid(0).snapshot(); + + snpMgr.createSnapshot("test_snapshot").get(getTestTimeout()); + + grid(0).destroyCache(DEFAULT_CACHE_NAME); + + awaitPartitionMapExchange(); + + TestRecordingCommunicationSpi cm = ((TestRecordingCommunicationSpi)grid(1).configuration().getCommunicationSpi()); + + cm.blockMessages((node, msg) -> msg instanceof SingleNodeMessage sm + && (sm.type() == CHECK_SNAPSHOT_PARTS.ordinal())); + + IgniteFutureImpl restoreFut = snpMgr.restoreSnapshot("test_snapshot", null, null, 0, true); + + assertTrue(waitForCondition(cm::hasBlockedMessages, getTestTimeout())); + + injectTestSystemOut(); + + int code = execute("--snapshot", "status"); + + assertEquals("Unexpected exit code", EXIT_CODE_OK, code); + + String out = testOut.toString(); + + assertContains(log, out, "Check snapshot operation is in progress"); + assertContains(log, out, "Snapshot name: test_snapshot"); + + assertFalse("Snapshot future has finished", restoreFut.isDone()); + + 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, 100, 32, null); + snpMgr.createSnapshot("testSnapshot0").get(getTestTimeout()); + + if (incremental) { + try (IgniteDataStreamer streamer = grid(0).dataStreamer(DEFAULT_CACHE_NAME)) { + for (int i = 100; i < 110; i++) + streamer.addData(i, i); + } + + snpMgr.createIncrementalSnapshot("testSnapshot0").get(getTestTimeout()); + } + + if (twoSnapshots) { + snpMgr.createSnapshot("testSnapshot1").get(getTestTimeout()); + + if (incremental) { + try (IgniteDataStreamer streamer = grid(0).dataStreamer(DEFAULT_CACHE_NAME)) { + for (int i = 110; i < 120; i++) + streamer.addData(i, i); + } + + snpMgr.createIncrementalSnapshot("testSnapshot1").get(getTestTimeout()); + } + } + + TestRecordingCommunicationSpi cm = ((TestRecordingCommunicationSpi)grid(1).configuration().getCommunicationSpi()); + + cm.blockMessages((node, msg) -> msg instanceof SingleNodeMessage sm && + (sm.type() == CHECK_SNAPSHOT_PARTS.ordinal())); + + IgniteInternalFuture checkFut0 = runAsync(() -> incremental + ? execute("--snapshot", "check", "testSnapshot0", "--increment", "1") + : execute("--snapshot", "check", "testSnapshot0")); + + IgniteInternalFuture checkFut1 = twoSnapshots + ? runAsync(() -> incremental + ? execute("--snapshot", "check", "testSnapshot1", "--increment", "1") + : execute("--snapshot", "check", "testSnapshot1")) + : null; + + 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")); + + String out = testOut.toString(); + + if (log.isInfoEnabled()) + log.info("Test out:" + U.nl() + out); + + String checkSnpErr = "Check snapshot operation is in progress"; + + int errSum = Arrays.stream(out.split(U.nl())) + .mapToInt(l -> (l.length() - l.replace(checkSnpErr, "").length()) / checkSnpErr.length()) + .sum(); + + assertEquals(twoSnapshots ? 2 : 1, errSum); + + 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); + } + } + + cm.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, 100, 32, null); + snpMgr.createSnapshot("testSnapshot0").get(getTestTimeout()); + + if (twoChecks) { + createCacheAndPreload(grid(1), "cache2", 100, 32, null); + snpMgr.createSnapshot("testSnapshot1").get(getTestTimeout()); + } + + createCacheAndPreload(grid(1), "cache3", 100, 32, null); + + TestRecordingCommunicationSpi cm = ((TestRecordingCommunicationSpi)grid(1).configuration().getCommunicationSpi()); + + cm.blockMessages((node, msg) -> + msg instanceof SingleNodeMessage sm && (sm.type() == CHECK_SNAPSHOT_PARTS.ordinal() + || sm.type() == END_SNAPSHOT.ordinal())); + + IgniteFuture createFut = snpMgr.createSnapshot("testSnapshot2"); + + IgniteInternalFuture checkFut0 = runAsync(() -> execute("--snapshot", "check", "testSnapshot0")); + IgniteInternalFuture checkFut1 = twoChecks ? runAsync(() -> execute("--snapshot", "check", "testSnapshot1")) : null; + + // Wait for the creation gets blocked. + assertTrue(waitForCondition(() -> cm.blockedMessages().stream().anyMatch(m -> + m.ioMessage().message() instanceof SingleNodeMessage sm && sm.type() == END_SNAPSHOT.ordinal()), + getTestTimeout() + )); + + // Wait for the checks get blocked. + 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")); + + String out = testOut.toString(); + + assertTrue(out.contains("Create snapshot operation is in progress")); + assertTrue(out.contains("Snapshot name: testSnapshot2")); + + String checkSnpLogErr = "Check snapshot operation is in progress"; + + int entrySum = Arrays.stream(out.split(U.nl())) + .mapToInt(l -> (l.length() - l.replace(checkSnpLogErr, "").length()) / checkSnpLogErr.length()) + .sum(); + + assertEquals(twoChecks ? 2 : 1, entrySum); + + assertTrue(out.contains("Snapshot name: testSnapshot0")); + + if (twoChecks) + assertTrue(out.contains("Snapshot name: testSnapshot1")); + + cm.stopBlock(); + + createFut.get(); + checkFut0.get(); + + if (twoChecks) + checkFut1.get(); + } + /** @throws Exception If fails. */ @Test public void testSnapshotStatusInMemory() throws Exception { @@ -3785,7 +4035,7 @@ private void checkSnapshotStatus(boolean isCreating, boolean isRestoring, boolea assertEquals(out, EXIT_CODE_OK, status); if (!isCreating && !isRestoring) { - assertContains(log, out, "There is no create or restore snapshot operation in progress."); + assertContains(log, out, SnapshotStatusCommand.NO_SNP_OP_IN_PROGRESS_MSG); return; } 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..08f6ccdc165fc 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 @@ -17,8 +17,10 @@ package org.apache.ignite.internal.management.snapshot; +import java.io.Serializable; import java.text.DateFormat; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.Date; import java.util.List; @@ -35,12 +37,16 @@ import org.apache.ignite.internal.util.typedef.T5; import org.apache.ignite.internal.util.typedef.X; import org.apache.ignite.internal.util.typedef.internal.U; +import org.jetbrains.annotations.Nullable; import static org.apache.ignite.internal.management.SystemViewTask.SimpleType.NUMBER; import static org.apache.ignite.internal.management.SystemViewTask.SimpleType.STRING; /** */ -public class SnapshotStatusCommand extends AbstractSnapshotCommand { +public class SnapshotStatusCommand extends AbstractSnapshotCommand { + /** */ + public static final String NO_SNP_OP_IN_PROGRESS_MSG = "There is no snapshot operation in progress."; + /** {@inheritDoc} */ @Override public String description() { return "Get the status of the current snapshot operation"; @@ -57,55 +63,81 @@ public class SnapshotStatusCommand extends AbstractSnapshotCommand printer) { - if (status == null) { + @Override public void printResult(NoArg arg, @Nullable Serializable rawResult, Consumer printer) { + // An old-style empty result is null. + if (rawResult == null) { printer.accept("There is no create or restore snapshot operation in progress."); return; } - boolean isCreating = status.operation() == SnapshotStatusTask.SnapshotOperation.CREATE; - boolean isIncremental = status.incrementIndex() > 0; + SnapshotStatus singleStatus = rawResult instanceof SnapshotStatus ? (SnapshotStatus)rawResult : null; + + Collection statuses = rawResult instanceof Collection + ? (Collection)rawResult + : null; - GridStringBuilder s = new GridStringBuilder(); + // A new-style result with no statuses is an empty collection, never null. + if (statuses != null && statuses.isEmpty()) { + printer.accept(NO_SNP_OP_IN_PROGRESS_MSG); - if (isCreating) - s.a("Create snapshot operation is in progress.").nl(); - else - s.a("Restore snapshot operation is in progress.").nl(); + return; + } - s.a("Snapshot name: ").a(status.name()).nl(); - s.a("Incremental: ").a(isIncremental).nl(); + statuses = statuses == null + ? Collections.singleton(new SnapshotStatusTask.SnapshotOperationStatus(singleStatus)) + : statuses; + + boolean separator = false; + + for (SnapshotStatusTask.SnapshotOperationStatus s0 : statuses) { + if (separator) + printer.accept(U.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."); - if (isIncremental) - s.a("Increment index: ").a(status.incrementIndex()).nl(); + printer.accept(""); - 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(); + GridStringBuilder s = new GridStringBuilder(); - printer.accept(s.toString()); + boolean incremental = s0.incrementIndex() > 0; - SnapshotTaskProgressDesc desc; + s.a("Snapshot name: ").a(s0.name()).nl(); + s.a("Incremental: ").a(incremental).nl(); - if (isCreating && isIncremental) - desc = new CreateIncrementalSnapshotTaskProgressDesc(); - else if (isCreating) - desc = new CreateFullSnapshotTaskProgressDesc(); - else if (isIncremental) - desc = new RestoreIncrementalSnapshotTaskProgressDesc(); - else - desc = new RestoreFullSnapshotTaskProgressDesc(); + if (incremental) + s.a("Increment index: ").a(s0.incrementIndex()).nl(); - List> rows = status.progress().entrySet().stream().sorted(Map.Entry.comparingByKey()) - .map(e -> desc.buildRow(e.getKey(), e.getValue())) - .collect(Collectors.toList()); + 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(); - SystemViewCommand.printTable(desc.titles(), desc.types(), rows, printer); + printer.accept(s.toString()); - printer.accept(U.nl()); + 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); + + 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); + + separator = true; + } } /** Describes progress of a snapshot task. */ @@ -114,17 +146,17 @@ private abstract static class SnapshotTaskProgressDesc { private final List titles; /** */ - SnapshotTaskProgressDesc(List titles) { + private SnapshotTaskProgressDesc(List titles) { this.titles = Collections.unmodifiableList(titles); } /** @return Progress table columns titles. */ - List titles() { + private List titles() { return titles; } /** @return Progress table columns types. */ - List types() { + private List types() { List types = new ArrayList<>(); types.add(STRING); @@ -136,13 +168,13 @@ List types() { } /** @return Progress table data row. */ - abstract List buildRow(UUID nodeId, T5 progress); + protected abstract List buildRow(UUID nodeId, T5 progress); } /** */ private static class CreateFullSnapshotTaskProgressDesc extends SnapshotTaskProgressDesc { /** */ - CreateFullSnapshotTaskProgressDesc() { + private CreateFullSnapshotTaskProgressDesc() { super(F.asList("Node ID", "Processed, bytes", "Total, bytes", "Percent")); } @@ -163,7 +195,7 @@ private static class CreateFullSnapshotTaskProgressDesc extends SnapshotTaskProg /** */ private static class CreateIncrementalSnapshotTaskProgressDesc extends SnapshotTaskProgressDesc { /** */ - CreateIncrementalSnapshotTaskProgressDesc() { + private CreateIncrementalSnapshotTaskProgressDesc() { super(F.asList("Node ID", "Progress")); } @@ -176,7 +208,7 @@ private static class CreateIncrementalSnapshotTaskProgressDesc extends SnapshotT /** */ private static class RestoreFullSnapshotTaskProgressDesc extends SnapshotTaskProgressDesc { /** */ - RestoreFullSnapshotTaskProgressDesc() { + private RestoreFullSnapshotTaskProgressDesc() { super(F.asList("Node ID", "Processed, partitions", "Total, partitions", "Percent")); } @@ -197,7 +229,7 @@ private static class RestoreFullSnapshotTaskProgressDesc extends SnapshotTaskPro /** */ private static class RestoreIncrementalSnapshotTaskProgressDesc extends SnapshotTaskProgressDesc { /** */ - RestoreIncrementalSnapshotTaskProgressDesc() { + private RestoreIncrementalSnapshotTaskProgressDesc() { super(F.asList( "Node ID", "Processed, partitions", @@ -246,4 +278,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..afec07774361b 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,47 +26,68 @@ 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.Order; +import org.apache.ignite.internal.dto.IgniteDataTransferObject; import org.apache.ignite.internal.management.api.NoArg; 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.CoreFeatureRegistry; 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.apache.ignite.spi.metric.ReadOnlyMetricRegistry; import org.jetbrains.annotations.Nullable; -import static org.apache.ignite.internal.management.snapshot.SnapshotStatusTask.SnapshotStatus; import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNAPSHOT_METRICS; import static org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreProcess.SNAPSHOT_RESTORE_METRICS; +import static org.apache.ignite.internal.processors.rollingupgrade.feature.CoreFeatureRegistry.SNAPSHOT_CHECK_STATUS_FEATURE; import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.nodeIds; /** - * Task to get the status of the current snapshot operation in the cluster. + * Task to get the statuses of the current snapshot operations in the cluster. + *

Before {@link CoreFeatureRegistry#SNAPSHOT_CHECK_STATUS_FEATURE}, returns single {@link SnapshotStatus} for single + * snapshot creation or snapshot restoration operation. + *

With {@link CoreFeatureRegistry#SNAPSHOT_CHECK_STATUS_FEATURE}, returns collection of {@link SnapshotOperationStatus} + * for possible several snapshot checks operations or parallel creation/restoration and checks operations. May still + * contain a single status. */ @GridInternal -public class SnapshotStatusTask extends VisorMultiNodeTask { +public class SnapshotStatusTask extends VisorMultiNodeTask { /** */ private static final long serialVersionUID = 0L; + /** Flag of the feature availability at the task start. */ + private transient boolean featureEnabled; + /** {@inheritDoc} */ - @Override protected VisorJob job(NoArg arg) { + @Override protected VisorJob job(NoArg arg) { return new SnapshotStatusJob(arg, debug); } /** {@inheritDoc} */ @Override protected Collection jobNodes(VisorTaskArgument arg) { + featureEnabled = ignite.context().rollingUpgrade().features().isActive(SNAPSHOT_CHECK_STATUS_FEATURE) && + arg.initiatorFeatures().contains(SNAPSHOT_CHECK_STATUS_FEATURE.id());; + return nodeIds(ignite.cluster().forServers().nodes()); } /** {@inheritDoc} */ - @Nullable @Override protected SnapshotStatus reduce0(List results) { + @Override protected @Nullable Serializable reduce0(List results) { if (results.isEmpty()) throw new IgniteException("Failed to get the snapshot status. Topology is empty."); @@ -75,30 +97,91 @@ public class SnapshotStatusTask extends VisorMultiNodeTask res = F.viewReadOnly(results, ComputeJobResult::getData, r -> r.getData() != null); + // All the jobs' statuses combined to one flat collection. + Collection allStatuses = new ArrayList<>(results.size()); + + for (ComputeJobResult jobRes : results) { + if (jobRes.getData() == null) + continue; + + Serializable rawResult = jobRes.getData(); + + if (rawResult instanceof SnapshotStatus singleStatus) + allStatuses.add(new SnapshotOperationStatus(singleStatus)); + else + allStatuses.addAll((Collection)rawResult); + } + + ArrayList taskResult = new ArrayList<>(); + + // First, find create or restore operation. + Collection filtered = F.viewReadOnly(allStatuses, s -> s, s -> s.operation() != SnapshotOperation.CHECK); + + if (!filtered.isEmpty()) { + // Merged create or restore. + SnapshotOperationStatus merged = F.first(filtered); + + String mergedRqId = merged.reqId; + + // Filter out differing requests due to concurrent updates on nodes. + filtered = F.view(filtered, s -> s.reqId.equals(mergedRqId)); + + // Progress per node. + Map> progress = new HashMap<>(); + + filtered.forEach(s -> progress.putAll(s.progress)); + + merged = new SnapshotOperationStatus(merged.op, merged.name, merged.incIdx, merged.reqId, merged.startTime, progress); + + taskResult.add(merged); + } + + if (!featureEnabled) { + if (taskResult.isEmpty()) + return null; + + SnapshotOperationStatus s0 = taskResult.get(0); - // There is no snapshot operation. - if (res.isEmpty()) - return null; + return new SnapshotStatus(s0.op, s0.name, s0.incIdx, s0.reqId, s0.startTime, s0.progress); + } + + // Check operation statuses go last. + filtered = F.viewReadOnly(allStatuses, s -> s, s -> s.operation() == SnapshotOperation.CHECK); - SnapshotStatus s0 = F.first(res); + // Found check statuses. + if (!filtered.isEmpty()) { + // Merged check statuses per snapshot. + Map mergedCheckStatuses = U.newHashMap(filtered.size()); - // Filter out differing requests due to concurrent updates on nodes. - res = F.view(res, s -> s.requestId.equals(s0.requestId)); + filtered.forEach(chkStatus -> { + SnapshotOperationStatus chkStatus0 = mergedCheckStatuses.computeIfAbsent(chkStatus.name, (snpName -> chkStatus)); - // Merge nodes progress. - Map> progress = new HashMap<>(); + if (chkStatus0 != chkStatus) + chkStatus0.progress.putAll(chkStatus.progress); + }); - res.forEach(s -> progress.putAll(s.progress)); + taskResult.addAll(mergedCheckStatuses.values()); + } - return new SnapshotStatus(s0.op, s0.name, s0.incIdx, s0.requestId, s0.startTime, progress); + return taskResult; } - /** */ - private static class SnapshotStatusJob extends SnapshotJob { + /** + * A job to get the statuses of the current snapshot operations on current node. + *

Before {@link CoreFeatureRegistry#SNAPSHOT_CHECK_STATUS_FEATURE}, returns single {@link SnapshotStatus} for single + * snapshot creation or snapshot restoration operation. + *

With {@link CoreFeatureRegistry#SNAPSHOT_CHECK_STATUS_FEATURE}, returns collection of {@link SnapshotOperationStatus} + * for possible several snapshot checks operations or parallel creation/restoration and checks operations. May still + * contain a single status. + */ + private static class SnapshotStatusJob extends SnapshotJob { /** */ private static final long serialVersionUID = 0L; + /** */ + @LoggerResource + private transient IgniteLogger log; + /** * @param arg Job argument. * @param debug Flag indicating whether debug information should be printed into node log. @@ -108,14 +191,16 @@ protected SnapshotStatusJob(@Nullable NoArg arg, boolean debug) { } /** {@inheritDoc} */ - @Override protected SnapshotStatus run(@Nullable NoArg arg) throws IgniteException { + @Override protected @Nullable Serializable run(@Nullable NoArg arg) throws IgniteException { if (!CU.isPersistenceEnabled(ignite.context().config())) return null; IgniteSnapshotManager snpMgr = ignite.context().cache().context().snapshotMgr(); - SnapshotOperationRequest req = snpMgr.currentCreateRequest(); + // Only one operation like creation or restoration is allowed at the same time. + SnapshotStatus createOrRestoreStatus = null; + if (req != null) { T5 metrics; @@ -130,7 +215,7 @@ protected SnapshotStatusJob(@Nullable NoArg arg, boolean debug) { -1L, -1L, -1L); } - return new SnapshotStatus( + createOrRestoreStatus = new SnapshotStatus( SnapshotOperation.CREATE, req.snapshotName(), req.incrementIndex(), @@ -139,32 +224,97 @@ 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()) { + createOrRestoreStatus = 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() + ) ) - ) + ); + } + } + + /** Return old single result. */ + if (!ignite.context().rollingUpgrade().features().isActive(SNAPSHOT_CHECK_STATUS_FEATURE)) { + log.warning("The snapshot check status feature isn't activated yet. The status is available only " + + "for snapshot creation and restoration."); + + return createOrRestoreStatus; + } + + // New many results. + ArrayList snpOpStatuses = null; + + if (createOrRestoreStatus != null) { + snpOpStatuses = new ArrayList<>(); + + // Status of creation or restoration operation always goes first. This is important for old command initiators (control.sh). + snpOpStatuses.add(new SnapshotOperationStatus(createOrRestoreStatus)); + } + + for (ReadOnlyMetricRegistry 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; + + 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 + ); + } + + SnapshotOperationStatus snpOpStatus = new SnapshotOperationStatus( + SnapshotOperation.CHECK, + MetricUtils.fromFullName(snpCheckMReg.name()).get2(), + incIdx, + rqIdMetric.getAsString(), + ((LongMetric)snpCheckMReg.findMetric("startTime")).value(), + F.asMap(ignite.localNode().id(), metrics) ); + + if (snpOpStatuses == null) + snpOpStatuses = new ArrayList<>(); + + snpOpStatuses.add(snpOpStatus); } - return null; + return snpOpStatuses; } } @@ -239,12 +389,109 @@ public Map> progress() { } } + /** + * Snapshot operation status after {@link CoreFeatureRegistry#SNAPSHOT_CHECK_STATUS_FEATURE}. + * Is now a {@link IgniteDataTransferObject}. + */ + public static class SnapshotOperationStatus extends IgniteDataTransferObject { + /** */ + private static final long serialVersionUID = 0L; + + /** Operation type. */ + @Order(0) + SnapshotOperation op; + + /** Snapshot name. */ + @Order(1) + String name; + + /** Incremental snapshot index. */ + @Order(2) + int incIdx; + + /** Request ID. */ + @Order(3) + String reqId; + + /** Start time. */ + @Order(4) + long startTime; + + /** Progress of operation on nodes. */ + @Order(5) + Map> progress; + + /** Empty constructor for serialization purposes. */ + public SnapshotOperationStatus() { + // No-op. + } + + /** */ + private SnapshotOperationStatus( + SnapshotOperation op, + String name, + int incIdx, + String reqId, + long startTime, + Map> progress + ) { + this.op = op; + this.name = name; + this.incIdx = incIdx; + this.reqId = reqId; + this.startTime = startTime; + this.progress = progress; + } + + /** */ + public SnapshotOperationStatus(SnapshotStatus s) { + this(s.op, s.name, s.incIdx, s.requestId, s.startTime, s.progress); + } + + /** @return Operation type. */ + public SnapshotOperation operation() { + return op; + } + + /** @return Snapshot name. */ + public String name() { + return name; + } + + /** @return Incremental snapshot index. */ + public int incrementIndex() { + return incIdx; + } + + /** @return Request ID. */ + public String requestId() { + return reqId; + } + + /** @return Start time. */ + public long startTime() { + return startTime; + } + + /** @return Progress of operation on nodes. */ + public Map> progress() { + return Collections.unmodifiableMap(progress); + } + } + /** Snapshot operation type. */ public enum SnapshotOperation { /** Create snapshot. */ CREATE, /** Restore snapshot. */ - RESTORE + RESTORE, + + /** + * Check snapshot. + * + * @see CoreFeatureRegistry#SNAPSHOT_CHECK_STATUS_FEATURE + */ + CHECK } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java index 11afe3182fb56..46a8a28262cbd 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java @@ -1731,6 +1731,8 @@ private IgniteInternalFuture cancelSnapshot0(String name) { } /** + * TODO: support cancellation of the check operation https://issues.apache.org/jira/browse/IGNITE-29240 + * * @param reqId Snapshot operation request ID. * @return Future which will be completed when cancel operation finished. */ @@ -1876,6 +1878,18 @@ public IgniteInternalFuture checkSnapshot(String }); } + /** {@link #checkSnapshot(String, String, Collection, boolean, int, boolean, UUID)} with default {@code reqId} == {@code null}. */ + public IgniteInternalFuture checkSnapshot( + String name, + @Nullable String snpPath, + @Nullable Collection grps, + boolean includeCustomHandlers, + int incIdx, + boolean check + ) { + return checkSnapshot(name, snpPath, grps, includeCustomHandlers, incIdx, check, null); + } + /** * The check snapshot procedure performs compute operation over the whole cluster to verify the snapshot * entirety and partitions consistency. The result future will be completed with an exception if this @@ -1888,16 +1902,18 @@ public IgniteInternalFuture checkSnapshot(String * handlers, otherwise only system consistency check will be performed. * @param incIdx Incremental snapshot index. * @param check If {@code true} check snapshot integrity. + * @param reqId Optional request ID to use. If {@code null}, own random one is generated. * @return Future with the result of execution snapshot partitions verify task, which besides calculating partition * hashes of {@link IdleVerifyResult} also contains the snapshot metadata distribution across the cluster. */ - public IgniteInternalFuture checkSnapshot( + IgniteInternalFuture checkSnapshot( String name, @Nullable String snpPath, @Nullable Collection grps, boolean includeCustomHandlers, int incIdx, - boolean check + boolean check, + @Nullable UUID reqId ) { A.notNullOrEmpty(name, "Snapshot name cannot be null or empty."); A.ensure(U.alphanumericUnderscore(name), "Snapshot name must satisfy the following name pattern: a-zA-Z0-9_"); @@ -1911,7 +1927,7 @@ public IgniteInternalFuture checkSnapshot( } IgniteInternalFuture res = checkSnpProc.start( - name, snpPath, grps, check, incIdx, includeCustomHandlers); + name, snpPath, grps, check, incIdx, includeCustomHandlers, reqId); res.listen(lsnr -> { if (log.isInfoEnabled()) { @@ -4477,6 +4493,7 @@ public CancelSnapshotCallable(UUID reqId, String snpName) { /** {@inheritDoc} */ @Override public Boolean call() { + // TODO: support cancellation of the check operation https://issues.apache.org/jira/browse/IGNITE-29240 if (reqId != null) return ignite.context().cache().context().snapshotMgr().cancelLocalSnapshotOperations(reqId); else { 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 cc5a85092d1e2..aabbdd226e23f 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 @@ -661,6 +661,7 @@ private List assingMetas(Map start( String snpName, @@ -668,16 +669,17 @@ public IgniteInternalFuture start( @Nullable Collection grpNames, boolean fullCheck, int incIdx, - boolean allRestoreHandlers + boolean allRestoreHandlers, + @Nullable UUID reqId ) { assert !F.isEmpty(snpName); - UUID reqId = UUID.randomUUID(); + UUID reqId0 = reqId != null ? reqId : UUID.randomUUID(); Set requiredNodes = new HashSet<>(F.viewReadOnly(kctx.discovery().discoCache().aliveBaselineNodes(), node2id())); SnapshotCheckProcessRequest req = new SnapshotCheckProcessRequest( - reqId, + reqId0, requiredNodes, snpName, snpPath, @@ -689,7 +691,7 @@ public IgniteInternalFuture start( GridFutureAdapter clusterOpFut = new GridFutureAdapter<>(); - clusterOpFut.listen(fut -> clusterOpFuts.remove(reqId)); + clusterOpFut.listen(fut -> clusterOpFuts.remove(reqId0)); clusterOpFuts.put(reqId, clusterOpFut); @@ -759,6 +761,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 metric marks that the registry is complete. + mreg.register( + "requestId", + () -> ctx.req.requestId().toString(), + String.class, + "The request ID of the current cluster snapshot check operation on this node." + ); } /** */ 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..bfd507a2c4d44 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 @@ -36,6 +36,9 @@ * Snapshot MBean features. */ public class SnapshotMXBeanImpl implements SnapshotMXBean { + /** */ + public static final String NO_CREATE_OR_RESTORE_IN_PROGRESS_MSG = "There is no create or restore snapshot operation in progress."; + /** Instance of snapshot cache shared manager. */ private final IgniteSnapshotManager mgr; @@ -138,6 +141,6 @@ public SnapshotMXBeanImpl(GridKernalContext ctx) { (incremental ? ", incrementIndex=" + incIdx : "") + ", id=" + id + ']'; } - return "There is no create or restore snapshot operation in progress."; + return NO_CREATE_OR_RESTORE_IN_PROGRESS_MSG; } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java index 2643a5575f6d4..31f0abc9c9fad 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java @@ -322,7 +322,7 @@ public IgniteFutureImpl start( snpMgr.recordSnapshotEvent(snpName, msg, EventType.EVT_CLUSTER_SNAPSHOT_RESTORE_STARTED); - snpMgr.checkSnapshot(snpName, snpPath, cacheGrpNames, incIdx < 1, incIdx, check).listen(f -> { + snpMgr.checkSnapshot(snpName, snpPath, cacheGrpNames, incIdx < 1, incIdx, check, fut.rqId).listen(f -> { if (f.error() != null) { finishProcess(fut0.rqId, f.error()); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/CoreFeatureRegistry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/CoreFeatureRegistry.java index 4d00489327fea..e4e186fffad72 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/CoreFeatureRegistry.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/CoreFeatureRegistry.java @@ -111,4 +111,7 @@ public class CoreFeatureRegistry { /** Feature that introduces the snapshot list command. */ public static final IgniteFeature SNAPSHOT_LIST_FEATURE = new IgniteCoreFeature(4); + + /** Feature that introduces statuses of the snapshot check operations. */ + public static final IgniteFeature SNAPSHOT_CHECK_STATUS_FEATURE = new IgniteCoreFeature(5); } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/VisorTaskArgument.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/VisorTaskArgument.java index 55602bbb8f853..4b1bd9671e664 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/visor/VisorTaskArgument.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/VisorTaskArgument.java @@ -137,6 +137,11 @@ public boolean isDebug() { return debug; } + /** */ + public IgniteCoreFeatureSet initiatorFeatures() { + return cmdInitiatorFeatures; + } + /** {@inheritDoc} */ @Override public void writeExternal(ObjectOutput out) throws IOException { out.writeInt(MAGIC); 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..1a0324d92b310 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,10 +59,10 @@ 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<>(); + private final Map, Set> blockCls = new HashMap<>(); /** */ private IgniteBiPredicate blockP; @@ -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. diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckStatusRollingUpgradeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckStatusRollingUpgradeTest.java new file mode 100644 index 0000000000000..2a39a83069050 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotCheckStatusRollingUpgradeTest.java @@ -0,0 +1,279 @@ +/* + * 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.processors.cache.persistence.snapshot; + +import java.io.File; +import java.io.Serializable; +import java.nio.file.DirectoryStream; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Collection; +import java.util.Collections; +import java.util.UUID; +import org.apache.ignite.Ignite; +import org.apache.ignite.IgniteDataStreamer; +import org.apache.ignite.Ignition; +import org.apache.ignite.cache.CacheAtomicityMode; +import org.apache.ignite.cache.CacheMode; +import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.client.Config; +import org.apache.ignite.client.IgniteClient; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.cluster.ClusterState; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.ClientConfiguration; +import org.apache.ignite.configuration.DataRegionConfiguration; +import org.apache.ignite.configuration.DataStorageConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.management.api.NoArg; +import org.apache.ignite.internal.management.snapshot.SnapshotStatusTask; +import org.apache.ignite.internal.management.snapshot.SnapshotStatusTask.SnapshotOperation; +import org.apache.ignite.internal.processors.rollingupgrade.AbstractRollingUpgradeTest; +import org.apache.ignite.internal.util.distributed.SingleNodeMessage; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.G; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.internal.visor.VisorTaskArgument; +import org.apache.ignite.internal.visor.VisorTaskResult; +import org.apache.ignite.lang.IgniteFuture; +import org.junit.Test; + +import static java.nio.file.Files.newDirectoryStream; +import static org.apache.ignite.internal.TestRecordingCommunicationSpi.spi; +import static org.apache.ignite.internal.processors.rollingupgrade.feature.ManagementApiVersionValidationTest.withCoreVersion; +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CHECK_SNAPSHOT_METAS; +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.END_SNAPSHOT; +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; + +/** */ +public class IgniteClusterSnapshotCheckStatusRollingUpgradeTest extends AbstractRollingUpgradeTest { + /** */ + private static final int ALL_GRIDS = 4; + + /** */ + private static final int CLIENTS = 1; + + /** */ + private static final String SNP_NAME = "testSnapshot"; + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + cleanPersistenceDir(); + } + + /** {@inheritDoc} */ + @Override protected void beforeTest() throws Exception { + super.beforeTest(); + + cleanPersistenceDir(); + } + + /** {@inheritDoc} */ + @Override protected void cleanPersistenceDir() throws Exception { + super.cleanPersistenceDir(); + + // Clean all: also separated snapshot working directories. + try (DirectoryStream files = newDirectoryStream(Paths.get(U.defaultWorkDirectory()))) { + for (Path path : files) + U.delete(path); + } + } + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName, String ver) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName, ver); + + cfg.setDataStorageConfiguration( + new DataStorageConfiguration() + .setDefaultDataRegionConfiguration( + new DataRegionConfiguration() + .setPersistenceEnabled(true) + .setMaxSize(DataStorageConfiguration.DFLT_DATA_REGION_INITIAL_SIZE)) + ); + + cfg.setWorkDirectory(new File(U.defaultWorkDirectory(), igniteInstanceName).getAbsolutePath()); + + return cfg; + } + + /** */ + @Test + public void testClientOfThePreviousVersionWithCreateInProgress() throws Exception { + doTestDifferentClusterAndClientVersions(MASTER_VER_REF, "2.18.0"); + } + + /** */ + @Test + public void testClientOfTheNextVersionWithCreateInProgress() throws Exception { + doTestDifferentClusterAndClientVersions("2.18.0", MASTER_VER_REF); + } + + /** */ + private void doTestDifferentClusterAndClientVersions(String clusterVer, String clientVer) throws Exception { + for (int i = 0; i < ALL_GRIDS - CLIENTS; i++) + startGrid(i, clusterVer, false); + + grid(0).cluster().state(ClusterState.ACTIVE); + + createCache(0); + + snp(0).createSnapshot("testSnapshotToCheck").get(getTestTimeout()); + + for (int i = 0; i < ALL_GRIDS - CLIENTS; i++) { + spi(grid(i)).blockMessages((node, msg) -> + msg instanceof SingleNodeMessage snm && + (snm.type() == END_SNAPSHOT.ordinal() || snm.type() == CHECK_SNAPSHOT_METAS.ordinal())); + } + + int chkNodeIdx = 1; + + IgniteInternalFuture chkFut = snp(chkNodeIdx).checkSnapshot("testSnapshotToCheck", null); + IgniteFuture createFut = snp(0).createSnapshot(SNP_NAME); + + try { + assertTrue(waitForCondition( + () -> snp(chkNodeIdx).isSnapshotCreating() && snp(chkNodeIdx).isSnapshotChecking("testSnapshotToCheck"), 10_000L)); + + Collection nodes = Collections.singleton(nodeId(0)); + + VisorTaskArgument vtArg = withCoreVersion(clientVer, () -> new VisorTaskArgument<>(nodes, new NoArg(), false)); + + SnapshotStatusTask.SnapshotStatus status = snapshotStatusTaskWithClient(vtArg); + + assertEquals(SnapshotOperation.CREATE, status.operation()); + assertEquals(SNP_NAME, status.name()); + assertEquals(ALL_GRIDS - CLIENTS, status.progress().size()); + } + finally { + for (int i = 0; i < ALL_GRIDS - CLIENTS; i++) + spi(grid(i)).stopBlock(); + } + + createFut.get(10_000L); + chkFut.get(10_000L); + } + + /** */ + @Test + public void testSnapshotStatusFeature() throws Exception { + for (int i = 0; i < ALL_GRIDS; i++) + startGrid(i, "2.18.0", i >= ALL_GRIDS - CLIENTS); + + grid(0).cluster().active(true); + + createCacheAndSnapshot(1); + + ru(grid(0)).enableVersionUpgrade(); + + validateSnapshotCheckStatus(false); + + for (int i = 0; i < ALL_GRIDS; i++) { + assertTrue(ru(grid(i)).isVersionUpgradeEnabled()); + + upgradeNodeVersion(i, "2.18.0", MASTER_VER_REF); + } + + validateSnapshotCheckStatus(false); + + ru(grid(1)).finalizeClusterVersion(); + + for (int i = 0; i < ALL_GRIDS; i++) { + int i0 = i; + + assertTrue(waitForCondition(() -> !ru(grid(i0)).isVersionUpgradeEnabled(), getTestTimeout())); + } + + validateSnapshotCheckStatus(true); + } + + /** */ + private void validateSnapshotCheckStatus(boolean supports) throws Exception { + int blockNodeIdx = 1; + + spi(grid(blockNodeIdx)).blockMessages((node, msg) -> msg instanceof SingleNodeMessage snm && + snm.type() == CHECK_SNAPSHOT_METAS.ordinal()); + + IgniteInternalFuture chkFut = snp(1).checkSnapshot(SNP_NAME, null); + + assertTrue(waitForCondition(() -> spi(grid(blockNodeIdx)).hasBlockedMessages(), getTestTimeout())); + + for (Ignite ig : G.allGrids()) { + Serializable rawRes = snapshotStatusTaskWithGrid(ig); + + if (supports) { + assertTrue(rawRes instanceof Collection cll && !cll.isEmpty() && + F.first(cll) instanceof SnapshotStatusTask.SnapshotOperationStatus); + } + else + assertTrue(rawRes == null); + } + + spi(grid(blockNodeIdx)).stopBlock(); + + chkFut.get(10_000L); + } + + /** */ + private static T snapshotStatusTaskWithGrid(Ignite grid) throws Exception { + Collection nodes = grid.cluster().nodes().stream().map(ClusterNode::id).toList(); + + VisorTaskArgument vtArg = new VisorTaskArgument<>(nodes, new NoArg(), false); + + return (T)grid.compute().execute(SnapshotStatusTask.class, vtArg).result(); + } + + /** */ + private static T snapshotStatusTaskWithClient(VisorTaskArgument vtArg) throws Exception { + try (IgniteClient cli = Ignition.startClient(new ClientConfiguration().setAddresses(Config.SERVER))) { + VisorTaskResult vtRes = cli.compute().execute(SnapshotStatusTask.class.getName(), vtArg); + + return vtRes.result(); + } + } + + /** */ + private IgniteSnapshotManager snp(int gridIdx) { + return grid(gridIdx).context().cache().context().snapshotMgr(); + } + + /** */ + private void createCacheAndSnapshot(int gridIdx) { + createCache(gridIdx); + + snp(gridIdx).createSnapshot(SNP_NAME).get(getTestTimeout()); + } + + /** Creates a replicated cache and fills it with data. */ + private void createCache(int gridIdx) { + int partsCnt = 4; + int keysCnt = partsCnt * 5; + + grid(gridIdx).createCache(new CacheConfiguration<>(DEFAULT_CACHE_NAME) + .setCacheMode(CacheMode.REPLICATED) + .setAffinity(new RendezvousAffinityFunction().setPartitions(partsCnt)) + .setAtomicityMode(CacheAtomicityMode.ATOMIC)); + + try (IgniteDataStreamer ds = grid(gridIdx).dataStreamer(DEFAULT_CACHE_NAME)) { + for (int i = 0; i < keysCnt; i++) + ds.addData(i, i); + } + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotMXBeanTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotMXBeanTest.java index cdb30ae79b828..a6538f646afcf 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotMXBeanTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotMXBeanTest.java @@ -328,7 +328,7 @@ private void checkSnapshotStatus(boolean isCreating, boolean isRestoring, boolea String status = bean.status(); if (!isCreating && !isRestoring) { - assertContains(log, status, "There is no create or restore snapshot operation in progress"); + assertContains(log, status, SnapshotMXBeanImpl.NO_CREATE_OR_RESTORE_IN_PROGRESS_MSG); continue; } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/ManagementApiVersionValidationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/ManagementApiVersionValidationTest.java index b8af58f9f7cda..6cd07a0cc81e7 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/ManagementApiVersionValidationTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/feature/ManagementApiVersionValidationTest.java @@ -94,7 +94,7 @@ private VisorTaskArgument createCommandArgument(int destNodeIdx, String } /** */ - private static R withCoreVersion(String ver, Callable action) throws Exception { + public static R withCoreVersion(String ver, Callable action) throws Exception { IgniteCoreFeatureSet prev = IgniteCoreFeatureSet.INSTANCE; IgniteCoreFeatureSet.INSTANCE = createCoreFeatureSet(ver); diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite9.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite9.java index 86802121999c7..6b2ccfa4ed443 100644 --- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite9.java +++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSnapshotTestSuite9.java @@ -20,6 +20,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteClusterSnapshotCheckStatusRollingUpgradeTest; import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteClusterSnapshotDeleteRollingUpgradeTest; import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteClusterSnapshotDeleteTest; import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteClusterSnapshotListRollingUpgradeTest; @@ -47,6 +48,7 @@ public static List> suite(Collection ignoredTests) { GridTestUtils.addTestIfNeeded(suite, IgniteClusterSnapshotDeleteTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, IgniteClusterSnapshotListTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, IgniteClusterSnapshotListRollingUpgradeTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, IgniteClusterSnapshotCheckStatusRollingUpgradeTest.class, ignoredTests); return suite; } diff --git a/modules/wire-compatibility-check/wire/enums/org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperation.json b/modules/wire-compatibility-check/wire/enums/org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperation.json new file mode 100644 index 0000000000000..e17d08090b831 --- /dev/null +++ b/modules/wire-compatibility-check/wire/enums/org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperation.json @@ -0,0 +1,8 @@ +{ + "class": "org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperation", + "fields": [ + {"order": 0, "type": "org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperation", "name": "CREATE"}, + {"order": 1, "type": "org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperation", "name": "RESTORE"}, + {"order": 2, "type": "org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperation", "name": "CHECK"} + ] +} diff --git a/modules/wire-compatibility-check/wire/idto/org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperationStatus.json b/modules/wire-compatibility-check/wire/idto/org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperationStatus.json new file mode 100644 index 0000000000000..0d052889397b2 --- /dev/null +++ b/modules/wire-compatibility-check/wire/idto/org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperationStatus.json @@ -0,0 +1,11 @@ +{ + "class": "org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperationStatus", + "fields": [ + {"order": 0, "type": "org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperation", "name": "op"}, + {"order": 1, "type": "java.lang.String", "name": "name"}, + {"order": 2, "type": "int", "name": "incIdx"}, + {"order": 3, "type": "java.lang.String", "name": "reqId"}, + {"order": 4, "type": "long", "name": "startTime"}, + {"order": 5, "type": "java.util.Map>", "name": "progress"} + ] +}