This is an automated email from the ASF dual-hosted git repository. sergeychugunov pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push: new 52bfe7b IGNITE-13366 Maintenance Mode core functionality: Java API to request MM, Ignite node entering MM - Fixes #8325. 52bfe7b is described below commit 52bfe7bda3507b1172134278c2f823bf3b58088b Author: Sergey Chugunov <sergey.chugu...@gmail.com> AuthorDate: Thu Oct 15 09:35:11 2020 +0300 IGNITE-13366 Maintenance Mode core functionality: Java API to request MM, Ignite node entering MM - Fixes #8325. Signed-off-by: Sergey Chugunov <sergey.chugu...@gmail.com> --- .../persistence/MetaStorageCompatibilityTest.java | 12 +- .../apache/ignite/internal/GridKernalContext.java | 9 + .../ignite/internal/GridKernalContextImpl.java | 14 + .../org/apache/ignite/internal/IgniteKernal.java | 22 +- .../internal/maintenance/MaintenanceFileStore.java | 229 ++++++++++++++ .../internal/maintenance/MaintenanceProcessor.java | 276 +++++++++++++++++ .../managers/discovery/GridDiscoveryManager.java | 11 +- .../pagemem/store/IgnitePageStoreManager.java | 5 - .../processors/cache/ClusterCachesInfo.java | 12 +- .../CleanCacheStoresMaintenanceAction.java | 75 +++++ .../CorruptedPdsMaintenanceCallback.java | 79 +++++ .../GridCacheDatabaseSharedManager.java | 22 ++ .../persistence/file/FilePageStoreManager.java | 85 +++-- .../wal/reader/StandaloneGridKernalContext.java | 6 + .../ignite/maintenance/MaintenanceAction.java | 52 ++++ .../ignite/maintenance/MaintenanceRegistry.java | 157 ++++++++++ .../apache/ignite/maintenance/MaintenanceTask.java | 88 ++++++ .../maintenance/MaintenanceWorkflowCallback.java | 68 ++++ .../apache/ignite/maintenance/package-info.java | 21 ++ .../discovery/isolated/IsolatedDiscoverySpi.java | 256 +++++++++++++++ .../spi/discovery/isolated/IsolatedNode.java | 153 +++++++++ .../ignite/cache/RebalanceCancellationTest.java | 26 +- .../cache/WalModeChangeAdvancedSelfTest.java | 192 +++++++++++- ...ocalWalModeChangeDuringRebalancingSelfTest.java | 244 ++++++++++++--- .../persistence/MaintenanceRegistrySimpleTest.java | 345 +++++++++++++++++++++ .../persistence/pagemem/NoOpPageStoreManager.java | 5 - .../ignite/testsuites/IgnitePdsTestSuite2.java | 5 + 27 files changed, 2367 insertions(+), 102 deletions(-) diff --git a/modules/compatibility/src/test/java/org/apache/ignite/compatibility/persistence/MetaStorageCompatibilityTest.java b/modules/compatibility/src/test/java/org/apache/ignite/compatibility/persistence/MetaStorageCompatibilityTest.java index 803e41d..846864b 100644 --- a/modules/compatibility/src/test/java/org/apache/ignite/compatibility/persistence/MetaStorageCompatibilityTest.java +++ b/modules/compatibility/src/test/java/org/apache/ignite/compatibility/persistence/MetaStorageCompatibilityTest.java @@ -38,9 +38,13 @@ import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.lang.IgniteInClosure; import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi; import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.WithSystemProperty; import org.jetbrains.annotations.Nullable; import org.junit.Test; +import static org.apache.ignite.IgniteSystemProperties.IGNITE_DISABLE_WAL_DURING_REBALANCING; +import static org.apache.ignite.cluster.ClusterState.ACTIVE; + /** * Tests migration of metastorage. */ @@ -87,7 +91,7 @@ public class MetaStorageCompatibilityTest extends IgnitePersistenceCompatibility try (Ignite ig0 = IgnitionEx.start(prepareConfig(getConfiguration(), CONSISTENT_ID_1))) { try (Ignite ig1 = IgnitionEx.start(prepareConfig(getConfiguration(), CONSISTENT_ID_2))) { - assertTrue(GridTestUtils.waitForCondition(() -> ig1.cluster().active(), 10_000)); + assertTrue(GridTestUtils.waitForCondition(() -> ig1.cluster().state() == ACTIVE, 10_000)); } } } @@ -100,6 +104,7 @@ public class MetaStorageCompatibilityTest extends IgnitePersistenceCompatibility * Tests that BLT can be changed and persisted after metastorage migration. */ @Test + @WithSystemProperty(key = IGNITE_DISABLE_WAL_DURING_REBALANCING, value = "false") public void testMigrationToNewBaselineSetNewBaselineAfterMigration() throws Exception { try { U.delete(new File(U.defaultWorkDirectory())); @@ -119,7 +124,7 @@ public class MetaStorageCompatibilityTest extends IgnitePersistenceCompatibility try (Ignite ig0 = IgnitionEx.start(prepareConfig(getConfiguration(), CONSISTENT_ID_1))) { try (Ignite ig1 = IgnitionEx.start(prepareConfig(getConfiguration(), CONSISTENT_ID_2))) { - assertTrue(GridTestUtils.waitForCondition(() -> ig1.cluster().active(), 10_000)); + assertTrue(GridTestUtils.waitForCondition(() -> ig1.cluster().state() == ACTIVE, 10_000)); } } } @@ -132,6 +137,7 @@ public class MetaStorageCompatibilityTest extends IgnitePersistenceCompatibility * */ @Test + @WithSystemProperty(key = IGNITE_DISABLE_WAL_DURING_REBALANCING, value = "false") public void testMigrationWithExceptionDuringTheProcess() throws Exception { try { U.delete(new File(U.defaultWorkDirectory())); @@ -162,7 +168,7 @@ public class MetaStorageCompatibilityTest extends IgnitePersistenceCompatibility try (Ignite ig0 = IgnitionEx.start(prepareConfig(getConfiguration(), CONSISTENT_ID_1))) { try (Ignite ig1 = IgnitionEx.start(prepareConfig(getConfiguration(), CONSISTENT_ID_2))) { - assertTrue(GridTestUtils.waitForCondition(() -> ig1.cluster().active(), 10_000)); + assertTrue(GridTestUtils.waitForCondition(() -> ig1.cluster().state() == ACTIVE, 10_000)); } } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java index 1d65155..56f9765 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContext.java @@ -21,6 +21,7 @@ import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.ExecutorService; + import org.apache.ignite.IgniteLogger; import org.apache.ignite.configuration.IgniteConfiguration; import org.apache.ignite.internal.managers.checkpoint.GridCheckpointManager; @@ -78,6 +79,7 @@ import org.apache.ignite.internal.util.IgniteExceptionRegistry; import org.apache.ignite.internal.util.StripedExecutor; import org.apache.ignite.internal.util.tostring.GridToStringExclude; import org.apache.ignite.internal.worker.WorkersRegistry; +import org.apache.ignite.maintenance.MaintenanceRegistry; import org.apache.ignite.plugin.PluginNotFoundException; import org.apache.ignite.plugin.PluginProvider; import org.apache.ignite.thread.IgniteStripedThreadPoolExecutor; @@ -201,6 +203,13 @@ public interface GridKernalContext extends Iterable<GridComponent> { public GridMetricManager metric(); /** + * Gets maintenance registry. + * + * @return Maintenance registry. + */ + public MaintenanceRegistry maintenanceRegistry(); + + /** * Gets system view manager. * * @return Monitoring manager. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java index fd46dfa..ae589ad 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/GridKernalContextImpl.java @@ -31,6 +31,7 @@ import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.ExecutorService; + import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; import org.apache.ignite.IgniteLogger; @@ -38,6 +39,7 @@ import org.apache.ignite.IgniteSystemProperties; import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.configuration.IgniteConfiguration; import org.apache.ignite.failure.FailureType; +import org.apache.ignite.internal.maintenance.MaintenanceProcessor; import org.apache.ignite.internal.managers.checkpoint.GridCheckpointManager; import org.apache.ignite.internal.managers.collision.GridCollisionManager; import org.apache.ignite.internal.managers.communication.GridIoManager; @@ -104,6 +106,7 @@ import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.internal.worker.WorkersRegistry; import org.apache.ignite.lang.IgnitePredicate; +import org.apache.ignite.maintenance.MaintenanceRegistry; import org.apache.ignite.plugin.PluginNotFoundException; import org.apache.ignite.plugin.PluginProvider; import org.apache.ignite.thread.IgniteStripedThreadPoolExecutor; @@ -322,6 +325,10 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable /** */ @GridToStringExclude + private MaintenanceProcessor maintenanceProc; + + /** */ + @GridToStringExclude private List<GridComponent> comps = new LinkedList<>(); /** */ @@ -695,6 +702,8 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable diagnosticProcessor = (DiagnosticProcessor)comp; else if (comp instanceof DurableBackgroundTasksProcessor) durableBackgroundTasksProcessor = (DurableBackgroundTasksProcessor)comp; + else if (comp instanceof MaintenanceProcessor) + maintenanceProc = (MaintenanceProcessor) comp; else if (!(comp instanceof DiscoveryNodeValidationProcessor || comp instanceof PlatformPluginProcessor)) assert (comp instanceof GridPluginComponent) : "Unknown manager class: " + comp.getClass(); @@ -788,6 +797,11 @@ public class GridKernalContextImpl implements GridKernalContext, Externalizable } /** {@inheritDoc} */ + @Override public MaintenanceRegistry maintenanceRegistry() { + return maintenanceProc; + } + + /** {@inheritDoc} */ @Override public GridCacheProcessor cache() { return cacheProc; } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java index aaeb8cc..50c1a55 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java @@ -111,6 +111,7 @@ import org.apache.ignite.internal.binary.BinaryMarshaller; import org.apache.ignite.internal.binary.BinaryUtils; import org.apache.ignite.internal.cluster.ClusterGroupAdapter; import org.apache.ignite.internal.cluster.IgniteClusterEx; +import org.apache.ignite.internal.maintenance.MaintenanceProcessor; import org.apache.ignite.internal.managers.GridManager; import org.apache.ignite.internal.managers.IgniteMBeansManager; import org.apache.ignite.internal.managers.checkpoint.GridCheckpointManager; @@ -228,6 +229,7 @@ import org.apache.ignite.plugin.PluginProvider; import org.apache.ignite.spi.IgniteSpi; import org.apache.ignite.spi.IgniteSpiVersionCheckException; import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi; +import org.apache.ignite.spi.discovery.isolated.IsolatedDiscoverySpi; import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode; import org.apache.ignite.spi.tracing.TracingConfigurationManager; import org.apache.ignite.thread.IgniteStripedThreadPoolExecutor; @@ -1193,7 +1195,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable { // Assign discovery manager to context before other processors start so they // are able to register custom event listener. - final GridManager discoMgr = new GridDiscoveryManager(ctx); + GridManager discoMgr = new GridDiscoveryManager(ctx); ctx.add(discoMgr, false); @@ -1201,12 +1203,25 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable { // able to register custom event listener. startManager(new GridEncryptionManager(ctx)); + startProcessor(new PdsConsistentIdProcessor(ctx)); + + MaintenanceProcessor mntcProcessor = new MaintenanceProcessor(ctx); + + startProcessor(mntcProcessor); + + if (mntcProcessor.isMaintenanceMode()) { + ctx.config().setDiscoverySpi(new IsolatedDiscoverySpi()); + + discoMgr = new GridDiscoveryManager(ctx); + + ctx.add(discoMgr, false); + } + // Start processors before discovery manager, so they will // be able to start receiving messages once discovery completes. try { startProcessor(COMPRESSION.createOptional(ctx)); startProcessor(new GridMarshallerMappingProcessor(ctx)); - startProcessor(new PdsConsistentIdProcessor(ctx)); startProcessor(new MvccProcessorImpl(ctx)); startProcessor(createComponent(DiscoveryNodeValidationProcessor.class, ctx)); startProcessor(new GridAffinityProcessor(ctx)); @@ -1280,6 +1295,9 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable { throw e; } + // All components exept Discovery are started, time to check if maintenance is still needed + mntcProcessor.prepareAndExecuteMaintenance(); + gw.writeLock(); try { diff --git a/modules/core/src/main/java/org/apache/ignite/internal/maintenance/MaintenanceFileStore.java b/modules/core/src/main/java/org/apache/ignite/internal/maintenance/MaintenanceFileStore.java new file mode 100644 index 0000000..02698a1 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/maintenance/MaintenanceFileStore.java @@ -0,0 +1,229 @@ +/* + * 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.maintenance; + +import java.io.File; +import java.io.IOException; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIO; +import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory; +import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderSettings; +import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFoldersResolver; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.maintenance.MaintenanceTask; + +/** + * Provides API for durable storage of {@link MaintenanceTask}s and hides implementation details from higher levels. + * + * Human-readable storage format is rigid but simple. + * <ol> + * <li> + * Maintenance file with tasks is stored in work directory of node + * under persistent store root defined by consistentId of node. + * </li> + * <li> + * Each task is written to disk as a {@link String} on a separate line. + * </li> + * <li> + * Task consists of two or three parts: task UUID, task description and optional parameters. + * </li> + * </ol> + */ +public class MaintenanceFileStore { + /** */ + private static final String MAINTENANCE_FILE_NAME = "maintenance_tasks.mntc"; + + /** */ + private static final String TASKS_SEPARATOR = System.lineSeparator(); + + /** */ + private static final String TASK_PARTS_SEPARATOR = "\t"; + + /** Maintenance task consists of two or three parts: ID, description (user-readable part) + * and optional task parameters. */ + private static final int MAX_MNTC_TASK_PARTS_COUNT = 3; + + /** */ + private final boolean inMemoryMode; + + /** */ + private final PdsFoldersResolver pdsFoldersResolver; + + /** */ + private volatile File mntcTasksFile; + + /** */ + private volatile FileIO mntcTasksFileIO; + + /** */ + private final FileIOFactory ioFactory; + + /** */ + private final Map<String, MaintenanceTask> tasksInSync = new ConcurrentHashMap<>(); + + /** */ + private final IgniteLogger log; + + /** */ + public MaintenanceFileStore(boolean inMemoryMode, + PdsFoldersResolver pdsFoldersResolver, + FileIOFactory ioFactory, + IgniteLogger log) { + this.inMemoryMode = inMemoryMode; + this.pdsFoldersResolver = pdsFoldersResolver; + this.ioFactory = ioFactory; + this.log = log; + } + + /** */ + public void init() throws IgniteCheckedException, IOException { + if (inMemoryMode) + return; + + PdsFolderSettings folderSettings = pdsFoldersResolver.resolveFolders(); + File storeDir = new File(folderSettings.persistentStoreRootPath(), folderSettings.folderName()); + U.ensureDirectory(storeDir, "store directory for node persistent data", log); + + mntcTasksFile = new File(storeDir, MAINTENANCE_FILE_NAME); + + if (!mntcTasksFile.exists()) + mntcTasksFile.createNewFile(); + + mntcTasksFileIO = ioFactory.create(mntcTasksFile); + + readTasksFromFile(); + } + + /** + * Deletes file with maintenance tasks. + */ + public void clear() { + if (mntcTasksFile != null) + mntcTasksFile.delete(); + } + + /** + * Stops + */ + public void stop() throws IOException { + if (inMemoryMode) + return; + + if (mntcTasksFileIO != null) + mntcTasksFileIO.close(); + } + + /** */ + private void readTasksFromFile() throws IOException { + int len = (int) mntcTasksFileIO.size(); + + if (len == 0) + return; + + byte[] allBytes = new byte[len]; + + mntcTasksFileIO.read(allBytes, 0, len); + + String[] allTasks = new String(allBytes).split(TASKS_SEPARATOR); + + for (String taskStr : allTasks) { + String[] subStrs = taskStr.split(TASK_PARTS_SEPARATOR); + + int partsNum = subStrs.length; + + if (partsNum < MAX_MNTC_TASK_PARTS_COUNT - 1) { + log.info("Corrupted maintenance task found and will be skipped, " + + "mandatory parts are missing: " + taskStr); + + continue; + } + + if (partsNum > MAX_MNTC_TASK_PARTS_COUNT) { + log.info("Corrupted maintenance task found and will be skipped, " + + "too many parts in task: " + taskStr); + + continue; + } + + String name = subStrs[0]; + + MaintenanceTask task = new MaintenanceTask(name, subStrs[1], partsNum == 3 ? subStrs[2] : null); + + tasksInSync.put(name, task); + } + } + + /** */ + private void writeTasksToFile() throws IOException { + mntcTasksFileIO.clear(); + + String allTasks = tasksInSync.values().stream() + .map( + task -> task.name() + + TASK_PARTS_SEPARATOR + + task.description() + + TASK_PARTS_SEPARATOR + + (task.parameters() != null ? task.parameters() : "") + ) + .collect(Collectors.joining(System.lineSeparator())); + + byte[] allTasksBytes = allTasks.getBytes(); + + int left = allTasksBytes.length; + int len = allTasksBytes.length; + + while ((left -= mntcTasksFileIO.writeFully(allTasksBytes, len - left, left)) > 0) + ; + + mntcTasksFileIO.force(); + } + + /** */ + public Map<String, MaintenanceTask> getAllTasks() { + if (inMemoryMode) + return null; + + return Collections.unmodifiableMap(tasksInSync); + } + + /** */ + public void writeMaintenanceTask(MaintenanceTask task) throws IOException { + if (inMemoryMode) + return; + + tasksInSync.put(task.name(), task); + + writeTasksToFile(); + } + + /** */ + public void deleteMaintenanceTask(String taskName) throws IOException { + if (inMemoryMode) + return; + + tasksInSync.remove(taskName); + + writeTasksToFile(); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/maintenance/MaintenanceProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/maintenance/MaintenanceProcessor.java new file mode 100644 index 0000000..8f85ceb --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/maintenance/MaintenanceProcessor.java @@ -0,0 +1,276 @@ +/* + * 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.maintenance; + +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteException; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.processors.GridProcessorAdapter; +import org.apache.ignite.internal.util.typedef.internal.CU; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.maintenance.MaintenanceAction; +import org.apache.ignite.maintenance.MaintenanceRegistry; +import org.apache.ignite.maintenance.MaintenanceTask; +import org.apache.ignite.maintenance.MaintenanceWorkflowCallback; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +/** */ +public class MaintenanceProcessor extends GridProcessorAdapter implements MaintenanceRegistry { + /** */ + private static final String IN_MEMORY_MODE_ERR_MSG = "Maintenance Mode is not supported for in-memory clusters"; + + /** + * Active {@link MaintenanceTask}s are the ones that were read from disk when node entered Maintenance Mode. + */ + private final Map<String, MaintenanceTask> activeTasks = new ConcurrentHashMap<>(); + + /** + * Requested {@link MaintenanceTask}s are collection of tasks requested by user + * or other components when node operates normally (not in Maintenance Mode). + */ + private final Map<String, MaintenanceTask> requestedTasks = new ConcurrentHashMap<>(); + + /** */ + private final Map<String, MaintenanceWorkflowCallback> workflowCallbacks = new ConcurrentHashMap<>(); + + /** */ + private final MaintenanceFileStore fileStorage; + + /** */ + private final boolean inMemoryMode; + + /** + * @param ctx Kernal context. + */ + public MaintenanceProcessor(GridKernalContext ctx) { + super(ctx); + + inMemoryMode = !CU.isPersistenceEnabled(ctx.config()); + + if (inMemoryMode) { + fileStorage = new MaintenanceFileStore(true, + null, + null, + null); + + return; + } + + fileStorage = new MaintenanceFileStore(false, + ctx.pdsFolderResolver(), + ctx.config().getDataStorageConfiguration().getFileIOFactory(), + log); + } + + /** {@inheritDoc} */ + @Override public @Nullable MaintenanceTask registerMaintenanceTask(MaintenanceTask task) throws IgniteCheckedException { + if (inMemoryMode) + throw new IgniteCheckedException(IN_MEMORY_MODE_ERR_MSG); + + if (isMaintenanceMode()) + throw new IgniteCheckedException("Node is already in Maintenance Mode, " + + "registering additional maintenance task is not allowed in Maintenance Mode."); + + MaintenanceTask oldTask = requestedTasks.put(task.name(), task); + + if (oldTask != null) { + log.info( + "Maintenance Task with name " + task.name() + + " is already registered" + + oldTask.parameters() != null ? " with parameters " + oldTask.parameters() : "" + "." + + " It will be replaced with new task" + + task.parameters() != null ? " with parameters " + task.parameters() : "" + "." + ); + } + + try { + fileStorage.writeMaintenanceTask(task); + } + catch (IOException e) { + throw new IgniteCheckedException("Failed to register maintenance task " + task, e); + } + + return oldTask; + } + + /** {@inheritDoc} */ + @Override public void stop(boolean cancel) throws IgniteCheckedException { + try { + fileStorage.stop(); + } + catch (IOException e) { + log.warning("Failed to free maintenance file resources", e); + } + } + + /** {@inheritDoc} */ + @Override public void start() throws IgniteCheckedException { + if (inMemoryMode) + return; + + try { + fileStorage.init(); + + activeTasks.putAll(fileStorage.getAllTasks()); + } + catch (Throwable t) { + log.warning("Caught exception when starting MaintenanceProcessor," + + " maintenance mode won't be entered", t); + + activeTasks.clear(); + + fileStorage.clear(); + } + } + + /** {@inheritDoc} */ + @Override public void prepareAndExecuteMaintenance() { + if (isMaintenanceMode()) { + workflowCallbacks.entrySet().removeIf(cbE -> + { + if (!cbE.getValue().shouldProceedWithMaintenance()) { + unregisterMaintenanceTask(cbE.getKey()); + + return true; + } + + return false; + } + ); + } + + if (!workflowCallbacks.isEmpty()) + proceedWithMaintenance(); + else { + if (log.isInfoEnabled()) + log.info("All maintenance tasks are fixed, no need to enter maintenance mode. " + + "Restart the node to get it back to normal operations."); + } + } + + /** + * Handles all {@link MaintenanceTask maintenance tasks} left + * after {@link MaintenanceRegistry#prepareAndExecuteMaintenance()} check. + * + * If a task defines an action that should be started automatically (e.g. defragmentation starts automatically, + * no additional confirmation from user is required), it is executed. + * + * Otherwise waits for user to trigger actions for maintenance tasks. + */ + private void proceedWithMaintenance() { + for (Map.Entry<String, MaintenanceWorkflowCallback> cbE : workflowCallbacks.entrySet()) { + MaintenanceAction mntcAction = cbE.getValue().automaticAction(); + + if (mntcAction != null) { + try { + mntcAction.execute(); + } + catch (Throwable t) { + log.warning("Failed to execute automatic action for maintenance task: " + + activeTasks.get(cbE.getKey()), t); + + throw t; + } + } + } + } + + /** {@inheritDoc} */ + @Override public @Nullable MaintenanceTask activeMaintenanceTask(String maitenanceTaskName) { + return activeTasks.get(maitenanceTaskName); + } + + /** {@inheritDoc} */ + @Override public boolean isMaintenanceMode() { + return !activeTasks.isEmpty(); + } + + /** {@inheritDoc} */ + @Override public void unregisterMaintenanceTask(String maintenanceTaskName) { + if (inMemoryMode) + return; + + if (isMaintenanceMode()) + activeTasks.remove(maintenanceTaskName); + else + requestedTasks.remove(maintenanceTaskName); + + try { + fileStorage.deleteMaintenanceTask(maintenanceTaskName); + } + catch (IOException e) { + log.warning("Failed to clear maintenance task with name " + + maintenanceTaskName + + " from file, whole file will be deleted", e + ); + + fileStorage.clear(); + } + } + + /** {@inheritDoc} */ + @Override public void registerWorkflowCallback(@NotNull String maintenanceTaskName, @NotNull MaintenanceWorkflowCallback cb) { + if (inMemoryMode) + throw new IgniteException(IN_MEMORY_MODE_ERR_MSG); + + List<MaintenanceAction> actions = cb.allActions(); + + if (actions == null || actions.isEmpty()) + throw new IgniteException("Maintenance workflow callback should provide at least one mainetance action"); + + int size = actions.size(); + long distinctSize = actions.stream().map(a -> a.name()).distinct().count(); + + if (distinctSize < size) + throw new IgniteException("All actions of a single workflow should have unique names: " + + actions.stream().map(a -> a.name()).collect(Collectors.joining(", "))); + + Optional<String> wrongActionName = actions + .stream() + .map(MaintenanceAction::name) + .filter(name -> !U.alphanumericUnderscore(name)) + .findFirst(); + + if (wrongActionName.isPresent()) + throw new IgniteException( + "All actions' names should contain only alphanumeric and underscore symbols: " + + wrongActionName.get()); + + workflowCallbacks.put(maintenanceTaskName, cb); + } + + /** {@inheritDoc} */ + @Override public List<MaintenanceAction> actionsForMaintenanceTask(String maintenanceTaskName) { + if (inMemoryMode) + throw new IgniteException(IN_MEMORY_MODE_ERR_MSG); + + if (!activeTasks.containsKey(maintenanceTaskName)) + throw new IgniteException("Maintenance workflow callback for given task name not found, " + + "cannot retrieve maintenance actions for it: " + maintenanceTaskName); + + return workflowCallbacks.get(maintenanceTaskName).allActions(); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java index e791564..0a8c857 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java @@ -37,6 +37,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; + import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteClientDisconnectedException; import org.apache.ignite.IgniteException; @@ -292,10 +293,12 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> { public GridDiscoveryManager(GridKernalContext ctx) { super(ctx, ctx.config().getDiscoverySpi()); - ctx.systemView().registerView(NODES_SYS_VIEW, NODES_SYS_VIEW_DESC, - new ClusterNodeViewWalker(), - () -> F.concat(false, allNodes(), daemonNodes()), - ClusterNodeView::new); + if (ctx.systemView().view(NODES_SYS_VIEW) == null) { + ctx.systemView().registerView(NODES_SYS_VIEW, NODES_SYS_VIEW_DESC, + new ClusterNodeViewWalker(), + () -> F.concat(false, allNodes(), daemonNodes()), + ClusterNodeView::new); + } } /** {@inheritDoc} */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java index 243d99d..8216b96 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/pagemem/store/IgnitePageStoreManager.java @@ -224,11 +224,6 @@ public interface IgnitePageStoreManager extends GridCacheSharedManager, IgniteCh public boolean hasIndexStore(int grpId); /** - * @param grpDesc Cache group descriptor. - */ - public void beforeCacheGroupStart(CacheGroupDescriptor grpDesc); - - /** * Calculates number of pages currently allocated for given cache group. * * @param grpId cache group id. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java index f773f4d..88a9fde 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ClusterCachesInfo.java @@ -2222,6 +2222,13 @@ public class ClusterCachesInfo { Map<String, Integer> caches = Collections.singletonMap(startedCacheCfg.getName(), cacheId); boolean persistent = resolvePersistentFlag(exchActions, startedCacheCfg); + boolean walGloballyEnabled = false; + + // client nodes cannot read wal enabled/disabled status so they should use default one + if (ctx.clientNode()) + walGloballyEnabled = persistent; + else if (persistent) + walGloballyEnabled = ctx.cache().context().database().walEnabled(grpId, false); CacheGroupDescriptor grpDesc = new CacheGroupDescriptor( startedCacheCfg, @@ -2232,7 +2239,7 @@ public class ClusterCachesInfo { deploymentId, caches, persistent, - persistent, + walGloballyEnabled, null, cacheCfgEnrichment ); @@ -2240,9 +2247,6 @@ public class ClusterCachesInfo { if (startedCacheCfg.isEncryptionEnabled()) ctx.encryption().beforeCacheGroupStart(grpId, encKey); - if (ctx.cache().context().pageStore() != null) - ctx.cache().context().pageStore().beforeCacheGroupStart(grpDesc); - CacheGroupDescriptor old = registeredCacheGrps.put(grpId, grpDesc); assert old == null : old; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CleanCacheStoresMaintenanceAction.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CleanCacheStoresMaintenanceAction.java new file mode 100644 index 0000000..de94e6e --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CleanCacheStoresMaintenanceAction.java @@ -0,0 +1,75 @@ +/* + * 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; + +import java.io.File; + +import org.apache.ignite.maintenance.MaintenanceAction; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DATA_FILENAME; + +/** + * + */ +public class CleanCacheStoresMaintenanceAction implements MaintenanceAction<Void> { + /** */ + public static final String ACTION_NAME = "clean_data_files"; + + /** */ + private final File rootStoreDir; + + /** */ + private final String[] cacheStoreDirs; + + /** + * @param rootStoreDir + * @param cacheStoreDirs + */ + public CleanCacheStoresMaintenanceAction(File rootStoreDir, String[] cacheStoreDirs) { + this.rootStoreDir = rootStoreDir; + this.cacheStoreDirs = cacheStoreDirs; + } + + /** {@inheritDoc} */ + @Override public Void execute() { + for (String cacheStoreDirName : cacheStoreDirs) { + File cacheStoreDir = new File(rootStoreDir, cacheStoreDirName); + + if (cacheStoreDir.exists() && cacheStoreDir.isDirectory()) { + for (File file : cacheStoreDir.listFiles()) { + if (!file.getName().equals(CACHE_DATA_FILENAME)) + file.delete(); + } + } + } + + return null; + } + + /** {@inheritDoc} */ + @Override public @NotNull String name() { + return ACTION_NAME; + } + + /** {@inheritDoc} */ + @Override public @Nullable String description() { + return "Cleans data files of cache groups"; + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CorruptedPdsMaintenanceCallback.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CorruptedPdsMaintenanceCallback.java new file mode 100644 index 0000000..52a8f6f --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/CorruptedPdsMaintenanceCallback.java @@ -0,0 +1,79 @@ +/* + * 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; + +import java.io.File; +import java.util.Arrays; +import java.util.List; + +import org.apache.ignite.maintenance.MaintenanceAction; +import org.apache.ignite.maintenance.MaintenanceWorkflowCallback; +import org.jetbrains.annotations.NotNull; + +import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DATA_FILENAME; + +/** + * + */ +public class CorruptedPdsMaintenanceCallback implements MaintenanceWorkflowCallback { + /** */ + private final File workDir; + + /** */ + private final List<String> cacheStoreDirs; + + /** + * @param workDir + * @param cacheStoreDirs + */ + public CorruptedPdsMaintenanceCallback(@NotNull File workDir, + @NotNull List<String> cacheStoreDirs) + { + this.workDir = workDir; + this.cacheStoreDirs = cacheStoreDirs; + } + + /** {@inheritDoc} */ + @Override public boolean shouldProceedWithMaintenance() { + for (String cacheStoreDirName : cacheStoreDirs) { + File cacheStoreDir = new File(workDir, cacheStoreDirName); + + if (cacheStoreDir.exists() + && cacheStoreDir.isDirectory() + && cacheStoreDir.listFiles().length > 0 + ) { + for (File f : cacheStoreDir.listFiles()) { + if (!f.getName().equals(CACHE_DATA_FILENAME)) + return true; + } + } + } + + return false; + } + + /** {@inheritDoc} */ + @Override public List<MaintenanceAction> allActions() { + return Arrays.asList(new CleanCacheStoresMaintenanceAction(workDir, cacheStoreDirs.toArray(new String[0]))); + } + + /** {@inheritDoc} */ + @Override public MaintenanceAction automaticAction() { + return null; + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java index 2214c62..50224f3 100755 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java @@ -29,6 +29,7 @@ import java.nio.channels.OverlappingFileLockException; import java.nio.file.Path; import java.nio.file.Paths; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.Comparator; @@ -50,6 +51,7 @@ import java.util.function.Consumer; import java.util.function.Predicate; import java.util.function.ToLongFunction; import java.util.stream.Collectors; + import org.apache.ignite.DataRegionMetricsProvider; import org.apache.ignite.DataStorageMetrics; import org.apache.ignite.IgniteCheckedException; @@ -147,6 +149,8 @@ import org.apache.ignite.lang.IgniteBiTuple; import org.apache.ignite.lang.IgniteInClosure; import org.apache.ignite.lang.IgniteOutClosure; import org.apache.ignite.lang.IgnitePredicate; +import org.apache.ignite.maintenance.MaintenanceRegistry; +import org.apache.ignite.maintenance.MaintenanceTask; import org.apache.ignite.mxbean.DataStorageMetricsMXBean; import org.apache.ignite.spi.systemview.view.MetastorageView; import org.apache.ignite.transactions.TransactionState; @@ -170,6 +174,7 @@ import static org.apache.ignite.internal.processors.cache.distributed.dht.topolo import static org.apache.ignite.internal.processors.cache.persistence.CheckpointState.FINISHED; import static org.apache.ignite.internal.processors.cache.persistence.CheckpointState.LOCK_RELEASED; import static org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointReadWriteLock.CHECKPOINT_LOCK_HOLD_COUNT; +import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CORRUPTED_DATA_FILES_MNTC_TASK_NAME; import static org.apache.ignite.internal.util.IgniteUtils.checkpointBufferSize; /** @@ -1574,6 +1579,23 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan if (kctx.clientNode()) return; + MaintenanceRegistry mntcRegistry = kctx.maintenanceRegistry(); + + MaintenanceTask mntcTask = mntcRegistry.activeMaintenanceTask(CORRUPTED_DATA_FILES_MNTC_TASK_NAME); + + if (mntcTask != null) { + log.warning("Maintenance task found, stop restoring memory"); + + File workDir = ((FilePageStoreManager) cctx.pageStore()).workDir(); + + mntcRegistry.registerWorkflowCallback(CORRUPTED_DATA_FILES_MNTC_TASK_NAME, + new CorruptedPdsMaintenanceCallback(workDir, + Arrays.asList(mntcTask.parameters().split(File.separator))) + ); + + return; + } + checkpointReadLock(); try { diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java index 7f6269d..024f52c 100755 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/file/FilePageStoreManager.java @@ -33,6 +33,7 @@ import java.nio.file.Path; import java.nio.file.PathMatcher; import java.nio.file.StandardCopyOption; import java.util.AbstractList; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; @@ -50,6 +51,7 @@ import java.util.function.BiFunction; import java.util.function.Function; import java.util.function.Predicate; import java.util.stream.Collectors; + import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; import org.apache.ignite.IgniteLogger; @@ -80,12 +82,12 @@ import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteCa import org.apache.ignite.internal.processors.cache.persistence.tree.io.PageIO; import org.apache.ignite.internal.processors.metric.impl.LongAdderMetric; import org.apache.ignite.internal.util.GridStripedReadWriteLock; -import org.apache.ignite.internal.util.IgniteUtils; import org.apache.ignite.internal.util.typedef.X; import org.apache.ignite.internal.util.typedef.internal.CU; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.internal.util.worker.GridWorker; import org.apache.ignite.lang.IgniteOutClosure; +import org.apache.ignite.maintenance.MaintenanceTask; import org.apache.ignite.marshaller.Marshaller; import org.apache.ignite.marshaller.MarshallerUtils; import org.apache.ignite.thread.IgniteThread; @@ -143,6 +145,9 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen public static final PathMatcher TMP_FILE_MATCHER = FileSystems.getDefault().getPathMatcher("glob:**" + TMP_SUFFIX); + /** Unique name for corrupted data files maintenance task. */ + public static final String CORRUPTED_DATA_FILES_MNTC_TASK_NAME = "corrupted-cache-data-files-task"; + /** Listeners of configuration changes e.g. overwrite or remove actions. */ private final List<BiConsumer<String, File>> lsnrs = new CopyOnWriteArrayList<>(); @@ -358,6 +363,31 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen /** {@inheritDoc} */ @Override public void beginRecover() { + List<String> groupsWithWalDisabled = checkCachesWithDisabledWal(); + + if (!groupsWithWalDisabled.isEmpty()) { + String errorMsg = "Cache groups with potentially corrupted partition files found. " + + "To cleanup them maintenance is needed, node will enter maintenance mode on next restart. " + + "Cleanup cache group folders manually or trigger maintenance action to do that and restart the node. " + + "Corrupted files are located in subdirectories " + groupsWithWalDisabled + + " in a work dir " + storeWorkDir; + + log.warning(errorMsg); + + try { + cctx.kernalContext().maintenanceRegistry() + .registerMaintenanceTask( + new MaintenanceTask(CORRUPTED_DATA_FILES_MNTC_TASK_NAME, + "Corrupted cache groups found", + groupsWithWalDisabled.stream().collect(Collectors.joining(File.separator))) + ); + } catch (IgniteCheckedException e) { + log.warning("Failed to register maintenance record for corrupted partition files.", e); + } + + throw new IgniteException(errorMsg); + } + for (CacheStoreHolder holder : idxCacheStores.values()) { holder.idxStore.beginRecover(); @@ -366,6 +396,38 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen } } + /** + * Checks cache groups' settings and returns groups names with disabled WAL. + * + * @return List of cache groups names that had WAL disabled before node stop. + */ + private List<String> checkCachesWithDisabledWal() { + List<String> corruptedCachesDirs = new ArrayList<>(); + + for (Integer grpDescId : idxCacheStores.keySet()) { + CacheGroupDescriptor desc = cctx.cache().cacheGroupDescriptor(grpDescId); + + if (desc != null && desc.persistenceEnabled()) { + boolean localEnabled = cctx.database().walEnabled(grpDescId, true); + boolean globalEnabled = cctx.database().walEnabled(grpDescId, false); + + if (!localEnabled || !globalEnabled) { + File dir = cacheWorkDir(desc.config()); + + if (Arrays.stream( + dir.listFiles()) + .filter(f -> !f.getName().equals(CACHE_DATA_FILENAME)) + .count() > 0) + { + corruptedCachesDirs.add(cacheDirName(desc.config())); + } + } + } + } + + return corruptedCachesDirs; + } + /** {@inheritDoc} */ @Override public void finishRecover() throws IgniteCheckedException { try { @@ -1313,27 +1375,6 @@ public class FilePageStoreManager extends GridCacheSharedManagerAdapter implemen return store; } - /** {@inheritDoc} */ - @Override public void beforeCacheGroupStart(CacheGroupDescriptor grpDesc) { - if (grpDesc.persistenceEnabled()) { - boolean localEnabled = cctx.database().walEnabled(grpDesc.groupId(), true); - boolean globalEnabled = cctx.database().walEnabled(grpDesc.groupId(), false); - - if (!localEnabled || !globalEnabled) { - File dir = cacheWorkDir(grpDesc.config()); - - assert dir.exists(); - - boolean res = IgniteUtils.delete(dir); - - assert res; - - if (!globalEnabled) - grpDesc.walEnabled(false); - } - } - } - /** * @param pageStoreFileIoFactory File IO factory to override default, may be used for blocked read-write. * @param pageStoreV1FileIoFactory File IO factory for reading V1 page store and for fast touching page files diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java index e1cded2..d5a9ed0 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/wal/reader/StandaloneGridKernalContext.java @@ -96,6 +96,7 @@ import org.apache.ignite.internal.util.IgniteExceptionRegistry; import org.apache.ignite.internal.util.StripedExecutor; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.internal.worker.WorkersRegistry; +import org.apache.ignite.maintenance.MaintenanceRegistry; import org.apache.ignite.marshaller.Marshaller; import org.apache.ignite.plugin.PluginNotFoundException; import org.apache.ignite.plugin.PluginProvider; @@ -365,6 +366,11 @@ public class StandaloneGridKernalContext implements GridKernalContext { } /** {@inheritDoc} */ + @Override public MaintenanceRegistry maintenanceRegistry() { + return null; + } + + /** {@inheritDoc} */ @Override public IgniteRestProcessor rest() { return null; } diff --git a/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceAction.java b/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceAction.java new file mode 100644 index 0000000..381c660 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceAction.java @@ -0,0 +1,52 @@ +/* + * 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.maintenance; + +import org.apache.ignite.lang.IgniteExperimental; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +/** + * Maintenance action interface enables user to execute operations related to a particular {@link MaintenanceTask}. + * + * These operations may resolve maintenance situation (e.g. remove corrupted data files), get information + * about other ongoing maintenance action (e.g. if some action requires a lot of time and user wants to know + * current progress of the action) or cancel other ongoing action. + * + * List of maintenance actions available for each task is defined by {@link MaintenanceWorkflowCallback}. + * + * {@link MaintenanceRegistry} provides an access to maintenance actions for a {@link MaintenanceTask} with + * call {@link MaintenanceRegistry#actionsForMaintenanceTask(String)} + * + */ +@IgniteExperimental +public interface MaintenanceAction<T> { + /** Executes operations of current maintenance action and returns results. */ + public T execute(); + + /** + * Mandatory human-readable name of maintenance action. + * All actions of single {@link MaintenanceWorkflowCallback} should have unique names. + */ + @NotNull public String name(); + + /** + * Optional user-readable description of maintenance action. + */ + @Nullable public String description(); +} diff --git a/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceRegistry.java b/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceRegistry.java new file mode 100644 index 0000000..b264700 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceRegistry.java @@ -0,0 +1,157 @@ +/* + * 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.maintenance; + +import java.util.List; + +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteException; +import org.apache.ignite.lang.IgniteExperimental; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +/** + * {@link MaintenanceRegistry} is a service local to each Ignite node + * that allows to request performing maintenance actions on that particular node. + * + * <p> + * When a node gets into a situation when some specific actions are required + * it enters the special mode called maintenance mode. + * In maintenance mode it doesn't join to the rest of the cluster but still allows to connect to it + * with control.{sh|bat} script or via JXM interface and perform needed actions. + * </p> + * + * <p> + * Implementing new task for maintenance mode requires several pieces of code. + * + * <ul> + * <li> + * First, component requiring Maintenance Mode should be able to register new {@link MaintenanceTask} + * with {@link MaintenanceRegistry#registerMaintenanceTask(MaintenanceTask)} method. + * + * Registration could happen automatically (e.g. if component detects some emergency situation + * that requires user intervention) + * or by user request (e.g. for a planned maintenance that requires + * detaching node from the rest of the cluster). + * </li> + * <li> + * Component responsible for handling this {@link MaintenanceTask} + * on startup checks if the task is registered (thus it should go to Maintenance Mode). + * If task is found component provides to {@link MaintenanceRegistry} its own implementation + * of {@link MaintenanceWorkflowCallback} interface + * via method {@link MaintenanceRegistry#registerWorkflowCallback(String, MaintenanceWorkflowCallback)}. + * </li> + * <li> + * {@link MaintenanceWorkflowCallback} should provide {@link MaintenanceRegistry} with + * {@link MaintenanceAction}s that are able to resolve maintenance task, + * get information about it and so on. + * Logic of these actions is completely up to the component providing it + * and depends only on particular maintenance task. + * </li> + * <li> + * When maintenance task is fixed, it should be removed from {@link MaintenanceRegistry} + * with call {@link MaintenanceRegistry#unregisterMaintenanceTask(String)}. + * </li> + * </ul> + * </p> + */ +@IgniteExperimental +public interface MaintenanceRegistry { + /** + * @return {@code True} if any maintenance task was found. + */ + public boolean isMaintenanceMode(); + + /** + * Method to register {@link MaintenanceTask} locally on the node where method is called. + * <p> + * For now it is not allowed to register new Maintenance Tasks in Maintenance Mode + * so this method should be called only when node operates normally. + * This may change in the future so it will become possible to create other maintenance tasks + * on node that is already entered Maintenance Mode. + * </p> + * <p> + * When task is registered node continues to operate normally + * and will enter Maintenance Mode only after restart. + * </p> + * + * @param task {@link MaintenanceTask} object with maintenance information that needs + * to be stored to maintenance registry. + * + * @throws IgniteCheckedException If handling or storing maintenance task failed. + * + * @return Previously registered {@link MaintenanceTask} with the same ID + * or null if no tasks were registered for this ID. + */ + public @Nullable MaintenanceTask registerMaintenanceTask(MaintenanceTask task) throws IgniteCheckedException; + + /** + * Deletes {@link MaintenanceTask} of given ID from maintenance registry. + * + * @param maintenanceTaskName name of {@link MaintenanceTask} to be deleted. + */ + public void unregisterMaintenanceTask(String maintenanceTaskName); + + /** + * Returns active {@link MaintenanceTask} by its name. + * There are active tasks only when node entered Maintenance Mode. + * + * {@link MaintenanceTask} becomes active when node enters Maintenance Mode and doesn't resolve the task + * during maintenance prepare phase. + * + * @param maintenanceTaskName Maintenance Task name. + * + * @return {@link MaintenanceTask} object for given name or null if no maintenance task was found. + */ + @Nullable public MaintenanceTask activeMaintenanceTask(String maintenanceTaskName); + + /** + * Registers {@link MaintenanceWorkflowCallback} for a {@link MaintenanceTask} with a given name. + * + * Component registered {@link MaintenanceTask} automatically or by user request + * is responsible for providing {@link MaintenanceRegistry} with an implementation of + * {@link MaintenanceWorkflowCallback} where registry obtains {@link MaintenanceAction}s + * to be executed for this task and does a preliminary check before starting maintenance. + * + * @param maintenanceTaskName name of {@link MaintenanceTask} this callback is registered for. + * @param cb {@link MaintenanceWorkflowCallback} interface used by MaintenanceRegistry to execute + * maintenance steps by workflow. + */ + public void registerWorkflowCallback(@NotNull String maintenanceTaskName, @NotNull MaintenanceWorkflowCallback cb); + + /** + * All {@link MaintenanceAction}s provided by a component for {@link MaintenanceTask} with a given name. + * + * @param maintenanceTaskName name of Maintenance Task. + * @return {@link List} of all available {@link MaintenanceAction}s for given Maintenance Task. + * + * @throws IgniteException if no Maintenance Tasks are registered for provided name. + */ + public List<MaintenanceAction> actionsForMaintenanceTask(String maintenanceTaskName); + + /** + * Examine all components if they need to execute maintenance actions. + * + * As user may resolve some maintenance situations by hand when the node was turned off, + * component may find out that no maintenance is needed anymore. + * + * {@link MaintenanceTask Maintenance tasks} for these components are removed + * and their {@link MaintenanceAction maintenance actions} are not executed. + */ + public void prepareAndExecuteMaintenance(); +} diff --git a/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceTask.java b/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceTask.java new file mode 100644 index 0000000..49795f1 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceTask.java @@ -0,0 +1,88 @@ +/* + * 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.maintenance; + +import org.apache.ignite.lang.IgniteExperimental; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +/** + * Represents request to handle maintenance situation. + * + * It can be created automatically or by user request by any component needed maintenance and should be registered + * in Maintenance Registry with the method {@link MaintenanceRegistry#registerMaintenanceTask(MaintenanceTask)}. + * + * Lifecycle of Maintenance Task is managed by {@link MaintenanceRegistry}. + * + * Task contains unique ID of maintenance situation (e.g. situation of PDS corruption or defragmentation), + * description of task and optional parameters. + * + * When task is created node should be restarted to enter maintenance mode. + * In that mode node can start actions needed to resolve maintenance situation or wait for user to trigger them. + * + * Components that may need to perform maintenance actions as part of their recovery workflow should check + * maintenance status on startup and supply {@link MaintenanceWorkflowCallback} implementation to + * {@link MaintenanceRegistry#registerWorkflowCallback(String, MaintenanceWorkflowCallback)} to allow Maintenance Registry + * to find maintenance actions and start them automatically or by user request. + * + * Matching between {@link MaintenanceTask} and {@link MaintenanceWorkflowCallback} is performed based on + * the name of maintenance task that should be unique among all registered tasks. + */ +@IgniteExperimental +public class MaintenanceTask { + /** */ + private final String name; + + /** */ + private final String description; + + /** */ + private final String params; + + /** + * @param name Mandatory name of maintenance task. Name should be unique among all other tasks. + * @param description Mandatory description of maintenance situation. + * @param params Optional parameters that may be needed to perform maintenance actions. + */ + public MaintenanceTask(@NotNull String name, @NotNull String description, @Nullable String params) { + this.name = name; + this.description = description; + this.params = params; + } + + /** + * @return Name of Maintenance Task unique among all registered tasks. + */ + public @NotNull String name() { + return name; + } + + /** + * @return Human-readable not-nullable description of the task. + */ + public @NotNull String description() { + return description; + } + + /** + * @return Optional parameters that could be used by actions associated with this Maintenance Task. + */ + public @Nullable String parameters() { + return params; + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceWorkflowCallback.java b/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceWorkflowCallback.java new file mode 100644 index 0000000..26ba2a1 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/maintenance/MaintenanceWorkflowCallback.java @@ -0,0 +1,68 @@ +/* + * 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.maintenance; + +import java.util.List; + +import org.apache.ignite.lang.IgniteExperimental; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +/** + * Abstraction to decouple interaction between {@link MaintenanceRegistry} + * and components that may require maintenance. + * + * If a component may cause node to enter maintenance mode, it should register this callback + * in {@link MaintenanceRegistry} using method {@link MaintenanceRegistry#registerWorkflowCallback(String, MaintenanceWorkflowCallback)} + * + * {@link MaintenanceRegistry} during its workflow will collect necessary information about maintenance for components + * without knowing implementation details of the components. + */ +@IgniteExperimental +public interface MaintenanceWorkflowCallback { + /** + * Called by {@link MaintenanceRegistry} and enables it to check if maintenance is still needed + * for component that provided this callback. + * + * User may fix maintenance situation by hand when node was down thus before going to maintenance mode + * we should be able to check if it is still necessary. + * + * @return {@code True} if maintenance is still needed for the component. + */ + public boolean shouldProceedWithMaintenance(); + + /** + * Supplies list of {@link MaintenanceAction}s that user can call to fix maintenance situation for the component or + * get information about ongoing actions. Should not be null or empty. + * + * @return Not null and non-empty {@link List} of {@link MaintenanceAction}. + */ + @NotNull public List<MaintenanceAction> allActions(); + + /** + * Component can provide optional {@link MaintenanceAction} that will be executed automatically + * by {@link MaintenanceRegistry} when node enters maintenance mode. + * + * If no automatic actions are provided {@link MaintenanceRegistry} will wait for user + * to trigger {@link MaintenanceAction} with logic to fix the maintenance situation. + * + * @return {@link MaintenanceAction} for automatic execution or null if maintenance situation + * should not be fixed automatically. + */ + @Nullable public MaintenanceAction automaticAction(); +} diff --git a/modules/core/src/main/java/org/apache/ignite/maintenance/package-info.java b/modules/core/src/main/java/org/apache/ignite/maintenance/package-info.java new file mode 100644 index 0000000..71b2a1d --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/maintenance/package-info.java @@ -0,0 +1,21 @@ +/* + * 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. + */ + +/** + * Contains public interfaces for maintenance mechanism. + */ +package org.apache.ignite.maintenance; diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/isolated/IsolatedDiscoverySpi.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/isolated/IsolatedDiscoverySpi.java new file mode 100644 index 0000000..a1d706a --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/isolated/IsolatedDiscoverySpi.java @@ -0,0 +1,256 @@ +/* + * 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.spi.discovery.isolated; + +import java.io.Serializable; +import java.util.Collection; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import org.apache.ignite.IgniteException; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteFeatures; +import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpi; +import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpiInternalListener; +import org.apache.ignite.lang.IgniteFuture; +import org.apache.ignite.lang.IgniteProductVersion; +import org.apache.ignite.spi.IgniteSpiAdapter; +import org.apache.ignite.spi.IgniteSpiContext; +import org.apache.ignite.spi.IgniteSpiException; +import org.apache.ignite.spi.IgniteSpiMultipleInstancesSupport; +import org.apache.ignite.spi.discovery.DiscoveryMetricsProvider; +import org.apache.ignite.spi.discovery.DiscoveryNotification; +import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage; +import org.apache.ignite.spi.discovery.DiscoverySpiDataExchange; +import org.apache.ignite.spi.discovery.DiscoverySpiHistorySupport; +import org.apache.ignite.spi.discovery.DiscoverySpiListener; +import org.apache.ignite.spi.discovery.DiscoverySpiNodeAuthenticator; +import org.apache.ignite.spi.discovery.DiscoverySpiOrderSupport; +import org.jetbrains.annotations.Nullable; + +import static java.util.Collections.emptyList; +import static java.util.Collections.singleton; +import static org.apache.ignite.events.EventType.EVT_NODE_JOINED; +import static org.apache.ignite.internal.IgniteFeatures.allNodesSupports; +import static org.apache.ignite.internal.events.DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT; + +/** + * Special discovery SPI implementation to start a single-node cluster in "isolated" mode. + * + * When used, node doesn't try to seek or communicate to other nodes that may be up and running even in the same JVM. + * + * At the same time all functions like sending discovery messages are functional with only note that + * no messages are sent to network but are processed by local node immediately when they are created. + */ +@IgniteSpiMultipleInstancesSupport(true) +@DiscoverySpiHistorySupport(true) +@DiscoverySpiOrderSupport(true) +public class IsolatedDiscoverySpi extends IgniteSpiAdapter implements IgniteDiscoverySpi { + /** */ + private Serializable consistentId; + + /** */ + private final long startTime = System.currentTimeMillis(); + + /** */ + private IsolatedNode locNode; + + /** */ + private DiscoverySpiListener lsnr; + + /** */ + private ExecutorService exec = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()); + + /** {@inheritDoc} */ + @Override public Serializable consistentId() throws IgniteSpiException { + if (consistentId == null) { + IgniteConfiguration cfg = ignite.configuration(); + + final Serializable cfgId = cfg.getConsistentId(); + + consistentId = cfgId != null ? cfgId : UUID.randomUUID(); + } + + return consistentId; + } + + /** {@inheritDoc} */ + @Override public Collection<ClusterNode> getRemoteNodes() { + return emptyList(); + } + + /** {@inheritDoc} */ + @Override public ClusterNode getLocalNode() { + return locNode; + } + + /** {@inheritDoc} */ + @Override public ClusterNode getNode(UUID nodeId) { + return locNode.id().equals(nodeId) ? locNode : null; + } + + /** {@inheritDoc} */ + @Override public boolean pingNode(UUID nodeId) { + return locNode.id().equals(nodeId); + } + + /** {@inheritDoc} */ + @Override public void setNodeAttributes(Map<String, Object> attrs, IgniteProductVersion ver) { + locNode = new IsolatedNode(ignite.configuration().getNodeId(), attrs, ver); + } + + /** {@inheritDoc} */ + @Override public void setListener(@Nullable DiscoverySpiListener lsnr) { + this.lsnr = lsnr; + } + + /** {@inheritDoc} */ + @Override public void setDataExchange(DiscoverySpiDataExchange exchange) { + // No-op. + } + + /** {@inheritDoc} */ + @Override public void setMetricsProvider(DiscoveryMetricsProvider metricsProvider) { + // No-op. + } + + /** {@inheritDoc} */ + @Override public void disconnect() throws IgniteSpiException { + // No-op. + } + + /** {@inheritDoc} */ + @Override public void setAuthenticator(DiscoverySpiNodeAuthenticator auth) { + // No-op. + } + + /** {@inheritDoc} */ + @Override public long getGridStartTime() { + return startTime; + } + + /** {@inheritDoc} */ + @Override public void sendCustomEvent(DiscoverySpiCustomMessage msg) throws IgniteException { + exec.execute(() -> { + IgniteFuture<?> fut = lsnr.onDiscovery(new DiscoveryNotification( + EVT_DISCOVERY_CUSTOM_EVT, + 1, + locNode, + singleton(locNode), + null, + msg, + null)); + + // Acknowledge message must be send after initial message processed. + fut.listen((f) -> { + DiscoverySpiCustomMessage ack = msg.ackMessage(); + + if (ack != null) { + exec.execute(() -> lsnr.onDiscovery(new DiscoveryNotification( + EVT_DISCOVERY_CUSTOM_EVT, + 1, + locNode, + singleton(locNode), + null, + ack, + null)) + ); + } + }); + }); + } + + /** {@inheritDoc} */ + @Override public void failNode(UUID nodeId, @Nullable String warning) { + // No-op. + } + + /** {@inheritDoc} */ + @Override public boolean isClientMode() throws IllegalStateException { + return false; + } + + /** {@inheritDoc} */ + @Override public void spiStart(@Nullable String igniteInstanceName) throws IgniteSpiException { + exec.execute(() -> { + lsnr.onLocalNodeInitialized(locNode); + + lsnr.onDiscovery(new DiscoveryNotification( + EVT_NODE_JOINED, + 1, + locNode, + singleton(locNode)) + ); + }); + } + + /** {@inheritDoc} */ + @Override public void spiStop() throws IgniteSpiException { + exec.shutdownNow(); + } + + /** {@inheritDoc} */ + @Override protected void onContextInitialized0(final IgniteSpiContext spiCtx) throws IgniteSpiException { + // No-op. + } + + /** {@inheritDoc} */ + @Override public boolean knownNode(UUID nodeId) { + return getNode(nodeId) != null; + } + + /** {@inheritDoc} */ + @Override public boolean clientReconnectSupported() { + return false; + } + + /** {@inheritDoc} */ + @Override public void clientReconnect() { + // No-op. + } + + /** {@inheritDoc} */ + @Override public boolean allNodesSupport(IgniteFeatures feature) { + if (locNode == null) + return false; + + return allNodesSupports(singleton(locNode), feature); + } + + /** {@inheritDoc} */ + @Override public void simulateNodeFailure() { + // No-op. + } + + /** {@inheritDoc} */ + @Override public void setInternalListener(IgniteDiscoverySpiInternalListener lsnr) { + // No-op. + } + + /** {@inheritDoc} */ + @Override public boolean supportsCommunicationFailureResolve() { + return false; + } + + /** {@inheritDoc} */ + @Override public void resolveCommunicationFailure(ClusterNode node, Exception err) { + // No-op. + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/isolated/IsolatedNode.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/isolated/IsolatedNode.java new file mode 100644 index 0000000..d623650 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/isolated/IsolatedNode.java @@ -0,0 +1,153 @@ +/* + * 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.spi.discovery.isolated; + +import java.io.Serializable; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; +import org.apache.ignite.cache.CacheMetrics; +import org.apache.ignite.cluster.ClusterMetrics; +import org.apache.ignite.internal.ClusterMetricsSnapshot; +import org.apache.ignite.internal.managers.discovery.IgniteClusterNode; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.lang.IgniteProductVersion; + +import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_NODE_CONSISTENT_ID; + +/** + * Special isolated node. + */ +public class IsolatedNode implements IgniteClusterNode { + /** */ + private final UUID id; + + /** */ + private final IgniteProductVersion ver; + + /** Consistent ID. */ + private Object consistentId; + + /** Node attributes. */ + private Map<String, Object> attrs; + + /** */ + private volatile ClusterMetrics metrics = new ClusterMetricsSnapshot(); + + /** */ + private volatile Map<Integer, CacheMetrics> cacheMetrics = Collections.emptyMap(); + + /** + * @param id Node ID. + * @param attrs Node attributes. + * @param ver Node version. + */ + public IsolatedNode(UUID id, Map<String, Object> attrs, IgniteProductVersion ver) { + this.id = id; + this.attrs = U.sealMap(attrs); + this.ver = ver; + } + + /** {@inheritDoc} */ + @Override public UUID id() { + return id; + } + + /** {@inheritDoc} */ + @Override public Object consistentId() { + return consistentId; + } + + /** {@inheritDoc} */ + @Override public <T> T attribute(String name) { + return (T)attrs.get(name); + } + + /** {@inheritDoc} */ + @Override public ClusterMetrics metrics() { + return metrics; + } + + /** {@inheritDoc} */ + @Override public Map<String, Object> attributes() { + return attrs; + } + + /** {@inheritDoc} */ + @Override public Collection<String> addresses() { + return Collections.singleton("127.0.0.1"); + } + + /** {@inheritDoc} */ + @Override public Collection<String> hostNames() { + return Collections.singleton("localhost"); + } + + /** {@inheritDoc} */ + @Override public long order() { + return 1; + } + + /** {@inheritDoc} */ + @Override public IgniteProductVersion version() { + return ver; + } + + /** {@inheritDoc} */ + @Override public boolean isLocal() { + return true; + } + + /** {@inheritDoc} */ + @Override public boolean isDaemon() { + return false; + } + + /** {@inheritDoc} */ + @Override public boolean isClient() { + return false; + } + + /** {@inheritDoc} */ + @Override public void setConsistentId(Serializable consistentId) { + this.consistentId = consistentId; + + final Map<String, Object> map = new HashMap<>(attrs); + + map.put(ATTR_NODE_CONSISTENT_ID, consistentId); + + attrs = Collections.unmodifiableMap(map); + } + + /** {@inheritDoc} */ + @Override public void setMetrics(ClusterMetrics metrics) { + this.metrics = metrics; + } + + /** {@inheritDoc} */ + @Override public Map<Integer, CacheMetrics> cacheMetrics() { + return cacheMetrics; + } + + /** {@inheritDoc} */ + @Override public void setCacheMetrics(Map<Integer, CacheMetrics> cacheMetrics) { + this.cacheMetrics = cacheMetrics != null ? cacheMetrics : Collections.emptyMap(); + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/cache/RebalanceCancellationTest.java b/modules/core/src/test/java/org/apache/ignite/cache/RebalanceCancellationTest.java index 2dbf804..2ed03ff 100644 --- a/modules/core/src/test/java/org/apache/ignite/cache/RebalanceCancellationTest.java +++ b/modules/core/src/test/java/org/apache/ignite/cache/RebalanceCancellationTest.java @@ -26,6 +26,7 @@ import org.apache.ignite.IgniteCache; import org.apache.ignite.IgniteDataStreamer; import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.cluster.ClusterState; import org.apache.ignite.configuration.CacheConfiguration; import org.apache.ignite.configuration.DataRegionConfiguration; import org.apache.ignite.configuration.DataStorageConfiguration; @@ -68,7 +69,7 @@ public class RebalanceCancellationTest extends GridCommonAbstractTest { public boolean persistenceEnabled; /** Add additional non-persistence data region. */ - public boolean addtiotionalMemRegion; + public boolean additionalMemRegion; /** Filter node. */ public boolean filterNode; @@ -86,7 +87,7 @@ public class RebalanceCancellationTest extends GridCommonAbstractTest { .setAffinity(new RendezvousAffinityFunction(false, 15)) .setBackups(BACKUPS)); - if (addtiotionalMemRegion) { + if (additionalMemRegion) { cfg.setCacheConfiguration(cfg.getCacheConfiguration()[0], new CacheConfiguration(MEM_REGION_CACHE) .setDataRegionName(MEM_REGION) @@ -116,12 +117,15 @@ public class RebalanceCancellationTest extends GridCommonAbstractTest { /** {@inheritDoc} */ @Override protected void beforeTest() throws Exception { + stopAllGrids(); + cleanPersistenceDir(); } /** {@inheritDoc} */ @Override protected void afterTest() throws Exception { stopAllGrids(); + cleanPersistenceDir(); } @@ -249,7 +253,7 @@ public class RebalanceCancellationTest extends GridCommonAbstractTest { @Test public void testComplexCompatibilityInMemory() throws Exception { persistenceEnabled = false; - addtiotionalMemRegion = false; + additionalMemRegion = false; IgniteEx crd = startGrid(0); @@ -320,16 +324,16 @@ public class RebalanceCancellationTest extends GridCommonAbstractTest { * Trigger rebalance when dynamic caches stop/start. * * @param persistence Persistent flag. - * @param addtiotionalRegion Use additional (non default) region. + * @param additionalRegion Use additional (non default) region. * @throws Exception If failed. */ - public void testRebalanceDynamicCache(boolean persistence, boolean addtiotionalRegion) throws Exception { + public void testRebalanceDynamicCache(boolean persistence, boolean additionalRegion) throws Exception { persistenceEnabled = persistence; - addtiotionalMemRegion = addtiotionalRegion; + additionalMemRegion = additionalRegion; IgniteEx ignite0 = startGrids(NODES_CNT); - ignite0.cluster().active(true); + ignite0.cluster().state(ClusterState.ACTIVE); grid(1).close(); @@ -371,14 +375,14 @@ public class RebalanceCancellationTest extends GridCommonAbstractTest { * Trigger rebalance when non-blt node left topology. * * @param persistence Persistent flag. - * @param addtiotionalRegion Use additional (non default) region. + * @param additionalRegion Use additional (non default) region. * @param fail If true node forcibly falling. * @throws Exception If failed. */ - public void testRebalanceNoneBltNode(boolean persistence, boolean addtiotionalRegion, + public void testRebalanceNoneBltNode(boolean persistence, boolean additionalRegion, boolean fail) throws Exception { persistenceEnabled = persistence; - addtiotionalMemRegion = addtiotionalRegion; + additionalMemRegion = additionalRegion; IgniteEx ignite0 = startGrids(NODES_CNT); @@ -445,7 +449,7 @@ public class RebalanceCancellationTest extends GridCommonAbstractTest { */ public void testRebalanceFilteredNode(boolean persistence, boolean addtiotionalRegion) throws Exception { persistenceEnabled = persistence; - addtiotionalMemRegion = addtiotionalRegion; + additionalMemRegion = addtiotionalRegion; filterNode = true; IgniteEx ignite0 = startGrids(NODES_CNT); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/WalModeChangeAdvancedSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/WalModeChangeAdvancedSelfTest.java index 9069978..9eb3b7c 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/WalModeChangeAdvancedSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/WalModeChangeAdvancedSelfTest.java @@ -17,6 +17,7 @@ package org.apache.ignite.internal.processors.cache; +import java.io.File; import java.util.ArrayList; import java.util.Collection; import java.util.concurrent.CountDownLatch; @@ -26,9 +27,14 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.ignite.Ignite; import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; import org.apache.ignite.internal.IgniteClientReconnectAbstractTest; +import org.apache.ignite.internal.IgniteEx; import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.processors.cache.persistence.CheckpointState; +import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager; +import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.testframework.GridTestUtils; import org.apache.ignite.testframework.GridTestUtils.SF; import org.junit.Ignore; @@ -36,6 +42,8 @@ import org.junit.Test; import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL; import static org.apache.ignite.cache.CacheMode.PARTITIONED; +import static org.apache.ignite.cluster.ClusterState.ACTIVE; +import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_DATA_FILENAME; /** * Concurrent and advanced tests for WAL state change. @@ -64,6 +72,143 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe } /** + * Verifies that node with consistent partitions (fully synchronized with disk on a previous checkpoint) + * starts successfully even if WAL for that cache group is globally disabled. + * + * <p> + * Test scenario: + * </p> + * <ol> + * <li> + * Start a cluster from one server node, activate cluster. + * </li> + * <li> + * Create new cache. Disable WAL for the cache and put some data to it. + * </li> + * <li> + * Trigger checkpoint and wait for it finish. Restart node. + * </li> + * <li> + * Verify that node starts successfully and data is presented in the cache. + * </li> + * </ol> + * + * @throws Exception If failed. + */ + @Test + public void testConsistentDataPreserved() throws Exception { + Ignite srv = startGrid(config(SRV_1, false, false)); + + srv.cluster().state(ACTIVE); + + IgniteCache cache1 = srv.getOrCreateCache(cacheConfig(CACHE_NAME, PARTITIONED, TRANSACTIONAL)); + + srv.cluster().disableWal(CACHE_NAME); + + for (int i = 0; i < 10; i++) + cache1.put(i, i); + + GridCacheDatabaseSharedManager dbMrg0 = (GridCacheDatabaseSharedManager) ((IgniteEx)srv).context().cache().context().database(); + + dbMrg0.forceCheckpoint("cp").futureFor(CheckpointState.FINISHED).get(); + + stopGrid(SRV_1); + + srv = startGrid(config(SRV_1, false, false)); + + assertForAllNodes(CACHE_NAME, false); + + cache1 = srv.cache(CACHE_NAME); + + for (int i = 0; i < 10; i++) + assertNotNull(cache1.get(i)); + } + + /** + * If user manually clears corrupted files when node was down, node detects this and not enters maintenance mode + * (although still need another restart to get back to normal operations). + * + * <p> + * Test scenario: + * <ol> + * <li> + * Start server node, create cache, disable WAL for the cache, put some keys to it. + * </li> + * <li> + * Stop server node, remove checkpoint end markers from cp directory + * to make node think it has failed in the middle of checkpoint. + * </li> + * <li> + * Start the node, verify it fails to start because of corrupted PDS of the cache. + * </li> + * <li> + * Clean data directory of the cache, start node again, verify it doesn't report maintenance mode. + * </li> + * <li> + * Restart node and verify it is in normal operations mode. + * </li> + * </ol> + * </p> + * + * + * @throws Exception If failed. + */ + @Test + public void testMaintenanceIsSkippedIfWasFixedManuallyOnDowntime() throws Exception { + IgniteEx srv = startGrid(config(SRV_1, false, false)); + + File cacheToClean = cacheDir(srv, CACHE_NAME); + + String ig0Folder = srv.context().pdsFolderResolver().resolveFolders().folderName(); + File dbDir = U.resolveWorkDirectory(srv.configuration().getWorkDirectory(), "db", false); + + File ig0LfsDir = new File(dbDir, ig0Folder); + File ig0CpDir = new File(ig0LfsDir, "cp"); + + srv.cluster().state(ACTIVE); + + IgniteCache cache1 = srv.getOrCreateCache(cacheConfig(CACHE_NAME, PARTITIONED, TRANSACTIONAL)); + + srv.cluster().disableWal(CACHE_NAME); + + for (int i = 0; i < 10; i++) + cache1.put(i, i); + + stopAllGrids(true); + + File[] cpMarkers = ig0CpDir.listFiles(); + + for (File cpMark : cpMarkers) { + if (cpMark.getName().contains("-END")) + cpMark.delete(); + } + + // Node should fail as its PDS may be corrupted because of disabled WAL + GridTestUtils.assertThrows(null, + () -> startGrid(config(SRV_1, false, false)), + Exception.class, + null); + + cleanCacheDir(cacheToClean); + + // Node should start successfully and not enter maintenance mode as MaintenanceRecord will be cleaned + // automatically because corrupted PDS was deleted during downtime + srv = startGrid(config(SRV_1, false, false)); + assertFalse(srv.context().maintenanceRegistry().isMaintenanceMode()); + + stopAllGrids(false); + + // After restart node works normal mode even without executing maintenance action to clear corrupted PDS + srv = startGrid(config(SRV_1, false, false)); + assertFalse(srv.context().maintenanceRegistry().isMaintenanceMode()); + + srv.cluster().state(ACTIVE); + + cache1 = srv.getOrCreateCache(CACHE_NAME); + assertEquals(0, cache1.size()); + } + + /** * Test cache cleanup on restart. * * @throws Exception If failed. @@ -72,7 +217,9 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe public void testCacheCleanup() throws Exception { Ignite srv = startGrid(config(SRV_1, false, false)); - srv.cluster().active(true); + File cacheToClean = cacheDir(srv, CACHE_NAME_2); + + srv.cluster().state(ACTIVE); IgniteCache cache1 = srv.getOrCreateCache(cacheConfig(CACHE_NAME, PARTITIONED, TRANSACTIONAL)); IgniteCache cache2 = srv.getOrCreateCache(cacheConfig(CACHE_NAME_2, PARTITIONED, TRANSACTIONAL)); @@ -118,9 +265,11 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe stopAllGrids(true); + cleanCacheDir(cacheToClean); + srv = startGrid(config(SRV_1, false, false)); - srv.cluster().active(true); + srv.cluster().state(ACTIVE); cache1 = srv.cache(CACHE_NAME); cache2 = srv.cache(CACHE_NAME_2); @@ -132,6 +281,24 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe assertEquals(0, cache2.size()); } + /** */ + private File cacheDir(Ignite ig, String cacheName) throws IgniteCheckedException { + String igFolder = ((IgniteEx)ig).context().pdsFolderResolver().resolveFolders().folderName(); + File dbDir = U.resolveWorkDirectory(ig.configuration().getWorkDirectory(), "db", false); + + File igPdsFolder = new File(dbDir, igFolder); + + return new File(igPdsFolder, "cache-" + cacheName); + } + + /** */ + private void cleanCacheDir(File cacheDir) { + for (File f : cacheDir.listFiles()) { + if (!f.getName().equals(CACHE_DATA_FILENAME)) + f.delete(); + } + } + /** * Test simple node join. * @@ -162,7 +329,7 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe // Start node and disable WAL. Ignite srv = startGrid(config(SRV_1, false, crdFiltered)); - srv.cluster().active(true); + srv.cluster().state(ACTIVE); srv.getOrCreateCache(cacheConfig(PARTITIONED)); assertForAllNodes(CACHE_NAME, true); @@ -173,7 +340,9 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe } // Start other nodes. - startGrid(config(SRV_2, false, false)); + IgniteEx ig2 = startGrid(config(SRV_2, false, false)); + + File ig2CacheDir = cacheDir(ig2, CACHE_NAME); if (crdFiltered) srv.cluster().disableWal(CACHE_NAME); @@ -196,6 +365,8 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe assertForAllNodes(CACHE_NAME, true); } + cleanCacheDir(ig2CacheDir); + // Start other nodes again. startGrid(config(SRV_2, false, false)); @@ -244,7 +415,7 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe Ignite cli = startGrid(config(CLI, true, false)); - cli.cluster().active(true); + cli.cluster().state(ACTIVE); cli.getOrCreateCache(cacheConfig(PARTITIONED)); @@ -268,7 +439,12 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe victimName = SRV_2; try { + File cacheDir = cacheDir(grid(victimName), CACHE_NAME); + stopGrid(victimName); + + cleanCacheDir(cacheDir); + startGrid(config(victimName, false, false)); Thread.sleep(200); @@ -313,7 +489,7 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe final Ignite srv = startGrid(config(SRV_1, false, false)); Ignite cli = startGrid(config(CLI, true, false)); - cli.cluster().active(true); + cli.cluster().state(ACTIVE); cli.getOrCreateCache(cacheConfig(PARTITIONED)); @@ -373,7 +549,7 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe final Ignite srv = startGrid(config(SRV_1, false, false)); Ignite cli = startGrid(config(CLI, true, false)); - cli.cluster().active(true); + cli.cluster().state(ACTIVE); srv.createCache(cacheConfig(PARTITIONED)); @@ -439,7 +615,7 @@ public class WalModeChangeAdvancedSelfTest extends WalModeChangeCommonAbstractSe final Ignite cacheCli = startGrid(config(CLI_2, true, false)); - cacheCli.cluster().active(true); + cacheCli.cluster().state(ACTIVE); final IgniteCache cache = cacheCli.getOrCreateCache(cacheConfig(PARTITIONED)); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/LocalWalModeChangeDuringRebalancingSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/LocalWalModeChangeDuringRebalancingSelfTest.java index 1c40db3..03ad4dd 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/LocalWalModeChangeDuringRebalancingSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/LocalWalModeChangeDuringRebalancingSelfTest.java @@ -22,13 +22,15 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.nio.MappedByteBuffer; import java.nio.file.OpenOption; -import java.util.Collection; +import java.util.Arrays; import java.util.Collections; -import java.util.Set; -import java.util.TreeSet; +import java.util.List; +import java.util.Optional; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; + import org.apache.ignite.Ignite; import org.apache.ignite.IgniteCache; import org.apache.ignite.IgniteCheckedException; @@ -56,7 +58,11 @@ import org.apache.ignite.internal.util.typedef.G; import org.apache.ignite.internal.util.typedef.internal.CU; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.lang.IgniteInClosure; +import org.apache.ignite.lang.IgniteRunnable; +import org.apache.ignite.maintenance.MaintenanceAction; +import org.apache.ignite.maintenance.MaintenanceRegistry; import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.resources.IgniteInstanceResource; import org.apache.ignite.spi.IgniteSpiException; import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi; import org.apache.ignite.testframework.GridTestUtils; @@ -65,6 +71,9 @@ import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; import org.junit.Assert; import org.junit.Test; +import static org.apache.ignite.cluster.ClusterState.ACTIVE; +import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CORRUPTED_DATA_FILES_MNTC_TASK_NAME; +import static org.apache.ignite.testframework.GridTestUtils.assertThrows; import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; /** @@ -112,6 +121,7 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL) // Test checks internal state before and after rebalance, so it is configured to be triggered manually .setRebalanceDelay(-1) + .setAffinity(new RendezvousAffinityFunction(false, 32)) .setBackups(dfltCacheBackupCnt), new CacheConfiguration(REPL_CACHE) @@ -253,7 +263,7 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr Ignite ignite = startGrids(3); ignite.cluster().baselineAutoAdjustEnabled(false); - ignite.cluster().active(true); + ignite.cluster().state(ACTIVE); IgniteCache<Integer, Integer> cache = ignite.cache(DEFAULT_CACHE_NAME); IgniteCache<Integer, Integer> cache2 = ignite.cache(REPL_CACHE); @@ -338,7 +348,7 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr Ignite ignite = startGrids(3); ignite.cluster().baselineAutoAdjustEnabled(false); - ignite.cluster().active(true); + ignite.cluster().state(ACTIVE); ignite.cluster().setBaselineTopology(3); @@ -393,7 +403,7 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr Ignite ignite = startGrids(3); ignite.cluster().baselineAutoAdjustEnabled(false); - ignite.cluster().active(true); + ignite.cluster().state(ACTIVE); IgniteCache<Integer, Integer> cache = ignite.cache(DEFAULT_CACHE_NAME); @@ -436,7 +446,7 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr Ignite ignite = startGrids(nodeCnt); - ignite.cluster().active(true); + ignite.cluster().state(ACTIVE); IgniteCache<Integer, Integer> cache = ignite.cache(REPL_CACHE); @@ -507,7 +517,7 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr Ignite ignite = startGrids(3); ignite.cluster().baselineAutoAdjustEnabled(false); - ignite.cluster().active(true); + ignite.cluster().state(ACTIVE); IgniteCache<Integer, Integer> cache = ignite.cache(DEFAULT_CACHE_NAME); @@ -547,60 +557,218 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr } /** + * Node doesn't delete consistent PDS when WAL was turned off automatically (disable WAL during rebalancing feature). + * + * <p> + * Test scenario: + * <ol> + * <li> + * Two server nodes are started, cluster is activated, baseline is set. 2500 keys are put into cache. + * </li> + * <li> + * Checkpoint is started and finished on both nodes. + * </li> + * <li> + * Node n1 is stopped, another 2500 keys are put into the same cache. + * </li> + * <li> + * Node n1 is started back so rebalancing is triggered from n0 to n1. WAL is turned off on n1 automatically. + * </li> + * <li> + * Both nodes are stopped without checkpoint. + * </li> + * <li> + * Node n1 is started and activated. Lost partitions are reset. + * </li> + * <li> + * First 2500 keys are found in cache thus PDS wasn't removed on restart. + * </li> + * <li> + * Second 2500 keys are not found in cache as WAL was disabled during rebalancing + * and no checkpoint was triggered. + * </li> + * </ol> + * </p> + * * @throws Exception If failed. */ @Test - public void testDataClearedAfterRestartWithDisabledWal() throws Exception { - Ignite ignite = startGrid(0); + public void testConsistentPdsIsNotClearedAfterRestartWithDisabledWal() throws Exception { + dfltCacheBackupCnt = 1; - ignite.cluster().baselineAutoAdjustEnabled(false); - ignite.cluster().active(true); + IgniteEx ig0 = startGrid(0); + IgniteEx ig1 = startGrid(1); - IgniteCache<Integer, Integer> cache = ignite.cache(DEFAULT_CACHE_NAME); + ig0.cluster().baselineAutoAdjustEnabled(false); + ig0.cluster().state(ACTIVE); - int keysCnt = getKeysCount(); + IgniteCache<Integer, Integer> cache = ig0.cache(DEFAULT_CACHE_NAME); - for (int k = 0; k < keysCnt; k++) + for (int k = 0; k < 2500; k++) cache.put(k, k); - IgniteEx newIgnite = startGrid(1); + GridCacheDatabaseSharedManager dbMrg0 = (GridCacheDatabaseSharedManager) ig0.context().cache().context().database(); + GridCacheDatabaseSharedManager dbMrg1 = (GridCacheDatabaseSharedManager) ig1.context().cache().context().database(); - newIgnite.cluster().setBaselineTopology(2); + dbMrg0.forceCheckpoint("cp").futureFor(CheckpointState.FINISHED).get(); + dbMrg1.forceCheckpoint("cp").futureFor(CheckpointState.FINISHED).get(); - // Await fully exchange complete. - awaitExchange(newIgnite); + stopGrid(1); - CacheGroupContext grpCtx = newIgnite.cachex(DEFAULT_CACHE_NAME).context().group(); + for (int k = 2500; k < 5000; k++) + cache.put(k, k); - assertFalse(grpCtx.localWalEnabled()); + ig1 = startGrid(1); - stopGrid(1); - stopGrid(0); + awaitExchange(ig1); - newIgnite = startGrid(1); + stopAllGrids(false); - newIgnite.cluster().active(true); + ig1 = startGrid(1); + ig1.cluster().state(ACTIVE); - newIgnite.cluster().setBaselineTopology(newIgnite.cluster().nodes()); + ig1.resetLostPartitions(Arrays.asList(DEFAULT_CACHE_NAME)); - cache = newIgnite.cache(DEFAULT_CACHE_NAME); + awaitExchange(ig1); - Collection<Integer> lostParts = cache.lostPartitions(); + cache = ig1.cache(DEFAULT_CACHE_NAME); - Set<Integer> keys = new TreeSet<>(); + for (int k = 0; k < 2500; k++) + assertTrue(cache.containsKey(k)); - for (int k = 0; k < keysCnt; k++) { - // Skip lost partitions. - if (lostParts.contains(newIgnite.affinity(DEFAULT_CACHE_NAME).partition(k))) - continue; + for (int k = 2500; k < 5000; k++) + assertFalse(cache.containsKey(k)); + } + + /** + * Test is opposite to {@link #testConsistentPdsIsNotClearedAfterRestartWithDisabledWal()} + * + * <p> + * Test scenario: + * <ol> + * <li> + * Two server nodes are started, cluster is activated, baseline is set. 2500 keys are put into cache. + * </li> + * <li> + * Checkpoint is started and finished on both nodes. + * </li> + * <li> + * Node n1 is stopped, another 2500 keys are put into the same cache. + * </li> + * <li> + * Node n1 is started back so rebalancing is triggered from n0 to n1. WAL is turned off on n1 automatically. + * </li> + * <li> + * Both nodes are stopped without checkpoint. + * </li> + * <li> + * CP END marker for the first checkpoint is removed on node n1 so node will think it crushed during checkpoint + * on the next restart. + * </li> + * <li> + * Node n1 fails to start as it sees potentially corrupted files of one cache. Manual action is required. + * </li> + * <li> + * Cache files are cleaned up manually on node n1 and it starts successfully. + * </li> + * </ol> + * </p> + * + * @throws Exception If failed. + */ + @Test + public void testPdsWithBrokenBinaryConsistencyIsClearedAfterRestartWithDisabledWal() throws Exception { + dfltCacheBackupCnt = 1; + + IgniteEx ig0 = startGrid(0); + IgniteEx ig1 = startGrid(1); + + String ig1Folder = ig1.context().pdsFolderResolver().resolveFolders().folderName(); + File dbDir = U.resolveWorkDirectory(ig1.configuration().getWorkDirectory(), "db", false); + + File ig1LfsDir = new File(dbDir, ig1Folder); + File ig1CpDir = new File(ig1LfsDir, "cp"); + + ig0.cluster().baselineAutoAdjustEnabled(false); + ig0.cluster().state(ACTIVE); + + IgniteCache<Integer, Integer> cache = ig0.cache(DEFAULT_CACHE_NAME); + + for (int k = 0; k < 2500; k++) + cache.put(k, k); - keys.add(k); + GridCacheDatabaseSharedManager dbMrg0 = (GridCacheDatabaseSharedManager) ig0.context().cache().context().database(); + GridCacheDatabaseSharedManager dbMrg1 = (GridCacheDatabaseSharedManager) ig1.context().cache().context().database(); + + dbMrg0.forceCheckpoint("cp").futureFor(CheckpointState.FINISHED).get(); + dbMrg1.forceCheckpoint("cp").futureFor(CheckpointState.FINISHED).get(); + + stopGrid(1); + + for (int k = 2500; k < 5000; k++) + cache.put(k, k); + + ig1 = startGrid(1); + + awaitExchange(ig1); + + stopAllGrids(false); + + ig0 = startGrid(0); + + File[] cpMarkers = ig1CpDir.listFiles(); + + for (File cpMark : cpMarkers) { + if (cpMark.getName().contains("-END")) + cpMark.delete(); } - for (Integer k : keys) - assertFalse("k=" + k + ", v=" + cache.get(k), cache.containsKey(k)); + assertThrows(null, () -> startGrid(1), Exception.class, null); + + ig1 = startGrid(1); + + assertEquals(1, ig0.cluster().nodes().size()); + assertEquals(1, ig1.cluster().nodes().size()); + + AtomicBoolean actionNotFound = new AtomicBoolean(false); + + ig1.compute().run(new IgniteRunnable() { + @IgniteInstanceResource + private Ignite ig; + + @Override public void run() { + MaintenanceRegistry mntcRegistry = ((IgniteEx) ig).context().maintenanceRegistry(); + + List<MaintenanceAction> actions = mntcRegistry + .actionsForMaintenanceTask(CORRUPTED_DATA_FILES_MNTC_TASK_NAME); + + Optional<MaintenanceAction> optional = actions + .stream() + .filter(a -> a.name().equals(CleanCacheStoresMaintenanceAction.ACTION_NAME)).findFirst(); + + if (!optional.isPresent()) + actionNotFound.set(true); + else + optional.get().execute(); + + mntcRegistry.unregisterMaintenanceTask(CORRUPTED_DATA_FILES_MNTC_TASK_NAME); + } + }); + + assertFalse("Action to clear corrupted PDS is not found", actionNotFound.get()); + + stopAllGrids(); + + ig1 = startGrid(1); + + ig1.cluster().state(ACTIVE); + + assertEquals(1, ig1.cluster().nodes().size()); + + cache = ig1.cache(DEFAULT_CACHE_NAME); - assertFalse(cache.containsKeys(keys)); + for (int k = 0; k < 2500; k++) + assertFalse(cache.containsKey(k)); } /** @@ -611,7 +779,7 @@ public class LocalWalModeChangeDuringRebalancingSelfTest extends GridCommonAbstr Ignite ignite = startGrids(4); ignite.cluster().baselineAutoAdjustEnabled(false); - ignite.cluster().active(true); + ignite.cluster().state(ACTIVE); IgniteCache<Integer, Integer> cache = ignite.cache(DEFAULT_CACHE_NAME); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/MaintenanceRegistrySimpleTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/MaintenanceRegistrySimpleTest.java new file mode 100644 index 0000000..5866d43 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/MaintenanceRegistrySimpleTest.java @@ -0,0 +1,345 @@ +/* + * 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; + +import java.io.File; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.configuration.DataRegionConfiguration; +import org.apache.ignite.configuration.DataStorageConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.maintenance.MaintenanceProcessor; +import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderSettings; +import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFoldersResolver; +import org.apache.ignite.internal.processors.cache.persistence.wal.reader.StandaloneGridKernalContext; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.maintenance.MaintenanceAction; +import org.apache.ignite.maintenance.MaintenanceTask; +import org.apache.ignite.maintenance.MaintenanceWorkflowCallback; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.logger.GridTestLog4jLogger; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +/** + * Simple unit test to cover basic MaintenanceRegistry functionality like action validations, + * maintenance tasks structure etc. + */ +public class MaintenanceRegistrySimpleTest { + /** */ + private final IgniteLogger log = new GridTestLog4jLogger(); + + /** */ + @Before + public void beforeTest() throws Exception { + cleanMaintenanceRegistryFile(); + } + + /** */ + @After + public void afterTest() throws Exception { + cleanMaintenanceRegistryFile(); + } + + /** */ + private void cleanMaintenanceRegistryFile() throws Exception { + String dftlWorkDir = U.defaultWorkDirectory(); + + for (File f : new File(dftlWorkDir).listFiles()) { + if (f.getName().endsWith(".mntc")) + f.delete(); + } + } + + /** */ + private GridKernalContext initContext(boolean persistenceEnabled) throws IgniteCheckedException { + String dfltWorkDir = U.defaultWorkDirectory(); + + GridKernalContext kctx = new StandaloneGridKernalContext(log, null, null) + { + @Override protected IgniteConfiguration prepareIgniteConfiguration() { + IgniteConfiguration cfg = super.prepareIgniteConfiguration(); + + cfg.setDataStorageConfiguration(new DataStorageConfiguration().setDefaultDataRegionConfiguration( + new DataRegionConfiguration().setPersistenceEnabled(persistenceEnabled) + )); + + return cfg; + } + + @Override public PdsFoldersResolver pdsFolderResolver() { + return new PdsFoldersResolver() { + @Override public PdsFolderSettings resolveFolders() { + return new PdsFolderSettings(new File(dfltWorkDir), U.maskForFileName("")); + } + }; + } + }; + + return kctx; + } + + /** + * {@link MaintenanceTask} could be replaced with new parameters after registration, old task is deleted. + * + * @throws IgniteCheckedException If initialization failed. + */ + @Test + public void testMaintenanceTaskReplacement() throws IgniteCheckedException { + String name0 = "taskName0"; + String descr = "description"; + String oldParams = "oldParams"; + String newParams = "newParams"; + + MaintenanceProcessor proc = new MaintenanceProcessor(initContext(true)); + + proc.start(); + + assertFalse(proc.isMaintenanceMode()); + + proc.registerMaintenanceTask(new MaintenanceTask(name0, descr, oldParams)); + proc.registerMaintenanceTask(new MaintenanceTask(name0, descr, newParams)); + + proc.stop(false); + + proc.start(); + + assertTrue(proc.isMaintenanceMode()); + MaintenanceTask task = proc.activeMaintenanceTask(name0); + + assertNotNull(task); + assertEquals(newParams, task.parameters()); + } + + /** + * Registered {@link MaintenanceTask} can be deleted before node entered Maintenance Mode (before node restart). + * + * @throws IgniteCheckedException If initialization failed. + */ + @Test + public void testDeleteMaintenanceTask() throws IgniteCheckedException { + String name = "name0"; + + MaintenanceTask task = new MaintenanceTask(name, "description", null); + + MaintenanceProcessor proc = new MaintenanceProcessor(initContext(true)); + + proc.start(); + + proc.registerMaintenanceTask(task); + + assertFalse(proc.isMaintenanceMode()); + + proc.unregisterMaintenanceTask(name); + + proc.stop(false); + + proc.start(); + + assertNull(proc.activeMaintenanceTask(name)); + assertFalse(proc.isMaintenanceMode()); + } + + /** + * Maintenance actions provided by maintenance callback should all have unique names. + * + * @throws IgniteCheckedException If initialization failed. + */ + @Test + public void testMaintenanceCallbackProvidesActionsWithUniqueNames() throws IgniteCheckedException { + String actionName0 = "action0"; + String actionName1 = "action1"; + + String name0 = "name0"; + String name1 = "name1"; + + MaintenanceProcessor proc = new MaintenanceProcessor(initContext(true)); + + // attempt to register callback with actions with non-unique names throws exception + GridTestUtils.assertThrows(log, () -> + proc.registerWorkflowCallback(name0, new SimpleMaintenanceCallback(Arrays.asList(new SimpleAction(actionName0), new SimpleAction(actionName0)) + )), + IgniteException.class, + "unique names: " + actionName0 + ", " + actionName0); + + // Attempt to register callback with actions with unique names finishes succesfully + proc.registerWorkflowCallback(name1, new SimpleMaintenanceCallback(Arrays.asList(new SimpleAction(actionName0), new SimpleAction(actionName1)))); + } + + /** + * Smoke test for writing and restoring back maintenance tasks to/from file. + * + * @throws IgniteCheckedException If initialization of Maintenance Processor failed. + */ + @Test + public void testMultipleMaintenanceTasks() throws IgniteCheckedException { + MaintenanceProcessor proc = new MaintenanceProcessor(initContext(true)); + + String task0Name = "name0"; + String task1Name = "name1"; + + String desc0 = "task0"; + String desc1 = "task1"; + + String params0 = "task0_param"; + String params1 = "task1_param"; + + proc.start(); + + proc.registerMaintenanceTask(new MaintenanceTask(task0Name, desc0, params0)); + proc.registerMaintenanceTask(new MaintenanceTask(task1Name, desc1, params1)); + + proc.stop(false); + + proc.start(); + + MaintenanceTask task0 = proc.activeMaintenanceTask(task0Name); + MaintenanceTask task1 = proc.activeMaintenanceTask(task1Name); + + assertNotNull(task0); + assertNotNull(task1); + + assertEquals(desc0, task0.description()); + assertEquals(desc1, task1.description()); + + assertEquals(params0, task0.parameters()); + assertEquals(params1, task1.parameters()); + } + + /** + * + * @throws IgniteCheckedException If initialization of Maintenance Processor failed. + */ + @Test + public void testMaintenanceTasksWithoutParameters() throws IgniteCheckedException { + MaintenanceProcessor proc = new MaintenanceProcessor(initContext(true)); + + String task0Name = "name0"; + String task1Name = "name1"; + + String desc0 = "task0"; + String desc1 = "task1"; + + String params0 = "task0_param"; + + // call to initialize file for maintenance tasks + proc.start(); + + proc.registerMaintenanceTask(new MaintenanceTask(task0Name, desc0, params0)); + proc.registerMaintenanceTask(new MaintenanceTask(task1Name, desc1, null)); + + proc.stop(false); + + // call to force Maintenance Processor to read that file and fill internal collection of maintenance tasks + proc.start(); + + MaintenanceTask task0 = proc.activeMaintenanceTask(task0Name); + MaintenanceTask task1 = proc.activeMaintenanceTask(task1Name); + + assertNotNull(task0); + assertNotNull(task1); + + assertEquals(params0, task0.parameters()); + assertNull(task1.parameters()); + } + + /** + * Name of maintenance action should contain only alphanumeric and underscore symbols. + * + * @throws IgniteCheckedException If initialization of Maintenance Processor failed. + */ + @Test + public void testMaintenanceActionNameSymbols() throws IgniteCheckedException { + MaintenanceProcessor proc = new MaintenanceProcessor(initContext(true)); + + String name0 = "name0"; + String wrongName = "wrong*Name"; + + GridTestUtils.assertThrows(log, + () -> proc.registerWorkflowCallback(name0, new SimpleMaintenanceCallback(Arrays.asList(new SimpleAction(wrongName)))), + IgniteException.class, + "alphanumeric"); + + } + + /** */ + private final class SimpleMaintenanceCallback implements MaintenanceWorkflowCallback { + /** */ + private final List<MaintenanceAction> actions = new ArrayList<>(); + + SimpleMaintenanceCallback(List<MaintenanceAction> actions) { + this.actions.addAll(actions); + } + + /** {@inheritDoc} */ + @Override public boolean shouldProceedWithMaintenance() { + return true; + } + + /** {@inheritDoc} */ + @Override public @NotNull List<MaintenanceAction> allActions() { + return actions; + } + + /** {@inheritDoc} */ + @Override public @Nullable MaintenanceAction automaticAction() { + return null; + } + } + + /** */ + private final class SimpleAction implements MaintenanceAction<Void> { + /** */ + private final String name; + + /** */ + private SimpleAction(String name) { + this.name = name; + } + + /** {@inheritDoc} */ + @Override public Void execute() { + return null; + } + + /** {@inheritDoc} */ + @Override public @NotNull String name() { + return name; + } + + /** {@inheritDoc} */ + @Override public @Nullable String description() { + return null; + } + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java index cc7c6a9..3c7a2ed 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/pagemem/NoOpPageStoreManager.java @@ -211,11 +211,6 @@ public class NoOpPageStoreManager implements IgnitePageStoreManager { } /** {@inheritDoc} */ - @Override public void beforeCacheGroupStart(CacheGroupDescriptor grpDesc) { - // No-op. - } - - /** {@inheritDoc} */ @Override public void onActivate(GridKernalContext kctx) { // No-op. } diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java index 375efe7..00fc33a 100644 --- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java +++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java @@ -20,6 +20,7 @@ package org.apache.ignite.testsuites; import java.util.ArrayList; import java.util.Collection; import java.util.List; + import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.HistoricalRebalanceHeuristicsTest; import org.apache.ignite.internal.processors.cache.persistence.IgniteDataStorageMetricsSelfTest; import org.apache.ignite.internal.processors.cache.persistence.IgnitePdsCacheStartStopWithFreqCheckpointTest; @@ -32,6 +33,7 @@ import org.apache.ignite.internal.processors.cache.persistence.IgnitePersistentS import org.apache.ignite.internal.processors.cache.persistence.IgniteRebalanceScheduleResendPartitionsTest; import org.apache.ignite.internal.processors.cache.persistence.LocalWalModeChangeDuringRebalancingSelfTest; import org.apache.ignite.internal.processors.cache.persistence.LocalWalModeNoChangeDuringRebalanceOnNonNodeAssignTest; +import org.apache.ignite.internal.processors.cache.persistence.MaintenanceRegistrySimpleTest; import org.apache.ignite.internal.processors.cache.persistence.WALPreloadingWithCompactionTest; import org.apache.ignite.internal.processors.cache.persistence.WalPreloadingConcurrentTest; import org.apache.ignite.internal.processors.cache.persistence.baseline.ClientAffinityAssignmentWithBaselineTest; @@ -134,6 +136,9 @@ public class IgnitePdsTestSuite2 { GridTestUtils.addTestIfNeeded(suite, ClusterStateChangeEventTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, ClusterStateChangeEventWithPersistenceTest.class, ignoredTests); + // Maintenance tests + GridTestUtils.addTestIfNeeded(suite, MaintenanceRegistrySimpleTest.class, ignoredTests); + return suite; }