Compare commits
7
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
eb04621d4e
|
||
|
|
38e74cc07f | ||
|
|
8aa4511839
|
||
|
|
9b2158bf97 | ||
|
|
fd81c88f32
|
||
|
|
63c441e6cd
|
||
|
|
b13122f1b0
|
No files matched your search
@@ -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.<id>` (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_<table>_<column>`), 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`.
|
||||
|
||||
@@ -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.<id>` 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.
|
||||
|
||||
@@ -172,6 +172,23 @@
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<resources>
|
||||
<!-- help bundles carry the artifact version (QQQ help bundle provenance) -->
|
||||
<resource>
|
||||
<directory>src/main/resources</directory>
|
||||
<filtering>true</filtering>
|
||||
<includes>
|
||||
<include>help/*.yaml</include>
|
||||
</includes>
|
||||
</resource>
|
||||
<resource>
|
||||
<directory>src/main/resources</directory>
|
||||
<filtering>false</filtering>
|
||||
<excludes>
|
||||
<exclude>help/*.yaml</exclude>
|
||||
</excludes>
|
||||
</resource>
|
||||
</resources>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
@@ -262,7 +279,7 @@
|
||||
<activeByDefault>true</activeByDefault>
|
||||
</activation>
|
||||
<properties>
|
||||
<qqq.snapshot.version>4.1.0-RC.7</qqq.snapshot.version>
|
||||
<qqq.snapshot.version>4.1.0-RC.8</qqq.snapshot.version>
|
||||
</properties>
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
|
||||
@@ -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<ObjectSto
|
||||
output.add(app);
|
||||
}
|
||||
output.addSelfToInstance(qInstance);
|
||||
HealthChecksMetaData.ofOrWithNew(qInstance).withSupplier(new HealthCheckSupplierMetaData()
|
||||
.withNamePrefix("object_storage.connection").withComponent("object_storage")
|
||||
.withSupplier(new QCodeReference(ObjectStorageConnectionChecks.class)));
|
||||
if(Boolean.TRUE.equals(qBitConfig.getAdminApp()) && qBitConfig.getParentAppName() != null && qInstance.getApp(qBitConfig.getParentAppName()) != null)
|
||||
{
|
||||
///////////////////////////////////////////////////////////////////////////////
|
||||
|
||||
+121
@@ -0,0 +1,121 @@
|
||||
/*
|
||||
* 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 java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import com.kingsrook.qbits.objectstorage.internal.CapabilityProbe;
|
||||
import com.kingsrook.qbits.objectstorage.internal.ConfigSync;
|
||||
import com.kingsrook.qbits.objectstorage.internal.Db;
|
||||
import com.kingsrook.qbits.objectstorage.internal.ObjectStorageRuntime;
|
||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||
import com.kingsrook.qqq.backend.core.alerts.AlertSeverity;
|
||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||
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.HealthCheckSupplier;
|
||||
import com.kingsrook.qqq.backend.core.health.HealthDetails;
|
||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
||||
|
||||
|
||||
/*******************************************************************************
|
||||
** Dynamic, read-only checks for configured connection rows. Diagnostic text and
|
||||
** resolved settings never cross the health boundary.
|
||||
*******************************************************************************/
|
||||
public class ObjectStorageConnectionChecks implements HealthCheckSupplier
|
||||
{
|
||||
/*******************************************************************************
|
||||
** Resolve current rows on every refresh; removed configuration is not probed.
|
||||
*******************************************************************************/
|
||||
@Override
|
||||
public List<HealthCheckMetaData> supply(QInstance qInstance) throws QException
|
||||
{
|
||||
ConfigSync.ensureSynced();
|
||||
List<HealthCheckMetaData> 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))));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<String> 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<String> 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<Capability, CapabilityStatus> 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<Capability, CapabilityStatus> 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<String, String> details = new java.util.concurrent.ConcurrentHashMap<>();
|
||||
private static final ThreadLocal<Map<String, String>> 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<Capability, CapabilityStatus> 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 <T extends S3Request, ReturnT> ReturnT invokeOperation(T request, Function<T, ReturnT> 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<Capability, CapabilityStatus> results, Report report)
|
||||
private static void readProbes(ConnectionRuntime connection, StorageBucketConfig bucket, Map<Capability, CapabilityStatus> 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);
|
||||
|
||||
@@ -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.
|
||||
*******************************************************************************/
|
||||
|
||||
@@ -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())));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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.
|
||||
*******************************************************************************/
|
||||
|
||||
+1
-1
@@ -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)" : "")));
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
bundle: qbit-object-storage
|
||||
version: 0.1.0-SNAPSHOT
|
||||
version: ${project.version}
|
||||
layer: QBIT
|
||||
entries:
|
||||
- key: app:objectStorageAdmin;slot:help
|
||||
|
||||
+364
@@ -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<String> 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<String, ConnectionRuntime>) 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");
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user