diff --git a/CHANGELOG.md b/CHANGELOG.md index b5bd307..2093629 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. - Expanded the host-neutral in-app help bundle to cover storage fields, actions, widgets and possible values, with live references and a `HelpBundleAssertions` coverage gate. Registered it for QQQ startup sync and HOST overrides; retained inline table/app help for hosts without help-content tables. - Every table explains itself: a help bundle (`help/qbit-object-storage.yaml`, QQQ help-bundle format) says what each `object_storage_*` table is for and how its rows get created; it shows on the table screen and when the table is empty. Applied inline by `ObjectStorageHelp` for hosts without help-content tables; a host's own help is kept. - A Setup card on the Storage home (connections, buckets, header rules, protection rules) with the state and next step of each, resolved the way the runtime resolves them. @@ -24,7 +26,12 @@ All notable changes to this project are documented here. The format follows Keep - Foreign keys on every relation between the qbit tables (`fk__`), generated by `ObjectStorageSchema`: owned observed state (capabilities, folders, objects, versions, usage) cascades from its bucket or object, `object.folder_id` and `object_reference.object_id` are set null, and a bucket's connection, header and protection rules and operation log restrict. Each key is added by an idempotent block that first records rows whose parent is gone in `qqq_fk_quarantine` and clears (nullable column) or deletes (required column) them, so it also upgrades an existing schema. New indexes `ix_object_storage_bucket_1` (connection_id), `ix_object_storage_header_rule_1` and `ix_object_storage_protection_rule_1` (bucket_id) back the keys. Upgrade: run the generated statements again (or the host's copy of them). - 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 +- Rechecked the released RC.8 JARs: core still has no S3 session-token/provider hook or file-reference interpreter. Retained both compatibility shims and moved their review deadline to RC.9 under issue #10. +- Raised the QQQ BOM baseline to released `4.1.0-RC.8` for the central health API. - Raised the QQQ BOM baseline to released `4.1.0-RC.7` for the core help-bundle registry and coverage assertions. - 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 1b33c4e..a897d98 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ S3-compatible storage as a capability for any QQQ app, and an admin app to run i ## Requirements - Java 21. -- QQQ `4.1.0-RC.7` (released BOM), including `qqq-backend-module-filesystem`. `ConnectionRuntime` maps connection settings to core's `S3BackendMetaData` and uses `S3ClientFactory.buildClient`; core owns HTTP, TLS and checksum setup. The unused SFTP/SSH dependency is excluded. See the [dependency diff](docs/S3-FACTORY-MIGRATION.md). +- QQQ `4.1.0-RC.8` (released BOM), including `qqq-backend-module-filesystem`. `ConnectionRuntime` maps connection settings to core's `S3BackendMetaData` and uses `S3ClientFactory.buildClient`; core owns HTTP, TLS and checksum setup. The unused SFTP/SSH dependency is excluded. See the [dependency diff](docs/S3-FACTORY-MIGRATION.md). - A QQQ backend for the qbit's tables (PostgreSQL is the intended store). `ObjectStorageSchema.postgresScript(config)` generates the DDL from the same metadata the app runs on, for your Liquibase or Flyway migration. ## Install @@ -92,6 +92,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. + ## Setup and starter defaults The Storage home has a **Setup** card: connections and buckets (declared in code, synced into their tables), header rules, and protection rules, each with its state and next step. Every table has help (`help/qbit-object-storage.yaml`) saying what it is for and how its rows get created, which also shows when the table is empty. diff --git a/pom.xml b/pom.xml index 3f251a7..86fdf43 100644 --- a/pom.xml +++ b/pom.xml @@ -262,7 +262,7 @@ true - 4.1.0-RC.7 + 4.1.0-RC.8 diff --git a/src/main/java/com/kingsrook/qbits/objectstorage/ObjectStorageQBitProducer.java b/src/main/java/com/kingsrook/qbits/objectstorage/ObjectStorageQBitProducer.java index 023de6f..6d47ab2 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/ObjectStorageQBitProducer.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/ObjectStorageQBitProducer.java @@ -24,6 +24,7 @@ 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; @@ -31,8 +32,11 @@ import com.kingsrook.qbits.objectstorage.model.ObjectStorageHelp; 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.help.bundles.HelpBundles; import com.kingsrook.qqq.backend.core.model.metadata.joins.QJoinMetaData; @@ -188,6 +192,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/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/ConnectionRuntime.java b/src/main/java/com/kingsrook/qbits/objectstorage/internal/ConnectionRuntime.java index 36de2ce..84aae98 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/internal/ConnectionRuntime.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/internal/ConnectionRuntime.java @@ -291,8 +291,8 @@ public class ConnectionRuntime /******************************************************************************* - ** STAND-IN(core=QRun-IO/qqq#1165, removeBy=4.1.0-RC.8) - ** RC.7 S3BackendMetaData and S3ClientFactory still lack a session token or + ** STAND-IN(core=QRun-IO/qqq#1165, removeBy=4.1.0-RC.9) + ** RC.8 S3BackendMetaData and S3ClientFactory still lack a session token or ** credentials-provider hook (verified against the released JAR). ** Override request credentials only; core still owns the client and transport. *******************************************************************************/ 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/internal/SecretRefs.java b/src/main/java/com/kingsrook/qbits/objectstorage/internal/SecretRefs.java index 244a89a..8081658 100644 --- a/src/main/java/com/kingsrook/qbits/objectstorage/internal/SecretRefs.java +++ b/src/main/java/com/kingsrook/qbits/objectstorage/internal/SecretRefs.java @@ -36,8 +36,8 @@ import com.kingsrook.qqq.backend.core.utils.StringUtils; ** the allow-list. Resolved values live only in memory, never in a record, a log ** line or a process value. ** - ** STAND-IN(core=QRun-IO/qqq#1165, removeBy=4.1.0-RC.8) - ** RC.7 QMetaDataVariableInterpreter still supports env/prop, not file references. + ** STAND-IN(core=QRun-IO/qqq#1165, removeBy=4.1.0-RC.9) + ** RC.8 QMetaDataVariableInterpreter still supports env/prop, not file references. ** ${file.PATH} moves to QMetaDataVariableInterpreter in QQQ core ** (spec section 8.2); then this class delegates entirely. *******************************************************************************/ 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)" : ""))); 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"); + } +}