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..6739ecafea6d2 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 @@ -48,6 +48,7 @@ import java.util.function.BooleanSupplier; import java.util.function.Consumer; import java.util.function.Function; +import java.util.function.Supplier; import java.util.function.UnaryOperator; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -77,6 +78,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 +118,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 +141,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 +159,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 +194,7 @@ 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.RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE; import static org.apache.ignite.testframework.GridTestUtils.assertContains; import static org.apache.ignite.testframework.GridTestUtils.assertNotContains; @@ -233,20 +246,22 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb /** */ protected ListeningTestLogger listeningLog; + /** */ + protected @Nullable Supplier communicationSpiSupp; + + /** */ + protected @Nullable PluginProvider pluginProvider; + + /** */ + protected boolean walCompaction; + /** {@inheritDoc} */ @Override protected void beforeTest() throws Exception { super.beforeTest(); initDiagnosticDir(); - cleanDiagnosticDir(); - } - - /** {@inheritDoc} */ - @Override protected void afterTest() throws Exception { - super.afterTest(); - - listeningLog = null; + cleanPersistenceDir(); } /** {@inheritDoc} */ @@ -263,6 +278,14 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb if (listeningLog != null) cfg.setGridLogger(listeningLog); + if (communicationSpiSupp != null) + cfg.setCommunicationSpi(communicationSpiSupp.get()); + + if (pluginProvider != null) + cfg.setPluginProviders(pluginProvider); + + cfg.getDataStorageConfiguration().setWalCompactionEnabled(walCompaction); + return cfg; } @@ -3658,6 +3681,268 @@ 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(null); + } + + /** + * 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(false); + } + + /** + * Tests that snapshot metrics aren't empty when being restored snapshot waits for the check process. + * All nodes don't support the check status. + */ + @Test + public void testRestoreSnapshotMetricsAtStartNoSupport() throws Exception { + doTestRestoreSnapshotMetricsAtStart(true); + } + + /** + * @param allNodesNotSupporting Flag of how many nodes don't support the snapshot check status. + * If {@code null}, all nodes support. + * If {@code true}, none supports. + * If {@code false}, just one node doesn't support. + */ + private void doTestRestoreSnapshotMetricsAtStart(@Nullable Boolean allNodesNotSupporting) throws Exception { + communicationSpiSupp = TestRecordingCommunicationSpi::new; + + // Creates empty feature set unsupporting the snapshot check ststus if required. + pluginProvider = allNodesNotSupporting == null ? 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 = allNodesNotSupporting + || 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) { + 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 (Boolean.TRUE.equals(allNodesNotSupporting)) { + logLsnr = LogListener.matches("The snapshot-check-aware status feature isn't enabled").build(); + + listeningLog.registerListener(logLsnr); + } + else if (Boolean.FALSE.equals(allNodesNotSupporting)) { + 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 (allNodesNotSupporting == null) { + assertContains(log, out, "Check snapshot operation is in progress"); + assertContains(log, out, "Snapshot name: test_snapshot"); + } + else { + assertTrue(logLsnr.check(getTestTimeout())); + assertContains(log, out, "There is no create or restore snapshot operation in progress"); + } + + // 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 { + communicationSpiSupp = TestRecordingCommunicationSpi::new; + + walCompaction = 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()); + } + } + + grid(0).destroyCaches(twoSnapshots ? F.asList(DEFAULT_CACHE_NAME, "cache2") : F.asList(DEFAULT_CACHE_NAME)); + + awaitPartitionMapExchange(); + + 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); + + assertTrue(out.contains(twoSnapshots ? "Check snapshot operations are in progress" : "Check snapshot operation is in progress")); + 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(); + } + /** @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..53735c84000cf 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 @@ -52,8 +52,8 @@ public class SnapshotStatusCommand extends AbstractSnapshotCommand taskClass() { - return SnapshotStatusTask.class; + @Override public Class taskClass() { + return SnapshotStatusTaskV2.class; } /** {@inheritDoc} */ @@ -65,47 +65,62 @@ public class SnapshotStatusCommand extends AbstractSnapshotCommand 0; - GridStringBuilder s = new GridStringBuilder(); + assert (status instanceof SnapshotStatusTaskV2.SnapshotStatusV2) == !(isCreating || isRestoring) + : "No create or restore snapshot operation found but the status os not of V2 status."; + + // The check operation can be run in parallel for different snapshots. + List multipleOpsView = isCreating || isRestoring + ? Collections.singletonList(status) + : ((SnapshotStatusTaskV2.SnapshotStatusV2)status).checkStatuses(); + + assert multipleOpsView.size() == 1 || !(isCreating || isRestoring) : "Only snapshot check supports multiple operations."; if (isCreating) - s.a("Create snapshot operation is in progress.").nl(); + printer.accept("Create snapshot operation is in progress."); + else if (isRestoring) + printer.accept("Restore snapshot operation is in progress."); else - s.a("Restore snapshot operation is in progress.").nl(); + printer.accept("Check snapshot operation" + (multipleOpsView.size() < 2 ? " is " : "s are ") + "in progress."); - s.a("Snapshot name: ").a(status.name()).nl(); - s.a("Incremental: ").a(isIncremental).nl(); + printer.accept(U.nl()); - if (isIncremental) - s.a("Increment index: ").a(status.incrementIndex()).nl(); + multipleOpsView.forEach(s0 -> { + 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(); + s.a("Snapshot name: ").a(s0.name()).nl(); + s.a("Incremental: ").a(isIncremental).nl(); - printer.accept(s.toString()); + if (isIncremental) + s.a("Increment index: ").a(s0.incrementIndex()).nl(); - SnapshotTaskProgressDesc desc; + 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(); - if (isCreating && isIncremental) - desc = new CreateIncrementalSnapshotTaskProgressDesc(); - else if (isCreating) - desc = new CreateFullSnapshotTaskProgressDesc(); - else if (isIncremental) - desc = new RestoreIncrementalSnapshotTaskProgressDesc(); - else - desc = new RestoreFullSnapshotTaskProgressDesc(); + printer.accept(s.toString()); - List> rows = status.progress().entrySet().stream().sorted(Map.Entry.comparingByKey()) - .map(e -> desc.buildRow(e.getKey(), e.getValue())) - .collect(Collectors.toList()); + SnapshotTaskProgressDesc desc; - SystemViewCommand.printTable(desc.titles(), desc.types(), rows, printer); + if (isCreating) + desc = isIncremental ? new CreateIncrementalSnapshotTaskProgressDesc() : new CreateFullSnapshotTaskProgressDesc(); + else if (isRestoring) + desc = isIncremental ? new RestoreIncrementalSnapshotTaskProgressDesc() : new RestoreFullSnapshotTaskProgressDesc(); + else + desc = new CheckSnapshotTaskProgressDesc(isIncremental); - 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); + + printer.accept(U.nl()); + }); } /** Describes progress of a snapshot task. */ @@ -246,4 +261,55 @@ private static class RestoreIncrementalSnapshotTaskProgressDesc extends Snapshot return result; } } + + /** */ + private static class CheckSnapshotTaskProgressDesc extends SnapshotTaskProgressDesc { + /** */ + private final boolean incremental; + + /** */ + private CheckSnapshotTaskProgressDesc(boolean incremental) { + super(incremental + ? F.asList("Node ID", "processedWalSegments", "totalWalSegments", "percent") + : F.asList("Node ID", "fullCheck", "processedPartitions", "totalPartitions", + "processedSnapshotParts", "snapshotPartsToProcess", "percent") + ); + + this.incremental = incremental; + } + + /** {@inheritDoc} */ + @Override public List buildRow(UUID nodeId, T5 progress) { + if (incremental) { + long processed = progress.get1(); + long total = progress.get2(); + + if (total <= 0) + return F.asList(nodeId, "unknown", "unknown", "unknown"); + + String percent = (int)(processed * 100 / total) + "%"; + + return F.asList(nodeId, processed, total, percent); + } + + long partitionsToCheck = progress.get3(); + long partsToCheck = progress.get5(); + + if (partitionsToCheck <= 0 || partsToCheck <= 0) + return F.asList(nodeId, "unknown", "unknown", "unknown", "unknown", "unknown", "unknown"); + + // Ratio of checked partitions in current snapshot part * total parts ratio. + double totalRatio = ((double)progress.get2() / partitionsToCheck) * ((double)progress.get4() / partsToCheck); + + return F.asList( + nodeId, + progress.get1() == 0L ? "false" : "true", // Full check flag; + progress.get2(), // Checked partitions in current snapshot part; + partitionsToCheck, // Total partitions in current snapshot part; + progress.get4(), // Checked shapshot parts; + partsToCheck, // Total snapshot parts to check. + ((int)(totalRatio * 100.0d)) + '%' + ); + } + } } 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..ecc13fd9c7e95 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 @@ -19,7 +19,6 @@ import java.io.Serializable; import java.util.Collection; -import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -95,7 +94,7 @@ public class SnapshotStatusTask extends VisorMultiNodeTask { + protected static class SnapshotStatusJob extends SnapshotJob { /** */ private static final long serialVersionUID = 0L; @@ -205,7 +204,7 @@ public SnapshotStatus( this.incIdx = incIdx; this.requestId = requestId; this.startTime = startTime; - this.progress = Collections.unmodifiableMap(progress); + this.progress = progress; } /** @return Operation type. */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotStatusTaskV2.java b/modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotStatusTaskV2.java new file mode 100644 index 0000000000000..c699a097ab876 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotStatusTaskV2.java @@ -0,0 +1,239 @@ +/* + * 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.management.snapshot; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; +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.SnapshotCheckProcess; +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.lang.GridFunc; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.T5; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.internal.visor.VisorJob; +import org.apache.ignite.resources.LoggerResource; +import org.apache.ignite.spi.metric.BooleanMetric; +import org.apache.ignite.spi.metric.IntMetric; +import org.apache.ignite.spi.metric.LongMetric; +import org.jetbrains.annotations.Nullable; + +/** V2 of {@link SnapshotStatusTask} with support of the snapshot check status. */ +@GridInternal +public class SnapshotStatusTaskV2 extends SnapshotStatusTask { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + @LoggerResource + private transient IgniteLogger log; + + /** */ + private transient @Nullable Boolean checkStatusSupported; + + /** @inheritDoc} */ + @Override protected VisorJob job(NoArg arg) { + if (checkStatusSupported == null) + resolveCheckStatusSupported(); + + return checkStatusSupported ? new SnapshotStatusJobV2(arg, debug) : new SnapshotStatusJob(arg, debug); + } + + /** */ + private void resolveCheckStatusSupported() { + var feature = new IgniteCoreFeature(SupportedFeatureRegistry.SNAPSHOT_CHECK_STATUS_FEATURE.id()); + + if (!ignite.context().rollingUpgrade().features().isActive(feature)) { + log.warning("The snapshot-check-aware status feature isn't enabled. The status is available only for " + + "snapshot creation and restoration."); + + checkStatusSupported = false; + + return; + } + + for (var n : ignite.cluster().nodes()) { + if (!(n instanceof IgniteClusterNode cn) || !cn.features().contains(feature)) { + log.warning(String.format("Node %s doesn't support the snapshot check status feature. The status " + + "is available only for snapshot creation and restoration.", n.id())); + + checkStatusSupported = false; + + return; + } + } + + checkStatusSupported = true; + } + + /** {@inheritDoc} */ + @Override protected @Nullable SnapshotStatus reduce0(List results) { + SnapshotStatus res0 = super.reduce0(results); + + // No results received at all. + if (res0 == null) + return null; + + // Found create or restore result. + if (res0.operation() != null) + return res0; + + Collection sameRqRes = F.viewReadOnly(results, ComputeJobResult::getData, + r -> r.getData() != null && ((SnapshotStatus)r.getData()).requestId().equals(res0.requestId())); + + assert !F.isEmpty(sameRqRes); + + SnapshotStatus firstRes = F.first(sameRqRes); + + assert firstRes instanceof SnapshotStatusV2 : "Expected V2 snapshot status result"; + + SnapshotStatusV2 firstResV2 = (SnapshotStatusV2)firstRes; + + // Check status: snpName, per node collection. + Map statusesMap = U.newHashMap(sameRqRes.size()); + + sameRqRes.forEach(s -> { + assert s instanceof SnapshotStatusV2; + + for (SnapshotStatus s0 : ((SnapshotStatusV2)s).checkStatuses) { + var prev = statusesMap.putIfAbsent(s0.name(), s0); + + if (prev == null) + continue; + + // Merge nodes progress. + prev.progress().putAll(s0.progress()); + } + + firstResV2.checkStatuses = new ArrayList<>(statusesMap.values()); + }); + + return firstResV2; + } + + /** V2 of {@link SnapshotStatusJob} with support of the snapshot check status. */ + private static class SnapshotStatusJobV2 extends SnapshotStatusTask.SnapshotStatusJob { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + private SnapshotStatusJobV2(@Nullable NoArg arg, boolean debug) { + super(arg, debug); + } + + /** {@inheritDoc} */ + @Override protected @Nullable SnapshotStatus run(@Nullable NoArg arg) throws IgniteException { + var res1 = super.run(arg); + + // Create or restore status detected. + if (res1 != null) + return res1; + + List checkStatuses = null; + + for (var snpCheckMReg : ignite.context().metric()) { + if (!snpCheckMReg.name().startsWith(SnapshotCheckProcess.SNAPSHOT_CHECK_METRIC)) + continue; + + if (checkStatuses == null) + checkStatuses = new ArrayList<>(); + + int incIdx = snpCheckMReg.findMetric("incrementIndex") == null + ? 0 + : ((IntMetric)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<>( + snpCheckMReg.findMetric("checkPartitions").value() ? 1L : 0L, + (long)snpCheckMReg.findMetric("processedPartitions").value(), + (long)snpCheckMReg.findMetric("totalPartitions").value(), + (long)snpCheckMReg.findMetric("processedSnapshotParts").value(), + (long)snpCheckMReg.findMetric("snapshotPartsToProcess").value() + ); + } + + var status = new SnapshotStatus( + null, + MetricUtils.fromFullName(snpCheckMReg.name()).get2(), + incIdx, + snpCheckMReg.findMetric("requestId").getAsString(), + ((LongMetric)snpCheckMReg.findMetric("startTime")).value(), + GridFunc.asMap(ignite.localNode().id(), metrics) + ); + + checkStatuses.add(status); + } + + return checkStatuses == null ? null : new SnapshotStatusV2(checkStatuses); + } + } + + /** V2 of {@link SnapshotStatus} with support of the snapshot check status. */ + public static class SnapshotStatusV2 extends SnapshotStatusTask.SnapshotStatus { + /** */ + private static final long serialVersionUID = 0L; + + /** Nodes' statuses of all snapshot check operations. */ + private List checkStatuses; + + /** */ + private SnapshotStatusV2(SnapshotStatus s1) { + super(s1.operation(), s1.name(), s1.incrementIndex(), s1.requestId(), s1.startTime(), s1.progress()); + } + + /** */ + private SnapshotStatusV2(List checkStatuses) { + // Single, V1 status holds first check status. + super( + null, + checkStatuses.get(0).name(), + checkStatuses.get(0).incrementIndex(), + checkStatuses.get(0).requestId(), + checkStatuses.get(0).startTime(), + checkStatuses.get(0).progress() + ); + + this.checkStatuses = checkStatuses; + } + + /** */ + public List checkStatuses() { + return checkStatuses; + } + } +} 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..f069919248f91 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 @@ -732,6 +732,13 @@ private void registerMetrics(SnapshotCheckContext ctx) { mreg.register("startTime", U::currentTimeMillis, "The system time of the start of the cluster snapshot check operation on current node."); + mreg.register( + "requestId", + () -> ctx.req.requestId().toString(), + String.class, + "The request ID of the last running cluster snapshot restore operation on this node." + ); + if (ctx.req.incrementalIndex() > 0) { mreg.register("incrementIndex", ctx.req::incrementalIndex, "The index of incremental snapshot of the snapshot check operation."); 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/resources/META-INF/classnames.properties b/modules/core/src/main/resources/META-INF/classnames.properties index 65f0244da47b7..54dd1595b9b00 100644 --- a/modules/core/src/main/resources/META-INF/classnames.properties +++ b/modules/core/src/main/resources/META-INF/classnames.properties @@ -659,6 +659,9 @@ org.apache.ignite.internal.management.snapshot.SnapshotStatusTask org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotOperation org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotStatus org.apache.ignite.internal.management.snapshot.SnapshotStatusTask$SnapshotStatusJob +org.apache.ignite.internal.management.snapshot.SnapshotStatusTaskV2 +org.apache.ignite.internal.management.snapshot.SnapshotStatusTaskV2$SnapshotStatusV2 +org.apache.ignite.internal.management.snapshot.SnapshotStatusTaskV2$SnapshotStatusJobV2 org.apache.ignite.internal.management.tracing.TracingConfigurationCommand$TracingConfigurationCommandArg org.apache.ignite.internal.management.tracing.TracingConfigurationCommand$TracingConfigurationResetAllCommandArg org.apache.ignite.internal.management.tracing.TracingConfigurationCommand$TracingConfigurationResetCommandArg 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..c8e529f84b789 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 @@ -32,6 +32,7 @@ import org.apache.ignite.internal.managers.communication.GridIoMessage; import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionDemandMessage; import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage; +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.lang.IgniteBiInClosure; @@ -248,6 +249,13 @@ public void waitForRecorded() throws InterruptedException { } } + /** */ + public List blockedMessages() { + var res = blockedMsgs; + + return F.isEmpty(res) ? Collections.emptyList() : Collections.unmodifiableList(res); + } + /** * @param cls Message class. * @param nodeName Node name.