HDDS-15496. SCM async container ID export CLI

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java

@@ -641,6 +641,13 @@ public final class ScmConfigKeys {
       "ozone.scm.ratis.events.max.limit";
   public static final int OZONE_SCM_RATIS_EVENTS_MAX_LIMIT_DEFAULT = 100;
 
+  public static final String OZONE_SCM_CONTAINER_EXPORT_DIR =
+      "ozone.scm.container.export.dir";
+
+  public static final String OZONE_SCM_CONTAINER_EXPORT_MAX_TERMINAL_JOBS =
+      "ozone.scm.container.export.max.terminal.jobs";
+  public static final int OZONE_SCM_CONTAINER_EXPORT_MAX_TERMINAL_JOBS_DEFAULT = 10;
+
   /**
    * Never constructed.
    */

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportLimits.java

@@ -0,0 +1,34 @@
+/*
+ * 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.hdds.scm.container.export;
+
+/**
+ * Shared limits and defaults for container ID export (CLI and SCM server).
+ */
+public final class ContainerExportLimits {
+
+  public static final String EXPORT_SUBDIR = "exports";
+
+  public static final int DEFAULT_PAGE_SIZE = 100_000;
+  public static final int DEFAULT_SHARD_SIZE = 500_000;
+  public static final int MAX_PAGE_SIZE = 1_000_000;
+  public static final int MAX_SHARD_SIZE = 5_000_000;
+
+  private ContainerExportLimits() {
+  }
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportStatus.java

@@ -0,0 +1,91 @@
+/*
+ * 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.hdds.scm.container.export;
+
+/**
+ * Client-side view of a container ID export job on the SCM leader.
+ */
+public final class ContainerExportStatus {
+
+  private final String jobId;
+  private final State state;
+  private final String lifecycleState;
+  private final String healthState;
+  private final long totalRows;
+  private final long elapsedMs;
+  private final String tarPath;
+  private final String errorMessage;
+
+  /**
+   * States of export job.
+   */
+  public enum State {
+    RUNNING,
+    SUCCEEDED,
+    FAILED
+  }
+
+  @SuppressWarnings("checkstyle:ParameterNumber")
+  public ContainerExportStatus(String jobId, State state, String lifecycleState, String healthState,
+      long totalRows, long elapsedMs, String tarPath, String errorMessage) {
+    this.jobId = jobId;
+    this.state = state;
+    this.lifecycleState = lifecycleState;
+    this.healthState = healthState;
+    this.totalRows = totalRows;
+    this.elapsedMs = elapsedMs;
+    this.tarPath = tarPath;
+    this.errorMessage = errorMessage;
+  }
+
+  public String getJobId() {
+    return jobId;
+  }
+
+  public State getState() {
+    return state;
+  }
+
+  public String getLifecycleState() {
+    return lifecycleState;
+  }
+
+  public String getHealthState() {
+    return healthState;
+  }
+
+  public long getTotalRows() {
+    return totalRows;
+  }
+
+  public long getElapsedMs() {
+    return elapsedMs;
+  }
+
+  public String getTarPath() {
+    return tarPath;
+  }
+
+  public String getErrorMessage() {
+    return errorMessage;
+  }
+
+  public boolean isTerminal() {
+    return state == State.SUCCEEDED || state == State.FAILED;
+  }
+}

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java

@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+
+/**
+ * This package contains classes related to container export.
+ */
+package org.apache.hadoop.hdds.scm.container.export;

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

@@ -763,6 +763,25 @@
       then the default value of 700 will be used.
     </description>
   </property>
+  <property>
+    <name>ozone.scm.container.export.dir</name>
+    <value/>
+    <tag>OZONE, SCM, STORAGE</tag>
+    <description>
+      Directory where the SCM leader writes container ID export TAR files.
+      When unset, exports are stored under {ozone.scm.db.dirs}/exports.
+      Use a separate volume in production to avoid filling the SCM metadata disk.
+    </description>
+  </property>
+  <property>
+    <name>ozone.scm.container.export.max.terminal.jobs</name>
+    <value>10</value>
+    <tag>OZONE, SCM</tag>
+    <description>
+      Maximum number of completed (SUCCEEDED or FAILED) export jobs retained in
+      SCM memory for status lookup. Older terminal jobs are evicted on new submits.
+    </description>
+  </property>
   <property>
     <name>ozone.scm.block.client.address</name>
     <value/>

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/client/ScmClient.java

@@ -34,12 +34,14 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.DecommissionScmResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StartContainerBalancerResponseProto;
 import org.apache.hadoop.hdds.scm.DatanodeAdminError;
+import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerListResult;
 import org.apache.hadoop.hdds.scm.container.ContainerReplicaInfo;
 import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
+import org.apache.hadoop.hdds.scm.container.export.ContainerExportStatus;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.ozone.upgrade.UpgradeFinalization.StatusAndMessages;
 
@@ -494,4 +496,15 @@ DecommissionScmResponseProto decommissionScm(
    * @throws IOException
    */
   List<Long> suppressContainers(List<Long> containerIds, boolean suppress) throws IOException;
+
+  /**
+   * Submit an async container ID export job on SCM leader.
+   */
+  String submitContainerIdExport(ContainerID startContainerId, HddsProtos.LifeCycleState lifecycleState,
+      ContainerHealthState healthState, long maxRows, int pageSize, int shardSize) throws IOException;
+
+  /**
+   * Get status of a container ID export job.
+   */
+  ContainerExportStatus getContainerIdExportStatus(String jobId) throws IOException;
 }

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java

@@ -40,11 +40,13 @@
 import org.apache.hadoop.hdds.scm.DatanodeAdminError;
 import org.apache.hadoop.hdds.scm.ScmConfig;
 import org.apache.hadoop.hdds.scm.ScmInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerListResult;
 import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
+import org.apache.hadoop.hdds.scm.container.export.ContainerExportStatus;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.ozone.upgrade.UpgradeFinalization.StatusAndMessages;
 import org.apache.hadoop.security.KerberosInfo;
@@ -554,4 +556,15 @@ DecommissionScmResponseProto decommissionScm(
    * @throws IOException
    */
   List<Long> suppressContainers(List<Long> containerIds, boolean suppress) throws IOException;
+
+  /**
+   * Submit an async container ID export job on SCM leader.
+   */
+  String submitContainerIdExport(ContainerID startContainerId, HddsProtos.LifeCycleState lifecycleState,
+      ContainerHealthState healthState, long maxRows, int pageSize, int shardSize) throws IOException;
+
+  /**
+   * Get status of a container ID export job.
+   */
+  ContainerExportStatus getContainerIdExportStatus(String jobId) throws IOException;
 }

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/StorageContainerLocationProtocolClientSideTranslatorPB.java

@@ -74,6 +74,8 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.ForceExitSafeModeResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetContainerCountRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetContainerCountResponseProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetContainerIdExportStatusRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetContainerIdExportStatusResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetContainerReplicasRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetContainerRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.GetContainerTokenRequestProto;
@@ -127,16 +129,20 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StartReplicationManagerRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StopContainerBalancerRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.StopReplicationManagerRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.SubmitContainerIdExportRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.SubmitContainerIdExportResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.SuppressContainerRequestProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.SuppressContainerResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.Type;
 import org.apache.hadoop.hdds.scm.DatanodeAdminError;
 import org.apache.hadoop.hdds.scm.ScmInfo;
+import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.hdds.scm.container.ContainerInfo;
 import org.apache.hadoop.hdds.scm.container.ContainerListResult;
 import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
+import org.apache.hadoop.hdds.scm.container.export.ContainerExportStatus;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.protocol.StorageContainerLocationProtocol;
 import org.apache.hadoop.hdds.scm.proxy.SCMContainerLocationFailoverProxyProvider;
@@ -224,9 +230,9 @@ private ScmContainerLocationResponse submitRequest(
 
   private ScmContainerLocationResponse submitRpcRequest(
       ScmContainerLocationRequest wrapper) throws ServiceException {
-    // If targetScmNode has a specific node ID, route follower-readable requests to that node
-    if (targetScmNode != null && targetScmNode.hasNodeId() && 
-        FOLLOWER_READABLE_COMMAND_TYPES.contains(wrapper.getCmdType())) {
+    // If targetScmNode has a specific node ID, route non-admin requests to that node without
+    // HA failover. Used by follower-readable commands and leader-only commands such as export.
+    if (targetScmNode != null && targetScmNode.hasNodeId() && !ADMIN_COMMAND_TYPE.contains(wrapper.getCmdType())) {
       try {
         StorageContainerLocationProtocolPB proxy = fpp.getProxyForNode(targetScmNode.getNodeId());
         return proxy.submitRequest(NULL_RPC_CONTROLLER, wrapper);
@@ -1342,6 +1348,74 @@ public List<Long> suppressContainers(List<Long> containerIds, boolean suppress)
     return response.getFailedContainerIDsList();
   }
 
+  @Override
+  public String submitContainerIdExport(ContainerID startContainerId, HddsProtos.LifeCycleState lifecycleState,
+      ContainerHealthState healthState, long maxRows, int pageSize, int shardSize) throws IOException {
+    SubmitContainerIdExportRequestProto.Builder builder = SubmitContainerIdExportRequestProto.newBuilder();
+    if (startContainerId != null) {
+      builder.setStartContainerID(startContainerId.getProtobuf());
+    }
+    if (lifecycleState != null) {
+      builder.setLifecycleState(lifecycleState);
+    }
+    if (healthState != null) {
+      builder.setHealthState(healthState.name());
+    }
+    if (maxRows > 0) {
+      builder.setMaxRows(maxRows);
+    }
+    if (pageSize > 0) {
+      builder.setPageSize(pageSize);
+    }
+    if (shardSize > 0) {
+      builder.setShardSize(shardSize);
+    }
+    SubmitContainerIdExportResponseProto response = submitRequest(
+        Type.SubmitContainerIdExport,
+        req -> req.setSubmitContainerIdExportRequest(builder.build()))
+        .getSubmitContainerIdExportResponse();
+    return response.getJobId();
+  }
+
+  @Override
+  public ContainerExportStatus getContainerIdExportStatus(String jobId)
+      throws IOException {
+    GetContainerIdExportStatusRequestProto request =
+        GetContainerIdExportStatusRequestProto.newBuilder()
+            .setJobId(jobId)
+            .build();
+    GetContainerIdExportStatusResponseProto response = submitRequest(
+        Type.GetContainerIdExportStatus,
+        req -> req.setGetContainerIdExportStatusRequest(request))
+        .getGetContainerIdExportStatusResponse();
+    return fromProto(response);
+  }
+
+  private static ContainerExportStatus fromProto(
+      GetContainerIdExportStatusResponseProto response) {
+    ContainerExportStatus.State state;
+    switch (response.getState()) {
+    case RUNNING:
+      state = ContainerExportStatus.State.RUNNING;
+      break;
+    case SUCCEEDED:
+      state = ContainerExportStatus.State.SUCCEEDED;
+      break;
+    case FAILED:
+      state = ContainerExportStatus.State.FAILED;
+      break;
+    default:
+      throw new IllegalArgumentException("Unknown export state: " + response.getState());
+    }
+    String lifecycle = response.hasLifecycleState()
+        ? response.getLifecycleState().name() : null;
+    return new ContainerExportStatus(response.getJobId(), state, lifecycle,
+        response.hasHealthState() ? response.getHealthState() : null,
+        response.getTotalRows(), response.getElapsedMs(),
+        response.hasTarPath() ? response.getTarPath() : null,
+        response.hasErrorMessage() ? response.getErrorMessage() : null);
+  }
+
   /**
    * Holder class to store the target SCM node ID for routing requests.
    * This allows requests to be directed to specific SCM nodes in an HA cluster.