From b13122f1b04847bd7d8eaf43aa2e89b108629d5d Mon Sep 17 00:00:00 2001 From: James Maes Date: Wed, 14 Oct 2026 13:39:38 -0500 Subject: [PATCH 1/3] feat(health): deadline-aware read-only capability probe Co-Authored-By: Claude Opus 5.5 --- .../internal/CapabilityProbe.java | 162 ++++++++++++++++-- .../objectstorage/internal/RateLimiter.java | 50 +++++- .../processes/TestConnectionProcess.java | 2 +- 3 files changed, 192 insertions(+), 22 deletions(-) diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/internal/CapabilityProbe.java b/src/main/java/com/kingsrook/qbits/objectstorage/internal/CapabilityProbe.java index 14f93e0..4345197 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/internal/CapabilityProbe.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/internal/CapabilityProbe.java @@ -21,7 +21,9 @@ package com.kingsrook.qbits.objectstorage.internal; +import javax.net.ssl.SSLException; import java.net.URI; +import java.net.UnknownHostException; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; @@ -35,6 +37,7 @@ import java.util.EnumMap; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.function.Function; import com.kingsrook.qbits.objectstorage.api.Capability; import com.kingsrook.qbits.objectstorage.api.CapabilityStatus; import com.kingsrook.qbits.objectstorage.api.PresignedPost; @@ -45,7 +48,9 @@ import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig; import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables; import com.kingsrook.qqq.backend.core.logging.QLogger; import com.kingsrook.qqq.backend.core.model.data.QRecord; +import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration; import software.amazon.awssdk.core.sync.RequestBody; +import software.amazon.awssdk.services.s3.DelegatingS3Client; import software.amazon.awssdk.services.s3.S3Client; import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest; import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload; @@ -61,6 +66,7 @@ import software.amazon.awssdk.services.s3.model.ListObjectsV2Response; import software.amazon.awssdk.services.s3.model.ObjectIdentifier; import software.amazon.awssdk.services.s3.model.PutObjectRequest; import software.amazon.awssdk.services.s3.model.S3Exception; +import software.amazon.awssdk.services.s3.model.S3Request; import software.amazon.awssdk.services.s3.model.UploadPartRequest; import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest; import software.amazon.awssdk.services.s3.presigner.model.PutObjectPresignRequest; @@ -79,6 +85,7 @@ public final class CapabilityProbe private static final QLogger LOG = QLogger.getLogger(CapabilityProbe.class); public static final String PROBE_PREFIX = ".qqq-probe/"; + public static final int CLOCK_SKEW_THRESHOLD_SECONDS = 900; private static final int MIN_PART_BYTES = 5 * 1024 * 1024; @@ -94,6 +101,17 @@ public final class CapabilityProbe private final List problems = new ArrayList<>(); private Long clockSkewSeconds; private Boolean presignHostReachable; + private String problemKind; + + + + /******************************************************************************* + ** Safe connection failure category, without provider messages or identifiers. + *******************************************************************************/ + public String getProblemKind() + { + return (problemKind); + } @@ -209,7 +227,7 @@ public final class CapabilityProbe *******************************************************************************/ public static Report probeConnection(String connectionName, boolean includeWrite) { - details.clear(); + details.get().clear(); Report report = new Report(); ConnectionRuntime connection = ObjectStorageRuntime.connection(connectionName); StorageConnectionConfig config = connection.getConfig(); @@ -230,7 +248,7 @@ public final class CapabilityProbe List notes = new ArrayList<>(); report.notes.put(bucket.getAlias(), notes); - readProbes(connection, bucket, results, report); + readProbes(connection, bucket, results, report, null); recordVersioningState(bucket); if(includeWrite) @@ -251,7 +269,7 @@ public final class CapabilityProbe } for(Map.Entry e : results.entrySet()) { - CapabilityStore.record(bucket, e.getKey(), e.getValue(), details.getOrDefault(bucket.getAlias() + ":" + e.getKey(), null)); + CapabilityStore.record(bucket, e.getKey(), e.getValue(), details.get().getOrDefault(bucket.getAlias() + ":" + e.getKey(), null)); } } @@ -262,7 +280,7 @@ public final class CapabilityProbe report.byBucket.get(writable.getAlias()).putAll(presign); for(Map.Entry e : presign.entrySet()) { - CapabilityStore.record(writable, e.getKey(), e.getValue(), details.get(writable.getAlias() + ":" + e.getKey())); + CapabilityStore.record(writable, e.getKey(), e.getValue(), details.get().get(writable.getAlias() + ":" + e.getKey())); } } @@ -281,7 +299,116 @@ public final class CapabilityProbe return (report); } - private static final Map details = new java.util.concurrent.ConcurrentHashMap<>(); + private static final ThreadLocal> details = ThreadLocal.withInitial(java.util.HashMap::new); + + + + /******************************************************************************* + ** Reuse the read probes within a health deadline, without storing diagnostics + ** or running write/presign probes. Only the returned report is consumed. + *******************************************************************************/ + public static Report probeConnection(String connectionName, Instant deadline) + { + try + { + return (probeReadOnly(connectionName, deadline)); + } + finally + { + details.remove(); + } + } + + + + /******************************************************************************* + ** Run the shared read-only calls without updating capability or bucket rows. + *******************************************************************************/ + private static Report probeReadOnly(String connectionName, Instant deadline) + { + Report report = new Report(); + ConnectionRuntime connection = ObjectStorageRuntime.connection(connectionName); + for(StorageBucketConfig bucket : ObjectStorageRuntime.getConfig().getBuckets()) + { + if(!connectionName.equals(bucket.getConnection())) + { + continue; + } + if(!Instant.now().isBefore(deadline) || Thread.currentThread().isInterrupted()) + { + report.problemKind = "timeout"; + break; + } + Map results = new EnumMap<>(Capability.class); + report.byBucket.put(bucket.getAlias(), results); + readProbes(connection, bucket, results, report, deadline); + } + if(!Instant.now().isBefore(deadline)) + { + report.problemKind = "timeout"; + } + return (report); + } + + + + /******************************************************************************* + ** Classify failures using types and known protocol codes, never message text. + *******************************************************************************/ + public static String problemKind(Throwable error) + { + for(Throwable cause = error; cause != null; cause = cause.getCause()) + { + if(cause instanceof UnknownHostException) + { + return ("dns"); + } + if(cause instanceof SSLException) + { + return ("tls"); + } + String code = ErrorMapper.errorCode(cause); + if("RequestTimeTooSkewed".equals(code) || "RequestExpired".equals(code)) + { + return ("clock"); + } + if(cause instanceof S3Exception s3 && (s3.statusCode() == 401 || s3.statusCode() == 403 + || "InvalidAccessKeyId".equals(code) || "SignatureDoesNotMatch".equals(code))) + { + return ("auth"); + } + } + return ("connection"); + } + + + + /******************************************************************************* + ** Bound each SDK call (including retries) by the remaining probe budget. + *******************************************************************************/ + private static S3Client boundedClient(S3Client client, Instant deadline) + { + return (new DelegatingS3Client(client) + { + /******************************************************************************* + ** Preserve credentials and other overrides while limiting the request. + *******************************************************************************/ + @Override + @SuppressWarnings("unchecked") // toBuilder retains the concrete request type. + protected ReturnT invokeOperation(T request, Function operation) + { + Duration remaining = Duration.between(Instant.now(), deadline); + if(remaining.isNegative() || remaining.isZero() || Thread.currentThread().isInterrupted()) + { + throw (new IllegalStateException("Storage probe deadline exceeded")); + } + AwsRequestOverrideConfiguration.Builder overrides = request.overrideConfiguration() + .map(AwsRequestOverrideConfiguration::toBuilder).orElseGet(AwsRequestOverrideConfiguration::builder); + T bounded = (T) request.toBuilder().overrideConfiguration(overrides.apiCallTimeout(remaining).apiCallAttemptTimeout(remaining).build()).build(); + return (operation.apply(bounded)); + } + }); + } @@ -291,7 +418,7 @@ public final class CapabilityProbe *******************************************************************************/ private static void recordVersioningState(StorageBucketConfig bucket) { - String seen = details.get(bucket.getAlias() + ":" + Capability.VERSIONING); + String seen = details.get().get(bucket.getAlias() + ":" + Capability.VERSIONING); if(seen == null || !Db.available(ObjectStorageTables.BUCKET)) { return; @@ -317,14 +444,14 @@ public final class CapabilityProbe /******************************************************************************* ** The read-only probes of one bucket. *******************************************************************************/ - private static void readProbes(ConnectionRuntime connection, StorageBucketConfig bucket, Map results, Report report) + private static void readProbes(ConnectionRuntime connection, StorageBucketConfig bucket, Map results, Report report, Instant deadline) { - S3Client client = connection.client(); + S3Client client = deadline == null ? connection.client() : boundedClient(connection.client(), deadline); String name = name(bucket); String alias = bucket.getAlias(); String sampleKey = null; - try(RateLimiter.Permit ignored = connection.getRateLimiter().acquire()) + try(RateLimiter.Permit ignored = deadline == null ? connection.getRateLimiter().acquire() : connection.getRateLimiter().acquire(deadline)) { ListObjectsV2Response r = client.listObjectsV2(ListObjectsV2Request.builder().bucket(name).maxKeys(1).build()); results.put(Capability.LIST_V2, CapabilityStatus.SUPPORTED); @@ -352,6 +479,7 @@ public final class CapabilityProbe { results.put(Capability.LIST_V2, classify(alias, Capability.LIST_V2, e)); report.problems.add(alias + ": listing failed (" + shortMessage(e) + ")"); + report.problemKind = problemKind(e); } probe(results, alias, Capability.VERSIONING, () -> @@ -359,10 +487,10 @@ public final class CapabilityProbe String status = client.getBucketVersioning(b -> b.bucket(name)).statusAsString(); if(status == null && ProviderDefaults.defaultStatus(connection.getConfig().getProviderType(), Capability.VERSIONING) == CapabilityStatus.UNSUPPORTED) { - details.put(alias + ":" + Capability.VERSIONING, "the provider reports versioning off and documents it as a stub"); + details.get().put(alias + ":" + Capability.VERSIONING, "the provider reports versioning off and documents it as a stub"); return (CapabilityStatus.UNSUPPORTED); } - details.put(alias + ":" + Capability.VERSIONING, status == null ? "not enabled" : status); + details.get().put(alias + ":" + Capability.VERSIONING, status == null ? "not enabled" : status); return (CapabilityStatus.SUPPORTED); }); probe(results, alias, Capability.OBJECT_LOCK, () -> @@ -376,7 +504,7 @@ public final class CapabilityProbe { if("ObjectLockConfigurationNotFoundError".equals(ErrorMapper.errorCode(e))) { - details.put(alias + ":" + Capability.OBJECT_LOCK, "object lock is not enabled on this bucket"); + details.get().put(alias + ":" + Capability.OBJECT_LOCK, "object lock is not enabled on this bucket"); return (CapabilityStatus.UNSUPPORTED); } throw (e); @@ -391,7 +519,7 @@ public final class CapabilityProbe if(key == null) { results.put(Capability.OBJECT_TAGGING, CapabilityStatus.UNKNOWN); - details.put(alias + ":" + Capability.OBJECT_TAGGING, "no object in the bucket to test"); + details.get().put(alias + ":" + Capability.OBJECT_TAGGING, "no object in the bucket to test"); } else { @@ -505,7 +633,7 @@ public final class CapabilityProbe try { client.putObject(PutObjectRequest.builder().bucket(name).key(key).ifNoneMatch("*").build(), RequestBody.fromString("2")); - details.put(alias + ":" + Capability.CONDITIONAL_WRITE, "If-None-Match was ignored"); + details.get().put(alias + ":" + Capability.CONDITIONAL_WRITE, "If-None-Match was ignored"); return (CapabilityStatus.UNSUPPORTED); } catch(S3Exception e) @@ -564,7 +692,7 @@ public final class CapabilityProbe report.presignHostReachable = true; if(r.statusCode() == 200 && !"probe".equals(r.body())) { - details.put(alias + ":" + Capability.PRESIGNED_GET, "the presigned GET returned unexpected content (does it reach the same store?)"); + details.get().put(alias + ":" + Capability.PRESIGNED_GET, "the presigned GET returned unexpected content (does it reach the same store?)"); return (CapabilityStatus.UNKNOWN); } return (httpStatus(alias, Capability.PRESIGNED_GET, r.statusCode(), r.body())); @@ -618,7 +746,7 @@ public final class CapabilityProbe { return (CapabilityStatus.SUPPORTED); } - details.put(alias + ":" + capability, "HTTP " + status + (body == null ? "" : " " + body.replaceAll("\\s+", " ").substring(0, Math.min(200, body.length())))); + details.get().put(alias + ":" + capability, "HTTP " + status + (body == null ? "" : " " + body.replaceAll("\\s+", " ").substring(0, Math.min(200, body.length())))); if(status == 403) { return (CapabilityStatus.NOT_PERMITTED); @@ -669,7 +797,7 @@ public final class CapabilityProbe static CapabilityStatus classify(String alias, Capability capability, Exception e) { String detail = shortMessage(e); - details.put(alias + ":" + capability, detail); + details.get().put(alias + ":" + capability, detail); if(e instanceof S3Exception s3) { String code = ErrorMapper.errorCode(s3); diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/internal/RateLimiter.java b/src/main/java/com/kingsrook/qbits/objectstorage/internal/RateLimiter.java index aa23227..4920933 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/internal/RateLimiter.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/internal/RateLimiter.java @@ -21,7 +21,10 @@ package com.kingsrook.qbits.objectstorage.internal; +import java.time.Duration; +import java.time.Instant; import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; /******************************************************************************* @@ -57,17 +60,52 @@ public class RateLimiter *******************************************************************************/ public Permit acquire() { + return (acquire(null)); + } + + + + /******************************************************************************* + ** Wait no later than a caller's deadline, releasing the slot on interruption. + *******************************************************************************/ + public Permit acquire(Instant deadline) + { + boolean acquired = false; try { - concurrency.acquire(); - waitForToken(); + if(deadline == null) + { + concurrency.acquire(); + acquired = true; + } + else + { + acquired = concurrency.tryAcquire(Math.max(0, Duration.between(Instant.now(), deadline).toNanos()), TimeUnit.NANOSECONDS); + if(!acquired) + { + throw (new IllegalStateException("Storage probe deadline exceeded")); + } + } + waitForToken(deadline); return (new Permit(concurrency)); } catch(InterruptedException e) { + if(acquired) + { + concurrency.release(); + } Thread.currentThread().interrupt(); throw (new IllegalStateException("Interrupted while waiting for the storage rate limiter", e)); } + catch(RuntimeException e) + { + if(acquired) + { + concurrency.release(); + } + throw (e); + } } @@ -75,10 +113,14 @@ public class RateLimiter /******************************************************************************* ** Block until a token is available. *******************************************************************************/ - private void waitForToken() throws InterruptedException + private void waitForToken(Instant deadline) throws InterruptedException { while(true) { + if(deadline != null && !Instant.now().isBefore(deadline)) + { + throw (new IllegalStateException("Storage probe deadline exceeded")); + } long sleepMillis; synchronized(this) { @@ -93,7 +135,7 @@ public class RateLimiter throttledWaits++; sleepMillis = Math.max(1, (long) ((1.0 - tokens) / opsPerSecond * 1000)); } - Thread.sleep(sleepMillis); + Thread.sleep(deadline == null ? sleepMillis : Math.min(sleepMillis, Math.max(1, Duration.between(Instant.now(), deadline).toMillis()))); } } diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/processes/TestConnectionProcess.java b/src/main/java/com/kingsrook/qbits/objectstorage/processes/TestConnectionProcess.java index 066b1a7..41912e2 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/processes/TestConnectionProcess.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/processes/TestConnectionProcess.java @@ -227,7 +227,7 @@ public final class TestConnectionProcess } if(report.getClockSkewSeconds() != null) { - boolean skewed = report.getClockSkewSeconds() > 900; + boolean skewed = report.getClockSkewSeconds() > CapabilityProbe.CLOCK_SKEW_THRESHOLD_SECONDS; ok = ok && !skewed; lines.add(new ProcessSummaryLine(skewed ? Status.ERROR : Status.INFO, "Clock skew against the store: " + report.getClockSkewSeconds() + " seconds" + (skewed ? " (SigV4 fails beyond 15 minutes)" : ""))); -- 2.54.0 From 63c441e6cd0a1785199b97b424a0a28d77aac3ff Mon Sep 17 00:00:00 2001 From: James Maes Date: Wed, 14 Oct 2026 13:39:39 -0500 Subject: [PATCH 2/3] feat(health): register object storage connection checks with core central health Co-Authored-By: Claude Opus 5.5 --- .../ObjectStorageQBitProducer.java | 7 + .../health/ObjectStorageConnectionChecks.java | 121 ++++++ .../ObjectStorageConnectionChecksTest.java | 364 ++++++++++++++++++ 3 files changed, 492 insertions(+) create mode 100644 src/main/java/com/kingsrook/qbits/objectstorage/health/ObjectStorageConnectionChecks.java create mode 100644 src/test/java/com/kingsrook/qbits/objectstorage/health/ObjectStorageConnectionChecksTest.java diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/ObjectStorageQBitProducer.java b/src/main/java/com/kingsrook/qbits/objectstorage/ObjectStorageQBitProducer.java index 1ff7001..c7d1d73 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/ObjectStorageQBitProducer.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/ObjectStorageQBitProducer.java @@ -24,14 +24,18 @@ package com.kingsrook.qbits.objectstorage; import java.util.ArrayList; import java.util.List; import com.kingsrook.qbits.objectstorage.app.StorageWidgets; +import com.kingsrook.qbits.objectstorage.health.ObjectStorageConnectionChecks; import com.kingsrook.qbits.objectstorage.internal.ConfigSync; import com.kingsrook.qbits.objectstorage.internal.DefaultObjectHeaderPolicy; import com.kingsrook.qbits.objectstorage.internal.ObjectStorageRuntime; import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables; import com.kingsrook.qbits.objectstorage.processes.ObjectStorageProcesses; import com.kingsrook.qqq.backend.core.exceptions.QException; +import com.kingsrook.qqq.backend.core.health.HealthCheckSupplierMetaData; +import com.kingsrook.qqq.backend.core.health.HealthChecksMetaData; import com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerMultiOutput; import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference; import com.kingsrook.qqq.backend.core.model.metadata.dashboard.QWidgetMetaData; import com.kingsrook.qqq.backend.core.model.metadata.joins.QJoinMetaData; import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData; @@ -186,6 +190,9 @@ public class ObjectStorageQBitProducer implements QBitMetaDataProducer supply(QInstance qInstance) throws QException + { + ConfigSync.ensureSynced(); + List checks = new ArrayList<>(); + for(QRecord row : Db.asSystem(() -> Db.query(ObjectStorageTables.CONNECTION))) + { + if(ObjectStorageRuntime.getConfig().findConnection(row.getValueString("name")) == null) + { + continue; + } + Integer id = row.getValueInteger("id"); + String name = "object_storage.connection." + id.toString().toLowerCase(Locale.ROOT); + checks.add(new HealthCheckMetaData().withName(name).withLabel("Storage connection " + id) + .withComponent("object_storage").withCategory(HealthCategory.STORAGE).withKind(HealthCheckKind.DEPENDENCY) + .withIntervalSeconds(300).withTimeoutMs(20000).withSeverity(AlertSeverity.ERROR) + .withCheckInstance(input -> run(id, input))); + } + return (checks); + } + + + + /******************************************************************************* + ** Check the current row, reuse the bounded capability probe, and expose only + ** numeric facts and a fixed problem vocabulary (never exception messages). + *******************************************************************************/ + private static HealthCheckOutcome run(Integer id, HealthCheckInput input) throws QException + { + if(!Instant.now().isBefore(input.getDeadline()) || Thread.currentThread().isInterrupted()) + { + return (HealthCheckOutcome.down("Storage probe timed out").withDetails(new HealthDetails().withValue("problemKind", "timeout"))); + } + QRecord row = Db.asSystem(() -> Db.first(ObjectStorageTables.CONNECTION, "id", id)); + if(row == null || ObjectStorageRuntime.getConfig().findConnection(row.getValueString("name")) == null) + { + return (HealthCheckOutcome.unknown("Storage connection is disabled or no longer configured")); + } + try + { + CapabilityProbe.Report report = CapabilityProbe.probeConnection(row.getValueString("name"), input.getDeadline()); + HealthDetails details = new HealthDetails().withValue("bucketsProbed", report.getByBucket().size()); + if(report.getClockSkewSeconds() != null) + { + details.withValue("clockSkewSeconds", report.getClockSkewSeconds()); + } + if(report.getProblemKind() != null) + { + return (HealthCheckOutcome.down("Storage connection probe failed").withDetails(details.withValue("problemKind", report.getProblemKind()))); + } + if(report.getClockSkewSeconds() != null && report.getClockSkewSeconds() > CapabilityProbe.CLOCK_SKEW_THRESHOLD_SECONDS) + { + return (HealthCheckOutcome.degraded("Storage clock skew exceeds the signing threshold").withDetails(details)); + } + if(report.getByBucket().isEmpty()) + { + return (HealthCheckOutcome.unknown("No storage buckets configured to probe").withDetails(details)); + } + return (HealthCheckOutcome.up("Storage connection is reachable").withDetails(details)); + } + catch(RuntimeException e) + { + return (HealthCheckOutcome.down("Storage connection probe failed") + .withDetails(new HealthDetails().withValue("problemKind", CapabilityProbe.problemKind(e)))); + } + } +} diff --git a/src/test/java/com/kingsrook/qbits/objectstorage/health/ObjectStorageConnectionChecksTest.java b/src/test/java/com/kingsrook/qbits/objectstorage/health/ObjectStorageConnectionChecksTest.java new file mode 100644 index 0000000..2132286 --- /dev/null +++ b/src/test/java/com/kingsrook/qbits/objectstorage/health/ObjectStorageConnectionChecksTest.java @@ -0,0 +1,364 @@ +/* + * QQQ - Low-code Application Framework for Engineers. + * Copyright (C) 2021-2025. Kingsrook, LLC + * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States + * contact@kingsrook.com + * https://github.com/Kingsrook/ + * + * 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 + * + * https://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.objectstorage.health; + + +import javax.net.ssl.SSLHandshakeException; +import java.lang.reflect.Field; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Proxy; +import java.net.UnknownHostException; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig; +import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig; +import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig; +import com.kingsrook.qbits.objectstorage.internal.CapabilityProbe; +import com.kingsrook.qbits.objectstorage.internal.ConfigSync; +import com.kingsrook.qbits.objectstorage.internal.ConnectionRuntime; +import com.kingsrook.qbits.objectstorage.internal.Db; +import com.kingsrook.qbits.objectstorage.internal.ObjectStorageRuntime; +import com.kingsrook.qbits.objectstorage.internal.RateLimiter; +import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables; +import com.kingsrook.qbits.objectstorage.processes.TestConnectionProcess; +import com.kingsrook.qbits.objectstorage.testsupport.StorageTestSupport; +import com.kingsrook.qqq.backend.core.health.HealthCategory; +import com.kingsrook.qqq.backend.core.health.HealthCheckInput; +import com.kingsrook.qqq.backend.core.health.HealthCheckKind; +import com.kingsrook.qqq.backend.core.health.HealthCheckMetaData; +import com.kingsrook.qqq.backend.core.health.HealthCheckOutcome; +import com.kingsrook.qqq.backend.core.health.HealthChecksMetaData; +import com.kingsrook.qqq.backend.core.health.HealthStatus; +import com.kingsrook.qqq.backend.core.model.data.QRecord; +import com.kingsrook.qqq.backend.core.model.metadata.QInstance; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import software.amazon.awssdk.awscore.exception.AwsErrorDetails; +import software.amazon.awssdk.http.SdkHttpResponse; +import software.amazon.awssdk.services.s3.S3Client; +import software.amazon.awssdk.services.s3.model.ListObjectsV2Response; +import software.amazon.awssdk.services.s3.model.S3Exception; +import software.amazon.awssdk.services.s3.model.S3Request; +import software.amazon.awssdk.utils.builder.SdkBuilder; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + + +/******************************************************************************* + ** Exercise the real probe using an in-memory S3 client and QQQ record store. + *******************************************************************************/ +class ObjectStorageConnectionChecksTest +{ + private static final String SECRET = "S3NT1NEL-secret"; + private QInstance instance; + private ObjectStorageQBitConfig config; + private final List calls = new ArrayList<>(); + private RuntimeException failure; + private long skew; + private boolean expireAfterList; + private Instant deadline; + + + + /******************************************************************************* + ** Poison every credential reference; the fake prevents all network access. + *******************************************************************************/ + @BeforeEach + void setUp() throws Exception + { + System.setProperty("os.health.secret", SECRET); + String ref = "${prop.os.health.secret}"; + config = new ObjectStorageQBitConfig().withBackendName(StorageTestSupport.MEMORY) + .withAllowedSecretRefPatterns(List.of("prop.os.health.*")) + .withConnection(new StorageConnectionConfig("PRIMARY").withEndpoint("https://user:" + SECRET + "@private.example") + .withAccessKeyRef(ref).withSecretKeyRef(ref).withSessionTokenRef(ref) + .withPresignAccessKeyRef(ref).withPresignSecretKeyRef(ref).withCaBundleRef(ref).withMaxConcurrency(1)) + .withBucket(new StorageBucketConfig("files").withConnection("PRIMARY").withBucketName("private-bucket").withIndexed(false)); + instance = StorageTestSupport.setUp(config); + S3Client client = (S3Client) Proxy.newProxyInstance(S3Client.class.getClassLoader(), new Class[] { S3Client.class }, (proxy, method, args) -> + { + if(method.getName().equals("close")) + { + return (null); + } + if(args != null && args.length == 1 && args[0] instanceof java.util.function.Consumer) + { + return (InvocationHandler.invokeDefault(proxy, method, args)); + } + calls.add(method.getName()); + assertThat(method.getName()).matches("listObjectsV2|getBucket.*|getObjectLockConfiguration|getObjectTagging"); + S3Request request = (S3Request) args[0]; + Duration timeout = request.overrideConfiguration().orElseThrow().apiCallTimeout().orElseThrow(); + assertThat(timeout).isPositive().isLessThanOrEqualTo(Duration.ofSeconds(20)); + assertThat(request.overrideConfiguration().orElseThrow().apiCallAttemptTimeout()).contains(timeout); + if(failure != null) + { + throw (failure); + } + if(method.getName().equals("listObjectsV2")) + { + if(expireAfterList) + { + Thread.sleep(Math.max(1, Duration.between(Instant.now(), deadline).toMillis() + 10)); + } + return (ListObjectsV2Response.builder().sdkHttpResponse(SdkHttpResponse.builder().statusCode(200) + .putHeader("Date", DateTimeFormatter.RFC_1123_DATE_TIME.format(Instant.now().minusSeconds(skew).atZone(ZoneOffset.UTC))).build()).build()); + } + return (((SdkBuilder) method.getReturnType().getMethod("builder").invoke(null)).build()); + }); + Field field = ConnectionRuntime.class.getDeclaredField("client"); + field.setAccessible(true); + field.set(ObjectStorageRuntime.connection("PRIMARY"), client); + } + + + + /******************************************************************************* + ** Restore process-wide fixtures and the poisoned property. + *******************************************************************************/ + @AfterEach + void tearDown() + { + StorageTestSupport.tearDown(); + System.clearProperty("os.health.secret"); + } + + + + /******************************************************************************* + ** Producer registration includes the required metadata and keeps both actions. + *******************************************************************************/ + @Test + void registrationAndRefresh() throws Exception + { + var registry = HealthChecksMetaData.ofOrWithNew(instance); + assertThat(registry.getSuppliers()).singleElement().satisfies(supplier -> + { + assertThat(supplier.getNamePrefix()).isEqualTo("object_storage.connection"); + assertThat(supplier.getComponent()).isEqualTo("object_storage"); + }); + HealthCheckMetaData check = check(); + assertThat(check.getName()).isEqualTo("object_storage.connection." + ConfigSync.connectionId("PRIMARY")).matches("[a-z0-9]+([._-][a-z0-9]+)*"); + assertThat(check.getKind()).isEqualTo(HealthCheckKind.DEPENDENCY); + assertThat(check.getCategory()).isEqualTo(HealthCategory.STORAGE); + assertThat(check.getComponent()).isEqualTo("object_storage"); + assertThat(check.getIntervalSeconds()).isEqualTo(300); + assertThat(check.getTimeoutMs()).isEqualTo(20000); + assertThat(instance.getProcess(TestConnectionProcess.READ_NAME)).isNotNull(); + assertThat(instance.getProcess(TestConnectionProcess.WRITE_NAME)).isNotNull(); + ConfigSync.sync(); + assertThat(registry.resolveAll(instance)).hasSize(1); + config.withConnection(new StorageConnectionConfig("SECOND")); + ConfigSync.sync(); + assertThat(registry.resolveAll(instance)).hasSize(2); + config.getConnections().clear(); + assertThat(registry.resolveAll(instance)).isEmpty(); + assertClean(check.getCheckInstance().run(input()), HealthStatus.UNKNOWN); + } + + + + /******************************************************************************* + ** Successful checks read buckets without persisting any capability diagnostics. + *******************************************************************************/ + @Test + void upReadOnlyAndNoSecrets() throws Exception + { + HealthCheckOutcome outcome = run(); + assertClean(outcome, HealthStatus.UP); + assertThat(outcome.getDetails().getValues()).containsEntry("bucketsProbed", 1); + assertThat(calls).contains("listObjectsV2", "getBucketVersioning", "getBucketAcl"); + assertThat(Db.asSystem(() -> Db.query(ObjectStorageTables.CONNECTION_CAPABILITY))).isEmpty(); + assertThat(Db.asSystem(() -> Db.query(ObjectStorageTables.BUCKET_CAPABILITY))).isEmpty(); + assertThat(Db.asSystem(() -> Db.first(ObjectStorageTables.CONNECTION, "name", "PRIMARY")).getValue("lastTestedAt")).isNull(); + } + + + + /******************************************************************************* + ** Clock skew uses the same threshold as the record action. + *******************************************************************************/ + @Test + void clockSkew() throws Exception + { + skew = CapabilityProbe.CLOCK_SKEW_THRESHOLD_SECONDS - 2; + assertClean(run(), HealthStatus.UP); + skew = CapabilityProbe.CLOCK_SKEW_THRESHOLD_SECONDS + 10; + HealthCheckOutcome outcome = run(); + assertClean(outcome, HealthStatus.DEGRADED); + assertThat((Long) outcome.getDetails().getValues().get("clockSkewSeconds")).isGreaterThan(900); + } + + + + /******************************************************************************* + ** Expected failures carry only a fixed kind even when errors contain secrets. + *******************************************************************************/ + @ParameterizedTest + @ValueSource(strings = { "auth", "tls", "dns", "clock", "connection" }) + void downAndNoSecrets(String kind) throws Exception + { + failure = switch(kind) + { + case "auth" -> (S3Exception) S3Exception.builder().statusCode(403).message(SECRET).build(); + case "tls" -> new IllegalStateException(SECRET, new SSLHandshakeException(SECRET)); + case "dns" -> new IllegalStateException(SECRET, new UnknownHostException(SECRET)); + case "clock" -> (S3Exception) S3Exception.builder().statusCode(403).message(SECRET) + .awsErrorDetails(AwsErrorDetails.builder().errorCode("RequestTimeTooSkewed").build()).build(); + default -> new IllegalStateException(SECRET); + }; + skew = 1000; + HealthCheckOutcome outcome = run(); + assertClean(outcome, HealthStatus.DOWN); + assertThat(outcome.getDetails().getValues()).containsEntry("problemKind", kind); + assertThat(Db.asSystem(() -> Db.query(ObjectStorageTables.CONNECTION_CAPABILITY))).isEmpty(); + } + + + + /******************************************************************************* + ** An empty configuration cannot establish connectivity. + *******************************************************************************/ + @Test + void noBuckets() throws Exception + { + config.getBuckets().clear(); + assertClean(run(), HealthStatus.UNKNOWN); + assertThat(calls).isEmpty(); + } + + + + /******************************************************************************* + ** Client initialization errors also cannot expose poisoned configuration. + *******************************************************************************/ + @Test + @SuppressWarnings("unchecked") + void clientInitializationFailure() throws Exception + { + Field connections = ObjectStorageRuntime.class.getDeclaredField("CONNECTIONS"); + connections.setAccessible(true); + ((Map) connections.get(null)).put("PRIMARY", new ConnectionRuntime(config.getConnections().getFirst()) + { + /******************************************************************************* + ** Simulate a credential or endpoint failure before any request is sent. + *******************************************************************************/ + @Override + public synchronized S3Client client() + { + throw (new IllegalStateException(SECRET)); + } + }); + assertClean(run(), HealthStatus.DOWN); + assertThat(calls).isEmpty(); + } + + + + /******************************************************************************* + ** Expired checks perform no I/O; a deadline reached mid-probe stops SDK calls. + *******************************************************************************/ + @Test + void deadlines() throws Exception + { + assertClean(check().getCheckInstance().run(input().withDeadline(Instant.now().minusSeconds(1))), HealthStatus.DOWN); + assertThat(calls).isEmpty(); + expireAfterList = true; + deadline = Instant.now().plusMillis(500); + HealthCheckOutcome outcome = check().getCheckInstance().run(input().withDeadline(deadline)); + assertClean(outcome, HealthStatus.DOWN); + assertThat(outcome.getDetails().getValues()).containsEntry("problemKind", "timeout"); + assertThat(calls).containsExactly("listObjectsV2"); + } + + + + /******************************************************************************* + ** Rate limiting also respects the budget, both for slots and depleted tokens. + *******************************************************************************/ + @Test + void rateLimiterDeadline() throws Exception + { + RateLimiter limiter = ObjectStorageRuntime.connection("PRIMARY").getRateLimiter(); + try(var held = limiter.acquire()) + { + assertClean(check().getCheckInstance().run(input().withDeadline(Instant.now().plusMillis(50))), HealthStatus.DOWN); + assertThat(calls).isEmpty(); + } + RateLimiter onePerSecond = new RateLimiter(1, 1); + onePerSecond.acquire().close(); + assertThatThrownBy(() -> onePerSecond.acquire(Instant.now().plusMillis(20))).isInstanceOf(IllegalStateException.class); + try(var permit = onePerSecond.acquire(Instant.now().plusSeconds(2))) + { + assertThat(permit).isNotNull(); + } + } + + + + /******************************************************************************* + ** Fetch checks through the registry, exercising the supplier code reference. + *******************************************************************************/ + private HealthCheckMetaData check() throws Exception + { + return (HealthChecksMetaData.ofOrWithNew(instance).resolveAll(instance).getFirst()); + } + + + + /******************************************************************************* + ** Default bounded input. + *******************************************************************************/ + private HealthCheckInput input() + { + return (new HealthCheckInput().withQInstance(instance).withDeadline(Instant.now().plusSeconds(20))); + } + + + + /******************************************************************************* + ** Run the registered check. + *******************************************************************************/ + private HealthCheckOutcome run() throws Exception + { + return (check().getCheckInstance().run(input())); + } + + + + /******************************************************************************* + ** HealthRunner.sanitize is package-private; enforce cleanliness on raw output. + *******************************************************************************/ + private void assertClean(HealthCheckOutcome outcome, HealthStatus expected) + { + assertThat(outcome.getStatus()).isEqualTo(expected); + assertThat(outcome.getMessage() + outcome.getDetails().toJson()).doesNotContain(SECRET, "private.example", "private-bucket"); + assertThat(outcome.getMessage()).doesNotContain("\n", "\r"); + } +} -- 2.54.0 From fd81c88f32a5b1ee200f6cdea5c26c432252cbff Mon Sep 17 00:00:00 2001 From: James Maes Date: Wed, 14 Oct 2026 13:39:40 -0500 Subject: [PATCH 3/3] docs: object storage health checks Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 5 +++++ README.md | 15 +++++++++++++++ 2 files changed, 20 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e26966..86fb626 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,8 @@ All notable changes to this project are documented here. The format follows Keep ## [0.1.0] - Unreleased ### Added +- Central health supplier `object_storage.connection.` (DEPENDENCY/STORAGE, 300s interval, 20000ms timeout), reusing bounded read-only capability probes with safe problem kinds and clock-skew details. Turn a connection's check off in the central `health_check` table. +- In-memory health tests for registration, refresh, UP/DOWN/DEGRADED/UNKNOWN outcomes, deadlines, read-only execution and sentinel-secret exclusion. - `StorageBucket.maxUploadBytes()` in the `api` package: the bucket's upload limit in bytes (null when none), so other qbits need not call `config()`, which returns an internal type. - **The service (P1).** `ObjectStorage.bucket(alias)` and `ObjectStorage.asSystem(caller, runId)`: put (multipart above the part size), get with ranges, head, copy (in parts for very large objects), header and metadata changes, delete (delete marker, private trash, or permanent), restore, list, presigned GET, PUT and POST, public URLs, tags. Key building and URL encoding, the header policy (`object_storage_header_rule`), `PublicUrlResolver`, `StorageChangeListener`, `StorageReferenceProvider`, `StorageAlertSink`. - Configuration in code, upserted into visible `object_storage_*` tables; secret references with an allow-list; per-connection rate limits; per-bucket and per-connection record security locks. @@ -16,6 +18,9 @@ All notable changes to this project are documented here. The format follows Keep - `ObjectStorageSchema`: PostgreSQL DDL generated from the table metadata. - Conformance suite against MinIO and Garage (Testcontainers), service, index and action tests, and the index and actions on PostgreSQL. +### Removed +- None for the health migration; both Test connection record actions, including on-demand write tests, are retained. + ### Changed - Raised the QQQ BOM baseline to released `4.1.0-RC.5` and delegated S3 client construction to core `S3ClientFactory.buildClient`. Removed local HTTP client and trust-manager implementations; existing endpoint, region, addressing, checksum, TLS/CA, credentials and timeout settings remain supported. - Clients remain owned by each connection runtime and are rebuilt on reset/reinstall. Retry behavior now follows core/SDK defaults instead of forcing `ADAPTIVE_V2`; hosts can select the SDK retry mode through `AWS_RETRY_MODE` or `aws.retryMode`. diff --git a/README.md b/README.md index 4ead1a2..a57910f 100644 --- a/README.md +++ b/README.md @@ -78,6 +78,21 @@ Every write is a standard streamed ETL: **Review** (the dry run, with summary li Permissions: `object_storage.view`, `.write`, `.delete`, `.admin` (admin implies the rest). Per-bucket scoping uses record security locks on the `object_storage_bucket` and `object_storage_connection` security key types (switch off with `withBucketScoping(false)` for a single-administrator app). +## Health checks + +The producer registers `object_storage.connection.` for each configured +`object_storage_connection` row: **DEPENDENCY**, **STORAGE**, every **300s**, +with a **20000ms** timeout and component `object_storage`. The central health runner +refreshes the supplier as rows change. The existing `CapabilityProbe` runs only +read-only calls within the deadline; it never runs write or presign tests or stores +free-form capability diagnostics. Connection failures are DOWN, clock skew above +900 seconds is DEGRADED, and connections without buckets are UNKNOWN. Details +contain only problem kind, clock skew seconds and bucket count. + +Turn a connection's check off or change its interval in the central `health_check` +table. Both **Test connection** record actions remain available, including +on-demand write tests. + ## Deployment notes - **Uploads** go through the app server (100 MB per file by default, per bucket `maxUploadBytes`), using QQQ's standard file-upload field, so the app needs `uploadedFileArchiveTableName` set in its Javalin metadata, as for any file upload. -- 2.54.0