diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportLimits.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportLimits.java new file mode 100644 index 000000000000..49e7b914b99b --- /dev/null +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportLimits.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. + */ +public final class ExportLimits { + + 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 ExportLimits() { + } +} diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java new file mode 100644 index 000000000000..103c9519fcab --- /dev/null +++ b/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; 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 000000000000..e3bc0f33924c --- /dev/null +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ContainerExportManager.java @@ -0,0 +1,508 @@ +/* + * 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.apache.hadoop.hdds.scm.container.export.ExportLimits.DEFAULT_PAGE_SIZE; +import static org.apache.hadoop.hdds.scm.container.export.ExportLimits.DEFAULT_SHARD_SIZE; +import static org.apache.hadoop.hdds.scm.container.export.ExportLimits.EXPORT_SUBDIR; + +import java.io.BufferedWriter; +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +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.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BooleanSupplier; +import java.util.stream.Collectors; +import org.apache.commons.io.FileUtils; +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.apache.hadoop.hdds.server.ServerUtils; +import org.apache.hadoop.hdds.utils.Archiver; +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. Completed TAR files are kept on disk while their + * job status remains in {@code jobTracker}; when a terminal job is evicted past {@code maxTerminalJobs}, + * its TAR is deleted as well. On startup, incomplete work ({@code {jobId}.in-progress} markers, + * {@code export-{jobId}} work directories, and matching partial TAR files) is removed. + */ +public class ContainerExportManager { + + private static final Logger LOG = LoggerFactory.getLogger(ContainerExportManager.class); + + static final String IN_PROGRESS_MARKER_SUFFIX = ".in-progress"; + static final String EXPORT_JOB_DIR_PREFIX = "export-"; + + //TODO: make ozone.scm.container.export.max.terminal.jobs configurable. + private static final int DEFAULT_MAX_TERMINAL_JOBS = 10; + private static final long SHUTDOWN_TIMEOUT_MS = 5_000; + + private static final DateTimeFormatter METADATA_TIMESTAMP_FORMAT = + DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss'Z'").withZone(ZoneOffset.UTC); + private static final DateTimeFormatter FILENAME_TIMESTAMP_FORMAT = + DateTimeFormatter.ofPattern("yyyyMMdd'T'HHmmss'Z'").withZone(ZoneOffset.UTC); + + private final Map jobTracker = new ConcurrentHashMap<>(); + private final AtomicReference runningJobId = new AtomicReference<>(); + private final ExecutorService workerPool; + private final ContainerManager containerManager; + private final ExportMetrics metrics; + private final BooleanSupplier isLeaderReady; + private final String exportDirectory; + private final int defaultShardSize; + private final int defaultPageSize; + private final int maxTerminalJobs; + /** SCM node id, used in log messages and the export worker thread name. */ + private final String scmId; + + public ContainerExportManager(ContainerManager containerManager, BooleanSupplier isLeaderReady, + OzoneConfiguration conf, String scmId) { + this.containerManager = Objects.requireNonNull(containerManager, "containerManager == null"); + this.isLeaderReady = Objects.requireNonNull(isLeaderReady, "isLeaderReady == null"); + this.exportDirectory = Objects.requireNonNull(resolveExportDirectory(conf), "exportDirectory == null"); + this.scmId = Objects.requireNonNull(scmId, "scmId == null"); + this.defaultShardSize = DEFAULT_SHARD_SIZE; + this.defaultPageSize = DEFAULT_PAGE_SIZE; + this.maxTerminalJobs = DEFAULT_MAX_TERMINAL_JOBS; + this.metrics = ExportMetrics.create(); + this.workerPool = newWorkerPool(this.scmId); + } + + ContainerExportManager(ContainerManager containerManager, BooleanSupplier isLeaderReady, + String exportDirectory, int defaultShardSize, int defaultPageSize, int maxTerminalJobs, + String scmId) { + this.containerManager = Objects.requireNonNull(containerManager, "containerManager == null"); + this.isLeaderReady = Objects.requireNonNull(isLeaderReady, "isLeaderReady == null"); + this.exportDirectory = Objects.requireNonNull(exportDirectory, "exportDirectory == null"); + this.scmId = Objects.requireNonNull(scmId, "scmId == null"); + this.defaultShardSize = defaultShardSize; + this.defaultPageSize = defaultPageSize; + this.maxTerminalJobs = maxTerminalJobs; + this.metrics = null; + this.workerPool = newWorkerPool(this.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 { + Files.createDirectories(Paths.get(exportDirectory)); + cleanupOrphanedExportArtifacts(); + LOG.info("{}: ContainerExportManager started (dir={}, defaultShardSize={}, defaultPageSize={}, maxTerminalJobs={})", + scmId, exportDirectory, defaultShardSize, defaultPageSize, maxTerminalJobs); + } + + // TODO: make ozone.scm.container.export.dir configurable + // Set path separate from scm.db.dirs to avoid large export TAR I/O + // contending with SCM metadata DB access as the disk fills. + private static String resolveExportDirectory(OzoneConfiguration conf) { + File scmDbDir = ServerUtils.getScmDbDir(conf); + return new File(scmDbDir, EXPORT_SUBDIR).getAbsolutePath(); + } + + /** + * Submit a container ID export job on the SCM leader. + * Batch sizing is described by {@link ExportSizing}. + * + * @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, long maxRows, int pageSize, int shardSize) { + if (!isLeaderReady.getAsBoolean()) { + return null; + } + if (lifeCycleState == null && healthState == null) { + throw new IllegalArgumentException("At least one of healthState or lifecycleState filter is required."); + } + validateRequest(start); + ExportSizing.validate(maxRows, pageSize, shardSize); + ExportSizing sizing = ExportSizing.resolve(maxRows, pageSize, shardSize, defaultPageSize, defaultShardSize); + + ExportJob.Id jobId = ExportJob.Id.newId(); + if (!runningJobId.compareAndSet(null, jobId)) { + return null; + } + + ExportScope scope = ExportScope.of(lifeCycleState, healthState); + Instant now = Instant.now(); + String metadataTimestamp = METADATA_TIMESTAMP_FORMAT.format(now); + String fileTimestamp = FILENAME_TIMESTAMP_FORMAT.format(now); + String tarFileName = String.format("container-ids-%s-%s-%s.tar", scope.getValue(), fileTimestamp, jobId.getValue()); + String tarPath = exportDirectory + File.separator + tarFileName; + + ExportJob job = new ExportJob(jobId, scope, metadataTimestamp, tarPath, start, sizing); + + evictOldTerminalJobs(); + jobTracker.put(jobId, job); + + if (metrics != null) { + metrics.incrExportJobsSubmitted(); + } + + workerPool.submit(() -> executeExport(job)); + LOG.info("{}: submitted container ID export job {} (scope={}, start={}, maxRows={}, pageSize={}, shardSize={})", + scmId, jobId, scope, start, sizing.getMaxRows(), sizing.getPageSize(), sizing.getShardSize()); + return jobId; + } + + private static void validateRequest(ContainerID start) { + if (start != null && start.getProtobuf().getId() < 0) { + throw new IllegalArgumentException("start container ID must be non-negative: " + start.getProtobuf().getId()); + } + } + + public ExportJob.Status getExportStatus(ExportJob.Id jobId) { + ExportJob job = jobTracker.get(jobId); + return job != null ? job.toStatus() : null; + } + + public void shutdown() { + LOG.info("{}: shutting down ContainerExportManager", scmId); + workerPool.shutdownNow(); + try { + if (!workerPool.awaitTermination(SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS)) { + LOG.warn("{}: timed out waiting for export worker shutdown", scmId); + } + } catch (InterruptedException e) { + LOG.warn("{}: interrupted waiting for export worker shutdown", scmId); + Thread.currentThread().interrupt(); + } + if (metrics != null) { + metrics.unRegister(); + } + } + + Map getJobTracker() { + return jobTracker; + } + + private void evictOldTerminalJobs() { + List> terminalJobs = jobTracker.entrySet().stream() + .filter(e -> e.getValue().getExecutionState() != ExportJob.ExecutionState.RUNNING) + .sorted(Comparator.comparingLong(e -> e.getValue().getEndTimeNs())) + .collect(Collectors.toList()); + int excess = terminalJobs.size() - maxTerminalJobs; + for (int i = 0; i < excess; i++) { + ExportJob evicted = terminalJobs.get(i).getValue(); + deleteExportTar(evicted); + jobTracker.remove(evicted.getId()); + } + } + + private void deleteExportTar(ExportJob job) { + String tarPath = job.getTarPath(); + if (tarPath == null) { + return; + } + File tar = new File(tarPath); + if (tar.isFile() && FileUtils.deleteQuietly(tar)) { + LOG.debug("Removed container export TAR for evicted job {}: {}", job.getId(), tar.getName()); + } + } + + private void cleanupOrphanedExportArtifacts() { + File exportDir = new File(exportDirectory); + File[] children = exportDir.listFiles(); + if (children == null) { + return; + } + for (File child : children) { + if (child.isFile() && child.getName().endsWith(IN_PROGRESS_MARKER_SUFFIX)) { + String jobId = child.getName().substring( + 0, child.getName().length() - IN_PROGRESS_MARKER_SUFFIX.length()); + if (isUuidDirectoryName(jobId)) { + removeIncompleteExportArtifacts(jobId); + } + } + } + for (File child : children) { + if (child.isDirectory()) { + String jobId = jobIdFromExportDirName(child.getName()); + if (jobId != null) { + removeIncompleteExportArtifacts(jobId); + } + } + } + } + + private static String exportJobDirName(String jobId) { + return EXPORT_JOB_DIR_PREFIX + jobId; + } + + private static String jobIdFromExportDirName(String dirName) { + if (!dirName.startsWith(EXPORT_JOB_DIR_PREFIX)) { + return null; + } + String jobId = dirName.substring(EXPORT_JOB_DIR_PREFIX.length()); + return isUuidDirectoryName(jobId) ? jobId : null; + } + + private void removeIncompleteExportArtifacts(String jobId) { + LOG.info("Removing incomplete container export artifacts for job {}", jobId); + FileUtils.deleteQuietly(inProgressMarkerFile(jobId)); + File tar = findTarForJobId(jobId); + if (tar != null) { + FileUtils.deleteQuietly(tar); + LOG.info("Removed incomplete container export TAR for job {}: {}", jobId, tar.getName()); + } + File jobWorkDir = new File(exportDirectory, exportJobDirName(jobId)); + if (jobWorkDir.isDirectory()) { + FileUtils.deleteQuietly(jobWorkDir); + LOG.info("Removed orphaned container export work directory: {}", + jobWorkDir.getAbsolutePath()); + } + } + + private File findTarForJobId(String jobId) { + File exportDir = new File(exportDirectory); + File[] matches = exportDir.listFiles( + (dir, fileName) -> fileName.endsWith("-" + jobId + ".tar")); + if (matches == null || matches.length == 0) { + return null; + } + return matches[0]; + } + + private File inProgressMarkerFile(String jobId) { + return new File(exportDirectory, jobId + IN_PROGRESS_MARKER_SUFFIX); + } + + private void markExportInProgress(String jobId) throws IOException { + Files.createFile(inProgressMarkerFile(jobId).toPath()); + } + + private void clearExportInProgress(String jobId) { + FileUtils.deleteQuietly(inProgressMarkerFile(jobId)); + } + + private static boolean isUuidDirectoryName(String directoryName) { + try { + return directoryName.equals(UUID.fromString(directoryName).toString()); + } catch (IllegalArgumentException e) { + return false; + } + } + + private void executeExport(ExportJob job) { + String jobIdValue = job.getId().getValue(); + Path jobDir = Paths.get(exportDirectory, exportJobDirName(jobIdValue)); + Path workDir = jobDir.resolve("work"); + File tarFile = new File(job.getTarPath()); + job.setStartTimeNs(System.nanoTime()); + boolean succeeded = false; + Archiver.AppendableTar appendableTar = null; + + try { + Files.createDirectories(workDir); + job.setExecutionState(ExportJob.ExecutionState.RUNNING); + markExportInProgress(jobIdValue); + + ContainerID cursor = job.getStartContainerId(); + int fileIndex = 1; + long totalRows = 0; + long recordsInCurrentFile = 0; + BufferedWriter writer = null; + Path currentShardPath = null; + // Pre-allocated buffer: ~12 chars per ID (up to 20 digits + newline) per page. + StringBuilder buf = new StringBuilder(job.getPageSize() * 12); + + try { + while (true) { + if (Thread.currentThread().isInterrupted()) { + throw new InterruptedException("Export job " + jobIdValue + " cancelled"); + } + if (!isLeaderReady.getAsBoolean()) { + throw new IOException(scmId + ": SCM lost leadership during export job " + jobIdValue); + } + + int fetchCount = job.getPageSize(); + if (job.getMaxRows() > 0) { + long remaining = job.getMaxRows() - totalRows; + if (remaining <= 0) { + break; + } + fetchCount = (int) Math.min(fetchCount, remaining); + } + + List page = containerManager.getContainerIDs( + cursor, fetchCount, job.getLifeCycleState(), job.getHealthState()); + if (page.isEmpty()) { + break; + } + + for (ContainerID containerId : page) { + if (recordsInCurrentFile == 0) { + writer = closeWriter(writer); + currentShardPath = workDir.resolve(job.shardFileName(fileIndex)); + writer = Files.newBufferedWriter(currentShardPath, StandardCharsets.UTF_8); + job.writeMetadataHeader(writer, fileIndex, containerId); + LOG.info("{}: export job {} created shard part{}", scmId, jobIdValue, fileIndex); + } + + buf.append(containerId.getProtobuf().getId()).append('\n'); + totalRows++; + recordsInCurrentFile++; + job.setTotalRows(totalRows); + + if (recordsInCurrentFile >= job.getShardSize()) { + writer.write(buf.toString()); + buf.setLength(0); + writer = closeWriter(writer); + if (appendableTar == null) { + appendableTar = Archiver.openForAppend(tarFile); + } + appendShardToTar(appendableTar, currentShardPath, job, fileIndex); + currentShardPath = null; + recordsInCurrentFile = 0; + fileIndex++; + } + } + + // Flush the batch buffer at the end of each page. + if (buf.length() > 0 && writer != null) { + writer.write(buf.toString()); + buf.setLength(0); + } + + cursor = ContainerID.valueOf( + page.get(page.size() - 1).getProtobuf().getId() + 1); + } + + writer = closeWriter(writer); + if (totalRows == 0) { + clearExportInProgress(jobIdValue); + FileUtils.deleteQuietly(workDir.toFile()); + FileUtils.deleteQuietly(jobDir.toFile()); + job.setExecutionState(ExportJob.ExecutionState.SUCCEEDED); + job.setTarPath(null); + succeeded = true; + LOG.info("{}: export job {} completed with zero matching containers", scmId, jobIdValue); + return; + } + + if (currentShardPath != null) { + if (appendableTar == null) { + appendableTar = Archiver.openForAppend(tarFile); + } + appendShardToTar(appendableTar, currentShardPath, job, fileIndex); + } + + FileUtils.deleteQuietly(workDir.toFile()); + FileUtils.deleteQuietly(jobDir.toFile()); + job.setExecutionState(ExportJob.ExecutionState.SUCCEEDED); + succeeded = true; + LOG.info("{}: export job {} completed ({} rows, tar={}).", + scmId, jobIdValue, totalRows, tarFile.getAbsolutePath()); + } finally { + closeWriter(writer); + if (appendableTar != null) { + appendableTar.close(); + if (succeeded) { + clearExportInProgress(jobIdValue); + } + } + } + } catch (InterruptedException e) { + job.setExecutionState(ExportJob.ExecutionState.FAILED); + job.setErrorMessage(e.getMessage()); + cleanupFailedArtifacts(jobDir, tarFile, jobIdValue); + LOG.info("{}: export job {} was cancelled", scmId, jobIdValue); + Thread.currentThread().interrupt(); + } catch (IOException | RuntimeException e) { + succeeded = false; + job.setExecutionState(ExportJob.ExecutionState.FAILED); + job.setErrorMessage(e.getMessage() != null ? e.getMessage() : e.toString()); + cleanupFailedArtifacts(jobDir, tarFile, jobIdValue); + LOG.error("{}: export job {} failed", scmId, jobIdValue, e); + } finally { + runningJobId.compareAndSet(job.getId(), null); + if (metrics != null) { + if (succeeded) { + metrics.incrExportJobsSucceeded(); + long bytesWritten = tarFile.isFile() ? tarFile.length() : 0L; + metrics.recordLastSuccessfulExport(job.getTotalRows(), bytesWritten); + } else if (job.getExecutionState() == ExportJob.ExecutionState.FAILED) { + metrics.incrExportJobsFailed(); + } + } + if (job.getExecutionState() == ExportJob.ExecutionState.SUCCEEDED + || job.getExecutionState() == ExportJob.ExecutionState.FAILED) { + evictOldTerminalJobs(); + } + } + } + + private void appendShardToTar(Archiver.AppendableTar tar, Path shardPath, ExportJob job, int partIndex) + throws IOException { + tar.appendFile(shardPath.toFile(), job.shardEntryName(partIndex)); + FileUtils.deleteQuietly(shardPath.toFile()); + } + + /** Remove partial work artifacts after a failed or cancelled export. */ + private void cleanupFailedArtifacts(Path jobDir, File tarFile, String jobId) { + if (jobDir != null) { + FileUtils.deleteQuietly(jobDir.toFile()); + } + if (tarFile != null) { + FileUtils.deleteQuietly(tarFile); + } + clearExportInProgress(jobId); + } + + 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/ExportJob.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java new file mode 100644 index 000000000000..4d2ba160a893 --- /dev/null +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportJob.java @@ -0,0 +1,269 @@ +/* + * 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.util.Objects; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +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; + +/** + * In-memory state for a container ID export job. + */ +public final class ExportJob { + + private final Id id; + private final ExportScope scope; + private final String timestamp; + private final ContainerID startContainerId; + private final ExportSizing sizing; + private volatile String tarPath; + private volatile ExecutionState executionState = ExecutionState.RUNNING; + private volatile long totalRows; + private volatile long startTimeNs; + private volatile long endTimeNs; + private volatile String errorMessage; + + /** + * Unique job identifier. + */ + public static final class Id { + private final String value; + + private Id(String value) { + this.value = Objects.requireNonNull(value, "value == null"); + } + + public static Id newId() { + return new Id(UUID.randomUUID().toString()); + } + + public static Id of(String value) { + return new Id(value); + } + + public String getValue() { + return value; + } + + @Override + public String toString() { + return value; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (!(obj instanceof Id)) { + return false; + } + return value.equals(((Id) obj).value); + } + + @Override + public int hashCode() { + return value.hashCode(); + } + } + + /** + * Snapshot of export progress returned to callers. Reads live fields from the enclosing job. + */ + public final class Status { + private Status() { + } + + public Id getId() { + return id; + } + + public ExecutionState getExecutionState() { + return executionState; + } + + public LifeCycleState getLifeCycleState() { + return scope.getLifeCycleState(); + } + + public ContainerHealthState getHealthState() { + return scope.getHealthState(); + } + + public long getTotalRows() { + return totalRows; + } + + public long getElapsedMs() { + if (startTimeNs <= 0) { + return 0; + } + long endNs = endTimeNs > 0 ? endTimeNs : System.nanoTime(); + return TimeUnit.NANOSECONDS.toMillis(endNs - startTimeNs); + } + + public String getTarPath() { + return tarPath; + } + + public String getErrorMessage() { + return errorMessage; + } + + public boolean isTerminal() { + return executionState == ExecutionState.SUCCEEDED + || executionState == ExecutionState.FAILED; + } + } + + /** + * Job execution state. + */ + public enum ExecutionState { + RUNNING, + SUCCEEDED, + FAILED + } + + ExportJob(Id id, ExportScope scope, String timestamp, String tarPath, ContainerID startContainerId, + ExportSizing sizing) { + this.id = id; + this.scope = scope; + this.timestamp = timestamp; + this.tarPath = tarPath; + this.startContainerId = startContainerId != null ? startContainerId : ContainerID.valueOf(0); + this.sizing = sizing; + } + + Id getId() { + return id; + } + + ExportScope getScope() { + return scope; + } + + String getTimestamp() { + return timestamp; + } + + ContainerID getStartContainerId() { + return startContainerId; + } + + LifeCycleState getLifeCycleState() { + return scope.getLifeCycleState(); + } + + ContainerHealthState getHealthState() { + return scope.getHealthState(); + } + + long getMaxRows() { + return sizing.getMaxRows(); + } + + int getPageSize() { + return sizing.getPageSize(); + } + + int getShardSize() { + return sizing.getShardSize(); + } + + ExecutionState getExecutionState() { + return executionState; + } + + void setExecutionState(ExecutionState newState) { + this.executionState = newState; + if ((newState == ExecutionState.SUCCEEDED || newState == ExecutionState.FAILED) + && endTimeNs == 0) { + this.endTimeNs = System.nanoTime(); + } + } + + long getEndTimeNs() { + return endTimeNs; + } + + long getTotalRows() { + return totalRows; + } + + void setTotalRows(long rows) { + this.totalRows = rows; + } + + void setStartTimeNs(long startTimeNs) { + this.startTimeNs = startTimeNs; + } + + void setErrorMessage(String message) { + this.errorMessage = message; + } + + String getTarPath() { + return tarPath; + } + + void setTarPath(String path) { + this.tarPath = path; + } + + Status toStatus() { + return new Status(); + } + + String shardFileName(int partIndex) { + return String.format("container-ids-%s-%s-part%03d.txt", + scope.getValue(), timestamp, partIndex); + } + + String shardEntryName(int partIndex) { + return shardFileName(partIndex); + } + + void writeMetadataHeader(BufferedWriter writer, int partNumber, ContainerID shardStartContainerId) + throws IOException { + writer.write("# jobId=" + id.getValue()); + writer.newLine(); + writer.write("# timestamp=" + timestamp); + writer.newLine(); + if (scope.getHealthState() != null) { + writer.write("# healthState=" + scope.getHealthState().name()); + writer.newLine(); + } + if (scope.getLifeCycleState() != null) { + writer.write("# lifecycleState=" + scope.getLifeCycleState().name()); + writer.newLine(); + } + writer.write("# startContainerId=" + shardStartContainerId.getProtobuf().getId()); + writer.newLine(); + writer.write("# part=" + partNumber); + writer.newLine(); + writer.write("# format=container-id-per-line"); + writer.newLine(); + writer.newLine(); + } +} diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportMetrics.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportMetrics.java new file mode 100644 index 000000000000..7e0df1b6f54c --- /dev/null +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportMetrics.java @@ -0,0 +1,80 @@ +/* + * 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 org.apache.hadoop.metrics2.MetricsSystem; +import org.apache.hadoop.metrics2.annotation.Metric; +import org.apache.hadoop.metrics2.annotation.Metrics; +import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem; +import org.apache.hadoop.metrics2.lib.MutableCounterLong; +import org.apache.hadoop.metrics2.lib.MutableGaugeLong; +import org.apache.hadoop.ozone.OzoneConsts; + +/** + * Metrics for async container ID export jobs on SCM. + */ +@Metrics(about = "SCM Container Export Metrics", context = OzoneConsts.OZONE) +public final class ExportMetrics { + + private static final String SOURCE_NAME = ExportMetrics.class.getSimpleName(); + + @Metric(about = "Number of export jobs submitted") + private MutableCounterLong numExportJobsSubmitted; + + @Metric(about = "Number of export jobs succeeded") + private MutableCounterLong numExportJobsSucceeded; + + @Metric(about = "Number of export jobs failed") + private MutableCounterLong numExportJobsFailed; + + @Metric(about = "Rows exported by the most recent successful export job.") + private MutableGaugeLong lastExportRows; + + @Metric(about = "TAR bytes written by the most recent successful export job.") + private MutableGaugeLong lastExportBytesWritten; + + private ExportMetrics() { + } + + public static ExportMetrics create() { + MetricsSystem ms = DefaultMetricsSystem.instance(); + return ms.register(SOURCE_NAME, "SCM Container Export IDs Metrics", new ExportMetrics()); + } + + public void unRegister() { + MetricsSystem ms = DefaultMetricsSystem.instance(); + ms.unregisterSource(SOURCE_NAME); + } + + public void incrExportJobsSubmitted() { + numExportJobsSubmitted.incr(); + } + + public void incrExportJobsSucceeded() { + numExportJobsSucceeded.incr(); + } + + public void incrExportJobsFailed() { + numExportJobsFailed.incr(); + } + + public void recordLastSuccessfulExport(long rows, long bytesWritten) { + lastExportRows.set(rows); + lastExportBytesWritten.set(bytesWritten); + } +} diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportScope.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportScope.java new file mode 100644 index 000000000000..c88cf886cc3f --- /dev/null +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportScope.java @@ -0,0 +1,74 @@ +/* + * 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 org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState; +import org.apache.hadoop.hdds.scm.container.ContainerHealthState; + +/** + * Container listing filters for an export job. + * An export job filters containers by {@link ContainerHealthState}, {@link LifeCycleState} or both. + * Example TAR name: + * {@code container-ids-health-MISSING_lifecycle-OPEN-20260101T120000Z-{jobId}.tar} + */ +public final class ExportScope { + + private final LifeCycleState lifeCycleState; + private final ContainerHealthState healthState; + private final String value; + + private ExportScope(LifeCycleState lifeCycleState, ContainerHealthState healthState, String value) { + this.lifeCycleState = lifeCycleState; + this.healthState = healthState; + this.value = value; + } + + public static ExportScope of(LifeCycleState lifeCycleState, ContainerHealthState healthState) { + StringBuilder sb = new StringBuilder(); + if (healthState != null) { + sb.append("health-").append(healthState.name()); + } + if (lifeCycleState != null) { + if (sb.length() > 0) { + sb.append('_'); + } + sb.append("lifecycle-").append(lifeCycleState.name()); + } + return new ExportScope(lifeCycleState, healthState, sb.toString()); + } + + public LifeCycleState getLifeCycleState() { + return lifeCycleState; + } + + public ContainerHealthState getHealthState() { + return healthState; + } + + /** + * Stable filter name segment used in export TAR and shard file names. + */ + public String getValue() { + return value; + } + + @Override + public String toString() { + return value; + } +} diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportSizing.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportSizing.java new file mode 100644 index 000000000000..21536020b883 --- /dev/null +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/ExportSizing.java @@ -0,0 +1,75 @@ +/* + * 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.apache.hadoop.hdds.scm.container.export.ExportLimits.MAX_PAGE_SIZE; +import static org.apache.hadoop.hdds.scm.container.export.ExportLimits.MAX_SHARD_SIZE; + +/** + * Per-job batch sizing for container ID export. + * These values are always used together when reading container IDs from SCM and writing them into the export TAR: + * {@code pageSize} container IDs returned per {@code ContainerManager} listing query. + * {@code shardSize} - container IDs written into each shard text file before that file is appended to the TAR. + * {@code maxRows} - upper bound on total exported container IDs for the job; {@code 0} means no limit. + *

+ * A {@code pageSize} or {@code shardSize} of {@code 0} in a submit request means use the + * manager defaults from {@link ExportLimits}. + */ +final class ExportSizing { + + private final long maxRows; + private final int pageSize; + private final int shardSize; + + ExportSizing(long maxRows, int pageSize, int shardSize) { + this.maxRows = maxRows; + this.pageSize = pageSize; + this.shardSize = shardSize; + } + + static void validate(long maxRows, int pageSize, int shardSize) { + if (maxRows < 0) { + throw new IllegalArgumentException("maxRows must be non-negative: " + maxRows); + } + if (pageSize < 0 || pageSize > MAX_PAGE_SIZE) { + throw new IllegalArgumentException("pageSize must be between 0 and " + MAX_PAGE_SIZE + ": " + pageSize); + } + if (shardSize < 0 || shardSize > MAX_SHARD_SIZE) { + throw new IllegalArgumentException("shardSize must be between 0 and " + MAX_SHARD_SIZE + ": " + shardSize); + } + } + + static ExportSizing resolve(long maxRows, int pageSize, int shardSize, int defaultPageSize, + int defaultShardSize) { + return new ExportSizing(maxRows, + pageSize > 0 ? pageSize : defaultPageSize, + shardSize > 0 ? shardSize : defaultShardSize); + } + + long getMaxRows() { + return maxRows; + } + + int getPageSize() { + return pageSize; + } + + int getShardSize() { + return shardSize; + } +} diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/export/package-info.java new file mode 100644 index 000000000000..103c9519fcab --- /dev/null +++ b/hadoop-hdds/server-scm/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; 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 100c9feb9b67..e1082e7aee82 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; @@ -282,6 +283,8 @@ public final class StorageContainerManager extends ServiceRuntimeInfoImpl private ReplicationManager replicationManager; + private final ContainerExportManager containerExportManager; + private SCMSafeModeManager scmSafeModeManager; private SCMCertificateClient scmCertificateClient; private RootCARotationManager rootCARotationManager; @@ -436,6 +439,10 @@ private StorageContainerManager(OzoneConfiguration conf, initializeSystemManagers(conf, configurator); + containerExportManager = new ContainerExportManager( + containerManager, this::checkLeader, conf, getScmId()); + containerExportManager.start(); + if (isSecretKeyEnable(securityConfig)) { secretKeyManagerService = new SecretKeyManagerService(scmContext, conf, scmHAManager.getRatisServer()); @@ -1679,6 +1686,8 @@ public void stop() { stopReplicationManager(); + containerExportManager.shutdown(); + try { LOG.info("Stopping the Datanode Admin Monitor."); scmDecommissionManager.stop(); @@ -1930,6 +1939,10 @@ public ReplicationManager getReplicationManager() { return replicationManager; } + public ContainerExportManager getContainerExportManager() { + return containerExportManager; + } + public PlacementPolicy getContainerPlacementPolicy() { return containerPlacementPolicy; } 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 000000000000..afc8662e6621 --- /dev/null +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/export/TestContainerExportManager.java @@ -0,0 +1,336 @@ +/* + * 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.when; + +import java.io.File; +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.UUID; +import java.util.function.BooleanSupplier; +import java.util.stream.Collectors; +import org.apache.commons.io.FileUtils; +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_DEFAULT_PAGE_SIZE = 2; + private static final int TEST_DEFAULT_SHARD_SIZE = 3; + private static final int TEST_MAX_TERMINAL_JOBS = 10; + 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_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, + TEST_MAX_TERMINAL_JOBS, () -> true); + } + + @AfterEach + public void teardown() { + exportManager.shutdown(); + } + + @Test + public void testMultiShardExportCreatesTar() 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, 0, 0, 0); + 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.getTarPath()); + assertTrue(status.getTarPath().endsWith(".tar")); + File tarPath = new File(status.getTarPath()); + assertTrue(tarPath.exists()); + + Path extractDir = Files.createTempDirectory("export-tar"); + try { + Archiver.extract(tarPath, 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 TAR")); + } + assertTrue(Files.readAllLines(extractDir.resolve(part2Name)).contains( + "# startContainerId=4")); + } finally { + FileUtils.deleteQuietly(extractDir.toFile()); + } + } + + @Test + public void testSingleShardExportCreatesTar() throws Exception { + exportManager.shutdown(); + exportManager = newExportManager(100, TEST_DEFAULT_PAGE_SIZE, + TEST_MAX_TERMINAL_JOBS, () -> 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, 0, 0, 0); + assertNotNull(jobId); + + ExportJob.Status status = waitForTerminal(jobId); + if (status.getExecutionState() == ExportJob.ExecutionState.FAILED) { + fail(status.getErrorMessage()); + } + assertEquals(ExportJob.ExecutionState.SUCCEEDED, status.getExecutionState()); + assertTrue(status.getTarPath().endsWith(".tar")); + assertTrue(new File(status.getTarPath()).exists()); + assertFalse(new File(status.getTarPath().replace(".tar", ".txt")).exists()); + } + + @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, 0, 0, 0); + assertNotNull(first); + assertNull(exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.EMPTY, 0, 0, 0)); + } + + @Test + public void testRejectMissingFilters() { + assertThrows(IllegalArgumentException.class, () -> + exportManager.submitJob(ContainerID.valueOf(0), null, null, 0, 0, 0)); + } + + @Test + public void testExportScope() { + assertEquals("health-MISSING", + ExportScope.of(null, ContainerHealthState.MISSING).getValue()); + } + + @Test + public void testMetadataTimestampIsHumanReadable() 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, 0, 0, 0); + assertNotNull(jobId); + waitForTerminal(jobId); + + ExportJob job = exportManager.getJobTracker().get(jobId); + assertTrue(job.getTimestamp().matches("\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z")); + } + + @Test + public void testElapsedMsStableAfterCompletion() 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()); + + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.MISSING, 0, 0, 0); + assertNotNull(jobId); + ExportJob.Status first = waitForTerminal(jobId); + Thread.sleep(200); + ExportJob.Status second = exportManager.getExportStatus(jobId); + assertEquals(first.getElapsedMs(), second.getElapsedMs()); + } + + @Test + public void testTerminalJobEvictionWhenOverCap() throws Exception { + exportManager.shutdown(); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, 2, () -> true); + + 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()); + ExportJob.Id job1 = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); + ExportJob.Status status1 = waitForTerminal(job1); + File tar1 = new File(status1.getTarPath()); + assertTrue(tar1.exists()); + + 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 job2 = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.EMPTY, 0, 0, 0); + waitForTerminal(job2); + + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(0)), anyInt(), isNull(), eq(ContainerHealthState.UNDER_REPLICATED))) + .thenReturn(ids(20)); + when(containerManager.getContainerIDs( + eq(ContainerID.valueOf(21)), anyInt(), isNull(), eq(ContainerHealthState.UNDER_REPLICATED))) + .thenReturn(Collections.emptyList()); + ExportJob.Id job3 = exportManager.submitJob(ContainerID.valueOf(0), null, + ContainerHealthState.UNDER_REPLICATED, 0, 0, 0); + waitForTerminal(job3); + + assertNotNull(exportManager.getExportStatus(job2)); + assertNotNull(exportManager.getExportStatus(job3)); + assertNull(exportManager.getExportStatus(job1)); + assertFalse(tar1.exists()); //evicting a terminal job deletes its TAR too + } + + @Test + public void testOrphanWorkDirRemovedOnStartup() throws Exception { + String jobId = UUID.randomUUID().toString(); + Path orphan = tempDir.toPath().resolve(ContainerExportManager.EXPORT_JOB_DIR_PREFIX + jobId).resolve("work"); + Files.createDirectories(orphan); + exportManager.shutdown(); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, + TEST_MAX_TERMINAL_JOBS, () -> true); + assertFalse(Files.exists(orphan)); + } + + @Test + public void testIncompleteExportArtifactsRemovedOnStartup() throws Exception { + String jobId = UUID.randomUUID().toString(); + Path jobDir = tempDir.toPath().resolve(ContainerExportManager.EXPORT_JOB_DIR_PREFIX + jobId).resolve("work"); + Files.createDirectories(jobDir); + File partialTar = new File(tempDir, "container-ids-health-MISSING-20260101T000000Z-" + jobId + ".tar"); + assertTrue(partialTar.createNewFile()); + File inProgress = new File(tempDir, jobId + ContainerExportManager.IN_PROGRESS_MARKER_SUFFIX); + assertTrue(inProgress.createNewFile()); + + exportManager.shutdown(); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, + TEST_MAX_TERMINAL_JOBS, () -> true); + + assertFalse(Files.exists(jobDir)); + assertFalse(partialTar.exists()); + assertFalse(inProgress.exists()); + } + + @Test + public void testCompletedExportTarRetainedOnRestart() 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()); + + ExportJob.Id jobId = exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0); + ExportJob.Status status = waitForTerminal(jobId); + assertEquals(ExportJob.ExecutionState.SUCCEEDED, status.getExecutionState()); + File tar = new File(status.getTarPath()); + assertTrue(tar.exists()); + + exportManager.shutdown(); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, + TEST_MAX_TERMINAL_JOBS, () -> true); + assertTrue(tar.exists()); + assertNull(exportManager.getExportStatus(jobId)); + } + + @Test + public void testRejectSubmitWhenNotLeader() throws Exception { + exportManager.shutdown(); + exportManager = newExportManager(TEST_DEFAULT_SHARD_SIZE, TEST_DEFAULT_PAGE_SIZE, 2, () -> false); + assertNull(exportManager.submitJob(ContainerID.valueOf(0), null, ContainerHealthState.MISSING, 0, 0, 0)); + } + + private ContainerExportManager newExportManager(int defaultShardSize, int defaultPageSize, int maxTerminalJobs, + BooleanSupplier isLeaderReady) throws Exception { + ContainerExportManager manager = new ContainerExportManager(containerManager, isLeaderReady, + tempDir.getAbsolutePath(), defaultShardSize, defaultPageSize, maxTerminalJobs, TEST_SCM_ID); + 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.isTerminal(); + }, 100, 30_000); + return exportManager.getExportStatus(jobId); + } +}