diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java new file mode 100644 index 00000000000..473d08cfa09 --- /dev/null +++ b/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. + * + *

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. + * + *

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. + * + *

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 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 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; + } +} diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportFileManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportFileManager.java index 5ebbf012970..4eda172c70c 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportFileManager.java +++ b/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 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 diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java index ba66a4907f1..917f3d0ef82 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java @@ -21,15 +21,32 @@ import java.io.IOException; import java.util.Objects; import java.util.UUID; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; +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; /** - * Metadata for a container ID export job. + * In-memory state for a container ID export job. + *

Mutable fields are guarded by a {@link ReadWriteLock} so {@link #toStatus()} returns a + * consistent snapshot while the worker updates progress. */ public final class ExportJob { + private final ReadWriteLock lock = new ReentrantReadWriteLock(); + private final Id id; private final ExportScope scope; private final String jobStartTime; + private final ContainerID startContainerId; + private final int batchSize; + private final int partSize; + // Planned .tar.gz output path, fixed at job creation; used by the worker to write the archive. + private final String plannedArchivePath; + private ExecutionState executionState = ExecutionState.RUNNING; + private long totalRows; + private String errorMessage; /** * Unique job identifier. @@ -75,10 +92,158 @@ public int hashCode() { } } - ExportJob(Id id, ExportScope scope, String jobStartTime) { + /** + * Immutable snapshot of export progress. + */ + public static final class Status { + private final Id id; + private final ExecutionState executionState; + private final long totalRows; + // plannedArchivePath when the job SUCCEEDED with rows; null while running, on failure, or zero matches. + private final String completedArchivePath; + private final String errorMessage; + + private Status(Id id, ExecutionState executionState, long totalRows, String completedArchivePath, + String errorMessage) { + this.id = id; + this.executionState = executionState; + this.totalRows = totalRows; + this.completedArchivePath = completedArchivePath; + this.errorMessage = errorMessage; + } + + public Id getId() { + return id; + } + + public ExecutionState getExecutionState() { + return executionState; + } + + public long getTotalRows() { + return totalRows; + } + + public String getCompletedArchivePath() { + return completedArchivePath; + } + + public String getErrorMessage() { + return errorMessage; + } + } + + /** + * Job execution state. + */ + public enum ExecutionState { + RUNNING(false), + SUCCEEDED(true), + FAILED(true); + + private final boolean terminal; + + ExecutionState(boolean terminal) { + this.terminal = terminal; + } + + public boolean isTerminal() { + return terminal; + } + } + + ExportJob(Id id, ExportScope scope, String jobStartTime, String plannedArchivePath, ContainerID startContainerId, + int batchSize, int partSize) { this.id = id; this.scope = scope; this.jobStartTime = jobStartTime; + this.plannedArchivePath = plannedArchivePath; + this.startContainerId = startContainerId != null ? startContainerId : ContainerID.valueOf(0); + this.batchSize = batchSize; + this.partSize = partSize; + } + + Id getId() { + return id; + } + + LifeCycleState getLifeCycleState() { + return scope.getLifeCycleState(); + } + + ContainerHealthState getHealthState() { + return scope.getHealthState(); + } + + String getPlannedArchivePath() { + return plannedArchivePath; + } + + ContainerID getStartContainerId() { + return startContainerId; + } + + int getBatchSize() { + return batchSize; + } + + int getPartSize() { + return partSize; + } + + void updateTotalRows(long rows) { + lock.writeLock().lock(); + try { + totalRows = rows; + } finally { + lock.writeLock().unlock(); + } + } + + void completeWithNoMatches() { + lock.writeLock().lock(); + try { + transitionToTerminal(ExecutionState.SUCCEEDED); + } finally { + lock.writeLock().unlock(); + } + } + + void completeSucceeded() { + lock.writeLock().lock(); + try { + transitionToTerminal(ExecutionState.SUCCEEDED); + } finally { + lock.writeLock().unlock(); + } + } + + void fail(String message) { + lock.writeLock().lock(); + try { + errorMessage = message; + transitionToTerminal(ExecutionState.FAILED); + } finally { + lock.writeLock().unlock(); + } + } + + Status toStatus() { + lock.readLock().lock(); + try { + String completedArchivePath = + executionState == ExecutionState.SUCCEEDED && totalRows > 0 ? plannedArchivePath : null; + return new Status(id, executionState, totalRows, completedArchivePath, errorMessage); + } finally { + lock.readLock().unlock(); + } + } + + private void transitionToTerminal(ExecutionState terminalState) { + if (executionState.isTerminal()) { + throw new IllegalStateException("Export job " + id + " is already terminal: " + executionState); + } + executionState = terminalState; } String partFileName(int partIndex) { diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java index 745bea6ee1d..3c5f4458dd1 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java @@ -90,6 +90,7 @@ import org.apache.hadoop.hdds.scm.container.IncrementalContainerReportHandler; import org.apache.hadoop.hdds.scm.container.balancer.ContainerBalancer; import org.apache.hadoop.hdds.scm.container.balancer.MoveManager; +import org.apache.hadoop.hdds.scm.container.export.ContainerExportManager; import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementPolicyFactory; import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementMetrics; import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMMetrics; @@ -312,6 +313,7 @@ public final class StorageContainerManager extends ServiceRuntimeInfoImpl private final ContainerBalancer containerBalancer; // MoveManager is used by ContainerBalancer to schedule container moves private final MoveManager moveManager; + private final ContainerExportManager containerExportManager; private StatefulServiceStateManager statefulServiceStateManager; // Used to keep track of pending replication and pending deletes for // container replicas. @@ -436,6 +438,9 @@ private StorageContainerManager(OzoneConfiguration conf, initializeSystemManagers(conf, configurator); + containerExportManager = new ContainerExportManager( + getScmId(), containerManager, this::checkLeader, conf); + if (isSecretKeyEnable(securityConfig)) { secretKeyManagerService = new SecretKeyManagerService(scmContext, conf, scmHAManager.getRatisServer()); @@ -1607,6 +1612,7 @@ public void start() throws IOException { scmBlockManager.start(); leaseManager.start(); + containerExportManager.start(); try { httpServer = new StorageContainerManagerHttpServer(configuration, this); @@ -1735,6 +1741,13 @@ public void stop() { LOG.error("SCM block manager service stop failed.", ex); } + try { + LOG.info("Shutting down Container Export Manager."); + containerExportManager.shutdown(); + } catch (Exception ex) { + LOG.error("Container Export Manager shutdown failed.", ex); + } + if (metrics != null) { metrics.unRegister(); } @@ -1948,6 +1961,10 @@ public MoveManager getMoveManager() { return moveManager; } + public ContainerExportManager getContainerExportManager() { + return containerExportManager; + } + /** * Returns SCM root CA rotation manager. */ diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java new file mode 100644 index 00000000000..5e38728b036 --- /dev/null +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java @@ -0,0 +1,376 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.isNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.File; +import java.io.InputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.BooleanSupplier; +import java.util.stream.Collectors; +import java.util.zip.GZIPInputStream; +import org.apache.commons.compress.archivers.ArchiveInputStream; +import org.apache.commons.compress.archivers.tar.TarArchiveEntry; +import org.apache.commons.io.FileUtils; +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.apache.hadoop.hdds.utils.Archiver; +import org.apache.ozone.test.GenericTestUtils; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Tests for {@link ContainerExportManager}. + */ +public class TestContainerExportManager { + + private static final int TEST_BATCH_SIZE = 2; + private static final int TEST_PART_SIZE = 3; + private static final String TEST_SCM_ID = "test-scm"; + + @TempDir + private File tempDir; + + private ContainerManager containerManager; + private ContainerExportManager exportManager; + + @BeforeEach + public void setup() throws Exception { + containerManager = mock(ContainerManager.class); + exportManager = newExportManager(TEST_PART_SIZE, TEST_BATCH_SIZE, () -> true); + } + + @AfterEach + public void teardown() { + if (exportManager != null) { + exportManager.shutdown(); + } + } + + @Test + public void testRejectMissingFilters() { + assertThrows(IllegalArgumentException.class, () -> + exportManager.submitJob(ContainerID.valueOf(0), null, null)); + } + + @Test + public void testRejectSubmitWhenNotLeader() throws Exception { + exportManager.shutdown(); + exportManager = newExportManager(TEST_PART_SIZE, TEST_BATCH_SIZE, () -> false); + assertNull(exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING)); + } + + @Test + public void testGetExportStatusUnknownJobReturnsNull() { + assertNull(exportManager.getExportStatus(ExportJob.Id.newId())); + } + + @Test + public void testRejectConcurrentExport() { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(1)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(2)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenAnswer(invocation -> { + Thread.sleep(60_000); + return Collections.emptyList(); + }); + + ExportJob.Id first = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING); + assertNotNull(first); + assertNull(exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.EMPTY)); + } + + @Test + public void testNullStartDefaultsToZeroCursor() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + + ExportJob.Id jobId = exportManager.submitJob(null, null, ContainerHealthState.MISSING); + assertNotNull(jobId); + waitForTerminal(jobId); + + verify(containerManager).getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING)); + } + + @Test + public void testExportStartsFromRequestedCursor() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(5)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(5, 6)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(7)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(5), null, + ContainerHealthState.MISSING); + assertNotNull(jobId); + + ExportJob.Status status = waitForTerminal(jobId); + assertEquals(ExportJob.ExecutionState.SUCCEEDED, status.getExecutionState()); + assertEquals(2, status.getTotalRows()); + + verify(containerManager).getContainerIDs( + eq(ContainerID.valueOf(5)), anyInt(), isNull(), eq(ContainerHealthState.MISSING)); + } + + @Test + public void testExportWithLifeCycleFilter() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), eq(LifeCycleState.OPEN), isNull())) + .thenReturn(ids(1)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(2)), anyInt(), eq(LifeCycleState.OPEN), isNull())) + .thenReturn(Collections.emptyList()); + + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), LifeCycleState.OPEN, null); + assertNotNull(jobId); + waitForTerminal(jobId); + + verify(containerManager).getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), eq(LifeCycleState.OPEN), isNull()); + } + + @Test + public void testZeroMatchesSucceedsWithoutArchive() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING); + assertNotNull(jobId); + + ExportJob.Status status = waitForTerminal(jobId); + assertEquals(ExportJob.ExecutionState.SUCCEEDED, status.getExecutionState()); + assertEquals(0, status.getTotalRows()); + assertNull(status.getCompletedArchivePath()); + } + + @Test + public void testSinglePartExportCreatesTar() throws Exception { + exportManager.shutdown(); + exportManager = newExportManager(100, TEST_BATCH_SIZE, () -> true); + + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(1, 2, 3, 4)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(5)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING); + assertNotNull(jobId); + + ExportJob.Status status = waitForTerminal(jobId); + if (status.getExecutionState() == ExportJob.ExecutionState.FAILED) { + fail(status.getErrorMessage()); + } + assertEquals(ExportJob.ExecutionState.SUCCEEDED, status.getExecutionState()); + assertTrue(status.getCompletedArchivePath().endsWith(ExportFileManager.EXPORT_ARCHIVE_SUFFIX)); + assertTrue(new File(status.getCompletedArchivePath()).exists()); + } + + @Test + public void testMultiPartExportCreatesTar() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(1, 2)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(3)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(3, 4)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(5)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING); + assertNotNull(jobId); + + ExportJob.Status status = waitForTerminal(jobId); + if (status.getExecutionState() == ExportJob.ExecutionState.FAILED) { + fail(status.getErrorMessage()); + } + assertEquals(ExportJob.ExecutionState.SUCCEEDED, status.getExecutionState()); + assertEquals(4, status.getTotalRows()); + assertNotNull(status.getCompletedArchivePath()); + assertTrue(status.getCompletedArchivePath().endsWith(ExportFileManager.EXPORT_ARCHIVE_SUFFIX)); + File archive = new File(status.getCompletedArchivePath()); + assertTrue(archive.exists()); + assertFalse(Files.exists(tempDir.toPath().resolve(ExportFileManager.exportJobDirName(jobId)))); + + Path extractDir = Files.createTempDirectory("export-archive"); + try { + extractGzTar(archive, extractDir); + String part2Name; + try (java.util.stream.Stream stream = Files.list(extractDir)) { + part2Name = stream.map(path -> path.getFileName().toString()) + .filter(name -> name.endsWith("part002.txt")) + .findFirst() + .orElseThrow(() -> new AssertionError("part002.txt not found in archive")); + } + assertTrue(Files.readAllLines(extractDir.resolve(part2Name)).contains( + "# startContainerId=4")); + } finally { + FileUtils.deleteQuietly(extractDir.toFile()); + } + } + + @Test + public void testSubmitAfterPreviousJobCompletes() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(ids(1)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(2)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenReturn(Collections.emptyList()); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.EMPTY))) + .thenReturn(ids(10)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(11)), anyInt(), isNull(), eq(ContainerHealthState.EMPTY))) + .thenReturn(Collections.emptyList()); + + ExportJob.Id first = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING); + waitForTerminal(first); + + ExportJob.Id second = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.EMPTY); + assertNotNull(second); + ExportJob.Status status = waitForTerminal(second); + assertEquals(ExportJob.ExecutionState.SUCCEEDED, status.getExecutionState()); + assertEquals(1, status.getTotalRows()); + } + + @Test + public void testFailsWhenLeadershipLostDuringExport() throws Exception { + AtomicBoolean leader = new AtomicBoolean(true); + exportManager.shutdown(); + exportManager = newExportManager(TEST_PART_SIZE, TEST_BATCH_SIZE, leader::get); + + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenAnswer(invocation -> { + leader.set(false); + return ids(1); + }); + + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING); + assertNotNull(jobId); + + ExportJob.Status status = waitForTerminal(jobId); + assertEquals(ExportJob.ExecutionState.FAILED, status.getExecutionState()); + assertTrue(status.getErrorMessage().contains("lost leadership")); + } + + @Test + public void testFailsWhenContainerManagerThrows() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenThrow(new RuntimeException("container listing failed")); + + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING); + assertNotNull(jobId); + + ExportJob.Status status = waitForTerminal(jobId); + assertEquals(ExportJob.ExecutionState.FAILED, status.getExecutionState()); + assertTrue(status.getErrorMessage().contains("container listing failed")); + assertFalse(Files.exists(tempDir.toPath().resolve(ExportFileManager.exportJobDirName(jobId)))); + } + + @Test + public void testShutdownCancelsRunningExport() throws Exception { + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.MISSING))) + .thenAnswer(invocation -> { + Thread.sleep(60_000); + return Collections.emptyList(); + }); + + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING); + assertNotNull(jobId); + + exportManager.shutdown(); + ExportJob.Status status = waitForTerminal(jobId); + assertEquals(ExportJob.ExecutionState.FAILED, status.getExecutionState()); + exportManager = null; + } + + private ContainerExportManager newExportManager(int partSize, int batchSize, + BooleanSupplier isLeaderReady) throws Exception { + ContainerExportManager manager = new ContainerExportManager(TEST_SCM_ID, containerManager, isLeaderReady, + tempDir.getAbsolutePath(), partSize, batchSize); + manager.start(); + return manager; + } + + private static List ids(long... values) { + return Arrays.stream(values).mapToObj(ContainerID::valueOf) + .collect(Collectors.toList()); + } + + private ExportJob.Status waitForTerminal(ExportJob.Id jobId) throws Exception { + GenericTestUtils.waitFor(() -> { + ExportJob.Status status = exportManager.getExportStatus(jobId); + return status != null && status.getExecutionState().isTerminal(); + }, 100, 30_000); + return exportManager.getExportStatus(jobId); + } + + private static void extractGzTar(File archive, Path extractDir) throws Exception { + Files.createDirectories(extractDir); + try (InputStream in = new GZIPInputStream(Files.newInputStream(archive.toPath())); + ArchiveInputStream tarIn = Archiver.untar(in)) { + TarArchiveEntry entry; + while ((entry = tarIn.getNextEntry()) != null) { + Archiver.extractEntry(entry, tarIn, entry.getSize(), extractDir, extractDir.resolve(entry.getName())); + } + } + } +} diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportFileManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportFileManager.java index a97255ba198..780d579e138 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportFileManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportFileManager.java @@ -38,6 +38,7 @@ import org.apache.commons.io.FileUtils; 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.utils.Archiver; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -134,8 +135,9 @@ public void testListCompletedArchivePaths() throws Exception { public void testWriteArchiveFromPartFiles() throws Exception { ExportJob.Id jobId = ExportJob.Id.newId(); ExportScope scope = ExportScope.of(null, ContainerHealthState.MISSING); - ExportJob job = new ExportJob(jobId, scope, TEST_JOB_START_TIME); File archive = fileManager.resolveArchiveFile(scope, TEST_JOB_START_TIME, jobId); + ExportJob job = new ExportJob(jobId, scope, TEST_JOB_START_TIME, archive.getAbsolutePath(), + ContainerID.valueOf(0), 100, 500); fileManager.createJobDirectory(jobId); try (BufferedWriter writer = fileManager.newPartWriter(jobId, job.partFileName(1))) { diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportJob.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportJob.java index fdad57b4dc3..832fd6d1cb2 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportJob.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestExportJob.java @@ -18,6 +18,7 @@ package org.apache.hadoop.hdds.scm.container.export; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.BufferedWriter; @@ -26,6 +27,7 @@ import java.util.List; 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.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -34,7 +36,10 @@ */ public class TestExportJob { + private static final int TEST_BATCH_SIZE = 100; + private static final int TEST_PART_SIZE = 500; private static final String TEST_JOB_START_TIME = "2026-01-01-12-00-00"; + private static final ExportJob.Id TEST_JOB_ID = ExportJob.Id.of("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"); @TempDir private File tempDir; @@ -60,8 +65,28 @@ public void testWriteMetadataHeader() throws Exception { assertTrue(lines.contains("# part=2")); } + @Test + public void testStatusCompletedArchivePathDerivedFromOutcome() { + ExportJob job = newJob(ContainerHealthState.MISSING, null); + job.completeWithNoMatches(); + assertNull(job.toStatus().getCompletedArchivePath()); + + ExportJob succeededJob = newJob(ContainerHealthState.MISSING, null); + succeededJob.updateTotalRows(10); + succeededJob.completeSucceeded(); + assertEquals(testPlannedArchivePath(ExportScope.of(null, ContainerHealthState.MISSING)), + succeededJob.toStatus().getCompletedArchivePath()); + } + private static ExportJob newJob(ContainerHealthState healthState, LifeCycleState lifeCycleState) { ExportScope scope = ExportScope.of(lifeCycleState, healthState); - return new ExportJob(ExportJob.Id.newId(), scope, TEST_JOB_START_TIME); + return new ExportJob(TEST_JOB_ID, scope, TEST_JOB_START_TIME, testPlannedArchivePath(scope), + ContainerID.valueOf(0), TEST_BATCH_SIZE, TEST_PART_SIZE); + } + + private static String testPlannedArchivePath(ExportScope scope) { + return "/" + ExportFileManager.EXPORT_SUBDIR + "/container-ids_" + scope.getValue() + "_" + + TEST_JOB_START_TIME + ExportFileManager.EXPORT_ARCHIVE_JOB_INFIX + TEST_JOB_ID.getValue() + + ExportFileManager.EXPORT_ARCHIVE_SUFFIX; } }