diff --git a/pom.xml b/pom.xml index f2e31e8..147d0d8 100644 --- a/pom.xml +++ b/pom.xml @@ -86,6 +86,22 @@ true + + + + com.kingsrook.qbits + qbit-object-storage + 0.1.0-SNAPSHOT + true + + + + org.testcontainers + testcontainers + 2.0.3 + test + + org.junit.jupiter junit-jupiter @@ -101,6 +117,17 @@ + + org.apache.maven.plugins + maven-surefire-plugin + + + + minioadmin + minioadmin + + + org.jacoco jacoco-maven-plugin diff --git a/src/main/java/com/kingsrook/qbits/backups/storage/objectstorage/ObjectStorageBackupStorage.java b/src/main/java/com/kingsrook/qbits/backups/storage/objectstorage/ObjectStorageBackupStorage.java new file mode 100644 index 0000000..dddcaaa --- /dev/null +++ b/src/main/java/com/kingsrook/qbits/backups/storage/objectstorage/ObjectStorageBackupStorage.java @@ -0,0 +1,453 @@ +/* + * Copyright 2026 Kingsrook, LLC + * + * Licensed 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 com.kingsrook.qbits.backups.storage.objectstorage; + + +import java.io.IOException; +import java.io.InputStream; +import java.io.InterruptedIOException; +import java.io.OutputStream; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; +import java.time.Duration; +import java.util.ArrayList; +import java.util.EnumSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.function.Function; +import com.kingsrook.qbits.backups.storage.BackupStorage; +import com.kingsrook.qbits.backups.storage.BackupStorageException; +import com.kingsrook.qbits.backups.storage.StoredObjectInfo; +import com.kingsrook.qbits.backups.storage.WriteOptions; +import com.kingsrook.qbits.objectstorage.api.Capability; +import com.kingsrook.qbits.objectstorage.api.CapabilityStatus; +import com.kingsrook.qbits.objectstorage.api.CopyRequest; +import com.kingsrook.qbits.objectstorage.api.DeleteMode; +import com.kingsrook.qbits.objectstorage.api.DeleteRequest; +import com.kingsrook.qbits.objectstorage.api.ListRequest; +import com.kingsrook.qbits.objectstorage.api.LockMode; +import com.kingsrook.qbits.objectstorage.api.ObjectStorage; +import com.kingsrook.qbits.objectstorage.api.ObjectStream; +import com.kingsrook.qbits.objectstorage.api.ObjectSummary; +import com.kingsrook.qbits.objectstorage.api.Page; +import com.kingsrook.qbits.objectstorage.api.PresignGet; +import com.kingsrook.qbits.objectstorage.api.PutRequest; +import com.kingsrook.qbits.objectstorage.api.Range; +import com.kingsrook.qbits.objectstorage.api.StorageBucket; +import com.kingsrook.qbits.objectstorage.api.StorageException; +import com.kingsrook.qbits.objectstorage.api.StoredObject; + + +/******************************************************************************* + ** BackupStorage over a qbit-object-storage StorageBucket. Backups never builds + ** an S3 client and never holds a credential: every byte goes through the + ** storage qbit, in a system context tagged with the backups caller name (so + ** the object_storage_operation log shows "system: backups run ..."). + ** + ** A write of unknown length is bridged from an OutputStream to the put of an + ** InputStream (the storage qbit uses multipart above its threshold, so memory + ** stays bounded); closing the stream waits for the put and rethrows its failure. + ** Permanent delete uses DeleteRequest.permanent() (system callers only), and a + ** locked object comes back as LOCKED. + ** + ** This class is optional: it only loads when qbit-object-storage is on the classpath. + *******************************************************************************/ +public class ObjectStorageBackupStorage implements BackupStorage +{ + private final StorageBucket bucket; + + + + /******************************************************************************* + ** Constructor over a bucket (already in the right user or system context). + *******************************************************************************/ + public ObjectStorageBackupStorage(StorageBucket bucket) + { + this.bucket = bucket; + } + + + + /******************************************************************************* + ** A resolver (bucket alias to storage) that talks to the storage qbit as the + ** system caller "backups", tagged with the given run id (any text). + *******************************************************************************/ + public static Function systemResolver(String runId) + { + return (alias -> new ObjectStorageBackupStorage(ObjectStorage.asSystem("backups", runId).bucket(alias))); + } + + + + /******************************************************************************* + ** {@inheritDoc} + *******************************************************************************/ + @Override + public OutputStream openWrite(String key, WriteOptions options) + { + try + { + PipedInputStream in = new PipedInputStream(1 << 20); + PipedOutputStream out = new PipedOutputStream(in); + + PutRequest request = PutRequest.of(key, in).withContentType(options.getContentType()).withMetadata(new LinkedHashMap<>(options.getMetadata())); + if(options.getObjectLockMode() != null) + { + request.withObjectLock(LockMode.valueOf(options.getObjectLockMode()), options.getRetainUntil()); + } + + CompletableFuture put = CompletableFuture.runAsync(() -> + { + try + { + bucket.put(request); + } + finally + { + try + { + in.close(); + } + catch(IOException e) + { + // nothing more to do: the writer side sees the broken pipe + } + } + }); + + return (new OutputStream() + { + /******************************************************************************* + ** + *******************************************************************************/ + @Override + public void write(int b) throws IOException + { + ensureWriterAlive(); + out.write(b); + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + @Override + public void write(byte[] b, int off, int len) throws IOException + { + ensureWriterAlive(); + try + { + out.write(b, off, len); + } + catch(IOException e) + { + throw (failureOr(e)); + } + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + @Override + public void flush() throws IOException + { + out.flush(); + } + + + + /******************************************************************************* + ** Close the pipe, then wait for the put to finish, rethrowing its failure. + *******************************************************************************/ + @Override + public void close() throws IOException + { + try + { + out.close(); + put.join(); + } + catch(java.util.concurrent.CompletionException e) + { + throw (asIOException(e.getCause())); + } + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + private void ensureWriterAlive() throws IOException + { + if(put.isCompletedExceptionally()) + { + try + { + put.join(); + } + catch(java.util.concurrent.CompletionException e) + { + throw (asIOException(e.getCause())); + } + } + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + private IOException failureOr(IOException original) throws IOException + { + try + { + put.join(); + } + catch(java.util.concurrent.CompletionException e) + { + return (asIOException(e.getCause())); + } + return (original); + } + }); + } + catch(IOException e) + { + throw (new BackupStorageException(BackupStorageException.Reason.PROVIDER_ERROR, "Could not open a write stream for " + key, e)); + } + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + private static IOException asIOException(Throwable cause) + { + if(cause instanceof StorageException se) + { + return (new IOException(convert(se).getMessage(), convert(se))); + } + return (cause instanceof IOException io ? io : new IOException(String.valueOf(cause), cause)); + } + + + + /******************************************************************************* + ** {@inheritDoc} + *******************************************************************************/ + @Override + public InputStream openRead(String key) + { + try + { + ObjectStream stream = bucket.get(key, Range.all()); + return (new java.io.FilterInputStream(stream.getInputStream()) + { + /******************************************************************************* + ** + *******************************************************************************/ + @Override + public void close() throws IOException + { + try + { + super.close(); + } + finally + { + stream.close(); + } + } + }); + } + catch(StorageException e) + { + throw (convert(e)); + } + } + + + + /******************************************************************************* + ** {@inheritDoc} + *******************************************************************************/ + @Override + public Optional head(String key) + { + try + { + return (bucket.head(key).map(ObjectStorageBackupStorage::info)); + } + catch(StorageException e) + { + throw (convert(e)); + } + } + + + + /******************************************************************************* + ** {@inheritDoc} + *******************************************************************************/ + @Override + public List list(String prefix) + { + try + { + List rs = new ArrayList<>(); + String after = null; + do + { + Page page = bucket.list(ListRequest.prefix(prefix == null ? "" : prefix).pageSize(1000).after(after)); + for(ObjectSummary summary : page.getItems()) + { + rs.add(new StoredObjectInfo(summary.getKey(), summary.getSize() == null ? 0 : summary.getSize(), null, new LinkedHashMap<>(), null)); + } + after = page.getNextToken(); + } + while(after != null); + return (rs); + } + catch(StorageException e) + { + throw (convert(e)); + } + } + + + + /******************************************************************************* + ** {@inheritDoc} + *******************************************************************************/ + @Override + public void copy(String sourceKey, String destinationKey) + { + try + { + bucket.copy(CopyRequest.of(sourceKey, destinationKey)); + } + catch(StorageException e) + { + throw (convert(e)); + } + } + + + + /******************************************************************************* + ** {@inheritDoc} + *******************************************************************************/ + @Override + public DeleteResult deletePermanent(String key) + { + try + { + DeleteMode mode = bucket.delete(DeleteRequest.of(key).permanent()).getMode(); + return (switch(mode) + { + case SKIPPED_OBJECT_LOCK -> DeleteResult.LOCKED; + case NOT_FOUND -> DeleteResult.NOT_FOUND; + default -> DeleteResult.DELETED; + }); + } + catch(StorageException e) + { + throw (convert(e)); + } + } + + + + /******************************************************************************* + ** {@inheritDoc} + *******************************************************************************/ + @Override + public String presignGet(String key, Duration ttl, String downloadName) + { + try + { + return (bucket.presignGet(PresignGet.of(key).withTtl(ttl).withDownloadAs(downloadName)).toString()); + } + catch(StorageException e) + { + throw (convert(e)); + } + } + + + + /******************************************************************************* + ** {@inheritDoc} + *******************************************************************************/ + @Override + public Set capabilities() + { + Set rs = EnumSet.of(StorageCapability.SERVER_SIDE_COPY); + if(bucket.capability(Capability.OBJECT_LOCK) == CapabilityStatus.SUPPORTED) + { + rs.add(StorageCapability.OBJECT_LOCK); + } + if(bucket.capability(Capability.VERSIONING) == CapabilityStatus.SUPPORTED || bucket.isVersioned()) + { + rs.add(StorageCapability.VERSIONING); + } + if(bucket.capability(Capability.MULTIPART) == CapabilityStatus.SUPPORTED) + { + rs.add(StorageCapability.MULTIPART); + } + return (rs); + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + private static StoredObjectInfo info(StoredObject object) + { + java.time.Instant lockedUntil = object.getLockRetainUntil(); + return (new StoredObjectInfo(object.getKey(), object.getSize() == null ? 0 : object.getSize(), object.getLastModified(), + object.getHeaders() == null || object.getHeaders().getUserMetadata() == null ? new LinkedHashMap<>() : new LinkedHashMap<>(object.getHeaders().getUserMetadata()), lockedUntil)); + } + + + + /******************************************************************************* + ** Map a storage qbit failure onto a BackupStorageException. + *******************************************************************************/ + static BackupStorageException convert(StorageException e) + { + BackupStorageException.Reason reason = switch(e.getReason()) + { + case NOT_FOUND -> BackupStorageException.Reason.NOT_FOUND; + case ACCESS_DENIED -> BackupStorageException.Reason.ACCESS_DENIED; + case UNSUPPORTED -> BackupStorageException.Reason.UNSUPPORTED; + case PRECONDITION_FAILED -> BackupStorageException.Reason.PRECONDITION_FAILED; + case THROTTLED -> BackupStorageException.Reason.THROTTLED; + default -> BackupStorageException.Reason.PROVIDER_ERROR; + }; + return (new BackupStorageException(reason, e.getMessage() + (e.getRequestId() == null ? "" : " (request " + e.getRequestId() + ")"), e)); + } +} diff --git a/src/test/java/com/kingsrook/qbits/backups/storage/objectstorage/ObjectStorageBackupStorageMinioTest.java b/src/test/java/com/kingsrook/qbits/backups/storage/objectstorage/ObjectStorageBackupStorageMinioTest.java new file mode 100644 index 0000000..6285e39 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/backups/storage/objectstorage/ObjectStorageBackupStorageMinioTest.java @@ -0,0 +1,193 @@ +/* + * Copyright 2026 Kingsrook, LLC + * + * Licensed 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 com.kingsrook.qbits.backups.storage.objectstorage; + + +import java.io.OutputStream; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import com.kingsrook.qbits.backups.BackupsRuntime; +import com.kingsrook.qbits.backups.BackupsTestUtils; +import com.kingsrook.qbits.backups.manifest.BackupManifest; +import com.kingsrook.qbits.backups.model.BackupsTables; +import com.kingsrook.qbits.backups.run.BackupCatalog; +import com.kingsrook.qbits.backups.run.BackupEngine; +import com.kingsrook.qbits.backups.run.BackupRunFactory; +import com.kingsrook.qbits.backups.run.BackupRunResult; +import com.kingsrook.qbits.backups.storage.BackupStorage; +import com.kingsrook.qbits.backups.storage.BackupStorageException; +import com.kingsrook.qbits.backups.storage.StoredObjectInfo; +import com.kingsrook.qbits.backups.storage.WriteOptions; +import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig; +import com.kingsrook.qbits.objectstorage.api.ObjectStorage; +import com.kingsrook.qbits.objectstorage.config.BucketRole; +import com.kingsrook.qbits.objectstorage.config.ProviderType; +import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig; +import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig; +import com.kingsrook.qqq.backend.core.model.data.QRecord; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.containers.wait.strategy.Wait; +import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; +import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.S3Configuration; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + + +/******************************************************************************* + ** The backups qbit against a real S3-compatible store (MinIO, with object lock), + ** through the object-storage qbit: the adapter contract, and a full backup run. + *******************************************************************************/ +class ObjectStorageBackupStorageMinioTest +{ + private static GenericContainer minio; + private static BackupStorage storage; + + + + /******************************************************************************* + ** + *******************************************************************************/ + @BeforeAll + static void startMinio() throws Exception + { + org.junit.jupiter.api.Assumptions.assumeTrue(org.testcontainers.DockerClientFactory.instance().isDockerAvailable(), "Docker is not available: skipping the MinIO tests"); + minio = new GenericContainer<>("minio/minio:latest").withCommand("server", "/data").withExposedPorts(9000) + .withEnv("MINIO_ROOT_USER", "minioadmin").withEnv("MINIO_ROOT_PASSWORD", "minioadmin") + .waitingFor(Wait.forHttp("/minio/health/live").forPort(9000)); + try + { + minio.start(); + } + catch(RuntimeException e) + { + // for example CI with a remote Docker that cannot pull the image: skip, never fail the build + minio = null; + org.junit.jupiter.api.Assumptions.assumeTrue(false, "MinIO could not be started: " + e.getMessage()); + } + String endpoint = "http://" + minio.getHost() + ":" + minio.getMappedPort(9000); + + try(S3Client s3 = S3Client.builder().endpointOverride(URI.create(endpoint)).region(Region.US_EAST_1) + .credentialsProvider(StaticCredentialsProvider.create(AwsBasicCredentials.create("minioadmin", "minioadmin"))) + .serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(true).build()).build()) + { + s3.createBucket(b -> b.bucket("backups").objectLockEnabledForBucket(true)); + } + + ObjectStorage.configureStandalone(new ObjectStorageQBitConfig().withBackendName("memory") + .withConnection(new StorageConnectionConfig().withName("minio").withProviderType(ProviderType.MINIO).withEndpoint(endpoint).withRegion("us-east-1") + .withAccessKeyRef("${env.STORAGE_TEST_ACCESS_KEY}").withSecretKeyRef("${env.STORAGE_TEST_SECRET_KEY}")) + .withBucket(new StorageBucketConfig().withAlias("backups").withConnection("minio").withBucketName("backups").withRole(BucketRole.BACKUP).withIndexed(false))); + storage = ObjectStorageBackupStorage.systemResolver("test").apply("backups"); + } + + + + /******************************************************************************* + ** + *******************************************************************************/ + @AfterAll + static void stopMinio() + { + if(minio != null) + { + minio.stop(); + } + } + + + + /******************************************************************************* + ** Streamed write of unknown length, head, ranged-free read, list, copy, delete, presign. + *******************************************************************************/ + @Test + void testAdapterContract() throws Exception + { + byte[] big = new byte[20 * 1024 * 1024]; + new java.util.Random(7).nextBytes(big); + try(OutputStream out = storage.openWrite("contract/big.bin", WriteOptions.create().withContentType("application/octet-stream").withMetadata("qqq-backup-run", "r1"))) + { + out.write(big, 0, 5_000_000); + out.write(big, 5_000_000, big.length - 5_000_000); + } + StoredObjectInfo head = storage.head("contract/big.bin").orElseThrow(); + assertThat(head.size()).isEqualTo(big.length); + assertThat(storage.openRead("contract/big.bin").readAllBytes()).isEqualTo(big); + assertThat(storage.head("contract/none")).isEmpty(); + assertThatThrownBy(() -> storage.openRead("contract/none")).isInstanceOf(BackupStorageException.class) + .extracting(e -> ((BackupStorageException) e).getReason()).isEqualTo(BackupStorageException.Reason.NOT_FOUND); + + storage.copy("contract/big.bin", "contract/copy.bin"); + assertThat(storage.list("contract/")).extracting(StoredObjectInfo::key).containsExactly("contract/big.bin", "contract/copy.bin"); + assertThat(storage.presignGet("contract/copy.bin", Duration.ofMinutes(15), "x.bin")).startsWith("http"); + assertThat(storage.deletePermanent("contract/copy.bin")).isEqualTo(BackupStorage.DeleteResult.DELETED); + assertThat(storage.deletePermanent("contract/copy.bin")).isEqualTo(BackupStorage.DeleteResult.NOT_FOUND); + } + + + + /******************************************************************************* + ** Object lock: a locked object is not deleted, and says so. + *******************************************************************************/ + @Test + void testObjectLockedObjectIsNotDeleted() throws Exception + { + try(OutputStream out = storage.openWrite("lock/a.txt", WriteOptions.create().withObjectLock("GOVERNANCE", Instant.now().plus(Duration.ofDays(1))))) + { + out.write("locked".getBytes(StandardCharsets.UTF_8)); + } + assertThat(storage.head("lock/a.txt").orElseThrow().lockedUntil()).isNotNull(); + assertThat(storage.deletePermanent("lock/a.txt")).isEqualTo(BackupStorage.DeleteResult.LOCKED); + assertThat(storage.head("lock/a.txt")).isPresent(); + } + + + + /******************************************************************************* + ** A whole backup run through the storage qbit: manifest last, parts and files in + ** the bucket, and the manifest is readable back. + *******************************************************************************/ + @Test + void testBackupRunAgainstMinio() throws Exception + { + BackupsTestUtils.init(); + BackupsTestUtils.loadBlog(5); + BackupsRuntime.getConfig().withStorageResolver(alias -> "backups".equals(alias) ? storage : BackupsTestUtils.MEDIA); + + Integer id = new BackupCatalog().insert(BackupsTables.PLAN, Map.of("name", "Nightly", "planKey", "minio", "isActive", true, "scopeType", "INSTANCE", + "storageBucketAlias", "backups", "slaHours", 26)); + QRecord plan = new BackupCatalog().get(BackupsTables.PLAN, id).orElseThrow(); + + BackupRunResult result = new BackupEngine().execute(BackupRunFactory.forPlan(plan, "MANUAL")); + assertThat(result.isUsable()).as(result.getErrorMessage()).isTrue(); + + BackupManifest manifest = BackupManifest.read(storage.openRead(result.getManifestKey())); + assertThat(manifest.table("blogPost").rowCount()).isEqualTo(5); + List objects = storage.list(result.getRunPrefix()); + assertThat(objects).extracting(StoredObjectInfo::key).contains(result.getRunPrefix() + "manifest.json", result.getRunPrefix() + "checksums.sha256"); + assertThat(storage.list("qqq-backups/testapp/_files/sha256/")).isNotEmpty(); + } +}