timoninmaxim commented on code in PR #11897: URL: https://github.com/apache/ignite/pull/11897#discussion_r2307044928
########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java: ########## @@ -0,0 +1,752 @@ +/* + * 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.Serializable; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.TreeSet; +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import org.apache.ignite.IgniteException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.NodeStoppingException; +import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; +import org.apache.ignite.internal.management.cache.IdleVerifyResult; +import org.apache.ignite.internal.management.cache.PartitionKey; +import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree; +import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord; +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.tostring.GridToStringInclude; +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.S; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.jetbrains.annotations.Nullable; + +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CHECK_SNAPSHOT_METAS; +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CHECK_SNAPSHOT_PARTS; +import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.node2id; + +/** Distributed process of snapshot checking (with the partition hashes). */ +public class SnapshotCheckProcess { + /** */ + private final IgniteLogger log; + + /** */ + private final GridKernalContext kctx; + + /** Snapshot checker. */ + private final SnapshotChecker snpChecker; + + /** Operation contexts by name. */ + private final Map<String, SnapshotCheckContext> contexts = new ConcurrentHashMap<>(); + + /** Cluster-wide operation futures per snapshot called from current node. */ + private final Map<UUID, GridFutureAdapter<SnapshotPartitionsVerifyTaskResult>> clusterOpFuts = new ConcurrentHashMap<>(); + + /** Check metas first phase subprocess. */ + private final DistributedProcess<SnapshotCheckProcessRequest, SnapshotCheckResponse> phase1CheckMetas; + + /** Partition hashes second phase subprocess. */ + private final DistributedProcess<SnapshotCheckProcessRequest, SnapshotCheckResponse> phase2PartsHashes; + + /** Stop node lock. */ + private boolean nodeStopping; + + /** */ + public SnapshotCheckProcess(GridKernalContext kctx) { + this.kctx = kctx; + + snpChecker = new SnapshotChecker(kctx); + + log = kctx.log(getClass()); + + phase1CheckMetas = new DistributedProcess<>(kctx, CHECK_SNAPSHOT_METAS, this::prepareAndCheckMetas, + this::reducePreparationAndMetasCheck); + + phase2PartsHashes = new DistributedProcess<>(kctx, CHECK_SNAPSHOT_PARTS, this::validateParts, + this::reduceValidatePartsAndFinish); + } + + /** + * Stops all the processes with the passed exception. + * + * @param err The interrupt reason. + */ + void interrupt(Throwable err) { + // Prevents starting new processes in #prepareAndCheckMetas. + synchronized (contexts) { + nodeStopping = true; + } + + contexts.forEach((snpName, ctx) -> ctx.locProcFut.onDone(err)); + + clusterOpFuts.forEach((reqId, fut) -> fut.onDone(err)); + } + + /** Phase 2 and process finish. */ + private IgniteInternalFuture<?> reduceValidatePartsAndFinish( + UUID reqId, + Map<UUID, SnapshotCheckResponse> results, + Map<UUID, Throwable> errors + ) { + SnapshotCheckContext ctx = context(null, reqId); + + if (ctx == null) + return new GridFinishedFuture<>(); + + contexts.remove(ctx.req.snapshotName()); + + if (log.isInfoEnabled()) + log.info("Finished snapshot validation [req=" + ctx.req + ']'); + + GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> clusterOpFut = clusterOpFuts.get(reqId); + + if (clusterOpFut == null) + return new GridFinishedFuture<>(); + + if (F.isEmpty(errors)) { + ClusterTopologyCheckedException ex = checkNodeLeft(ctx.req.nodes(), results.keySet()); + + if (ex != null) { + clusterOpFut.onDone(ex); + + return new GridFinishedFuture<>(); + } + } + + if (ctx.req.incrementalIndex() > 0) { + SnapshotFileTree sft; + + if (ctx.locFileTree == null) + sft = null; + else if (kctx.config().getConsistentId() == null) + sft = null; + else + sft = ctx.locFileTree.get(kctx.config().getConsistentId().toString()); + + reduceIncrementalResults(sft, ctx.req.incrementalIndex(), ctx.req.nodes(), ctx.clusterMetas, results, errors, clusterOpFut); + } + else if (ctx.req.allRestoreHandlers()) + reduceCustomHandlersResults(ctx, results, errors, clusterOpFut); + else + reducePartitionsHashesResults(ctx.clusterMetas, results, errors, clusterOpFut); + + return new GridFinishedFuture<>(); + } + + /** */ + private void reduceIncrementalResults( + SnapshotFileTree sft, + int incIdx, + Set<UUID> requiredNodes, + Map<ClusterNode, List<SnapshotMetadata>> clusterMetas, + Map<UUID, SnapshotCheckResponse> results, + Map<UUID, Throwable> errors, + GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> fut + ) { + Map<ClusterNode, IncrementalSnapshotVerificationTaskResult> perNodeResults = new HashMap<>(); + + for (Map.Entry<UUID, SnapshotCheckResponse> resEntry : results.entrySet()) { + UUID nodeId = resEntry.getKey(); + + SnapshotCheckResponse incResp = resEntry.getValue(); + + if (incResp == null || !requiredNodes.contains(nodeId)) + continue; + + perNodeResults.put(kctx.cluster().get().node(nodeId), incResp.result()); + + if (F.isEmpty(incResp.exceptions())) + continue; + + errors.putIfAbsent(nodeId, asException(F.firstValue(incResp.exceptions()))); + } + + IdleVerifyResult chkRes = snpChecker.reduceIncrementalResults(sft, incIdx, perNodeResults, mapErrors(errors)); + + fut.onDone(new SnapshotPartitionsVerifyTaskResult(clusterMetas, chkRes)); + } + + /** */ + private void reduceCustomHandlersResults( + SnapshotCheckContext ctx, + Map<UUID, SnapshotCheckResponse> results, + Map<UUID, Throwable> errors, + GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> fut + ) { + try { + if (!errors.isEmpty()) + throw F.firstValue(errors); + + // Check responses: checking node -> snapshot part's consistent id -> handler name -> handler result. + Map<ClusterNode, Map<Object, Map<String, SnapshotHandlerResult<?>>>> reduced = new HashMap<>(); + + for (Map.Entry<UUID, SnapshotCheckResponse> respEntry : results.entrySet()) { + SnapshotCheckResponse nodeResp = respEntry.getValue(); + + if (nodeResp == null) + continue; + + if (!F.isEmpty(nodeResp.exceptions())) + throw F.firstValue(nodeResp.exceptions()); + + UUID nodeId = respEntry.getKey(); + + Map<String, Map<String, SnapshotHandlerResult<Object>>> cstHndRes = nodeResp.result(); + + cstHndRes.forEach((consId, respPerConsIdMap) -> { + // Reduced map of the handlers results per snapshot part's consistent id for certain node. + Map<Object, Map<String, SnapshotHandlerResult<?>>> nodePerConsIdResultMap + = reduced.computeIfAbsent(kctx.cluster().get().node(nodeId), n -> new HashMap<>()); + + respPerConsIdMap.forEach((hndId, hndRes) -> + nodePerConsIdResultMap.computeIfAbsent(consId, cstId -> new HashMap<>()).put(hndId, hndRes)); + }); + } + + snpChecker.checkCustomHandlersResults(ctx.req.snapshotName(), reduced); + + fut.onDone(new SnapshotPartitionsVerifyTaskResult(ctx.clusterMetas, null)); + } + catch (Throwable err) { + fut.onDone(err); + } + } + + /** */ + private void reducePartitionsHashesResults( + Map<ClusterNode, List<SnapshotMetadata>> clusterMetas, + Map<UUID, SnapshotCheckResponse> results, + Map<UUID, Throwable> errors, + GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> fut + ) { + IdleVerifyResult.Builder bldr = IdleVerifyResult.builder(); + + Map<ClusterNode, Exception> errors0 = mapErrors(errors); + + if (!results.isEmpty()) { + if (!errors0.isEmpty()) + bldr.exceptions(errors0); + + for (Map.Entry<UUID, SnapshotCheckResponse> respEntry : results.entrySet()) { + SnapshotCheckResponse resp = respEntry.getValue(); + + if (resp == null) + continue; + + if (!F.isEmpty(resp.exceptions())) { + ClusterNode node = kctx.cluster().get().node(respEntry.getKey()); + + bldr.addException(node, asException(F.firstValue(resp.exceptions()))); + } + + Map<String, Map<PartitionKey, PartitionHashRecord>> partsHashesRes = resp.result(); + + partsHashesRes.forEach((consId, partsPerConsId) -> bldr.addPartitionHashes(partsPerConsId)); + } + + fut.onDone(new SnapshotPartitionsVerifyTaskResult(clusterMetas, bldr.build())); + } + else + fut.onDone(new IgniteSnapshotVerifyException(errors0)); + } + + /** Phase 2 beginning. */ + private IgniteInternalFuture<SnapshotCheckResponse> validateParts(SnapshotCheckProcessRequest req) { + if (!req.nodes().contains(kctx.localNodeId())) + return new GridFinishedFuture<>(); + + SnapshotCheckContext ctx = context(req.snapshotName(), req.requestId()); + + assert ctx != null; + + if (F.isEmpty(ctx.metas)) + return new GridFinishedFuture<>(); + + GridFutureAdapter<SnapshotCheckResponse> phaseFut = ctx.phaseFuture(); + + // Might be already finished by asynchronous leave of a required node. + if (!phaseFut.isDone()) { + CompletableFuture<SnapshotCheckResponse> workingFut; + + if (req.incrementalIndex() > 0) { + assert !req.allRestoreHandlers() : "Snapshot handlers aren't supported for incremental snapshot."; + + workingFut = incrementalFuture(ctx); + } + else if (req.allRestoreHandlers()) + workingFut = allHandlersFuture(ctx); + else + workingFut = partitionsHashesFuture(ctx); + + workingFut.whenComplete((res, err) -> { + if (err != null) + phaseFut.onDone(err); + else + phaseFut.onDone(res); + }); + } + + return phaseFut; + } + + /** @return A composed future of increment checks for each consistent id regarding {@link SnapshotCheckContext#metas}. */ + private CompletableFuture<SnapshotCheckResponse> incrementalFuture(SnapshotCheckContext ctx) { + // Incremental snapshots do not support working on other topology. Only single meta and snapshot part can be processed. + SnapshotMetadata meta = ctx.metas.get(0); + + CompletableFuture<SnapshotCheckResponse> resFut = new CompletableFuture<>(); + + CompletableFuture<IncrementalSnapshotVerificationTaskResult> workingFut = snpChecker.checkIncrementalSnapshot( + ctx.locFileTree.get(meta.consistentId()), ctx.req.incrementalIndex()); + + workingFut.whenComplete((res, err) -> { + if (err != null) + resFut.completeExceptionally(err); + else + resFut.complete(new SnapshotCheckResponse(res, null)); + }); + + return resFut; + } + + /** @return A composed future of partitions checks for each consistent id regarding {@link SnapshotCheckContext#metas}. */ + private CompletableFuture<SnapshotCheckResponse> partitionsHashesFuture(SnapshotCheckContext ctx) { + // Per metas result: consistent id -> check results per partition key. + Map<String, Map<PartitionKey, PartitionHashRecord>> perMetaResults = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f); + // Per consistent id. + Map<String, Throwable> exceptions = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f); + CompletableFuture<SnapshotCheckResponse> composedFut = new CompletableFuture<>(); + AtomicInteger metasProcessed = new AtomicInteger(ctx.metas.size()); + + for (SnapshotMetadata meta : ctx.metas) { + CompletableFuture<Map<PartitionKey, PartitionHashRecord>> metaFut = snpChecker.checkPartitions( + meta, + ctx.locFileTree.get(meta.consistentId()), + ctx.req.groups(), + false, + ctx.req.fullCheck() + ); + + metaFut.whenComplete((res, err) -> { + if (err != null) + exceptions.put(meta.consistentId(), err); + else if (!F.isEmpty(res)) + perMetaResults.put(meta.consistentId(), res); + + if (metasProcessed.decrementAndGet() == 0) + composedFut.complete(new SnapshotCheckResponse(perMetaResults, exceptions)); + }); + } + + return composedFut; + } + + /** + * @return A composed future of all the snapshot handlers for each consistent id regarding {@link SnapshotCheckContext#metas}. + * @see IgniteSnapshotManager#handlers() + */ + private CompletableFuture<SnapshotCheckResponse> allHandlersFuture(SnapshotCheckContext ctx) { + // Per metas result: snapshot part's consistent id -> check result per handler name. + Map<String, Map<String, SnapshotHandlerResult<Object>>> perMetaResults = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f); + // Per consistent id. + Map<String, Throwable> exceptions = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f); + CompletableFuture<SnapshotCheckResponse> composedFut = new CompletableFuture<>(); + AtomicInteger metasProcessed = new AtomicInteger(ctx.metas.size()); + + for (SnapshotMetadata meta : ctx.metas) { + CompletableFuture<Map<String, SnapshotHandlerResult<Object>>> metaFut = snpChecker.invokeCustomHandlers(meta, + ctx.locFileTree.get(meta.consistentId()), ctx.req.groups(), ctx.req.fullCheck()); + + metaFut.whenComplete((res, err) -> { + if (err != null) + exceptions.put(meta.consistentId(), err); + else if (!F.isEmpty(res)) + perMetaResults.put(meta.consistentId(), res); + + if (metasProcessed.decrementAndGet() == 0) + composedFut.complete(new SnapshotCheckResponse(perMetaResults, exceptions)); + }); + } + + return composedFut; + } + + /** */ + private Map<ClusterNode, Exception> mapErrors(Map<UUID, Throwable> errors) { + return errors.entrySet().stream().collect(Collectors.toMap(e -> kctx.cluster().get().node(e.getKey()), + e -> asException(e.getValue()))); + } + + /** */ + private static Exception asException(Throwable th) { + return th instanceof Exception ? (Exception)th : new IgniteException(th); + } + + /** + * @param snpName Snapshot name. If {@code null}, ignored. + * @param reqId If {@code ctxId} is {@code null}, is used to find the operation context. + * @return Current snapshot checking context by {@code ctxId} or {@code reqId}. + */ + private @Nullable SnapshotCheckContext context(@Nullable String snpName, UUID reqId) { + return snpName == null + ? contexts.values().stream().filter(ctx0 -> ctx0.req.requestId().equals(reqId)).findFirst().orElse(null) + : contexts.get(snpName); + } + + /** Phase 1 beginning: prepare, collect and check local metas. */ + private IgniteInternalFuture<SnapshotCheckResponse> prepareAndCheckMetas(SnapshotCheckProcessRequest req) { + if (!req.nodes().contains(kctx.localNodeId()) && clusterOpFuts.get(req.requestId()) == null) + return new GridFinishedFuture<>(); + + SnapshotCheckContext ctx; + + synchronized (contexts) { + if (nodeStopping) + return new GridFinishedFuture<>(new NodeStoppingException("The node is stopping: " + kctx.localNodeId())); + + ctx = contexts.computeIfAbsent(req.snapshotName(), snpName -> new SnapshotCheckContext(req)); + } + + if (!ctx.req.requestId().equals(req.requestId())) { + return new GridFinishedFuture<>(new IllegalStateException("Validation of snapshot '" + req.snapshotName() + + "' has already started [ctx=" + ctx + ']')); + } + + // Excludes non-baseline initiator. + if (!baseline(kctx.localNodeId())) + return new GridFinishedFuture<>(); + + Collection<Integer> grpIds = F.isEmpty(req.groups()) ? null : F.viewReadOnly(req.groups(), CU::cacheId); + + GridFutureAdapter<SnapshotCheckResponse> phaseFut = ctx.phaseFuture(); + + // Might be already finished by asynchronous leave of a required node. + if (!phaseFut.isDone()) { + snpChecker.checkLocalMetas( + new SnapshotFileTree(kctx, req.snapshotName(), req.snapshotPath()), + req.incrementalIndex(), + grpIds + ).whenComplete((locMetas, err) -> { + if (err != null) + phaseFut.onDone(err); + else + phaseFut.onDone(new SnapshotCheckResponse(locMetas, null)); + }); + } + + return phaseFut; + } + + /** Phase 1 end. */ + private void reducePreparationAndMetasCheck( + UUID reqId, + Map<UUID, SnapshotCheckResponse> results, + Map<UUID, Throwable> errors + ) { + SnapshotCheckContext ctx = context(null, reqId); + + // The context is not stored in the case of concurrent check of the same snapshot but the operation future is registered. + GridFutureAdapter<SnapshotPartitionsVerifyTaskResult> clusterOpFut = clusterOpFuts.get(reqId); + + try { + if (!errors.isEmpty()) + throw new IgniteSnapshotVerifyException(mapErrors(errors)); + + if (ctx == null) { + assert clusterOpFut == null; + + return; + } + + ClusterTopologyCheckedException ex; + + if ((ex = checkNodeLeft(ctx.req.nodes(), results.keySet())) != null) + throw ex; + + Map<ClusterNode, List<SnapshotMetadata>> metas = new HashMap<>(); + + results.forEach((nodeId, nodeRes) -> { + // A node might be not required. It gives null result. But a required node might have invalid empty result + // which must be validated. + if (ctx.req.nodes().contains(nodeId) && baseline(nodeId) && !F.isEmpty((Collection<?>)nodeRes.result())) { + assert nodeRes != null; + + metas.put(kctx.cluster().get().node(nodeId), nodeRes.result()); + } + }); + + Map<ClusterNode, Exception> metasCheck = snpChecker.reduceMetasResults( + new SnapshotFileTree(kctx, ctx.req.snapshotName(), ctx.req.snapshotPath()), metas); + + if (!metasCheck.isEmpty()) + throw new IgniteSnapshotVerifyException(metasCheck); + + // If the topology is lesser that the snapshot's, we have to check another partitions parts. + ctx.metas = assingMetas(metas); + + if (!F.isEmpty(ctx.metas)) { + ctx.locFileTree = new HashMap<>(ctx.metas.size(), 1.0f); + + for (SnapshotMetadata metaToProc : ctx.metas) { + SnapshotFileTree sft = new SnapshotFileTree(kctx, ctx.req.snapshotName(), ctx.req.snapshotPath(), + metaToProc.folderName(), metaToProc.consistentId()); + + ctx.locFileTree.put(metaToProc.consistentId(), sft); + } + } + + if (clusterOpFut != null) + ctx.clusterMetas = metas; + + if (U.isLocalNodeCoordinator(kctx.discovery())) + phase2PartsHashes.start(reqId, ctx.req); + } + catch (Throwable th) { + if (ctx != null) { + contexts.remove(ctx.req.snapshotName()); + + if (log.isInfoEnabled()) Review Comment: Make message similar to the message written after second phase -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
