HDDS-15839. Implement SCM container ID export manager

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java

@@ -0,0 +1,276 @@
+/*
+ * 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;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.BooleanSupplier;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
+import org.apache.hadoop.hdds.scm.container.ContainerHealthState;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.scm.container.ContainerManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Manages asynchronous container ID export jobs on the SCM leader.
+ *
+ * <p>Health filters read {@link org.apache.hadoop.hdds.scm.container.ContainerInfo#getHealthState()}
+ * as last written by Replication Manager; they are not recomputed during export and may be stale
+ * if RM has not yet evaluated a container.
+ *
+ * <p>Job status is kept in memory only. On SCM restart or leader failover, in-flight jobs are lost
+ * and the operator must re-submit on the new leader. {@link ExportFileManager} owns on-disk layout,
+ * locking, part files, and completed archives; this class tracks {@link ExportJob} state and
+ * schedules work.
+ *
+ * <p>Job submission and the running-job slot are guarded by a {@link ReadWriteLock}. Per-job
+ * progress is read via {@link ExportJob#toStatus()}, which uses its own read lock for a consistent
+ * snapshot.
+ */
+public class ContainerExportManager {
+
+  private static final Logger LOG = LoggerFactory.getLogger(ContainerExportManager.class);
+
+  private static final int DEFAULT_BATCH_SIZE = 100_000;
+  private static final int DEFAULT_PART_SIZE = 500_000;
+  private static final long SHUTDOWN_TIMEOUT_MS = 5_000;
+
+  private final ReadWriteLock lock = new ReentrantReadWriteLock();
+  private final Map<ExportJob.Id, ExportJob> jobMap = new ConcurrentHashMap<>();
+  private ExportJob.Id runningJobId;
+  private final ExecutorService workerPool;
+  private final ContainerManager containerManager;
+  private final ExportFileManager fileManager;
+  private final BooleanSupplier isLeaderReady;
+  private final int partSize;
+  private final int batchSize;
+
+  public ContainerExportManager(String scmId, ContainerManager containerManager, BooleanSupplier isLeaderReady,
+      OzoneConfiguration conf) {
+    this(scmId, containerManager, isLeaderReady,
+        ExportFileManager.resolveExportDirectory(conf), DEFAULT_PART_SIZE, DEFAULT_BATCH_SIZE);
+  }
+
+  ContainerExportManager(String scmId, ContainerManager containerManager, BooleanSupplier isLeaderReady,
+      String exportDirectory, int partSize, int batchSize) {
+    this.containerManager = Objects.requireNonNull(containerManager, "containerManager == null");
+    this.isLeaderReady = Objects.requireNonNull(isLeaderReady, "isLeaderReady == null");
+    this.fileManager = new ExportFileManager(exportDirectory);
+    this.partSize = partSize;
+    this.batchSize = batchSize;
+    this.workerPool = newWorkerPool(scmId);
+  }
+
+  private static ExecutorService newWorkerPool(String scmId) {
+    return Executors.newSingleThreadExecutor(r -> {
+      Thread t = new Thread(r, scmId + "-ContainerExportWorker");
+      t.setDaemon(true);
+      return t;
+    });
+  }
+
+  /**
+   * Initializes the export directory. Must be called once before submitting jobs.
+   */
+  public void start() throws IOException {
+    fileManager.start();
+    LOG.info("ContainerExportManager started (dir={}, partSize={}, batchSize={})",
+        fileManager.getExportDirectory(), partSize, batchSize);
+  }
+
+  /**
+   * Submit a container ID export job on the SCM leader.
+   *
+   * @return job id, or {@code null} if not leader or another export is already running
+   */
+  public ExportJob.Id submitJob(ContainerID start, LifeCycleState lifeCycleState,
+      ContainerHealthState healthState) {
+    if (!isLeaderReady.getAsBoolean()) {
+      return null;
+    }
+
+    final ExportScope scope = ExportScope.of(lifeCycleState, healthState);
+
+    lock.writeLock().lock();
+    try {
+      if (runningJobId != null) {
+        return null;
+      }
+
+      ExportJob.Id jobId = ExportJob.Id.newId();
+      Instant now = Instant.now();
+      String jobStartTime = ExportFileManager.formatJobStartTime(now);
+      String plannedArchivePath = fileManager.resolveArchiveFile(scope, jobStartTime, jobId).getAbsolutePath();
+
+      ExportJob job = new ExportJob(jobId, scope, jobStartTime, plannedArchivePath, start, batchSize, partSize);
+      runningJobId = jobId;
+      jobMap.put(jobId, job);
+
+      workerPool.submit(() -> executeExport(job));
+      LOG.info("Submitted container ID export job {} (scope={}, start={}, batchSize={}, partSize={})",
+          jobId, scope, start, batchSize, partSize);
+      return jobId;
+    } finally {
+      lock.writeLock().unlock();
+    }
+  }
+
+  public ExportJob.Status getExportStatus(ExportJob.Id jobId) {
+    ExportJob job = jobMap.get(jobId);
+    return job != null ? job.toStatus() : null;
+  }
+
+  public void shutdown() {
+    LOG.info("Shutting down ContainerExportManager");
+    workerPool.shutdownNow();
+    try {
+      if (!workerPool.awaitTermination(SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
+        LOG.warn("Timed out waiting for export worker shutdown");
+      }
+    } catch (InterruptedException e) {
+      LOG.warn("Interrupted waiting for export worker shutdown");
+      Thread.currentThread().interrupt();
+    }
+    try {
+      fileManager.unlock();
+    } catch (IOException e) {
+      LOG.warn("Failed to unlock container export directory", e);
+    }
+  }
+
+  private void executeExport(ExportJob job) {
+    String jobIdValue = job.getId().getValue();
+    String plannedArchivePath = job.getPlannedArchivePath();
+
+    try {
+      fileManager.createJobDirectory(job.getId());
+
+      ContainerID cursor = job.getStartContainerId();
+      int partIndex = 1;
+      long totalRows = 0;
+      long recordsInCurrentPart = 0;
+      BufferedWriter writer = null;
+      // Pre-allocated buffer: ~12 chars per ID (up to 20 digits + newline) per batch.
+      StringBuilder buf = new StringBuilder(job.getBatchSize() * 12);
+
+      try {
+        while (true) {
+          if (Thread.currentThread().isInterrupted()) {
+            throw new InterruptedException("Export job " + jobIdValue + " cancelled");
+          }
+          if (!isLeaderReady.getAsBoolean()) {
+            throw new IOException("SCM lost leadership during export job " + jobIdValue);
+          }
+
+          List<ContainerID> batch = containerManager.getContainerIDs(
+              cursor, job.getBatchSize(), job.getLifeCycleState(), job.getHealthState());
+          if (batch.isEmpty()) {
+            break;
+          }
+
+          for (ContainerID containerId : batch) {
+            if (recordsInCurrentPart == 0) {
+              writer = closeWriter(writer);
+              writer = fileManager.newPartWriter(job.getId(), job.partFileName(partIndex));
+              job.writeMetadataHeader(writer, partIndex, containerId.getProtobuf().getId());
+              LOG.info("Export job {} created part{}", jobIdValue, partIndex);
+            }
+
+            buf.append(containerId.getProtobuf().getId()).append('\n');
+            totalRows++;
+            recordsInCurrentPart++;
+            job.updateTotalRows(totalRows);
+
+            if (recordsInCurrentPart >= job.getPartSize()) {
+              writer.write(buf.toString());
+              buf.setLength(0);
+              writer = closeWriter(writer);
+              recordsInCurrentPart = 0;
+              partIndex++;
+            }
+          }
+
+          if (buf.length() > 0 && writer != null) {
+            writer.write(buf.toString());
+            buf.setLength(0);
+          }
+
+          cursor = ContainerID.valueOf(
+              batch.get(batch.size() - 1).getProtobuf().getId() + 1);
+        }
+
+        writer = closeWriter(writer);
+      } finally {
+        closeWriter(writer);
+      }
+
+      if (totalRows == 0) {
+        job.completeWithNoMatches();
+        LOG.info("Export job {} completed with zero matching containers", jobIdValue);
+      } else {
+        fileManager.writeArchive(job.getId(), plannedArchivePath);
+        job.completeSucceeded();
+        LOG.info("Export job {} completed ({} rows, archive={}).",
+            jobIdValue, totalRows, plannedArchivePath);
+      }
+      fileManager.deleteJobDirectory(job.getId());
+    } catch (InterruptedException e) {
+      fileManager.cleanupFailedJob(job.getId(), plannedArchivePath);
+      job.fail(e.getMessage());
+      LOG.info("Export job {} was cancelled", jobIdValue);
+      Thread.currentThread().interrupt();
+    } catch (IOException | RuntimeException e) {
+      fileManager.cleanupFailedJob(job.getId(), plannedArchivePath);
+      job.fail(e.getMessage() != null ? e.getMessage() : e.toString());
+      LOG.error("Export job {} failed", jobIdValue, e);
+    } finally {
+      clearRunningJob(job.getId());
+    }
+  }
+
+  private void clearRunningJob(ExportJob.Id jobId) {
+    lock.writeLock().lock();
+    try {
+      if (runningJobId != null && runningJobId.equals(jobId)) {
+        runningJobId = null;
+      }
+    } finally {
+      lock.writeLock().unlock();
+    }
+  }
+
+  private static BufferedWriter closeWriter(BufferedWriter writer) throws IOException {
+    if (writer != null) {
+      writer.flush();
+      writer.close();
+    }
+    return null;
+  }
+}

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportFileManager.java

@@ -27,6 +27,9 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -37,6 +40,8 @@
 import org.apache.commons.compress.archivers.ArchiveOutputStream;
 import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
 import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.server.ServerUtils;
 import org.apache.hadoop.hdds.utils.Archiver;
 import org.apache.hadoop.ozone.util.UUIDUtil;
 import org.apache.ratis.util.AtomicFileOutputStream;
@@ -77,12 +82,15 @@ final class ExportFileManager {
 
   private static final Logger LOG = LoggerFactory.getLogger(ExportFileManager.class);
 
+  static final String EXPORT_SUBDIR = "exports";
   static final String EXPORT_JOB_DIR_PREFIX = "export_";
   static final String EXPORT_ARCHIVE_JOB_INFIX = "_job";
   static final String EXPORT_ARCHIVE_SUFFIX = ".tar.gz";
   static final String EXPORT_ARCHIVE_TMP_SUFFIX = EXPORT_ARCHIVE_SUFFIX + AtomicFileOutputStream.TMP_EXTENSION;
   static final String EXPORT_LOCK_NAME = "in_use.lock";
   private static final int EXPORT_JOB_START_TIME_LENGTH = 19;
+  private static final DateTimeFormatter EXPORT_JOB_START_TIME_FORMAT =
+      DateTimeFormatter.ofPattern("yyyy-MM-dd-HH-mm-ss").withZone(ZoneOffset.UTC);
 
   private final String exportDirectory;
   private FileLock exportDirectoryLock;
@@ -91,6 +99,15 @@ final class ExportFileManager {
     this.exportDirectory = Objects.requireNonNull(exportDirectory, "exportDirectory == null");
   }
 
+  static String resolveExportDirectory(OzoneConfiguration conf) {
+    File scmDbDir = ServerUtils.getScmDbDir(conf);
+    return new File(scmDbDir, EXPORT_SUBDIR).getAbsolutePath();
+  }
+
+  String getExportDirectory() {
+    return exportDirectory;
+  }
+
   void start() throws IOException {
     Files.createDirectories(Paths.get(exportDirectory));
     lock();
@@ -117,6 +134,15 @@ private void lock() throws IOException {
     }
   }
 
+  void unlock() throws IOException {
+    if (exportDirectoryLock == null) {
+      return;
+    }
+    exportDirectoryLock.release();
+    exportDirectoryLock.channel().close();
+    exportDirectoryLock = null;
+  }
+
   File resolveArchiveFile(ExportScope scope, String jobStartTime, ExportJob.Id jobId) {
     return new File(exportDirectory, String.format("container-ids_%s_%s%s%s%s",
         scope.getValue(), jobStartTime, EXPORT_ARCHIVE_JOB_INFIX, jobId.getValue(), EXPORT_ARCHIVE_SUFFIX));
@@ -177,6 +203,10 @@ List<String> listCompletedArchivePaths() {
     return archivePaths;
   }
 
+  static String formatJobStartTime(Instant jobStartTime) {
+    return EXPORT_JOB_START_TIME_FORMAT.format(jobStartTime);
+  }
+
   static String jobStartTimeFromArchiveFileName(String fileName) {
     int jobIndex = fileName.lastIndexOf(EXPORT_ARCHIVE_JOB_INFIX);
     if (jobIndex < EXPORT_JOB_START_TIME_LENGTH + 1