devmadhuu commented on code in PR #11855:
URL: https://github.com/apache/ozone/pull/11855#discussion_r4424727885
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/BucketEndpoint.java:
##########
@@ -63,10 +69,21 @@ public Response getBuckets(
) throws IOException {
List<OmBucketInfo> buckets = omMetadataManager.listBucketsUnderVolume(
volume, prevKey, limit);
+ Map<String, BucketDeletedBytesMetrics> trappedBytesMetrics =
+ bucketDeletedBytesMetricsService.getPerBucketMetrics();
Review Comment:
This is quite expensive. A paginated bucket request calls
`getPerBucketMetrics()`, which starts a full collection. The fixed pool limits
running threads, but its queue is unbounded. Imagine, With 10,000 buckets and
10,000 snapshots, a collection can perform approximately 100 million
snapshot-entry checks, before accounting for deleted-key and directory
traversal. Requesting just 100 buckets can trigger this work.
Can ordinary bucket listing read cached metrics without synchronously
enumerating every bucket and triggering a full-cluster OM scan? Please move
enumeration off the request thread, bound outstanding tasks, and control
collection independently of pagination. The current per-bucket OM calculator
traverses the global snapshot chain, making repeated full collections expensive
at scale.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/BucketDeletedBytesMetricsService.java:
##########
@@ -0,0 +1,419 @@
+/*
+ * 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.recon.api;
+
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY_DEFAULT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT_DEFAULT;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.TableIterator;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import
org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesCompleteResponse;
+import org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesMetrics;
+import
org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesProgressResponse;
+import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
+import org.apache.hadoop.ozone.recon.spi.OzoneManagerServiceProvider;
+import org.apache.hadoop.ozone.recon.tasks.BucketDeletedBytesCollectionTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Service for collecting and managing the snapshot-trapped vs purgeable
+ * deleted-bytes split across all buckets. Collects metrics asynchronously
+ * from OM, one RPC per bucket, and provides aggregated and per-bucket results.
+ */
+@Singleton
+public class BucketDeletedBytesMetricsService {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(BucketDeletedBytesMetricsService.class);
+ private static final int POLL_INTERVAL_MS = 200;
+
+ private final ExecutorService executorService;
+ private final ReconOMMetadataManager omMetadataManager;
+ private final OzoneManagerServiceProvider ozoneManagerServiceProvider;
+ private final int minimumApiDelayMs;
+ private final int maximumTaskTimeout;
+ private final AtomicBoolean isRunning = new AtomicBoolean(false);
+
+ private volatile MetricCollectionStatus currentStatus =
MetricCollectionStatus.NOT_STARTED;
+ private volatile String failedMessage = "Metrics collection task failed.
Please retry after some time.";
+ private Map<String, BucketDeletedBytesMetrics> perBucketMetrics = new
HashMap<>();
+ private long totalPurgeableSize = 0L;
+ private long totalSnapshotTrappedSize = 0L;
+ private long totalPurgeableKeys = 0L;
+ private long totalPurgeableDirs = 0L;
+ private long totalPurgeableKeySize = 0L;
+ private long totalPurgeableDirSize = 0L;
+ private int totalBucketsQueried;
+ private int totalBucketQueriesFailed;
+ private AtomicLong lastCollectionEndTime = new AtomicLong(0L);
+
+ @Inject
+ public BucketDeletedBytesMetricsService(
+ ReconOMMetadataManager omMetadataManager,
+ OzoneManagerServiceProvider ozoneManagerServiceProvider,
+ OzoneConfiguration config) {
+
+ this.omMetadataManager = omMetadataManager;
+ this.ozoneManagerServiceProvider = ozoneManagerServiceProvider;
+ this.minimumApiDelayMs = (int) config.getTimeDuration(
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY_DEFAULT,
+ TimeUnit.MILLISECONDS);
+ this.maximumTaskTimeout = (int) config.getTimeDuration(
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT_DEFAULT,
TimeUnit.MILLISECONDS);
+ this.lastCollectionEndTime.set(-minimumApiDelayMs);
+ int corePoolSize =
config.getInt(OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT);
+ corePoolSize = corePoolSize > 0
+ ? corePoolSize
+ : OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT;
+ ThreadFactory threadFactory = new ThreadFactoryBuilder()
+ .setNameFormat("BucketDeletedBytesCollector-%d")
+ .build();
+ this.executorService = Executors.newFixedThreadPool(corePoolSize,
threadFactory);
+ }
+
+ /**
+ * Starts the metrics collection task if not already running and rate limit
allows.
+ */
+ public void startTask() {
+ if (!isRunning.compareAndSet(false, true)) {
+ LOG.warn("Bucket deleted bytes collection already in progress,
skipping");
+ return;
+ }
+
+ if (System.currentTimeMillis() - lastCollectionEndTime.get() <
minimumApiDelayMs) {
+ LOG.debug("Rate limit active, skipping collection (delay: {}ms)",
minimumApiDelayMs);
+ isRunning.set(false);
+ return;
+ }
+
+ List<OmBucketInfo> buckets;
+ try {
+ buckets = listAllBuckets();
+ } catch (IOException | RuntimeException e) {
+ // Any failure here must release isRunning, otherwise every later
request is skipped as "in progress".
+ LOG.error("Failed to list buckets for deleted bytes collection", e);
+ resetState();
+ currentStatus = MetricCollectionStatus.FAILED;
+ failedMessage = e.getLocalizedMessage();
+ isRunning.set(false);
+ return;
+ }
+
+ if (buckets.isEmpty()) {
+ LOG.warn("No buckets found to query");
+ resetState();
+ currentStatus = MetricCollectionStatus.FINISHED;
+ isRunning.set(false);
+ return;
+ }
+
+ currentStatus = MetricCollectionStatus.IN_PROGRESS;
+ LOG.debug("Starting deleted bytes collection for {} buckets",
buckets.size());
+
+ CompletableFuture.runAsync(() -> collectMetrics(buckets), executorService)
+ .exceptionally(throwable -> {
+ LOG.error("Bucket deleted bytes collection failed", throwable);
+ synchronized (BucketDeletedBytesMetricsService.this) {
+ resetState();
+ currentStatus = MetricCollectionStatus.FAILED;
+ failedMessage = throwable.getLocalizedMessage();
+ isRunning.set(false);
+ }
+ return null;
+ });
+ }
+
+ private List<OmBucketInfo> listAllBuckets() throws IOException {
Review Comment:
How are pending deleted keys accounted for after their bucket has been
deleted? Enumerating only `bucketTable` excludes this valid state. Please
preserve accounting for deleted-key entries belonging to removed buckets and
cover “`delete keys`, `delete bucket`, `cleanup still pending`.” The previous
deleted-key total used statistics for the entire deleted-key table, so it
included these entries. Now this logic will introduce regression.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/BucketDeletedBytesMetricsService.java:
##########
@@ -0,0 +1,419 @@
+/*
+ * 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.recon.api;
+
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY_DEFAULT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT_DEFAULT;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.TableIterator;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import
org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesCompleteResponse;
+import org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesMetrics;
+import
org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesProgressResponse;
+import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
+import org.apache.hadoop.ozone.recon.spi.OzoneManagerServiceProvider;
+import org.apache.hadoop.ozone.recon.tasks.BucketDeletedBytesCollectionTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Service for collecting and managing the snapshot-trapped vs purgeable
+ * deleted-bytes split across all buckets. Collects metrics asynchronously
+ * from OM, one RPC per bucket, and provides aggregated and per-bucket results.
+ */
+@Singleton
+public class BucketDeletedBytesMetricsService {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(BucketDeletedBytesMetricsService.class);
+ private static final int POLL_INTERVAL_MS = 200;
+
+ private final ExecutorService executorService;
+ private final ReconOMMetadataManager omMetadataManager;
+ private final OzoneManagerServiceProvider ozoneManagerServiceProvider;
+ private final int minimumApiDelayMs;
+ private final int maximumTaskTimeout;
+ private final AtomicBoolean isRunning = new AtomicBoolean(false);
+
+ private volatile MetricCollectionStatus currentStatus =
MetricCollectionStatus.NOT_STARTED;
+ private volatile String failedMessage = "Metrics collection task failed.
Please retry after some time.";
+ private Map<String, BucketDeletedBytesMetrics> perBucketMetrics = new
HashMap<>();
+ private long totalPurgeableSize = 0L;
+ private long totalSnapshotTrappedSize = 0L;
+ private long totalPurgeableKeys = 0L;
+ private long totalPurgeableDirs = 0L;
+ private long totalPurgeableKeySize = 0L;
+ private long totalPurgeableDirSize = 0L;
+ private int totalBucketsQueried;
+ private int totalBucketQueriesFailed;
+ private AtomicLong lastCollectionEndTime = new AtomicLong(0L);
+
+ @Inject
+ public BucketDeletedBytesMetricsService(
+ ReconOMMetadataManager omMetadataManager,
+ OzoneManagerServiceProvider ozoneManagerServiceProvider,
+ OzoneConfiguration config) {
+
+ this.omMetadataManager = omMetadataManager;
+ this.ozoneManagerServiceProvider = ozoneManagerServiceProvider;
+ this.minimumApiDelayMs = (int) config.getTimeDuration(
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY_DEFAULT,
+ TimeUnit.MILLISECONDS);
+ this.maximumTaskTimeout = (int) config.getTimeDuration(
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT_DEFAULT,
TimeUnit.MILLISECONDS);
+ this.lastCollectionEndTime.set(-minimumApiDelayMs);
+ int corePoolSize =
config.getInt(OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT);
+ corePoolSize = corePoolSize > 0
+ ? corePoolSize
+ : OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT;
+ ThreadFactory threadFactory = new ThreadFactoryBuilder()
+ .setNameFormat("BucketDeletedBytesCollector-%d")
+ .build();
+ this.executorService = Executors.newFixedThreadPool(corePoolSize,
threadFactory);
+ }
+
+ /**
+ * Starts the metrics collection task if not already running and rate limit
allows.
+ */
+ public void startTask() {
+ if (!isRunning.compareAndSet(false, true)) {
+ LOG.warn("Bucket deleted bytes collection already in progress,
skipping");
+ return;
+ }
+
+ if (System.currentTimeMillis() - lastCollectionEndTime.get() <
minimumApiDelayMs) {
+ LOG.debug("Rate limit active, skipping collection (delay: {}ms)",
minimumApiDelayMs);
+ isRunning.set(false);
+ return;
+ }
+
+ List<OmBucketInfo> buckets;
+ try {
+ buckets = listAllBuckets();
+ } catch (IOException | RuntimeException e) {
+ // Any failure here must release isRunning, otherwise every later
request is skipped as "in progress".
+ LOG.error("Failed to list buckets for deleted bytes collection", e);
+ resetState();
+ currentStatus = MetricCollectionStatus.FAILED;
+ failedMessage = e.getLocalizedMessage();
+ isRunning.set(false);
+ return;
+ }
+
+ if (buckets.isEmpty()) {
+ LOG.warn("No buckets found to query");
+ resetState();
+ currentStatus = MetricCollectionStatus.FINISHED;
+ isRunning.set(false);
+ return;
+ }
+
+ currentStatus = MetricCollectionStatus.IN_PROGRESS;
+ LOG.debug("Starting deleted bytes collection for {} buckets",
buckets.size());
+
+ CompletableFuture.runAsync(() -> collectMetrics(buckets), executorService)
+ .exceptionally(throwable -> {
+ LOG.error("Bucket deleted bytes collection failed", throwable);
+ synchronized (BucketDeletedBytesMetricsService.this) {
+ resetState();
+ currentStatus = MetricCollectionStatus.FAILED;
+ failedMessage = throwable.getLocalizedMessage();
+ isRunning.set(false);
+ }
+ return null;
+ });
+ }
+
+ private List<OmBucketInfo> listAllBuckets() throws IOException {
+ List<OmBucketInfo> buckets = new ArrayList<>();
+ Table<String, OmBucketInfo> bucketTable =
omMetadataManager.getBucketTable();
+ if (bucketTable == null) {
+ // Recon's OM DB copy is not loaded yet, e.g. right after Recon startup.
+ throw new IOException("Recon OM DB is not initialized yet. Please retry
after some time.");
+ }
+ try (TableIterator<String, Table.KeyValue<String, OmBucketInfo>> iterator =
+ bucketTable.iterator()) {
+ while (iterator.hasNext()) {
+ OmBucketInfo omBucketInfo = iterator.next().getValue();
+ if (omBucketInfo != null) {
+ buckets.add(omBucketInfo);
+ }
+ }
+ }
+ return buckets;
+ }
+
+ private void collectMetrics(List<OmBucketInfo> buckets) {
+ try {
+ CollectionContext context = submitCollectionTasks(buckets);
+ processCollectionFutures(context);
+ updateFinalState(context);
+ } catch (Exception e) {
+ resetState();
+ currentStatus = MetricCollectionStatus.FAILED;
+ failedMessage = e.getLocalizedMessage();
+ isRunning.set(false);
+ }
+ }
+
+ private CollectionContext submitCollectionTasks(List<OmBucketInfo> buckets) {
+ Map<String, Future<BucketDeletedBytesMetrics>> futures = new HashMap<>();
+ long submissionTime = System.currentTimeMillis();
+ for (OmBucketInfo bucket : buckets) {
+ String key = bucket.getVolumeName() + "/" + bucket.getBucketName();
Review Comment:
Can we exclude linked buckets from cluster aggregation or deduplicate by the
resolved source bucket? `getBucketDeletedBytes()` resolves links in OM, so
collecting both a source bucket and its link adds the same data twice. Please
add coverage for a source bucket with one or more links.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/BucketDeletedBytesMetricsService.java:
##########
@@ -0,0 +1,419 @@
+/*
+ * 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.recon.api;
+
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY_DEFAULT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT_DEFAULT;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.TableIterator;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import
org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesCompleteResponse;
+import org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesMetrics;
+import
org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesProgressResponse;
+import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
+import org.apache.hadoop.ozone.recon.spi.OzoneManagerServiceProvider;
+import org.apache.hadoop.ozone.recon.tasks.BucketDeletedBytesCollectionTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Service for collecting and managing the snapshot-trapped vs purgeable
+ * deleted-bytes split across all buckets. Collects metrics asynchronously
+ * from OM, one RPC per bucket, and provides aggregated and per-bucket results.
+ */
+@Singleton
+public class BucketDeletedBytesMetricsService {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(BucketDeletedBytesMetricsService.class);
+ private static final int POLL_INTERVAL_MS = 200;
+
+ private final ExecutorService executorService;
+ private final ReconOMMetadataManager omMetadataManager;
+ private final OzoneManagerServiceProvider ozoneManagerServiceProvider;
+ private final int minimumApiDelayMs;
+ private final int maximumTaskTimeout;
+ private final AtomicBoolean isRunning = new AtomicBoolean(false);
+
+ private volatile MetricCollectionStatus currentStatus =
MetricCollectionStatus.NOT_STARTED;
+ private volatile String failedMessage = "Metrics collection task failed.
Please retry after some time.";
+ private Map<String, BucketDeletedBytesMetrics> perBucketMetrics = new
HashMap<>();
+ private long totalPurgeableSize = 0L;
+ private long totalSnapshotTrappedSize = 0L;
+ private long totalPurgeableKeys = 0L;
+ private long totalPurgeableDirs = 0L;
+ private long totalPurgeableKeySize = 0L;
+ private long totalPurgeableDirSize = 0L;
+ private int totalBucketsQueried;
+ private int totalBucketQueriesFailed;
+ private AtomicLong lastCollectionEndTime = new AtomicLong(0L);
+
+ @Inject
+ public BucketDeletedBytesMetricsService(
+ ReconOMMetadataManager omMetadataManager,
+ OzoneManagerServiceProvider ozoneManagerServiceProvider,
+ OzoneConfiguration config) {
+
+ this.omMetadataManager = omMetadataManager;
+ this.ozoneManagerServiceProvider = ozoneManagerServiceProvider;
+ this.minimumApiDelayMs = (int) config.getTimeDuration(
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY_DEFAULT,
+ TimeUnit.MILLISECONDS);
+ this.maximumTaskTimeout = (int) config.getTimeDuration(
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT_DEFAULT,
TimeUnit.MILLISECONDS);
+ this.lastCollectionEndTime.set(-minimumApiDelayMs);
+ int corePoolSize =
config.getInt(OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT);
+ corePoolSize = corePoolSize > 0
+ ? corePoolSize
+ : OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT;
+ ThreadFactory threadFactory = new ThreadFactoryBuilder()
+ .setNameFormat("BucketDeletedBytesCollector-%d")
+ .build();
+ this.executorService = Executors.newFixedThreadPool(corePoolSize,
threadFactory);
+ }
+
+ /**
+ * Starts the metrics collection task if not already running and rate limit
allows.
+ */
+ public void startTask() {
+ if (!isRunning.compareAndSet(false, true)) {
+ LOG.warn("Bucket deleted bytes collection already in progress,
skipping");
+ return;
+ }
+
+ if (System.currentTimeMillis() - lastCollectionEndTime.get() <
minimumApiDelayMs) {
+ LOG.debug("Rate limit active, skipping collection (delay: {}ms)",
minimumApiDelayMs);
+ isRunning.set(false);
+ return;
+ }
+
+ List<OmBucketInfo> buckets;
+ try {
+ buckets = listAllBuckets();
+ } catch (IOException | RuntimeException e) {
+ // Any failure here must release isRunning, otherwise every later
request is skipped as "in progress".
+ LOG.error("Failed to list buckets for deleted bytes collection", e);
+ resetState();
+ currentStatus = MetricCollectionStatus.FAILED;
+ failedMessage = e.getLocalizedMessage();
+ isRunning.set(false);
+ return;
+ }
+
+ if (buckets.isEmpty()) {
+ LOG.warn("No buckets found to query");
+ resetState();
+ currentStatus = MetricCollectionStatus.FINISHED;
+ isRunning.set(false);
+ return;
+ }
+
+ currentStatus = MetricCollectionStatus.IN_PROGRESS;
+ LOG.debug("Starting deleted bytes collection for {} buckets",
buckets.size());
+
+ CompletableFuture.runAsync(() -> collectMetrics(buckets), executorService)
+ .exceptionally(throwable -> {
+ LOG.error("Bucket deleted bytes collection failed", throwable);
+ synchronized (BucketDeletedBytesMetricsService.this) {
+ resetState();
+ currentStatus = MetricCollectionStatus.FAILED;
+ failedMessage = throwable.getLocalizedMessage();
+ isRunning.set(false);
+ }
+ return null;
+ });
+ }
+
+ private List<OmBucketInfo> listAllBuckets() throws IOException {
+ List<OmBucketInfo> buckets = new ArrayList<>();
+ Table<String, OmBucketInfo> bucketTable =
omMetadataManager.getBucketTable();
+ if (bucketTable == null) {
+ // Recon's OM DB copy is not loaded yet, e.g. right after Recon startup.
+ throw new IOException("Recon OM DB is not initialized yet. Please retry
after some time.");
+ }
+ try (TableIterator<String, Table.KeyValue<String, OmBucketInfo>> iterator =
+ bucketTable.iterator()) {
+ while (iterator.hasNext()) {
+ OmBucketInfo omBucketInfo = iterator.next().getValue();
+ if (omBucketInfo != null) {
+ buckets.add(omBucketInfo);
+ }
+ }
+ }
+ return buckets;
+ }
+
+ private void collectMetrics(List<OmBucketInfo> buckets) {
+ try {
+ CollectionContext context = submitCollectionTasks(buckets);
+ processCollectionFutures(context);
+ updateFinalState(context);
+ } catch (Exception e) {
+ resetState();
+ currentStatus = MetricCollectionStatus.FAILED;
+ failedMessage = e.getLocalizedMessage();
+ isRunning.set(false);
+ }
+ }
+
+ private CollectionContext submitCollectionTasks(List<OmBucketInfo> buckets) {
+ Map<String, Future<BucketDeletedBytesMetrics>> futures = new HashMap<>();
+ long submissionTime = System.currentTimeMillis();
+ for (OmBucketInfo bucket : buckets) {
+ String key = bucket.getVolumeName() + "/" + bucket.getBucketName();
+ BucketDeletedBytesCollectionTask task = new
BucketDeletedBytesCollectionTask(
+ bucket.getVolumeName(), bucket.getBucketName(),
ozoneManagerServiceProvider);
+ futures.put(key, executorService.submit(task));
Review Comment:
The coordinator and bucket requests use the same thread pool.
With:
`ozone.recon.bucket.deleted.bytes.collection.thread.count=1`
the execution becomes:
Only worker: coordinator waits for bucket requests
Queue: bucket requests wait for the only worker
Please run the coordinator separately from the bucket RPC executor. With
`thread.count=1`, the coordinator occupies the only worker while waiting for
tasks submitted to that same executor, so no bucket RPC runs. Add coverage for
a single-worker configuration.
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/BucketDeletedBytesMetricsService.java:
##########
@@ -0,0 +1,419 @@
+/*
+ * 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.recon.api;
+
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY_DEFAULT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT;
+import static
org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT_DEFAULT;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.TableIterator;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import
org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesCompleteResponse;
+import org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesMetrics;
+import
org.apache.hadoop.ozone.recon.api.types.BucketDeletedBytesProgressResponse;
+import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
+import org.apache.hadoop.ozone.recon.spi.OzoneManagerServiceProvider;
+import org.apache.hadoop.ozone.recon.tasks.BucketDeletedBytesCollectionTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Service for collecting and managing the snapshot-trapped vs purgeable
+ * deleted-bytes split across all buckets. Collects metrics asynchronously
+ * from OM, one RPC per bucket, and provides aggregated and per-bucket results.
+ */
+@Singleton
+public class BucketDeletedBytesMetricsService {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(BucketDeletedBytesMetricsService.class);
+ private static final int POLL_INTERVAL_MS = 200;
+
+ private final ExecutorService executorService;
+ private final ReconOMMetadataManager omMetadataManager;
+ private final OzoneManagerServiceProvider ozoneManagerServiceProvider;
+ private final int minimumApiDelayMs;
+ private final int maximumTaskTimeout;
+ private final AtomicBoolean isRunning = new AtomicBoolean(false);
+
+ private volatile MetricCollectionStatus currentStatus =
MetricCollectionStatus.NOT_STARTED;
+ private volatile String failedMessage = "Metrics collection task failed.
Please retry after some time.";
+ private Map<String, BucketDeletedBytesMetrics> perBucketMetrics = new
HashMap<>();
+ private long totalPurgeableSize = 0L;
+ private long totalSnapshotTrappedSize = 0L;
+ private long totalPurgeableKeys = 0L;
+ private long totalPurgeableDirs = 0L;
+ private long totalPurgeableKeySize = 0L;
+ private long totalPurgeableDirSize = 0L;
+ private int totalBucketsQueried;
+ private int totalBucketQueriesFailed;
+ private AtomicLong lastCollectionEndTime = new AtomicLong(0L);
+
+ @Inject
+ public BucketDeletedBytesMetricsService(
+ ReconOMMetadataManager omMetadataManager,
+ OzoneManagerServiceProvider ozoneManagerServiceProvider,
+ OzoneConfiguration config) {
+
+ this.omMetadataManager = omMetadataManager;
+ this.ozoneManagerServiceProvider = ozoneManagerServiceProvider;
+ this.minimumApiDelayMs = (int) config.getTimeDuration(
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_MINIMUM_API_DELAY_DEFAULT,
+ TimeUnit.MILLISECONDS);
+ this.maximumTaskTimeout = (int) config.getTimeDuration(
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_TIMEOUT_DEFAULT,
TimeUnit.MILLISECONDS);
+ this.lastCollectionEndTime.set(-minimumApiDelayMs);
+ int corePoolSize =
config.getInt(OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT,
+ OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT);
+ corePoolSize = corePoolSize > 0
+ ? corePoolSize
+ : OZONE_RECON_BUCKET_DELETED_BYTES_COLLECTION_THREAD_COUNT_DEFAULT;
+ ThreadFactory threadFactory = new ThreadFactoryBuilder()
+ .setNameFormat("BucketDeletedBytesCollector-%d")
+ .build();
+ this.executorService = Executors.newFixedThreadPool(corePoolSize,
threadFactory);
+ }
+
+ /**
+ * Starts the metrics collection task if not already running and rate limit
allows.
+ */
+ public void startTask() {
+ if (!isRunning.compareAndSet(false, true)) {
+ LOG.warn("Bucket deleted bytes collection already in progress,
skipping");
+ return;
+ }
+
+ if (System.currentTimeMillis() - lastCollectionEndTime.get() <
minimumApiDelayMs) {
+ LOG.debug("Rate limit active, skipping collection (delay: {}ms)",
minimumApiDelayMs);
+ isRunning.set(false);
+ return;
+ }
+
+ List<OmBucketInfo> buckets;
+ try {
+ buckets = listAllBuckets();
+ } catch (IOException | RuntimeException e) {
+ // Any failure here must release isRunning, otherwise every later
request is skipped as "in progress".
+ LOG.error("Failed to list buckets for deleted bytes collection", e);
+ resetState();
+ currentStatus = MetricCollectionStatus.FAILED;
+ failedMessage = e.getLocalizedMessage();
+ isRunning.set(false);
+ return;
+ }
+
+ if (buckets.isEmpty()) {
+ LOG.warn("No buckets found to query");
+ resetState();
+ currentStatus = MetricCollectionStatus.FINISHED;
+ isRunning.set(false);
+ return;
+ }
+
+ currentStatus = MetricCollectionStatus.IN_PROGRESS;
+ LOG.debug("Starting deleted bytes collection for {} buckets",
buckets.size());
+
+ CompletableFuture.runAsync(() -> collectMetrics(buckets), executorService)
+ .exceptionally(throwable -> {
+ LOG.error("Bucket deleted bytes collection failed", throwable);
+ synchronized (BucketDeletedBytesMetricsService.this) {
+ resetState();
+ currentStatus = MetricCollectionStatus.FAILED;
+ failedMessage = throwable.getLocalizedMessage();
+ isRunning.set(false);
+ }
+ return null;
+ });
+ }
+
+ private List<OmBucketInfo> listAllBuckets() throws IOException {
+ List<OmBucketInfo> buckets = new ArrayList<>();
+ Table<String, OmBucketInfo> bucketTable =
omMetadataManager.getBucketTable();
+ if (bucketTable == null) {
+ // Recon's OM DB copy is not loaded yet, e.g. right after Recon startup.
+ throw new IOException("Recon OM DB is not initialized yet. Please retry
after some time.");
+ }
+ try (TableIterator<String, Table.KeyValue<String, OmBucketInfo>> iterator =
+ bucketTable.iterator()) {
+ while (iterator.hasNext()) {
+ OmBucketInfo omBucketInfo = iterator.next().getValue();
+ if (omBucketInfo != null) {
+ buckets.add(omBucketInfo);
+ }
+ }
+ }
+ return buckets;
+ }
+
+ private void collectMetrics(List<OmBucketInfo> buckets) {
+ try {
+ CollectionContext context = submitCollectionTasks(buckets);
+ processCollectionFutures(context);
+ updateFinalState(context);
+ } catch (Exception e) {
+ resetState();
+ currentStatus = MetricCollectionStatus.FAILED;
+ failedMessage = e.getLocalizedMessage();
+ isRunning.set(false);
+ }
+ }
+
+ private CollectionContext submitCollectionTasks(List<OmBucketInfo> buckets) {
+ Map<String, Future<BucketDeletedBytesMetrics>> futures = new HashMap<>();
+ long submissionTime = System.currentTimeMillis();
+ for (OmBucketInfo bucket : buckets) {
+ String key = bucket.getVolumeName() + "/" + bucket.getBucketName();
+ BucketDeletedBytesCollectionTask task = new
BucketDeletedBytesCollectionTask(
+ bucket.getVolumeName(), bucket.getBucketName(),
ozoneManagerServiceProvider);
+ futures.put(key, executorService.submit(task));
+ }
+ int totalQueried = futures.size();
+ LOG.debug("Submitted {} bucket deleted bytes collection tasks",
totalQueried);
+ return new CollectionContext(totalQueried, futures, submissionTime);
+ }
+
+ private void processCollectionFutures(CollectionContext context) {
+ while (!context.futures.isEmpty()) {
+ long currentTime = System.currentTimeMillis();
+ Iterator<Map.Entry<String, Future<BucketDeletedBytesMetrics>>> iterator =
+ context.futures.entrySet().iterator();
+ boolean processedAny = false;
+ while (iterator.hasNext()) {
+ Map.Entry<String, Future<BucketDeletedBytesMetrics>> entry =
iterator.next();
+ String key = entry.getKey();
+ Future<BucketDeletedBytesMetrics> future = entry.getValue();
+ if (checkAndHandleTimeout(key, future, context, currentTime)) {
+ iterator.remove();
+ processedAny = true;
+ continue;
+ }
+ if (future.isDone()) {
+ handleCompletedFuture(key, future, context);
+ iterator.remove();
+ processedAny = true;
+ }
+ }
+ if (!context.futures.isEmpty() && !processedAny) {
+ try {
+ Thread.sleep(POLL_INTERVAL_MS);
+ } catch (InterruptedException e) {
+ LOG.warn("Collection polling interrupted");
+ Thread.currentThread().interrupt();
+ break;
+ }
+ }
+ }
+ }
+
+ private boolean checkAndHandleTimeout(
+ String key, Future<BucketDeletedBytesMetrics> future, CollectionContext
context, long currentTime) {
+ long elapsedTime = currentTime - context.submissionTime;
+ if (elapsedTime > maximumTaskTimeout && !future.isDone()) {
+ LOG.warn("Task for bucket {} timed out after {}ms", key, elapsedTime);
+ future.cancel(true);
+ context.failed++;
+ return true;
+ }
+ return false;
+ }
+
+ private void handleCompletedFuture(
+ String key, Future<BucketDeletedBytesMetrics> future, CollectionContext
context) {
+ try {
+ BucketDeletedBytesMetrics result = future.get();
+ if (result.getPurgeableBytes() < 0 && result.getSnapshotTrappedBytes() <
0) {
+ context.failed++;
+ } else {
+ context.totalPurgeableSize += Math.max(result.getPurgeableBytes(), 0L);
+ context.totalSnapshotTrappedSize +=
Math.max(result.getSnapshotTrappedBytes(), 0L);
+ context.totalPurgeableKeys += Math.max(result.getPurgeableKeys(), 0L);
+ context.totalPurgeableDirs += Math.max(result.getPurgeableDirs(), 0L);
+ // An OM that predates the key/dir bytes split reports -1; the split
total is then unknown.
+ if (result.getPurgeableKeyBytes() < 0 || result.getPurgeableDirBytes()
< 0) {
+ context.purgeableSplitUnknown = true;
+ } else {
+ context.totalPurgeableKeySize += result.getPurgeableKeyBytes();
+ context.totalPurgeableDirSize += result.getPurgeableDirBytes();
+ }
+ }
+ context.results.put(key, result);
+ LOG.debug("Processed deleted bytes result for bucket {}", key);
+ } catch (ExecutionException | InterruptedException e) {
+ String errorType = e instanceof InterruptedException ? "interrupted" :
"execution failed";
+ LOG.error("Task {} for bucket {} failed", errorType, key, e);
+ context.failed++;
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ private void updateFinalState(CollectionContext context) {
+ synchronized (this) {
+ perBucketMetrics = context.results;
+ totalPurgeableSize = context.totalPurgeableSize;
+ totalSnapshotTrappedSize = context.totalSnapshotTrappedSize;
+ totalPurgeableKeys = context.totalPurgeableKeys;
+ totalPurgeableDirs = context.totalPurgeableDirs;
+ totalPurgeableKeySize = context.purgeableSplitUnknown ? -1L :
context.totalPurgeableKeySize;
+ totalPurgeableDirSize = context.purgeableSplitUnknown ? -1L :
context.totalPurgeableDirSize;
+ totalBucketsQueried = context.totalQueried;
+ totalBucketQueriesFailed = context.failed;
+ currentStatus = MetricCollectionStatus.FINISHED;
Review Comment:
Can we explicitly distinguish complete results from partial or fully failed
collection? For example, if bucket A returns 100 GB and bucket B’s request
fails, we return FINISHED with 100 GB, although B’s contribution is unknown. If
every request fails, we return FINISHED with zero totals.
`totalBucketQueriesFailed` already exposes the failures, but please ensure
the UI uses it to show “incomplete” or “unavailable” instead of presenting
these values as complete cluster totals. Could we add coverage for the
all-requests-failed case?
--
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]
