HDDS-14010. [Recon] Endpoint to retrieve pending deletion metrics from DataNodes, SCM, and OM. (#9413)

Author: priyeshkaratha

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -3504,6 +3504,24 @@
       If the buffer overflows, task reinitialization will be triggered.
     </description>
   </property>
+  <property>
+    <name>ozone.recon.dn.metrics.collection.minimum.api.delay</name>
+    <value>30s</value>
+    <tag>OZONE, RECON, DN</tag>
+    <description>
+      Minimum delay in API to start a new task for Jmx collection.
+      It behaves like a rate limiter to avoid unnecessary task creation.
+    </description>
+  </property>
+  <property>
+    <name>ozone.recon.dn.metrics.collection.timeout</name>
+    <value>10m</value>
+    <tag>OZONE, RECON, DN</tag>
+    <description>
+      Maximum time taken for the api to complete.
+      If it exceeds pending tasks will be cancelled.
+    </description>
+  </property>
   <property>
     <name>ozone.scm.datanode.admin.monitor.interval</name>
     <value>30s</value>

hadoop-ozone/dist/src/main/compose/ozone/docker-config

@@ -58,7 +58,7 @@ OZONE-SITE.XML_ozone.http.basedir=/tmp/ozone_http
 OZONE-SITE.XML_hdds.container.ratis.datastream.enabled=true
 
 OZONE-SITE.XML_ozone.fs.hsync.enabled=true
-
+OZONE-SITE.XML_ozone.recon.dn.metrics.collection.minimum.api.delay=5s
 OZONE_CONF_DIR=/etc/hadoop
 OZONE_LOG_DIR=/var/log/hadoop
 

hadoop-ozone/integration-test-recon/src/test/java/org/apache/hadoop/ozone/recon/TestStorageDistributionEndpoint.java

@@ -22,19 +22,23 @@
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT;
 import static org.apache.hadoop.hdds.client.ReplicationFactor.THREE;
 import static org.apache.hadoop.hdds.client.ReplicationType.RATIS;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HA_DBTRANSACTIONBUFFER_FLUSH_INTERVAL;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HA_RATIS_SNAPSHOT_GAP;
 import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_INTERVAL;
-import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_TIMEOUT;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_DIR_DELETING_SERVICE_INTERVAL;
 import static org.apache.hadoop.ozone.recon.TestReconEndpointUtil.getReconWebAddress;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.FSDataOutputStream;
@@ -46,6 +50,8 @@
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.scm.ScmConfig;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.events.SCMEvents;
 import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
 import org.apache.hadoop.hdds.utils.IOUtils;
 import org.apache.hadoop.ozone.HddsDatanodeService;
@@ -66,6 +72,9 @@
 import org.apache.hadoop.ozone.om.helpers.OmKeyArgs;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup;
+import org.apache.hadoop.ozone.recon.api.DataNodeMetricsService;
+import org.apache.hadoop.ozone.recon.api.types.DataNodeMetricsServiceResponse;
+import org.apache.hadoop.ozone.recon.api.types.ScmPendingDeletion;
 import org.apache.hadoop.ozone.recon.api.types.StorageCapacityDistributionResponse;
 import org.apache.hadoop.ozone.recon.spi.impl.OzoneManagerServiceProviderImpl;
 import org.apache.ozone.test.GenericTestUtils;
@@ -100,6 +109,7 @@ public class TestStorageDistributionEndpoint {
   private static final ObjectMapper MAPPER = new ObjectMapper();
 
   private static final String STORAGE_DIST_ENDPOINT = "/api/v1/storageDistribution";
+  private static final String PENDING_DELETION_ENDPOINT = "/api/v1/pendingDeletion";
 
   static List<Arguments> replicationConfigs() {
     return Collections.singletonList(
@@ -110,17 +120,14 @@ static List<Arguments> replicationConfigs() {
   @BeforeAll
   public static void setup() throws Exception {
     conf = new OzoneConfiguration();
-    conf.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, 100,
-        TimeUnit.MILLISECONDS);
-    conf.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_TIMEOUT, 100,
-        TimeUnit.MILLISECONDS);
-    conf.setTimeDuration(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL,
-        100, TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(OZONE_DIR_DELETING_SERVICE_INTERVAL, 100, TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(OZONE_BLOCK_DELETING_SERVICE_INTERVAL, 100, TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL, 100, TimeUnit.MILLISECONDS);
     conf.setLong(OZONE_SCM_HA_RATIS_SNAPSHOT_GAP, 1L);
-    conf.setTimeDuration(HDDS_HEARTBEAT_INTERVAL, 50,
-        TimeUnit.MILLISECONDS);
-    conf.setTimeDuration(HDDS_CONTAINER_REPORT_INTERVAL, 200,
-        TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(HDDS_HEARTBEAT_INTERVAL, 50, TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(HDDS_CONTAINER_REPORT_INTERVAL, 200, TimeUnit.MILLISECONDS);
+    conf.setTimeDuration(OZONE_SCM_HA_DBTRANSACTIONBUFFER_FLUSH_INTERVAL, 500, TimeUnit.MILLISECONDS);
+    conf.set(ReconServerConfigKeys.OZONE_RECON_DN_METRICS_COLLECTION_MINIMUM_API_DELAY, "5s");
 
     // Enhanced SCM configuration for faster block deletion processing
     ScmConfig scmConfig = conf.getObject(ScmConfig.class);
@@ -129,18 +136,9 @@ public static void setup() throws Exception {
     conf.set(HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT, "0s");
 
     // Enhanced DataNode configuration to move pending deletion from SCM to DN faster
-    DatanodeConfiguration dnConf =
-        conf.getObject(DatanodeConfiguration.class);
-    dnConf.setBlockDeletionInterval(Duration.ofMillis(100));
-    // Increase block delete queue limit to allow more queued commands on DN
-    dnConf.setBlockDeleteQueueLimit(50);
-    // Reduce the interval for delete command worker processing
-    dnConf.setBlockDeleteCommandWorkerInterval(Duration.ofMillis(100));
-    // Increase blocks deleted per interval to speed up deletion
-    dnConf.setBlockDeletionLimit(5000);
+    DatanodeConfiguration dnConf = conf.getObject(DatanodeConfiguration.class);
+    dnConf.setBlockDeletionInterval(Duration.ofMillis(30000));
     conf.setFromObject(dnConf);
-    // Increase DN delete threads for faster parallel processing
-    conf.setInt("ozone.datanode.block.delete.threads.max", 10);
 
     recon = new ReconService(conf);
     cluster = MiniOzoneCluster.newBuilder(conf)
@@ -190,19 +188,133 @@ public void testStorageDistributionEndpoint(ReplicationConfig replicationConfig)
       }
     }
     waitForKeysCreated(replicationConfig);
-    Thread.sleep(10000);
-    StringBuilder urlBuilder = new StringBuilder();
-    urlBuilder.append(getReconWebAddress(conf))
-        .append(STORAGE_DIST_ENDPOINT);
-    String response = TestReconEndpointUtil.makeHttpCall(conf, urlBuilder);
-    StorageCapacityDistributionResponse storageResponse =
-        MAPPER.readValue(response, StorageCapacityDistributionResponse.class);
-
-    assertEquals(20, storageResponse.getGlobalNamespace().getTotalKeys());
-    assertEquals(60, storageResponse.getGlobalNamespace().getTotalUsedSpace());
-    assertEquals(0, storageResponse.getUsedSpaceBreakDown().getOpenKeyBytes());
-    assertEquals(60, storageResponse.getUsedSpaceBreakDown().getCommittedKeyBytes());
-    assertEquals(3, storageResponse.getDataNodeUsage().size());
+    GenericTestUtils.waitFor(this::verifyStorageDistributionAfterKeyCreation, 1000, 30000);
+    closeAllContainers();
+    fs.delete(dir1, true);
+    GenericTestUtils.waitFor(this::verifyPendingDeletionAfterKeyDeletionOm, 1000, 30000);
+    GenericTestUtils.waitFor(this::verifyPendingDeletionAfterKeyDeletionScm, 2000, 30000);
+    GenericTestUtils.waitFor(() ->
+            Objects.requireNonNull(scm.getClientProtocolServer().getDeletedBlockSummary()).getTotalBlockCount() == 0,
+        1000, 30000);
+    GenericTestUtils.waitFor(this::verifyPendingDeletionAfterKeyDeletionDn, 2000, 60000);
+    GenericTestUtils.waitFor(this::verifyPendingDeletionClearsAtDn, 2000, 60000);
+    cluster.getHddsDatanodes().get(0).stop();
+    GenericTestUtils.waitFor(this::verifyPendingDeletionAfterKeyDeletionOnDnFailure, 2000, 60000);
+  }
+
+  private boolean verifyStorageDistributionAfterKeyCreation() {
+    try {
+      StringBuilder urlBuilder = new StringBuilder();
+      urlBuilder.append(getReconWebAddress(conf)).append(STORAGE_DIST_ENDPOINT);
+      String response = TestReconEndpointUtil.makeHttpCall(conf, urlBuilder);
+      StorageCapacityDistributionResponse storageResponse =
+          MAPPER.readValue(response, StorageCapacityDistributionResponse.class);
+
+      assertEquals(20, storageResponse.getGlobalNamespace().getTotalKeys());
+      assertEquals(60, storageResponse.getGlobalNamespace().getTotalUsedSpace());
+      assertEquals(0, storageResponse.getUsedSpaceBreakDown().getOpenKeyBytes());
+      assertEquals(60, storageResponse.getUsedSpaceBreakDown().getCommittedKeyBytes());
+      assertEquals(3, storageResponse.getDataNodeUsage().size());
+
+      return true;
+    } catch (Exception e) {
+      LOG.debug("Waiting for storage distribution assertions to pass", e);
+      return false;
+    }
+  }
+
+  private boolean verifyPendingDeletionAfterKeyDeletionOm() {
+    try {
+      syncDataFromOM();
+      StringBuilder urlBuilder = new StringBuilder();
+      urlBuilder.append(getReconWebAddress(conf)).append(PENDING_DELETION_ENDPOINT).append("?component=om");
+      String response = TestReconEndpointUtil.makeHttpCall(conf, urlBuilder);
+      Map<String, Number> pendingDeletionMap = MAPPER.readValue(response, Map.class);
+      assertEquals(30L, pendingDeletionMap.get("totalSize").longValue());
+      assertEquals(30L, pendingDeletionMap.get("pendingDirectorySize").longValue() +
+          pendingDeletionMap.get("pendingKeySize").longValue());
+      return true;
+    } catch (Exception e) {
+      LOG.debug("Waiting for storage distribution assertions to pass", e);
+      return false;
+    }
+  }
+
+  private boolean verifyPendingDeletionAfterKeyDeletionScm() {
+    try {
+      StringBuilder urlBuilder = new StringBuilder();
+      urlBuilder.append(getReconWebAddress(conf)).append(PENDING_DELETION_ENDPOINT).append("?component=scm");
+      String response = TestReconEndpointUtil.makeHttpCall(conf, urlBuilder);
+      ScmPendingDeletion pendingDeletion = MAPPER.readValue(response, ScmPendingDeletion.class);
+      assertEquals(30, pendingDeletion.getTotalReplicatedBlockSize());
+      assertEquals(10, pendingDeletion.getTotalBlocksize());
+      assertEquals(10, pendingDeletion.getTotalBlocksCount());
+      return true;
+    } catch (Throwable e) {
+      LOG.debug("Waiting for storage distribution assertions to pass", e);
+      return false;
+    }
+  }
+
+  private boolean verifyPendingDeletionAfterKeyDeletionDn() {
+    try {
+      scm.getScmHAManager().asSCMHADBTransactionBuffer().flush();
+      StringBuilder urlBuilder = new StringBuilder();
+      urlBuilder.append(getReconWebAddress(conf)).append(PENDING_DELETION_ENDPOINT).append("?component=dn");
+      String response = TestReconEndpointUtil.makeHttpCall(conf, urlBuilder);
+      DataNodeMetricsServiceResponse pendingDeletion = MAPPER.readValue(response, DataNodeMetricsServiceResponse.class);
+      assertNotNull(pendingDeletion);
+      assertEquals(30, pendingDeletion.getTotalPendingDeletionSize());
+      assertEquals(DataNodeMetricsService.MetricCollectionStatus.FINISHED, pendingDeletion.getStatus());
+      assertEquals(pendingDeletion.getTotalNodesQueried(), pendingDeletion.getPendingDeletionPerDataNode().size());
+      assertEquals(0, pendingDeletion.getTotalNodeQueryFailures());
+      pendingDeletion.getPendingDeletionPerDataNode().forEach(dn -> {
+        assertEquals(10, dn.getPendingBlockSize());
+      });
+      return true;
+    } catch (Throwable e) {
+      LOG.debug("Waiting for storage distribution assertions to pass", e);
+      return false;
+    }
+  }
+
+  private boolean verifyPendingDeletionClearsAtDn() {
+    try {
+      scm.getScmHAManager().asSCMHADBTransactionBuffer().flush();
+      StringBuilder urlBuilder = new StringBuilder();
+      urlBuilder.append(getReconWebAddress(conf)).append(PENDING_DELETION_ENDPOINT).append("?component=dn");
+      String response = TestReconEndpointUtil.makeHttpCall(conf, urlBuilder);
+      DataNodeMetricsServiceResponse pendingDeletion = MAPPER.readValue(response, DataNodeMetricsServiceResponse.class);
+      assertNotNull(pendingDeletion);
+      assertEquals(0, pendingDeletion.getTotalPendingDeletionSize());
+      assertEquals(DataNodeMetricsService.MetricCollectionStatus.FINISHED, pendingDeletion.getStatus());
+      assertEquals(pendingDeletion.getTotalNodesQueried(), pendingDeletion.getPendingDeletionPerDataNode().size());
+      assertEquals(0, pendingDeletion.getTotalNodeQueryFailures());
+      pendingDeletion.getPendingDeletionPerDataNode().forEach(dn -> {
+        assertEquals(0, dn.getPendingBlockSize());
+      });
+      return true;
+    } catch (Throwable e) {
+      LOG.debug("Waiting for storage distribution assertions to pass", e);
+      return false;
+    }
+  }
+
+  private boolean verifyPendingDeletionAfterKeyDeletionOnDnFailure() {
+    try {
+      StringBuilder urlBuilder = new StringBuilder();
+      urlBuilder.append(getReconWebAddress(conf)).append(PENDING_DELETION_ENDPOINT).append("?component=dn");
+      String response = TestReconEndpointUtil.makeHttpCall(conf, urlBuilder);
+      DataNodeMetricsServiceResponse pendingDeletion = MAPPER.readValue(response, DataNodeMetricsServiceResponse.class);
+      assertNotNull(pendingDeletion);
+      assertEquals(1, pendingDeletion.getTotalNodeQueryFailures());
+      assertTrue(pendingDeletion.getPendingDeletionPerDataNode()
+          .stream()
+          .anyMatch(dn -> dn.getPendingBlockSize() == -1));
+      return true;
+    } catch (Throwable e) {
+      return false;
+    }
   }
 
   private void verifyBlocksCreated(
@@ -286,4 +398,12 @@ public static void tear() {
       cluster.shutdown();
     }
   }
+
+  private static void closeAllContainers() {
+    for (ContainerInfo container :
+        scm.getContainerManager().getContainers()) {
+      scm.getEventQueue().fireEvent(SCMEvents.CLOSE_CONTAINER,
+          container.containerID());
+    }
+  }
 }

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/MetricsServiceProviderFactory.java

@@ -17,12 +17,20 @@
 
 package org.apache.hadoop.ozone.recon;
 
+import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_METRICS_HTTP_CONNECTION_REQUEST_TIMEOUT;
+import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_METRICS_HTTP_CONNECTION_REQUEST_TIMEOUT_DEFAULT;
+import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_METRICS_HTTP_CONNECTION_TIMEOUT;
+import static org.apache.hadoop.ozone.recon.ReconServerConfigKeys.OZONE_RECON_METRICS_HTTP_CONNECTION_TIMEOUT_DEFAULT;
+
+import java.util.concurrent.TimeUnit;
 import javax.inject.Inject;
 import javax.inject.Singleton;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.hdds.recon.ReconConfigKeys;
+import org.apache.hadoop.hdfs.web.URLConnectionFactory;
 import org.apache.hadoop.ozone.recon.spi.MetricsServiceProvider;
+import org.apache.hadoop.ozone.recon.spi.impl.JmxServiceProviderImpl;
 import org.apache.hadoop.ozone.recon.spi.impl.PrometheusServiceProviderImpl;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -39,12 +47,23 @@ public class MetricsServiceProviderFactory {
 
   private OzoneConfiguration configuration;
   private ReconUtils reconUtils;
+  private URLConnectionFactory connectionFactory;
 
   @Inject
   public MetricsServiceProviderFactory(OzoneConfiguration configuration,
                                        ReconUtils reconUtils) {
     this.configuration = configuration;
     this.reconUtils = reconUtils;
+    int connectionTimeout = (int) configuration.getTimeDuration(
+        OZONE_RECON_METRICS_HTTP_CONNECTION_TIMEOUT,
+        OZONE_RECON_METRICS_HTTP_CONNECTION_TIMEOUT_DEFAULT,
+        TimeUnit.MILLISECONDS);
+    int connectionRequestTimeout = (int) configuration.getTimeDuration(
+        OZONE_RECON_METRICS_HTTP_CONNECTION_REQUEST_TIMEOUT,
+        OZONE_RECON_METRICS_HTTP_CONNECTION_REQUEST_TIMEOUT_DEFAULT,
+        TimeUnit.MILLISECONDS);
+    connectionFactory = URLConnectionFactory.newDefaultURLConnectionFactory(connectionTimeout,
+        connectionRequestTimeout, configuration);
   }
 
   /**
@@ -62,11 +81,20 @@ public MetricsServiceProvider getMetricsServiceProvider() {
             String.format("Choosing Prometheus as Metrics service provider " +
                 "with configured endpoint: %s", prometheusEndpoint));
       }
-      return new PrometheusServiceProviderImpl(configuration, reconUtils);
+      return new PrometheusServiceProviderImpl(configuration, reconUtils, connectionFactory);
     }
     return null;
   }
 
+  /**
+   * Returns the configured MetricsServiceProvider implementation for Jmx.
+   * @param endpoint
+   * @return MetricsServiceProvider instance for Jmx
+   */
+  public MetricsServiceProvider getJmxMetricsServiceProvider(String endpoint) {
+    return new JmxServiceProviderImpl(reconUtils, endpoint, connectionFactory);
+  }
+
   /**
    * Returns the Prometheus endpoint if configured. Otherwise returns null.
    *

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/ReconServerConfigKeys.java

@@ -123,14 +123,14 @@ public final class  ReconServerConfigKeys {
 
   public static final String
       OZONE_RECON_METRICS_HTTP_CONNECTION_TIMEOUT_DEFAULT =
-      "10s";
+      "30s";
 
   public static final String
       OZONE_RECON_METRICS_HTTP_CONNECTION_REQUEST_TIMEOUT =
       "ozone.recon.metrics.http.connection.request.timeout";
 
   public static final String
-      OZONE_RECON_METRICS_HTTP_CONNECTION_REQUEST_TIMEOUT_DEFAULT = "10s";
+      OZONE_RECON_METRICS_HTTP_CONNECTION_REQUEST_TIMEOUT_DEFAULT = "60s";
 
   public static final String OZONE_RECON_SCM_CONTAINER_THRESHOLD =
       "ozone.recon.scm.container.threshold";
@@ -213,6 +213,14 @@ public final class  ReconServerConfigKeys {
   public static final int
       OZONE_RECON_SCM_CLIENT_FAILOVER_MAX_RETRY_DEFAULT = 3;
 
+  public static final String OZONE_RECON_DN_METRICS_COLLECTION_MINIMUM_API_DELAY =
+      "ozone.recon.dn.metrics.collection.minimum.api.delay";
+  public static final String OZONE_RECON_DN_METRICS_COLLECTION_MINIMUM_API_DELAY_DEFAULT = "30s";
+
+  public static final String OZONE_RECON_DN_METRICS_COLLECTION_TIMEOUT =
+      "ozone.recon.dn.metrics.collection.timeout";
+  public static final String OZONE_RECON_DN_METRICS_COLLECTION_TIMEOUT_DEFAULT = "10m";
+
   /**
    * Private constructor for utility class.
    */