Vladsz83 commented on code in PR #13666: URL: https://github.com/apache/ignite/pull/13666#discussion_r4324576118
########## modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotListTest.java: ########## @@ -0,0 +1,936 @@ +/* + * 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.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 the external storages. */ + private boolean extStorages; + + /** Resolved external storages 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 the nodes (a cache has the single storage paths + // 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 uses the thread blocking and conditional waiting. Won't proceed with 1 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(SnapshotListTask.calculateDirectorySize(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 = SnapshotListTask.calculateDirectorySize(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 = SnapshotListTask.calculateDirectorySize(testSnpFt.root()); + + assertTrue(curSnpSz < testSnpSz); + } + } + + assertTrue(victimNodeFound); + assertEquals(grids, snpsCnt); + } + } + + /** */ + @Test + public void testMissingSnapshotPath() throws Exception { + doTestWrongSnapshotPath(true); + } + + /** */ + @Test + public void testEmptySnapshotPath() throws Exception { + doTestWrongSnapshotPath(false); + } + + /** */ + private void doTestWrongSnapshotPath(boolean missing) throws Exception { + // Doesn't matter here, speeds up the tests. + assumeFalse(snpThrdPoolSz < 2); Review Comment: Why `snpThrdPoolSz = DFLT_SNAPSHOT_THREAD_POOL_SIZE;`? It would be configured. -- 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]
