Mmuzaf commented on a change in pull request #8648: URL: https://github.com/apache/ignite/pull/8648#discussion_r599024865
########## File path: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java ########## @@ -0,0 +1,777 @@ +///* +// * Licensed to the Apache Software Foundation (ASF) under one or more +// * contributor license agreements. See the NOTICE file distributed with +// * this work for additional information regarding copyright ownership. +// * The ASF licenses this file to You under the Apache License, Version 2.0 +// * (the "License"); you may not use this file except in compliance with +// * the License. You may obtain a copy of the License at +// * +// * http://www.apache.org/licenses/LICENSE-2.0 +// * +// * Unless required by applicable law or agreed to in writing, software +// * distributed under the License is distributed on an "AS IS" BASIS, +// * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// * See the License for the specific language governing permissions and +// * limitations under the License. +// */ + +package org.apache.ignite.internal.processors.cache.persistence.snapshot; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BooleanSupplier; +import java.util.function.Consumer; +import java.util.stream.Collectors; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteException; +import org.apache.ignite.IgniteIllegalStateException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.cluster.ClusterState; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteFeatures; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.IgniteInterruptedCheckedException; +import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; +import org.apache.ignite.internal.processors.cache.GridCacheSharedContext; +import org.apache.ignite.internal.processors.cache.StoredCacheData; +import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager; +import org.apache.ignite.internal.processors.cache.verify.IdleVerifyResultV2; +import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState; +import org.apache.ignite.internal.util.distributed.DistributedProcess; +import org.apache.ignite.internal.util.future.GridFinishedFuture; +import org.apache.ignite.internal.util.future.GridFutureAdapter; +import org.apache.ignite.internal.util.future.IgniteFinishedFutureImpl; +import org.apache.ignite.internal.util.future.IgniteFutureImpl; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.internal.CU; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.lang.IgniteFuture; +import org.jetbrains.annotations.Nullable; + +import static org.apache.ignite.internal.IgniteFeatures.SNAPSHOT_RESTORE_CACHE_GROUP; +import static org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl.binaryWorkDir; +import static org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager.CACHE_GRP_DIR_PREFIX; +import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.databaseRelativePath; +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE; +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.RESTORE_CACHE_GROUP_SNAPSHOT_ROLLBACK; +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.RESTORE_CACHE_GROUP_SNAPSHOT_START; + +/** + * Distributed process to restore cache group from the snapshot. + */ +public class SnapshotRestoreProcess { + /** Reject operation message. */ + private static final String OP_REJECT_MSG = "Cache group restore operation was rejected. "; + + /** Kernal context. */ + private final GridKernalContext ctx; + + /** Cache group restore prepare phase. */ + private final DistributedProcess<SnapshotRestorePrepareRequest, ArrayList<StoredCacheData>> prepareRestoreProc; + + /** Cache group restore cache start phase. */ + private final DistributedProcess<UUID, Boolean> cacheStartProc; + + /** Cache group restore rollback phase. */ + private final DistributedProcess<UUID, Boolean> rollbackRestoreProc; + + /** Logger. */ + private final IgniteLogger log; + + /** The future to be completed when the cache restore process is complete. */ + private volatile GridFutureAdapter<Void> fut; + + /** Snapshot restore operation context. */ + private volatile SnapshotRestoreContext opCtx; + + /** + * @param ctx Kernal context. + */ + public SnapshotRestoreProcess(GridKernalContext ctx) { + this.ctx = ctx; + + log = ctx.log(getClass()); + + prepareRestoreProc = new DistributedProcess<>( + ctx, RESTORE_CACHE_GROUP_SNAPSHOT_PREPARE, this::prepare, this::finishPrepare); + + cacheStartProc = new DistributedProcess<>( + ctx, RESTORE_CACHE_GROUP_SNAPSHOT_START, this::cacheStart, this::finishCacheStart); + + rollbackRestoreProc = new DistributedProcess<>( + ctx, RESTORE_CACHE_GROUP_SNAPSHOT_ROLLBACK, this::rollback, this::finishRollback); + } + + /** + * Start cache group restore operation. + * + * @param snpName Snapshot name. + * @param cacheGrpNames Name of the cache groups for restore. + * @return Future that will be completed when the restore operation is complete and the cache groups are started. + */ + public IgniteFuture<Void> start(String snpName, Collection<String> cacheGrpNames) { + if (ctx.clientNode()) { + return new IgniteFinishedFutureImpl<>( + new IgniteException(OP_REJECT_MSG + "Client and daemon nodes can not perform this operation.")); + } + + synchronized (this) { + GridFutureAdapter<Void> fut0 = fut; + + if (opCtx != null || (fut0 != null && !fut0.isDone())) { + return new IgniteFinishedFutureImpl<>( + new IgniteException(OP_REJECT_MSG + "The previous snapshot restore operation was not completed.")); + } + + fut = new GridFutureAdapter<>(); + } + + DiscoveryDataClusterState clusterState = ctx.state().clusterState(); + + if (clusterState.state() != ClusterState.ACTIVE || clusterState.transition()) + return new IgniteFinishedFutureImpl<>(new IgniteException(OP_REJECT_MSG + "The cluster should be active.")); + + if (!clusterState.hasBaselineTopology()) { + return new IgniteFinishedFutureImpl<>( + new IgniteException(OP_REJECT_MSG + "The baseline topology is not configured for cluster.")); + } + + IgniteSnapshotManager snpMgr = ctx.cache().context().snapshotMgr(); + + if (snpMgr.isSnapshotCreating()) { + return new IgniteFinishedFutureImpl<>( + new IgniteException(OP_REJECT_MSG + "A cluster snapshot operation is in progress.")); + } + + if (!IgniteFeatures.allNodesSupports(ctx.grid().cluster().nodes(), SNAPSHOT_RESTORE_CACHE_GROUP)) { + return new IgniteFinishedFutureImpl<>( + new IgniteException(OP_REJECT_MSG + "Not all nodes in the cluster support restore operation.")); + } + + snpMgr.collectSnapshotMetadata(snpName).listen( + f -> { + if (f.error() != null) { + fut.onDone(f.error()); + + return; + } + + Set<UUID> dataNodes = new HashSet<>(); + Map<ClusterNode, List<SnapshotMetadata>> metas = f.result(); + Map<Integer, String> reqGrpIds = cacheGrpNames.stream().collect(Collectors.toMap(CU::cacheId, v -> v)); + + for (Map.Entry<ClusterNode, List<SnapshotMetadata>> entry : metas.entrySet()) { + SnapshotMetadata meta = F.first(entry.getValue()); + + assert meta != null : entry.getKey().id(); + + if (!entry.getKey().consistentId().equals(meta.consistentId())) + continue; + + dataNodes.add(entry.getKey().id()); + + reqGrpIds.keySet().removeAll(meta.partitions().keySet()); + } + + if (!reqGrpIds.isEmpty()) { + fut.onDone(new IllegalArgumentException(OP_REJECT_MSG + "Cache group(s) was not found in the " + + "snapshot [groups=" + reqGrpIds.values() + ", snapshot=" + snpName + ']')); + + return; + } + + snpMgr.runSnapshotVerfification(metas).listen( + f0 -> { + if (f0.error() != null) { + fut.onDone(f0.error()); + + return; + } + + IdleVerifyResultV2 res = f0.result(); + + if (!F.isEmpty(res.exceptions()) || res.hasConflicts()) { + StringBuilder sb = new StringBuilder(); + + res.print(sb::append, true); + + fut.onDone(new IgniteException(sb.toString())); + + return; + } + + SnapshotRestorePrepareRequest req = new SnapshotRestorePrepareRequest(UUID.randomUUID(), + snpName, dataNodes, cacheGrpNames, F.first(dataNodes)); + + prepareRestoreProc.start(req.requestId(), req); + } + ); + } + ); + + return new IgniteFutureImpl<>(fut); + } + + /** + * Check if snapshot restore process is currently running. + * + * @return {@code True} if the snapshot restore operation is in progress. + */ + public boolean isRestoring() { + return opCtx != null; + } + + /** + * Check if the cache or group with the specified name is currently being restored from the snapshot. + * + * @param cacheName Cache name. + * @param grpName Cache group name. + * @return {@code True} if the cache or group with the specified name is currently being restored. + */ + public boolean isRestoring(String cacheName, @Nullable String grpName) { + SnapshotRestoreContext opCtx0 = opCtx; + + if (opCtx0 == null) + return false; + + Map<Integer, StoredCacheData> cacheCfgs = opCtx0.cfgs; + + int cacheId = CU.cacheId(cacheName); + + if (cacheCfgs.containsKey(cacheId)) + return true; + + for (File grpDir : opCtx0.dirs) { + String locGrpName = FilePageStoreManager.cacheGroupName(grpDir); + + if (grpName != null) { + if (cacheName.equals(locGrpName)) + return true; + + if (CU.cacheId(locGrpName) == CU.cacheId(grpName)) + return true; + } + else if (CU.cacheId(locGrpName) == cacheId) + return true; + } + + return false; + } + + /** + * Finish local cache group restore process. + */ + private void finishProcess() { + finishProcess(null); + } + + /** + * Finish local cache group restore process. + * + * @param err Error, if any. + */ + private void finishProcess(@Nullable Throwable err) { + SnapshotRestoreContext opCtx0 = opCtx; + + if (err != null) { + log.error("Failed to restore snapshot cache group" + (opCtx0 == null ? "" : Review comment: Can you please fix the Intellij IDEA suggestions here? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
