zstan commented on code in PR #13666: URL: https://github.com/apache/ignite/pull/13666#discussion_r4374837219
########## modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotListCommand.java: ########## @@ -0,0 +1,223 @@ +/* + * 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.time.Instant; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.Date; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; +import java.util.stream.Collectors; +import java.util.stream.IntStream; +import org.apache.ignite.internal.management.SystemViewCommand; +import org.apache.ignite.internal.management.SystemViewTask; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotListJobResult; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotListTaskResult; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.internal.U; + +import static org.apache.ignite.internal.management.SystemViewTask.SimpleType.DATE; +import static org.apache.ignite.internal.management.SystemViewTask.SimpleType.NUMBER; +import static org.apache.ignite.internal.management.SystemViewTask.SimpleType.STRING; + +/** Snapshot list command. */ +public class SnapshotListCommand extends AbstractSnapshotCommand<SnapshotListCommandArg, SnapshotListTaskResult> { + /** */ + public static final String HEADER = "Snapshots were found on the following nodes:"; + + /** */ + public static final String DESC = "Lists all snapshots on all online server nodes, including their sizes"; + + /** */ + public static final String NO_SNAPSHOTS = "No snapshots found."; + + /** {@inheritDoc} */ + @Override public String description() { + return DESC; + } + + /** {@inheritDoc} */ + @Override public Class<SnapshotListCommandArg> argClass() { + return SnapshotListCommandArg.class; + } + + /** {@inheritDoc} */ + @Override public Class<SnapshotListTask> taskClass() { + return SnapshotListTask.class; + } + + /** {@inheritDoc} */ + @Override public void printResult(SnapshotListCommandArg arg, SnapshotListTaskResult res, Consumer<String> printer) { + printer.accept(HEADER); + + // Sorting the results by consistent id for better reading. + Integer[] nodesOrder = IntStream.range(0, res.nodesIds().length).boxed() + .sorted(Comparator.comparing(idx -> res.consistentIds()[idx])) + .toArray(Integer[]::new); + + // Flag whether the node's result has at least one incremental snapshot. + AtomicBoolean hasIncrementals = new AtomicBoolean(); + // Flag whether the node's result has at least one external storage. + AtomicBoolean hasExtStorages = new AtomicBoolean(); Review Comment: u not need to use Atomic here, not so good but more accurate is to use T2 or custom structure, plz fix it ########## modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotListTask.java: ########## @@ -0,0 +1,452 @@ +/* + * 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.io.File; +import java.io.IOException; +import java.nio.file.FileVisitResult; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.attribute.BasicFileAttributes; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.IgniteException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.compute.ComputeJobResult; +import org.apache.ignite.internal.NodeStoppingException; +import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotMetadata; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotListJobResult; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotListTaskResult; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadata; +import org.apache.ignite.internal.processors.rollingupgrade.feature.CoreFeatureRegistry; +import org.apache.ignite.internal.processors.task.GridInternal; +import org.apache.ignite.internal.thread.pool.IgniteThreadPoolExecutor; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.T2; +import org.apache.ignite.internal.util.typedef.X; +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.resources.LoggerResource; +import org.jetbrains.annotations.Nullable; + +import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.nodeIds; + +/** */ +@GridInternal +public class SnapshotListTask extends VisorMultiNodeTask<SnapshotListCommandArg, SnapshotListTaskResult, SnapshotListJobResult> { + /** Serial version uid. */ + private static final long serialVersionUID = 0L; + + /** */ + public static final String FEATURE_NOT_ACTIVE_ERR_MSG = "Won't search for local snapshots. " + + "The snapshot list feature isn't activated yet."; + + /** {@inheritDoc} */ + @Override protected VisorJob<SnapshotListCommandArg, SnapshotListJobResult> job(SnapshotListCommandArg arg) { + return new SnapshotListJob(arg, debug); + } + + /** {@inheritDoc} */ + @Override protected Collection<UUID> jobNodes(VisorTaskArgument<SnapshotListCommandArg> arg) { + if (!ignite.context().rollingUpgrade().features().isActive(CoreFeatureRegistry.SNAPSHOT_LIST_FEATURE)) + throw new IgniteException(FEATURE_NOT_ACTIVE_ERR_MSG); + + /** Allows {@link #map0(List, VisorTaskArgument)} to use the entire grid. */ + return nodeIds(ignite.cluster().forServers().nodes()); + } + + /** {@inheritDoc} */ + @Override protected SnapshotListTaskResult reduce0(List<ComputeJobResult> nodesJobsResults) throws IgniteException { + String[] cstIds = new String[nodesJobsResults.size()]; + UUID[] nodesIds = new UUID[nodesJobsResults.size()]; + SnapshotListJobResult[] nodesResults = new SnapshotListJobResult[nodesJobsResults.size()]; + + for (int i = 0; i < nodesJobsResults.size(); i++) { + ComputeJobResult nodeJobRes = nodesJobsResults.get(i); + + if (nodeJobRes.getException() != null) { + throw new IgniteException("Failed to execute snapshot list job on node [uuid=" + nodeJobRes.getNode().id() + ']', + nodeJobRes.getException()); + } + + assert nodeJobRes.getData() != null; + + cstIds[i] = nodeConsistentId(nodeJobRes.getNode()); + nodesIds[i] = nodeJobRes.getNode().id(); + nodesResults[i] = nodeJobRes.getData(); + } + + return new SnapshotListTaskResult(cstIds, nodesIds, nodesResults); + } + + /** */ + private String nodeConsistentId(ClusterNode n) { + UUID nodeId = n.id(); + + n = ignite.context().discovery().node(nodeId); + + if (n == null) + n = ignite.context().discovery().historicalNode(nodeId); + + return n == null ? "" : n.consistentId().toString(); + } + + /** + * Walk through a directory. Doesn't lock it or its content. Tries to find files and summarize their size. + * Tolerates and skips concurrent modification errors. + */ + public static long calculateDirectorySize(File path) throws IOException { + AtomicLong size = new AtomicLong(0); + AtomicBoolean entered = new AtomicBoolean(); + + Files.walkFileTree(path.toPath(), new SimpleFileVisitor<>() { + @Override public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) { + entered.compareAndSet(false, true); + + // Use attrs instead of Files.size() for efficiency. + if (attrs.isRegularFile()) + size.addAndGet(attrs.size()); + + return FileVisitResult.CONTINUE; + } + + @Override public FileVisitResult visitFileFailed(Path file, IOException err) throws IOException { + if (!entered.get() && file.toFile().equals(path)) { + // Cannot even start snapshot size calculation - can't enter snapshot directory. + throw err; + } + + return FileVisitResult.CONTINUE; + } + }); + + return size.get(); + } + + /** */ + private static class SnapshotListJob extends SnapshotJob<SnapshotListCommandArg, SnapshotListJobResult> { + /** Serial version uid. */ + private static final long serialVersionUID = 0L; + + /** */ + @LoggerResource + private IgniteLogger log; + + /** + * @param arg Snapshot list task argument. + * @param debug Flag indicating whether debug information should be printed into node log. + */ + protected SnapshotListJob(SnapshotListCommandArg arg, boolean debug) { + super(arg, debug); + } + + /** {@inheritDoc} */ + @Override protected SnapshotListJobResult run(SnapshotListCommandArg arg) { + assert !ignite.localNode().isClient(); + + if (ignite.context().isStopping()) + throw new IgniteException("Won't search for local snapshots.", new NodeStoppingException("Node is stopping.")); + + // Read local snapshots. + List<T2<SnapshotFileTree, Long>> locSnps = findLocalSnapshots(arg.src()); + + if (locSnps.isEmpty()) + return new SnapshotListJobResult(Collections.emptyMap()); + + // Main sizes of the local snapshots (per snapshot name). + Map<String, Long> sizes = new ConcurrentHashMap<>(locSnps.size(), 1.0f); + // Optional incremental parts descriptions (per snapshot name). + Map<String, SnapshotListJobResult.SnapshotInfo> incs = new ConcurrentHashMap<>(locSnps.size(), 1.0f); + // Optional external storages descriptions (per snapshot name). + Map<String, SnapshotListJobResult.SnapshotInfo> extStors = new ConcurrentHashMap<>(locSnps.size(), 1.0f); + + // Snapshot sizes, incremental parts and external storages futures. + List<Future<?>> futs = new ArrayList<>(locSnps.size() * 3); + // Names of snapshots with read failures. + Set<String> failedSnps = ConcurrentHashMap.newKeySet(locSnps.size() / 2); + + IgniteThreadPoolExecutor exec = ignite.context().pools().getSnapshotExecutorService(); + + // Descriptions of the local snapshots: main sizes, external storages and incremental parts. + for (T2<SnapshotFileTree, Long> snpPair : locSnps) { + SnapshotFileTree sft = snpPair.get1(); + String snpName = sft.name(); + + // Future for the snapshot main size. It is the largest part of the job, thus it goes first. + Future<?> fut = exec.submit(() -> { + if (ignite.context().isStopping()) + throw new IgniteException("Won't search for local snapshots.", new NodeStoppingException("Node is stopping.")); + + if (failedSnps.contains(snpName)) + return; + + try { + sizes.put(snpName, calculateDirectorySize(sft.root())); + } + catch (Exception e) { + failedSnps.add(snpName); + + log.warning("Failed to calculate snapshot's size, snapshot ignored [snpName='" + snpName + "']", e); + } + }); + + futs.add(fut); + + // Future for optional incremental parts. + fut = exec.submit(() -> { + if (ignite.context().isStopping()) + throw new IgniteException("Won't search for local snapshots.", new NodeStoppingException("Node is stopping.")); + + if (failedSnps.contains(snpName)) + return; + + try { + SnapshotListJobResult.SnapshotInfo incDesc = incrementals(sft); + + if (incDesc != null) + incs.put(snpName, incDesc); + } + catch (Exception e) { + failedSnps.add(snpName); + + log.warning("Failed to read snapshot's incremental parts, snapshot ignored [snpName='" + snpName + "']", e); + } + }); + + futs.add(fut); + + // Future for optional external storages. + fut = exec.submit(() -> { + if (ignite.context().isStopping()) + throw new IgniteException("Won't search for local snapshots.", new NodeStoppingException("Node is stopping.")); + + if (failedSnps.contains(snpName)) + return; + + try { + SnapshotListJobResult.SnapshotInfo extDesc = externalStorages(sft); + + if (extDesc != null) + extStors.put(snpName, extDesc); + } + catch (Exception e) { + failedSnps.add(snpName); + + log.warning("Failed to read snapshot's external storages, snapshot ignored [snpName='" + snpName + "']", e); + } + }); + + futs.add(fut); + } + + // Wait for the futures. + for (Future<?> fut : futs) { + try { + fut.get(); + } + catch (ExecutionException e) { + throw new IgniteException("Failed to read local node's snapshots.", e); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + + throw new IgniteException("Interrupted while reading local snapshots.", e); + } + } + + // Result snapshot descriptions. + Map<String, SnapshotListJobResult.SnapshotInfo> resMap = new ConcurrentHashMap<>(locSnps.size() - failedSnps.size(), 1.0f); + + // Reduce results. + for (T2<SnapshotFileTree, Long> snpPair : locSnps) { + String snpName = snpPair.get1().name(); + + if (failedSnps.contains(snpName)) + continue; + + Long size = sizes.get(snpName); + + // The size must be calculated unless the snapshot has failed. + assert size != null; + + SnapshotListJobResult.SnapshotInfo snpDesc = new SnapshotListJobResult.SnapshotInfo( + size, + snpPair.get2(), + extStors.get(snpName), + incs.get(snpName) + ); + + resMap.put(snpName, snpDesc); + } + + return new SnapshotListJobResult(resMap); + } + + /** */ + private @Nullable SnapshotListJobResult.SnapshotInfo externalStorages(SnapshotFileTree sft) { + int extStoragesCnt = 0; + long extStoragesSize = 0; + + for (File extraStorage : sft.allStorages().toList()) { + if (sft.nodeStorage().equals(extraStorage)) + continue; + + try { + extStoragesSize += calculateDirectorySize(extraStorage); + } + catch (Exception e) { + log.warning("Failed to calculate snapshot's external storage size, storage ignored [extraStorage='" + + extraStorage + "']", e); + + continue; + } + + extStoragesCnt++; + } + + return extStoragesCnt == 0 ? null : new SnapshotListJobResult.SnapshotInfo(extStoragesCnt, extStoragesSize); + } + + /** @return Snapshot file tree and creation time from the snapshot metadata. */ + private List<T2<SnapshotFileTree, Long>> findLocalSnapshots(@Nullable String snpPath) { + // The tree is used only to extract the snapshots root directory. Such snapshot name is actually unallowed. + File[] dirsToParse = new SnapshotFileTree(ignite.context(), "snp", snpPath).root().getParentFile().listFiles(); Review Comment: Seems you can replace this strange code for (a bit clear) : `org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree#root` wdyt ? ########## modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotListTest.java: ########## @@ -0,0 +1,918 @@ +/* + * 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.IOException; +import java.io.RandomAccessFile; +import java.io.Serializable; +import java.nio.file.DirectoryStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.nio.file.attribute.PosixFilePermission; +import java.nio.file.attribute.PosixFilePermissions; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.stream.Stream; +import org.apache.ignite.IgniteDataStreamer; +import org.apache.ignite.IgniteException; +import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.NodeStoppingException; +import org.apache.ignite.internal.management.snapshot.SnapshotListCommandArg; +import org.apache.ignite.internal.management.snapshot.SnapshotListTask; +import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree; +import org.apache.ignite.internal.util.io.GridFileUtils; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.internal.visor.VisorTaskArgument; +import org.apache.ignite.lang.IgniteFuture; +import org.apache.ignite.plugin.AbstractTestPluginProvider; +import org.apache.ignite.plugin.PluginConfiguration; +import org.apache.ignite.plugin.PluginContext; +import org.apache.ignite.plugin.PluginProvider; +import org.jetbrains.annotations.Nullable; +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runners.Parameterized; + +import static java.nio.file.Files.newDirectoryStream; +import static org.apache.ignite.configuration.IgniteConfiguration.DFLT_SNAPSHOT_THREAD_POOL_SIZE; +import static org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause; +import static org.apache.ignite.testframework.GridTestUtils.cartesianProduct; +import static org.apache.ignite.testframework.GridTestUtils.runAsync; +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; +import static org.junit.Assume.assumeFalse; +import static org.junit.Assume.assumeTrue; + +/** Cluster-wide snapshot list procedure tests. */ +public class IgniteClusterSnapshotListTest extends AbstractSnapshotSelfTest { + /** Number of cache keys to pre-create at node startup. */ + private static final int CACHE_KEYS_RANGE = 15; + + /** Number of partitions of the test cache. */ + private static final int CACHE_PARTITIONS_COUNT = 4; + + /** Flag showing that the filesystem supports POSIX permissions. */ + private static boolean posixPermissions; + + /** Size of the snapshot utility thread pool. */ + @Parameterized.Parameter(2) + public int snpThrdPoolSz; + + /** */ + private PluginProvider<PluginConfiguration> pluginProvider; + + /** Flag to spread the test cache data over external storages. */ + private boolean extStorages; + + /** Resolved external storage paths. {@code null} if {@link #extStorages} is {@code false}. */ + private @Nullable String[] extStoragePaths; + + /** Parameters. */ + @Parameterized.Parameters(name = "encryption={0}, onlyPrimary={1}, snpThrdPoolSz={2}") + public static Collection<Object[]> params() { + return cartesianProduct( + encryptionParameters(), // Encryption. + F.asList(false, true), // Only primary. + F.asList(DFLT_SNAPSHOT_THREAD_POOL_SIZE, 1) // Snapshots thread pool size. + ); + } + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName) + .setWorkDirectory(new File(U.defaultWorkDirectory(), igniteInstanceName).getAbsolutePath()); + + if (pluginProvider != null) + cfg.setPluginProviders(pluginProvider); + + if (extStorages) { + // External storage paths must be identical on all nodes (a cache has a single storage path setting). + // Thus, a shared work directory is used, like in GridCommandHandlerListSnapshotTest. + cfg.setWorkDirectory(U.defaultWorkDirectory()); + + cfg.getDataStorageConfiguration().setExtraStoragePaths( + U.defaultWorkDirectory() + File.separator, + U.defaultWorkDirectory() + File.separator + "extStorage" + ); + + extStoragePaths = cfg.getDataStorageConfiguration().getExtraStoragePaths(); + + cfg.getDataStorageConfiguration().setExtraSnapshotPaths("", "extStorage"); + } + + cfg.setSnapshotThreadPoolSize(snpThrdPoolSz); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void cleanPersistenceDir() throws Exception { + super.cleanPersistenceDir(); + + // Also cleans separated snapshot working directories and custom snapshot paths. + try (DirectoryStream<Path> files = newDirectoryStream(Paths.get(U.defaultWorkDirectory()))) { + for (Path path : files) + U.delete(path); + } + } + + /** {@inheritDoc} */ + @Override protected void beforeTestsStarted() throws Exception { + super.beforeTestsStarted(); + + File workDir = new File(U.defaultWorkDirectory()); + + assertTrue(workDir.exists()); + + try { + Files.getPosixFilePermissions(workDir.toPath()); + + posixPermissions = true; + } + catch (UnsupportedOperationException ignored) { + // No-op. + } + } + + /** {@inheritDoc} */ + @Override protected <K, V> CacheConfiguration<K, V> txCacheConfig(CacheConfiguration<K, V> ccfg) { + // Speeds up the tests. + ccfg = super.txCacheConfig(ccfg) + .setAffinity(new RendezvousAffinityFunction(false, CACHE_PARTITIONS_COUNT)) + .setBackups(1); + + if (extStorages) { + assert !F.isEmpty(extStoragePaths); + + ccfg.setStoragePaths(extStoragePaths); + } + + return ccfg; + } + + /** */ + @Test + public void testConcurrentCreation() throws Exception { + // The test relies on thread blocking and conditional waiting, so it does not run with a single thread. + assumeTrue(snpThrdPoolSz > 1); + + int grids = 3; + int testNodeIdx = 1; + int testNodeOrder = testNodeIdx + 1; + + CountDownLatch beginSnpCreation = new CountDownLatch(grids); + CountDownLatch proceedSnpCreation = new CountDownLatch(1); + + // Delays snapshot creation after its metadata is written. + pluginProvider = new AbstractTestPluginProvider() { + @Override public String name() { + return "TestSnpMgrProvider"; + } + + @Override public <T> T createComponent(PluginContext ctx, Class<T> cls) { + if (IgniteSnapshotManager.class.isAssignableFrom(cls)) { + return (T)new IgniteSnapshotManager(((IgniteEx)ctx.grid()).context()) { + @Override public <M extends Serializable> void storeSnapshotMeta(M meta, File smf) { + super.storeSnapshotMeta(meta, smf); + + beginSnpCreation.countDown(); + + if (((IgniteEx)ctx.grid()).localNode().order() == testNodeOrder) { + try { + assertTrue(proceedSnpCreation.await(getTestTimeout(), TimeUnit.MILLISECONDS)); + } + catch (InterruptedException e) { + throw new IllegalStateException(e); + } + } + } + }; + } + + return super.createComponent(ctx, cls); + } + }; + + startGridsWithCache(grids, txCacheConfig(defaultCacheConfiguration()), CACHE_KEYS_RANGE); + + IgniteFuture<Void> createSnpFut = snp(grid(0)).createSnapshot(SNAPSHOT_NAME, null, false, onlyPrimary); + + assertTrue(beginSnpCreation.await(getTestTimeout(), TimeUnit.MILLISECONDS)); + + SnapshotListTaskResult lstOpRes = listSnapshots(grid(0)); + + int snpsCnt = Stream.of(lstOpRes.nodesSnapshots()).mapToInt(jr -> jr.snapshots().size()).sum(); + + assertEquals(grids, snpsCnt); + + SnapshotFileTree testSft = new SnapshotFileTree(grid(testNodeIdx).context(), SNAPSHOT_NAME, null); + + assertTrue(GridFileUtils.sizeOfDirectory(testSft.root()) > 0L); + + proceedSnpCreation.countDown(); + + createSnpFut.get(getTestTimeout()); + } + + /** */ + @Test + public void testCompletelyDeletedAfterMetaRead() throws Exception { + doTestDeletionAfterMetaRead(true); + } + + /** */ + @Test + public void testPartlyDeletedAfterMetaRead() throws Exception { + doTestDeletionAfterMetaRead(false); + } + + /** */ + private void doTestDeletionAfterMetaRead(boolean completeDeletion) throws Exception { + int grids = 3; + int testGridIdx = 1; + + CountDownLatch metaReadBeginLatch = new CountDownLatch(grids); + CountDownLatch metaReadProceedLatch = new CountDownLatch(1); + + pluginProvider = new AbstractTestPluginProvider() { + @Override public String name() { + return "TestSnpMgrProvider"; + } + + @Override public <T> T createComponent(PluginContext ctx, Class<T> cls) { + if (IgniteSnapshotManager.class.isAssignableFrom(cls)) { + return (T)new IgniteSnapshotManager(((IgniteEx)ctx.grid()).context()) { + @Override public List<SnapshotMetadata> readSnapshotMetadatas(SnapshotFileTree sft, boolean failIfCantRead) { + metaReadBeginLatch.countDown(); + + if (cctx.localNode().order() == testGridIdx + 1) { + try { + assertTrue(metaReadProceedLatch.await(getTestTimeout(), TimeUnit.MILLISECONDS)); + } + catch (InterruptedException e) { + throw new IllegalStateException(e); + } + } + + return super.readSnapshotMetadatas(sft, failIfCantRead); + } + }; + } + + return super.createComponent(ctx, cls); + } + }; + + startGridsWithSnapshot(grids, CACHE_KEYS_RANGE, true, false); + + SnapshotFileTree testSnpFt = new SnapshotFileTree(grid(testGridIdx).context(), SNAPSHOT_NAME, null); + + long testSnpSz = GridFileUtils.sizeOfDirectory(testSnpFt.root()); + + assertTrue(testSnpSz > 0); + + IgniteInternalFuture<SnapshotListTaskResult> lstOpFut = runAsync(() -> listSnapshots(grid(0))); + + assertTrue(metaReadBeginLatch.await(getTestTimeout(), TimeUnit.MILLISECONDS)); + + if (completeDeletion) + assertTrue(testSnpFt.root().exists() && U.delete(testSnpFt.root()) && !testSnpFt.root().exists()); + else + assertTrue(testSnpFt.nodeStorage().exists() && U.delete(testSnpFt.nodeStorage()) && !testSnpFt.nodeStorage().exists()); + + metaReadProceedLatch.countDown(); + + SnapshotListTaskResult lstOpRes = lstOpFut.get(); + + if (completeDeletion) { + int snpsCnt = Stream.of(lstOpRes.nodesSnapshots()).mapToInt(jr -> jr.snapshots().size()).sum(); + + assertEquals(grids - 1, snpsCnt); + } + else { + int snpsCnt = 0; + boolean victimNodeFound = false; + + for (int i = 0; i < lstOpRes.nodesIds().length; i++) { + UUID nid = lstOpRes.nodesIds()[i]; + + Map<String, SnapshotListJobResult.SnapshotInfo> nodeSnps = lstOpRes.nodesSnapshots()[i].snapshots(); + + snpsCnt += nodeSnps.size(); + + if (nid.equals(grid(testGridIdx).localNode().id())) { + victimNodeFound = true; + + assertEquals(1, nodeSnps.size()); + + long curSnpSz = GridFileUtils.sizeOfDirectory(testSnpFt.root()); + + assertTrue(curSnpSz < testSnpSz); + } + } + + assertTrue(victimNodeFound); + assertEquals(grids, snpsCnt); + } + } + + /** */ + @Test + public void testEmptySnapshotPath() throws Exception { + startGrids(3); + + File path = new File(U.defaultWorkDirectory(), "not_snapshots"); + + SnapshotListJobResult[] listOperationRes = listSnapshots(grid(0), path.getAbsolutePath()).nodesSnapshots(); + + int cnt = Stream.of(listOperationRes).mapToInt(nodeRes -> nodeRes.snapshots().size()).sum(); + + assertEquals(0, cnt); + } + + /** */ + @Test + public void testDefaultSnapshotPath() throws Exception { + doTestSnapshotPath(null); + } + + /** */ + @Test + @Ignore("https://issues.apache.org/jira/browse/IGNITE-29126";) + public void testRelativeSnapshotPath() throws Exception { + doTestSnapshotPath("ex_snapshots"); + } + + /** */ + @Test + public void testAbsoluteSnapshotPath() throws Exception { + doTestSnapshotPath(new File(U.defaultWorkDirectory(), "ex_snapshots").getAbsolutePath()); + } + + /** */ + private void doTestSnapshotPath(@Nullable String path) throws Exception { + int grids = 3; + + startGridsWithCache(grids, txCacheConfig(defaultCacheConfiguration()), CACHE_KEYS_RANGE); + + snp(grid(0)).createSnapshot(SNAPSHOT_NAME, path, false, onlyPrimary).get(getTestTimeout()); + + SnapshotListJobResult[] lstOpRes = listSnapshots(grid(0), path).nodesSnapshots(); + + int cnt = Stream.of(lstOpRes).mapToInt(nodeRes -> nodeRes.snapshots().size()).sum(); + + assertEquals(grids, cnt); + } + + /** */ + @Test + public void testMissingMeta() throws Exception { + doTestWithWrongMeta(false); + } + + /** */ + @Test + public void testCorruptedMeta() throws Exception { + doTestWithWrongMeta(true); + } + + /** + * Tests snapshot lists when one snapshot metadata cannot be read. + * + * @param corruptFile If {@code true}, corrupts metadata. Otherwise, deletes metadata. + */ + private void doTestWithWrongMeta(boolean corruptFile) throws Exception { + int grids = 3; + int testGridIdx = 1; + + startGridsWithSnapshot(grids, CACHE_KEYS_RANGE, true, false); + + SnapshotFileTree testSnpFt = new SnapshotFileTree(grid(testGridIdx).context(), SNAPSHOT_NAME, null); + + assertTrue(testSnpFt.meta().exists()); + + if (corruptFile) { + try (RandomAccessFile raf = new RandomAccessFile(testSnpFt.meta(), "rw")) { + raf.write(UUID.randomUUID().toString().getBytes()); + } + } + else + assertTrue(testSnpFt.meta().delete() && !testSnpFt.meta().exists()); + + SnapshotListTaskResult res = listSnapshots(grid(0)); + + int foundSnpsCnt = 0; + boolean victimNodeFound = false; + + for (int i = 0; i < res.nodesIds().length; i++) { + UUID nid = res.nodesIds()[i]; + + foundSnpsCnt += res.nodesSnapshots()[i].snapshots().size(); + + if (nid.equals(grid(testGridIdx).localNode().id())) { + victimNodeFound = true; + + assertTrue(res.nodesSnapshots()[i].snapshots().isEmpty()); + } + } + + assertTrue(victimNodeFound); + assertEquals(grids - 1, foundSnpsCnt); + } + + /** */ + @Test + public void testMissingIncrementalMeta() throws Exception { + doTestWithWrongIncrementalMeta(false); + } + + /** */ + @Test + public void testCorruptedIncrementalMeta() throws Exception { + doTestWithWrongIncrementalMeta(true); + } + + /** + * Tests snapshot list when incremental snapshot metadata cannot be read. + * The main snapshot should still be listed, but without incremental info on the affected node. + * + * @param corruptFile If {@code true}, corrupts metadata. Otherwise, deletes metadata. + */ + private void doTestWithWrongIncrementalMeta(boolean corruptFile) throws Exception { + // Incremental snapshots do not support the only-primary mode or encryption. + assumeFalse(onlyPrimary || encryption); + + int grids = 3; + int testGridIdx = 1; + int incsCnt = 3; + + IgniteEx ig = startGridsWithCache(grids, txCacheConfig(defaultCacheConfiguration()), CACHE_KEYS_RANGE); + + snp(ig).createSnapshot(SNAPSHOT_NAME, null, false, onlyPrimary).get(getTestTimeout()); + + for (int i = 0; i < incsCnt; ++i) { + try (IgniteDataStreamer<Integer, Integer> ds = grid(0).dataStreamer(DEFAULT_CACHE_NAME)) { + for (int kv = (i + 1) * CACHE_KEYS_RANGE; kv < (i + 1) * CACHE_KEYS_RANGE * 2; ++kv) + ds.addData(kv, kv); + } + + snp(ig).createSnapshot(SNAPSHOT_NAME, null, true, onlyPrimary).get(getTestTimeout()); + } + + SnapshotFileTree testSnpFt = new SnapshotFileTree(grid(testGridIdx).context(), SNAPSHOT_NAME, null); + SnapshotFileTree.IncrementalSnapshotFileTree incFt = testSnpFt.incrementalSnapshotFileTree(1); + File incMeta = incFt.meta(); + + assertTrue(incMeta.exists()); + + if (corruptFile) { + try (RandomAccessFile raf = new RandomAccessFile(incMeta, "rw")) { + raf.write(UUID.randomUUID().toString().getBytes()); + } + } + else + assertTrue(incMeta.delete() && !incMeta.exists()); + + SnapshotListTaskResult res = listSnapshots(grid(0)); + + for (int i = 0; i < res.nodesIds().length; i++) { + UUID nid = res.nodesIds()[i]; + + Map<String, SnapshotListJobResult.SnapshotInfo> snps = res.nodesSnapshots()[i].snapshots(); + + assertEquals(1, snps.size()); + + SnapshotListJobResult.SnapshotInfo info = snps.get(SNAPSHOT_NAME); + + assertNotNull(info); + assertNotNull(info.incrementals()); + + if (nid.equals(grid(testGridIdx).localNode().id())) + assertEquals(incsCnt - 1, info.incrementals().number().intValue()); + else + assertEquals(incsCnt, info.incrementals().number().intValue()); + } + } + + /** + * Test snapshot list operation when a node can't read some snapshot part due to insufficient permissions. + * That is, a test node can read the snapshot metadata but cannot read some of the snapshot data. + */ + @Test + public void testDeniedPermissions() throws Exception { + assumeTrue(posixPermissions); + // We rely on the snapshot sizes here. Better to avoid empty data nodes not to become flaky. + assumeFalse(onlyPrimary); + + int grids = 3; + int testGridIdx = 1; + + // Permissions to restore. + Map<Path, Set<PosixFilePermission>> oldPerms = new ConcurrentHashMap<>(); + // The 'change permissions' flag. + AtomicBoolean changePermissions = new AtomicBoolean(true); + + // Deny reading the whole snapshot part of one node. + pluginProvider = new AbstractTestPluginProvider() { + @Override public String name() { + return "TestSnpMgrProvider"; + } + + @Override public <T> T createComponent(PluginContext ctx, Class<T> cls) { + if (IgniteSnapshotManager.class.isAssignableFrom(cls)) { + return (T)new IgniteSnapshotManager(((IgniteEx)ctx.grid()).context()) { + @Override public List<SnapshotMetadata> readSnapshotMetadatas(SnapshotFileTree sft, + boolean failIfCantRead) { + if (changePermissions.get() && ctx.localNode().order() == testGridIdx + 1) { + File victimDir = sft.nodeStorage(); + + assertTrue(victimDir.exists()); + assertTrue(victimDir.isDirectory()); + + // Ensure the blocked snapshot part has a non-zero size; the sizes are compared later. + try { + assertTrue(GridFileUtils.sizeOfDirectory(victimDir) > 0L); + } + catch (IOException e) { + throw new IllegalStateException(e); + } + + Path victimDirPath = victimDir.toPath(); + + try { + Set<PosixFilePermission> perms = Files.getPosixFilePermissions(victimDirPath); + + assertFalse(perms.isEmpty()); + + // Deny reading. + Files.setPosixFilePermissions(victimDirPath, PosixFilePermissions.fromString("---------")); + + // Save actual permissions to restore. + oldPerms.put(victimDirPath, perms); + } + catch (Exception e) { + throw new IgniteException("Unable to set the test POSIX permissions.", e); + } + } + + return super.readSnapshotMetadatas(sft, failIfCantRead); + } + }; + } + + return super.createComponent(ctx, cls); + } + }; + + startGridsWithSnapshot(grids, CACHE_KEYS_RANGE, true, true); + + UUID testNodeId = grid(testGridIdx).localNode().id(); + + // Snapshot size with restricted permissions. + long testSize0 = 0L; + // Snapshot size with normal permissions. + long testSize1 = 0L; + + SnapshotListTaskResult listTaskRes; + + try { + // First run. + listTaskRes = listSnapshots(grid(0)); + + assertFalse(oldPerms.isEmpty()); + + for (int i = 0; i < listTaskRes.nodesIds().length; i++) { + UUID nodeId = listTaskRes.nodesIds()[i]; + + // Store the partly read snapshot size. + if (nodeId.equals(testNodeId)) { + Map<String, SnapshotListJobResult.SnapshotInfo> snps = listTaskRes.nodesSnapshots()[i].snapshots(); + + assertEquals(1, snps.size()); + + testSize0 = snps.get(SNAPSHOT_NAME).size(); + } + } + + // Ensure that we've found and read the snapshot on the test node. + assertTrue(testSize0 > 0L); + } + finally { + // Restore the permissions in any case. + oldPerms.forEach((path, perms) -> { + try { + Files.setPosixFilePermissions(path, perms); + } + catch (IOException e) { + throw new IllegalStateException(e); + } + }); + } + + // Relaunch the operation. + changePermissions.set(false); + + listTaskRes = listSnapshots(grid(0)); + + for (int i = 0; i < listTaskRes.nodesIds().length; i++) { + UUID nodeId = listTaskRes.nodesIds()[i]; + + // Store the fully read snapshot size. + if (nodeId.equals(testNodeId)) { + Map<String, SnapshotListJobResult.SnapshotInfo> snps = listTaskRes.nodesSnapshots()[i].snapshots(); + + assertEquals(1, snps.size()); + + testSize1 = snps.get(SNAPSHOT_NAME).size(); + } + } + + // Ensure that the newly calculated size is greater than in the previous run. + assertTrue(testSize1 > 0L); + assertTrue(testSize1 > testSize0); + } + + /** */ + @Test + public void testSnapshotListsDates() throws Exception { + // Incremental snapshots do not support the only-primary mode or encryption. + assumeFalse(onlyPrimary || encryption); + + int grids = 3; + + IgniteEx ig = startGridsWithCache(grids, txCacheConfig(defaultCacheConfiguration()), CACHE_KEYS_RANGE); + + long time0 = U.currentTimeMillis(); + + // Wait so that the timestamps differ. + U.sleep(300L); Review Comment: do we really want to chech for strong ">" , I mean : `assertTrue(info.date() > time0);` I just don\`t like to use sleep in tests without strong needs, probably you remove it and near too and change for : assertTrue(info.date() >= time0); also time0 - non informative and need to be renamed ########## modules/core/src/main/java/org/apache/ignite/internal/management/snapshot/SnapshotListTask.java: ########## @@ -0,0 +1,452 @@ +/* + * 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.io.File; +import java.io.IOException; +import java.nio.file.FileVisitResult; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.attribute.BasicFileAttributes; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.IgniteException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.compute.ComputeJobResult; +import org.apache.ignite.internal.NodeStoppingException; +import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotMetadata; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotListJobResult; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotListTaskResult; +import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadata; +import org.apache.ignite.internal.processors.rollingupgrade.feature.CoreFeatureRegistry; +import org.apache.ignite.internal.processors.task.GridInternal; +import org.apache.ignite.internal.thread.pool.IgniteThreadPoolExecutor; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.T2; +import org.apache.ignite.internal.util.typedef.X; +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.resources.LoggerResource; +import org.jetbrains.annotations.Nullable; + +import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.nodeIds; + +/** */ +@GridInternal +public class SnapshotListTask extends VisorMultiNodeTask<SnapshotListCommandArg, SnapshotListTaskResult, SnapshotListJobResult> { + /** Serial version uid. */ + private static final long serialVersionUID = 0L; + + /** */ + public static final String FEATURE_NOT_ACTIVE_ERR_MSG = "Won't search for local snapshots. " + + "The snapshot list feature isn't activated yet."; + + /** {@inheritDoc} */ + @Override protected VisorJob<SnapshotListCommandArg, SnapshotListJobResult> job(SnapshotListCommandArg arg) { + return new SnapshotListJob(arg, debug); + } + + /** {@inheritDoc} */ + @Override protected Collection<UUID> jobNodes(VisorTaskArgument<SnapshotListCommandArg> arg) { + if (!ignite.context().rollingUpgrade().features().isActive(CoreFeatureRegistry.SNAPSHOT_LIST_FEATURE)) + throw new IgniteException(FEATURE_NOT_ACTIVE_ERR_MSG); + + /** Allows {@link #map0(List, VisorTaskArgument)} to use the entire grid. */ + return nodeIds(ignite.cluster().forServers().nodes()); + } + + /** {@inheritDoc} */ + @Override protected SnapshotListTaskResult reduce0(List<ComputeJobResult> nodesJobsResults) throws IgniteException { + String[] cstIds = new String[nodesJobsResults.size()]; + UUID[] nodesIds = new UUID[nodesJobsResults.size()]; + SnapshotListJobResult[] nodesResults = new SnapshotListJobResult[nodesJobsResults.size()]; + + for (int i = 0; i < nodesJobsResults.size(); i++) { + ComputeJobResult nodeJobRes = nodesJobsResults.get(i); + + if (nodeJobRes.getException() != null) { + throw new IgniteException("Failed to execute snapshot list job on node [uuid=" + nodeJobRes.getNode().id() + ']', Review Comment: I\`d prefer to fix it, but if other places are uses the same approach - it\`s up to you ########## modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotListRollingUpgradeTest.java: ########## @@ -0,0 +1,230 @@ +/* + * 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.nio.file.DirectoryStream; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.UUID; +import org.apache.ignite.IgniteDataStreamer; +import org.apache.ignite.IgniteException; +import org.apache.ignite.cache.CacheAtomicityMode; +import org.apache.ignite.cache.CacheMode; +import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.configuration.CacheConfiguration; +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.snapshot.SnapshotListCommandArg; +import org.apache.ignite.internal.management.snapshot.SnapshotListTask; +import org.apache.ignite.internal.processors.rollingupgrade.AbstractRollingUpgradeTest; +import org.apache.ignite.internal.util.distributed.SingleNodeMessage; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.internal.visor.VisorTaskArgument; +import org.apache.ignite.testframework.GridTestUtils; +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.util.distributed.DistributedProcess.DistributedProcessType.RU_PREPARE_VERSION_FINALIZATION; +import static org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause; +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; + +/** */ +public class IgniteClusterSnapshotListRollingUpgradeTest extends AbstractRollingUpgradeTest { + /** */ + private static final int ALL_GRIDS = 4; + + /** */ + private static final int CLIENTS = 1; + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + // Clean everything, including the separated working directories. + try (DirectoryStream<Path> 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); + + DataRegionConfiguration dataRegionCfg = new DataRegionConfiguration() + .setPersistenceEnabled(true) + .setMaxSize(DataStorageConfiguration.DFLT_DATA_REGION_INITIAL_SIZE); + + cfg.setDataStorageConfiguration(new DataStorageConfiguration().setDefaultDataRegionConfiguration(dataRegionCfg)); + + cfg.setWorkDirectory(new File(U.defaultWorkDirectory(), igniteInstanceName).getAbsolutePath()); + + return cfg; + } + + /** */ + @Test + public void testParallelRollingUpgradeInProgress() throws Exception { + for (int i = 0; i < ALL_GRIDS; i++) + startGrid(i, "2.18.0", i >= ALL_GRIDS - CLIENTS); + + grid(0).cluster().active(true); + + int testNodeIx = ALL_GRIDS - CLIENTS - 1; + + createCacheAndSnapshot(testNodeIx); + + ru(grid(testNodeIx)).enableVersionUpgrade(); + + for (int i = 0; i < ALL_GRIDS; i++) { + assertTrue(ru(grid(i)).isVersionUpgradeEnabled()); + + upgradeNodeVersion(i, "2.18.0", MASTER_VER_REF); + } + + spi(grid(testNodeIx)).blockMessages((node, msg) -> msg instanceof SingleNodeMessage<?> snm && + snm.type() == RU_PREPARE_VERSION_FINALIZATION.ordinal()); + + IgniteInternalFuture<?> finalizeFut = GridTestUtils.runAsync(() -> ru(testNodeIx).finalizeClusterVersion()); + + assertTrue(spi(grid(testNodeIx)).waitForBlocked(1, getTestTimeout())); + + ensureSnapshotListCommandFails(); + + spi(grid(testNodeIx)).stopBlock(); + + assertFalse(spi(grid(testNodeIx)).hasBlockedMessages()); + + finalizeFut.get(getTestTimeout()); + + for (int i = 0; i < ALL_GRIDS; i++) { + int i0 = i; + + assertTrue(waitForCondition(() -> !ru(grid(i0)).isVersionUpgradeEnabled(), getTestTimeout())); + } + + ensureSnapshotListSucceeds(); + } + + /** */ + @Test + public void testSnapshotListFeature() 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); + + ensureSnapshotListCommandFails(); + + ru(grid(0)).enableVersionUpgrade(); + + for (int i = 0; i < ALL_GRIDS; i++) { + assertTrue(ru(grid(i)).isVersionUpgradeEnabled()); + + upgradeNodeVersion(i, "2.18.0", MASTER_VER_REF); + + ensureSnapshotListCommandFails(); + } + + ru(grid(1)).finalizeClusterVersion(); + + for (int i = 0; i < ALL_GRIDS; i++) { + int i0 = i; + + assertTrue(waitForCondition(() -> !ru(grid(i0)).isVersionUpgradeEnabled(), getTestTimeout())); + } + + ensureSnapshotListSucceeds(); + } + + /** */ + private void ensureSnapshotListSucceeds() throws Exception { + for (int i = 0; i < ALL_GRIDS; i++) { + for (int k = 0; k < ALL_GRIDS; k++) { + int i0 = i; + int k0 = k; Review Comment: redundant variables ########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotListJobResult.java: ########## @@ -0,0 +1,140 @@ +/* + * 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.util.Map; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.dto.IgniteDataTransferObject; +import org.jetbrains.annotations.Nullable; + +/** Per-node result of the snapshot list task. Contains information about the snapshots found on the node. */ +public final class SnapshotListJobResult extends IgniteDataTransferObject { + /** Serial version uid. */ + private static final long serialVersionUID = 0L; + + /** Local node's snapshot descriptions. */ + @Order(0) + Map<String, SnapshotInfo> snapshots; + + /** Default constructor for serialization purposes. */ + public SnapshotListJobResult() { + // No-op. + } + + /** */ + public SnapshotListJobResult(Map<String, SnapshotInfo> snapshots) { + this.snapshots = snapshots; + } + + /** @return Snapshot descriptions by snapshot name. */ + public Map<String, SnapshotInfo> snapshots() { + return snapshots; + } + + /** Holds information about a snapshot or its parts: size, creation time, number of incremental parts or external storages. */ + public static class SnapshotInfo extends IgniteDataTransferObject { + /** Serial version uid. */ + private static final long serialVersionUID = 0L; + + /** Total size of a snapshot. Or size of its external storages or incremental parts. */ + @Order(0) + long size; + + /** Creation date of the snapshot or of its incremental parts. */ + @Order(1) + @Nullable Long date; Review Comment: unboxing on constructor and boxing here, why do we need it ? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
