vldpyatkov commented on a change in pull request #7941: URL: https://github.com/apache/ignite/pull/7941#discussion_r502795743
########## File path: modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GroupKeyChangeProcess.java ########## @@ -0,0 +1,350 @@ +/* + * 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.managers.encryption; + +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import org.apache.ignite.IgniteException; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteFeatures; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.managers.encryption.GridEncryptionManager.EmptyResult; +import org.apache.ignite.internal.managers.encryption.GridEncryptionManager.KeyChangeFuture; +import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor; +import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor; +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.S; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.lang.IgniteFuture; +import org.apache.ignite.lang.IgniteFutureCancelledException; + +import static org.apache.ignite.internal.IgniteFeatures.CACHE_GROUP_KEY_CHANGE; +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CACHE_GROUP_KEY_CHANGE_FINISH; +import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CACHE_GROUP_KEY_CHANGE_PREPARE; + +/** + * A two-phase distributed process that rotates the encryption keys of specified cache groups and initiates + * re-encryption of those cache groups. + */ +class GroupKeyChangeProcess { + /** Grid kernal context. */ + private final GridKernalContext ctx; + + /** Cache group encyption key change prepare phase. */ + private final DistributedProcess<ChangeCacheEncryptionRequest, EmptyResult> prepareGKChangeProc; + + /** Cache group encyption key change perform phase. */ + private final DistributedProcess<ChangeCacheEncryptionRequest, EmptyResult> performGKChangeProc; + + /** Group encryption keys. */ + private final CacheGroupEncryptionKeys keys; + + /** Cache group key change future. */ + private volatile GroupKeyChangeFuture fut; + + /** Cache group key change request. */ + private volatile ChangeCacheEncryptionRequest req; + + /** + * @param ctx Grid kernal context. + */ + GroupKeyChangeProcess(GridKernalContext ctx, CacheGroupEncryptionKeys keys) { + this.ctx = ctx; + this.keys = keys; + + prepareGKChangeProc = + new DistributedProcess<>(ctx, CACHE_GROUP_KEY_CHANGE_PREPARE, this::prepare, this::finishPrepare); + performGKChangeProc = + new DistributedProcess<>(ctx, CACHE_GROUP_KEY_CHANGE_FINISH, this::perform, this::finishPerform); + } + + /** + * @return {@code True} if operation is still in progress. + */ + public boolean inProgress() { + return req != null; + } + + /** + * @param msg Error message. + */ + public void cancel(String msg) { + GridFutureAdapter<Void> keyChangeFut = fut; + + if (keyChangeFut != null && !keyChangeFut.isDone()) + keyChangeFut.onDone(new IgniteFutureCancelledException(msg)); + } + + /** + * Starts cache group encryption key change process. + * + * @param cacheOrGrpNames Cache or group names. + */ + public IgniteFuture<Void> start(Collection<String> cacheOrGrpNames) { + if (ctx.clientNode()) + throw new UnsupportedOperationException("Client and daemon nodes can not perform this operation."); + + if (!IgniteFeatures.allNodesSupports(ctx.grid().cluster().nodes(), CACHE_GROUP_KEY_CHANGE)) + throw new IllegalStateException("Not all nodes in the cluster support this operation."); + + if (!ctx.state().clusterState().state().active()) + throw new IgniteException("Operation was rejected. The cluster is inactive."); + + IgniteInternalFuture<Void> fut0 = fut; + + if (fut0 != null && !fut0.isDone()) { + return new IgniteFinishedFutureImpl<>(new IgniteException("Cache group key change was rejected. " + + "The previous change was not completed.")); + } + + int[] grpIds = new int[cacheOrGrpNames.size()]; + byte[] keyIds = new byte[grpIds.length]; + + int n = 0; + + for (String cacheOrGroupName : cacheOrGrpNames) { + CacheGroupDescriptor grpDesc = ctx.cache().cacheGroupDescriptor(CU.cacheId(cacheOrGroupName)); + + if (grpDesc == null) { + DynamicCacheDescriptor cacheDesc = ctx.cache().cacheDescriptor(cacheOrGroupName); + + if (cacheDesc == null) { + throw new IgniteException("Cache group key change was rejected. " + + "Cache or group \"" + cacheOrGroupName + "\" doesn't exists"); + } + + int grpId = cacheDesc.groupId(); + + grpDesc = ctx.cache().cacheGroupDescriptor(grpId); + + if (grpDesc.sharedGroup()) { + throw new IgniteException("Cache group key change was rejected. " + + "Cache or group \"" + cacheOrGroupName + "\" is a part of group \"" + + grpDesc.groupName() + "\". Provide group name instead of cache name for shared groups."); + } + } + + if (!grpDesc.config().isEncryptionEnabled()) { + throw new IgniteException("Cache group key change was rejected. " + + "Cache or group \"" + cacheOrGroupName + "\" is not encrypted."); + } + + if (ctx.encryption().reencryptionInProgress(grpDesc.groupId())) { + throw new IgniteException("Cache group key change was rejected. " + + "Cache group reencryption is in progress [grp=" + cacheOrGroupName + "]"); + } + + grpIds[n] = grpDesc.groupId(); + keyIds[n] = (byte)(ctx.encryption().groupKey(grpDesc.groupId()).unsignedId() + 1); + + n += 1; + } + + byte[][] keys = ctx.encryption().createKeys(grpIds.length).get1().toArray(new byte[grpIds.length][]); + + ChangeCacheEncryptionRequest req = + new ChangeCacheEncryptionRequest(grpIds, keys, keyIds, ctx.config().getEncryptionSpi().getMasterKeyName()); + + fut = new GroupKeyChangeFuture(req); + + prepareGKChangeProc.start(req.requestId(), req); + + return new IgniteFutureImpl<>(fut); + } + + /** + * Validates existing keys. + * + * @param req Request. + * @return Result future. + */ + private IgniteInternalFuture<EmptyResult> prepare(ChangeCacheEncryptionRequest req) { + if (ctx.clientNode()) + return new GridFinishedFuture<>(); + + if (inProgress()) { + return new GridFinishedFuture<>(new IgniteException("Cache group key change was rejected. " + + "The previous change was not completed.")); + } + + this.req = req; + + try { + for (int i = 0; i < req.groupIds().length; i++) { + int grpId = req.groupIds()[i]; + int keyId = req.keyIds()[i] & 0xff; + + if (ctx.encryption().reencryptionInProgress(grpId)) { + return new GridFinishedFuture<>(new IgniteException("Cache group key change was rejected. " + + "Cache group reencryption is in progress [grpId=" + grpId + "]")); + } + + List<Integer> keyIds = ctx.encryption().groupKeyIds(grpId); + + if (keyIds == null) { + return new GridFinishedFuture<>(new IgniteException("Cache group key change was rejected." + + "Encrypted cache group not found [grpId=" + grpId + "]")); + } + + GroupKey currKey = ctx.encryption().groupKey(grpId); + + for (int locKeyId : keyIds) { + if (locKeyId != keyId) + continue; + + Long walSegment = keys.reservedSegment(grpId, keyId); + + // Can overwrite inactive key if it was added during prepare phase. + if (walSegment == null && currKey.id() != (byte)keyId) + continue; + + return new GridFinishedFuture<>( + new IgniteException("Cache group key change was rejected. Cannot add new key identifier, " + + "it's already present. There existing WAL segments that encrypted with this key [" + + "grpId=" + grpId + ", newId=" + keyId + ", currId=" + currKey.unsignedId() + + ", walSegment=" + walSegment + "].")); + } + } + + return ctx.encryption().withMasterKeyChangeReadLock(() -> { + String curMasterKeyName = ctx.config().getEncryptionSpi().getMasterKeyName(); + + if (!curMasterKeyName.equals(req.masterKeyName())) { + return new GridFinishedFuture<>(new IgniteException("Cache group key change was rejected. " + + "Master key has been changed.")); + } + + for (int i = 0; i < req.groupIds().length; i++) { + // Store new key as inactive. + GroupKeyEncrypted grpKey = new GroupKeyEncrypted(req.keyIds()[i] & 0xff, req.keys()[i]); + + ctx.encryption().addGroupKey(req.groupIds()[i], grpKey); Review comment: Why we need to store the key here, we will still save it on the perform stage? ########## File path: modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java ########## @@ -430,6 +430,16 @@ private Process(UUID id) { * * @see IgniteSnapshotManager */ - END_SNAPSHOT + END_SNAPSHOT, + + /** + * Cache group encyption key change prepare phase. + */ + CACHE_GROUP_KEY_CHANGE_PREPARE, Review comment: If we often extend this enum, I recommend do not use native ordinal for it. Prefer to use enum with strongly fixed ordinal (like ShutdownPolicy) It will be able to lead to compatibility problems in the future. ########## File path: modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/CacheGroupPageScanner.java ########## @@ -0,0 +1,479 @@ +/* + * 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.managers.encryption; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.locks.ReentrantLock; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.configuration.DataStorageConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.NodeStoppingException; +import org.apache.ignite.internal.managers.communication.GridIoPolicy; +import org.apache.ignite.internal.pagemem.PageIdAllocator; +import org.apache.ignite.internal.pagemem.PageIdUtils; +import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager; +import org.apache.ignite.internal.processors.cache.CacheGroupContext; +import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager; +import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointListener; +import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryEx; +import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId; +import org.apache.ignite.internal.util.BasicRateLimiter; +import org.apache.ignite.internal.util.GridConcurrentHashSet; +import org.apache.ignite.internal.util.future.GridFinishedFuture; +import org.apache.ignite.internal.util.future.GridFutureAdapter; +import org.apache.ignite.internal.util.lang.IgniteInClosureX; +import org.apache.ignite.internal.util.typedef.X; +import org.apache.ignite.internal.util.typedef.internal.CU; +import org.apache.ignite.thread.IgniteThreadPoolExecutor; +import org.apache.ignite.thread.OomExceptionHandler; + +import static org.apache.ignite.internal.util.IgniteUtils.MB; + +/** + * Cache group page stores scanner. + * Scans a range of pages and marks them as dirty to re-encrypt them with the last encryption key on disk. + */ +public class CacheGroupPageScanner implements CheckpointListener { + /** Thread prefix for scanning tasks. */ + private static final String REENCRYPT_THREAD_PREFIX = "reencrypt"; + + /** Kernal context. */ + private final GridKernalContext ctx; + + /** Logger. */ + private final IgniteLogger log; + + /** Lock. */ + private final ReentrantLock lock = new ReentrantLock(); + + /** Mapping of cache group ID to group scanning task. */ + private final Map<Integer, GroupScanTask> grps = new ConcurrentHashMap<>(); + + /** Collection of groups waiting for a checkpoint. */ + private final Collection<GroupScanTask> cpWaitGrps = new ConcurrentLinkedQueue<>(); + + /** Page scanning speed limiter. */ + private final BasicRateLimiter limiter; + + /** Single-threaded executor to run cache group scan task. */ + private final ThreadPoolExecutor singleExecSvc; + + /** Number of pages that is scanned during reencryption under checkpoint lock. */ + private final int batchSize; + + /** Stop flag. */ + private boolean stopped; + + /** + * @param ctx Grid kernal context. + */ + public CacheGroupPageScanner(GridKernalContext ctx) { + this.ctx = ctx; + + log = ctx.log(getClass()); + + DataStorageConfiguration dsCfg = ctx.config().getDataStorageConfiguration(); + + if (!CU.isPersistenceEnabled(dsCfg)) { + batchSize = -1; + limiter = null; + singleExecSvc = null; + + return; + } + + double rateLimit = dsCfg.getEncryptionConfiguration().getReencryptionRateLimit(); + + limiter = rateLimit > 0 ? new BasicRateLimiter(rateLimit * MB / + (dsCfg.getPageSize() == 0 ? DataStorageConfiguration.DFLT_PAGE_SIZE : dsCfg.getPageSize())) : null; + + batchSize = dsCfg.getEncryptionConfiguration().getReencryptionBatchSize(); + + singleExecSvc = new IgniteThreadPoolExecutor(REENCRYPT_THREAD_PREFIX, + ctx.igniteInstanceName(), + 1, + 1, + IgniteConfiguration.DFLT_THREAD_KEEP_ALIVE_TIME, + new LinkedBlockingQueue<>(), + GridIoPolicy.SYSTEM_POOL, + new OomExceptionHandler(ctx)); + + singleExecSvc.allowCoreThreadTimeOut(true); + } + + /** {@inheritDoc} */ + @Override public void onCheckpointBegin(Context cpCtx) { + Set<GroupScanTask> completeCandidates = new HashSet<>(); + + cpWaitGrps.removeIf(completeCandidates::add); Review comment: The map (cpWaitGrps) does not synchronous with checkpoint process. ########## File path: modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/CacheGroupEncryptionKeys.java ########## @@ -0,0 +1,376 @@ +/* + * 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.managers.encryption; + +import java.io.Serializable; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.ListIterator; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CopyOnWriteArrayList; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.spi.encryption.EncryptionSpi; +import org.jetbrains.annotations.Nullable; + +/** + * Serves for managing encryption keys and related datastructure located in the heap. + */ +class CacheGroupEncryptionKeys { + /** Group encryption keys. */ + private final Map<Integer, List<GroupKey>> grpKeys = new ConcurrentHashMap<>(); + + /** + * WAL segments encrypted with previous encryption keys prevent keys from being deleted + * until the associated segment is deleted. + */ + private final Collection<TrackedWalSegment> trackedWalSegments = new ConcurrentLinkedQueue<>(); Review comment: What kind of WAL segment stored here? All segments which were encrypted or only one for one cache key. ########## File path: modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/CacheGroupPageScanner.java ########## @@ -0,0 +1,479 @@ +/* + * 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.managers.encryption; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.locks.ReentrantLock; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.configuration.DataStorageConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.NodeStoppingException; +import org.apache.ignite.internal.managers.communication.GridIoPolicy; +import org.apache.ignite.internal.pagemem.PageIdAllocator; +import org.apache.ignite.internal.pagemem.PageIdUtils; +import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager; +import org.apache.ignite.internal.processors.cache.CacheGroupContext; +import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager; +import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointListener; +import org.apache.ignite.internal.processors.cache.persistence.pagemem.PageMemoryEx; +import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId; +import org.apache.ignite.internal.util.BasicRateLimiter; +import org.apache.ignite.internal.util.GridConcurrentHashSet; +import org.apache.ignite.internal.util.future.GridFinishedFuture; +import org.apache.ignite.internal.util.future.GridFutureAdapter; +import org.apache.ignite.internal.util.lang.IgniteInClosureX; +import org.apache.ignite.internal.util.typedef.X; +import org.apache.ignite.internal.util.typedef.internal.CU; +import org.apache.ignite.thread.IgniteThreadPoolExecutor; +import org.apache.ignite.thread.OomExceptionHandler; + +import static org.apache.ignite.internal.util.IgniteUtils.MB; + +/** + * Cache group page stores scanner. + * Scans a range of pages and marks them as dirty to re-encrypt them with the last encryption key on disk. + */ +public class CacheGroupPageScanner implements CheckpointListener { + /** Thread prefix for scanning tasks. */ + private static final String REENCRYPT_THREAD_PREFIX = "reencrypt"; + + /** Kernal context. */ + private final GridKernalContext ctx; + + /** Logger. */ + private final IgniteLogger log; + + /** Lock. */ + private final ReentrantLock lock = new ReentrantLock(); + + /** Mapping of cache group ID to group scanning task. */ + private final Map<Integer, GroupScanTask> grps = new ConcurrentHashMap<>(); + + /** Collection of groups waiting for a checkpoint. */ + private final Collection<GroupScanTask> cpWaitGrps = new ConcurrentLinkedQueue<>(); + + /** Page scanning speed limiter. */ + private final BasicRateLimiter limiter; + + /** Single-threaded executor to run cache group scan task. */ + private final ThreadPoolExecutor singleExecSvc; + + /** Number of pages that is scanned during reencryption under checkpoint lock. */ + private final int batchSize; + + /** Stop flag. */ + private boolean stopped; + + /** + * @param ctx Grid kernal context. + */ + public CacheGroupPageScanner(GridKernalContext ctx) { + this.ctx = ctx; + + log = ctx.log(getClass()); + + DataStorageConfiguration dsCfg = ctx.config().getDataStorageConfiguration(); + + if (!CU.isPersistenceEnabled(dsCfg)) { + batchSize = -1; + limiter = null; + singleExecSvc = null; + + return; + } + + double rateLimit = dsCfg.getEncryptionConfiguration().getReencryptionRateLimit(); + + limiter = rateLimit > 0 ? new BasicRateLimiter(rateLimit * MB / + (dsCfg.getPageSize() == 0 ? DataStorageConfiguration.DFLT_PAGE_SIZE : dsCfg.getPageSize())) : null; + + batchSize = dsCfg.getEncryptionConfiguration().getReencryptionBatchSize(); + + singleExecSvc = new IgniteThreadPoolExecutor(REENCRYPT_THREAD_PREFIX, + ctx.igniteInstanceName(), + 1, + 1, + IgniteConfiguration.DFLT_THREAD_KEEP_ALIVE_TIME, + new LinkedBlockingQueue<>(), + GridIoPolicy.SYSTEM_POOL, + new OomExceptionHandler(ctx)); + + singleExecSvc.allowCoreThreadTimeOut(true); + } + + /** {@inheritDoc} */ + @Override public void onCheckpointBegin(Context cpCtx) { + Set<GroupScanTask> completeCandidates = new HashSet<>(); + + cpWaitGrps.removeIf(completeCandidates::add); + + cpCtx.finishedStateFut().listen( + f -> { + // Retry if error occurs. + if (f.error() != null || f.isCancelled()) { + cpWaitGrps.addAll(completeCandidates); + + return; + } + + lock.lock(); + + try { + for (GroupScanTask grpScanTask : completeCandidates) { + grps.remove(grpScanTask.groupId()); + + grpScanTask.onDone(); + + if (log.isInfoEnabled()) + log.info("Cache group reencryption is finished [grpId=" + grpScanTask.groupId() + "]"); + } + + if (!grps.isEmpty()) + return; + + ((GridCacheDatabaseSharedManager)ctx.cache().context().database()). + removeCheckpointListener(this); + } + finally { + lock.unlock(); + } + } + ); + } + + /** {@inheritDoc} */ + @Override public void beforeCheckpointBegin(Context cpCtx) { + // No-op. + } + + /** {@inheritDoc} */ + @Override public void onMarkCheckpointBegin(Context ctx) { + // No-op. + } + + /** + * Schedule scanning partitions. + * + * @param grpId Cache group ID. + */ + public IgniteInternalFuture<Void> schedule(int grpId) throws IgniteCheckedException { + CacheGroupContext grp = ctx.cache().cacheGroup(grpId); + + if (grp == null || !grp.affinityNode()) { + if (log.isInfoEnabled()) + log.info("Skip reencryption, cache group doesn't exist on the local node [grp=" + grpId + "]"); + + return new GridFinishedFuture<>(); + } + + lock.lock(); + + try { + if (stopped) + throw new NodeStoppingException("Operation has been cancelled (node is stopping)."); + + if (grps.isEmpty()) + ((GridCacheDatabaseSharedManager)ctx.cache().context().database()).addCheckpointListener(this); + + GroupScanTask prevState = grps.get(grpId); + + if (prevState != null && !prevState.isDone()) { + if (log.isDebugEnabled()) + log.debug("Reencryption already scheduled [grpId=" + grpId + "]"); + + return prevState; + } + + Set<Integer> parts = new HashSet<>(); + + forEachPageStore(grp, new IgniteInClosureX<Integer>() { + @Override public void applyx(Integer partId) { + if (ctx.encryption().getEncryptionState(grpId, partId) == 0) { + if (log.isDebugEnabled()) + log.debug("Skipping partition reencryption [grp=" + grpId + ", p=" + partId + "]"); + + return; + } + + parts.add(partId); + } + }); + + GroupScanTask grpScan = new GroupScanTask(grp, parts); + + singleExecSvc.submit(grpScan); + + if (log.isInfoEnabled()) + log.info("Scheduled reencryption [grpId=" + grpId + "]"); + + grps.put(grpId, grpScan); + + return grpScan; + } + finally { + lock.unlock(); + } + } + + /** + * @param grpId Cache group ID. + * @return Future that will be completed when all partitions have been scanned and pages have been written to disk. + */ + public IgniteInternalFuture<Void> statusFuture(int grpId) { + GroupScanTask grpScanTask = grps.get(grpId); + + return grpScanTask == null ? new GridFinishedFuture<>() : grpScanTask; + } + + /** + * Shutdown scanning and disable new tasks scheduling. + */ + public void stop() throws IgniteCheckedException { + lock.lock(); + + try { + stopped = true; + + for (GroupScanTask grpScanTask : grps.values()) + grpScanTask.cancel(); + + if (singleExecSvc != null) + singleExecSvc.shutdownNow(); + } finally { + lock.unlock(); + } + } + + /** + * Stop scannig the specified partition. + * + * @param grpId Cache group ID. + * @param partId Partition ID. + * @return {@code True} if reencryption was cancelled. + */ + public boolean excludePartition(int grpId, int partId) { + GroupScanTask grpScanTask = grps.get(grpId); + + if (grpScanTask == null) + return false; + + return grpScanTask.excludePartition(partId); + } + + /** + * Collect current number of pages in the specified cache group. + * + * @param grp Cache group. + * @return Partitions with current page count. + * @throws IgniteCheckedException If failed. + */ + public long[] pagesCount(CacheGroupContext grp) throws IgniteCheckedException { + // The last element of the array is used to store the status of the index partition. + long[] partStates = new long[grp.affinity().partitions() + 1]; + + ctx.cache().context().database().checkpointReadLock(); + + try { + forEachPageStore(grp, new IgniteInClosureX<Integer>() { + @Override public void applyx(Integer partId) throws IgniteCheckedException { + int pagesCnt = ctx.cache().context().pageStore().pages(grp.groupId(), partId); + + partStates[Math.min(partId, partStates.length - 1)] = pagesCnt; + } + }); + } finally { + ctx.cache().context().database().checkpointReadUnlock(); + } + + return partStates; + } + + /** + * @param grp Cache group. + * @param hnd Partition handler. + */ + private void forEachPageStore(CacheGroupContext grp, IgniteInClosureX<Integer> hnd) throws IgniteCheckedException { + int parts = grp.affinity().partitions(); + + IgnitePageStoreManager pageStoreMgr = ctx.cache().context().pageStore(); + + for (int p = 0; p < parts; p++) { + if (!pageStoreMgr.exists(grp.groupId(), p)) + continue; + + hnd.applyx(p); + } + + hnd.applyx(PageIdAllocator.INDEX_PARTITION); + } + + /** + * Cache group partition scanning task. + */ + private class GroupScanTask extends GridFutureAdapter<Void> implements Runnable { + /** Cache group ID. */ + private final CacheGroupContext grp; + + /** Partition IDs. */ + private final Set<Integer> parts; + + /** Page memory. */ + private final PageMemoryEx pageMem; + + /** + * @param grp Cache group. + */ + public GroupScanTask(CacheGroupContext grp, Set<Integer> parts) { + this.grp = grp; + this.parts = new GridConcurrentHashSet<>(parts); + + pageMem = (PageMemoryEx)grp.dataRegion().pageMemory(); + } + + /** {@inheritDoc} */ + @Override public synchronized boolean cancel() throws IgniteCheckedException { + return onCancelled(); + } + + /** + * Stop reencryption of the specified partition. + * + * @param partId Partition ID. + * @return {@code True} if reencryption was cancelled. + */ + public synchronized boolean excludePartition(int partId) { + return parts.remove(partId); + } + + /** + * @return Cache group ID. + */ + public int groupId() { + return grp.groupId(); + } + + /** {@inheritDoc} */ + @Override public void run() { + try { + for (int partId : parts) { + long state = ctx.encryption().getEncryptionState(grp.groupId(), partId); + + if (state == 0) + continue; + + scanPartition(partId, ReencryptStateUtils.pageIndex(state), ReencryptStateUtils.pageCount(state)); + + if (isDone()) + return; + } + + boolean added = cpWaitGrps.add(this); Review comment: cpWaitGrps is a list of those group which were check pointed. How do you guaranty that all pages were written to disk? ---------------------------------------------------------------- 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: [email protected]
