fmorg-git commented on code in PR #9468: URL: https://github.com/apache/ozone/pull/9468#discussion_r2674532529
########## hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/RevokedSTSTokenCleanupService.java: ########## @@ -0,0 +1,264 @@ +/* + * 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.hadoop.ozone.om.service; + +import com.google.common.annotations.VisibleForTesting; +import com.google.protobuf.ServiceException; +import java.io.IOException; +import java.time.Clock; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.hadoop.hdds.conf.StorageUnit; +import org.apache.hadoop.hdds.utils.BackgroundService; +import org.apache.hadoop.hdds.utils.BackgroundTask; +import org.apache.hadoop.hdds.utils.BackgroundTaskQueue; +import org.apache.hadoop.hdds.utils.BackgroundTaskResult; +import org.apache.hadoop.hdds.utils.db.Table; +import org.apache.hadoop.ozone.ClientVersion; +import org.apache.hadoop.ozone.om.OMConfigKeys; +import org.apache.hadoop.ozone.om.OMMetadataManager; +import org.apache.hadoop.ozone.om.OzoneManager; +import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.DeleteRevokedSTSTokensRequest; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type; +import org.apache.hadoop.util.Time; +import org.apache.ratis.protocol.ClientId; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Background service that periodically scans the revoked STS token table and submits OM requests to + * remove entries have been present past the cleanup threshold. + */ +public class RevokedSTSTokenCleanupService extends BackgroundService { + private static final Logger LOG = LoggerFactory.getLogger(RevokedSTSTokenCleanupService.class); + + // Use a single thread + private static final int REVOKED_STS_TOKEN_CLEANER_CORE_POOL_SIZE = 1; + private static final Clock CLOCK = Clock.system(ZoneOffset.UTC); + private static final long CLEANUP_THRESHOLD = 12 * 60 * 60 * 1000L; // 12 hours in milliseconds + + private final OzoneManager ozoneManager; + private final OMMetadataManager metadataManager; + private final AtomicBoolean suspended; + private final AtomicLong runCount; + private final AtomicLong submittedDeletedEntryCount; + // Dummy client ID to use for response, since this is triggered by a + // service, not the client. + private final ClientId clientId = ClientId.randomId(); + private final int ratisByteLimit; + + /** + * Creates a Revoked STS Token cleanup service. + * + * @param interval the interval between successive runs + * @param unit the time unit for {@code interval} + * @param serviceTimeout timeout for a single run + * @param ozoneManager the OzoneManager instance + */ + public RevokedSTSTokenCleanupService(long interval, TimeUnit unit, long serviceTimeout, OzoneManager ozoneManager) { + super( + "RevokedSTSTokenCleanupService", interval, unit, REVOKED_STS_TOKEN_CLEANER_CORE_POOL_SIZE, + serviceTimeout, ozoneManager.getThreadNamePrefix()); + this.ozoneManager = ozoneManager; + this.metadataManager = ozoneManager.getMetadataManager(); + this.suspended = new AtomicBoolean(false); + this.runCount = new AtomicLong(0); + this.submittedDeletedEntryCount = new AtomicLong(0); + int limit = (int) ozoneManager.getConfiguration().getStorageSize( + OMConfigKeys.OZONE_OM_RATIS_LOG_APPENDER_QUEUE_BYTE_LIMIT, + OMConfigKeys.OZONE_OM_RATIS_LOG_APPENDER_QUEUE_BYTE_LIMIT_DEFAULT, StorageUnit.BYTES); + // Always go to 90% of max limit for request as other header(s) will be added + this.ratisByteLimit = (int) (limit * 0.9); + } + + /** + * Returns the number of times this Background service has run. + * @return Long, run count. + */ + @VisibleForTesting + public long getRunCount() { + return runCount.get(); + } + + /** + * Returns the number of entries this Background service has submitted for deletion. + * @return Long, submitted for deletion entry count. + */ + @VisibleForTesting + public long getSubmittedDeletedEntryCount() { + return submittedDeletedEntryCount.get(); + } + + @Override + public BackgroundTaskQueue getTasks() { + final BackgroundTaskQueue queue = new BackgroundTaskQueue(); + queue.add(new RevokedSTSTokenCleanupTask()); + return queue; + } + + private boolean shouldRun() { + return !suspended.get() && ozoneManager.isLeaderReady(); + } + + private class RevokedSTSTokenCleanupTask implements BackgroundTask { + + @Override + public BackgroundTaskResult call() throws Exception { + if (!shouldRun()) { + return BackgroundTaskResult.EmptyTaskResult.newResult(); + } + + final long startTime = Time.monotonicNow(); + runCount.incrementAndGet(); + final Table<String, Long> revokedStsTokenTable = metadataManager.getS3RevokedStsTokenTable(); + + long deletedInRun = 0; + final List<String> batch = new ArrayList<>(); + + try (Table.KeyValueIterator<String, Long> iterator = revokedStsTokenTable.iterator()) { + iterator.seekToFirst(); + while (iterator.hasNext()) { + final Table.KeyValue<String, Long> entry = iterator.next(); + final String sessionToken = entry.getKey(); + final Long initialCreationTimeMillis = entry.getValue(); + + if (shouldCleanup(initialCreationTimeMillis)) { + // Calculate the size this token would add to the protobuf message. + // Make a copy of the batch to do the size check + final List<String> batchCopyWithCandidate = new ArrayList<>(batch); + batchCopyWithCandidate.add(sessionToken); + int batchWithCandidateSize = getBatchSerializedSize(batchCopyWithCandidate); + + // If adding this token would exceed the limit, submit the current batch + if (batchWithCandidateSize > ratisByteLimit) { + if (!batch.isEmpty()) { + if (submitCleanupRequest(batch)) { + deletedInRun += batch.size(); + } else { + LOG.warn("Failed to submit batch of {} revoked tokens.", batch.size()); + } + batch.clear(); + + // Re-calculate the size of the candidate token alone in an empty batch + // to check if it exceeds the limit by itself. + final List<String> singleCandidateBatch = new ArrayList<>(); + singleCandidateBatch.add(sessionToken); + batchWithCandidateSize = getBatchSerializedSize(singleCandidateBatch); + } + + // Check if the single token exceeds the limit (either strictly single or after flush) + if (batchWithCandidateSize > ratisByteLimit) { + LOG.error( + "Single revoked STS Token size ({}) would exceed the ratisByteLimit ({}). SessionToken " + + "initialCreationTimeMillis: {}", batchWithCandidateSize, ratisByteLimit, initialCreationTimeMillis); + continue; + } + } + batch.add(sessionToken); + } + } + } catch (IOException e) { + LOG.error("Failure while scanning s3RevokedStsTokenTable. It will be retried in the next interval", e); + if (deletedInRun == 0) { + return BackgroundTaskResult.EmptyTaskResult.newResult(); + } + } + + // Submit any remaining tokens + if (!batch.isEmpty()) { + if (submitCleanupRequest(batch)) { + deletedInRun += batch.size(); + } else { + LOG.warn("Failed to submit final batch of {} revoked tokens.", batch.size()); + } + } + + // Update stats + if (deletedInRun > 0) { + submittedDeletedEntryCount.addAndGet(deletedInRun); + LOG.info("Found and removed {} revoked STS token entries.", deletedInRun); + } + + final long elapsed = Time.monotonicNow() - startTime; + LOG.info("RevokedSTSTokenCleanupService run completed. deletedEntriesInRun={}, totalDeletedEntries={}, " + + "elapsedTimeMs={}", deletedInRun, submittedDeletedEntryCount.get(), elapsed); + + final long resultCount = deletedInRun; + return () -> (int) resultCount; + } + + /** + * Returns true if the given STS session token has been in the table past the cleanup threshold. + */ + private boolean shouldCleanup(long initialCreationTimeMillis) { + final long now = CLOCK.millis(); + + if (now - initialCreationTimeMillis > CLEANUP_THRESHOLD) { + if (LOG.isDebugEnabled()) { + LOG.debug( + "Revoked STS token entry created at {} is older than 12 hours, will clean up. Current time: {}", + initialCreationTimeMillis, now); + } + return true; + } + return false; + } + + /** + * Builds and submits an OMRequest to delete the provided revoked STS token(s). + */ + private boolean submitCleanupRequest(List<String> sessionTokens) { + final DeleteRevokedSTSTokensRequest request = DeleteRevokedSTSTokensRequest.newBuilder() + .addAllSessionToken(sessionTokens) + .build(); + + final OMRequest omRequest = OMRequest.newBuilder() + .setCmdType(Type.DeleteRevokedSTSTokens) + .setDeleteRevokedSTSTokensRequest(request) + .setClientId(clientId.toString()) + .setVersion(ClientVersion.CURRENT_VERSION) + .build(); + + try { + final OMResponse omResponse = OzoneManagerRatisUtils.submitRequest( + ozoneManager, omRequest, clientId, runCount.get()); Review Comment: oops, updated -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
