Compare commits
46
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c27034563c
|
||
|
|
a8e8750829
|
||
|
|
cfe190634d
|
||
|
|
ab16a1891b | ||
|
|
1c9a5c53b8
|
||
|
|
04d61ec5f0
|
||
|
|
8169d4c3c8
|
||
|
|
e451267495
|
||
|
|
adf3b7fe5d
|
||
|
|
55163ca402 | ||
|
|
9b7c907715
|
||
|
|
2dbcb4f6f0
|
||
|
|
c00d0e0d35
|
||
|
|
aec65dcdb9
|
||
|
|
7a949c32ab
|
||
|
|
95b45414d8
|
||
|
|
a7b5f19626 | ||
|
|
79bc8fcf34 | ||
|
|
4eec8b56f3 | ||
|
|
988f4c56b1 | ||
|
|
03ed383587 | ||
|
|
85e9ab1316 | ||
|
|
03215cab62 | ||
|
|
6866e6653f | ||
|
|
eceb2a6d59
|
||
|
|
3f8ca0ef1f | ||
|
|
e341a90df3
|
||
|
|
a7e6a40924 | ||
|
|
0222074cf9 | ||
|
|
4033590078
|
||
|
|
75b4647d08
|
||
|
|
9590619e0d | ||
|
|
4261382567
|
||
|
|
88f7f33fe7
|
||
|
|
2b5a84e805
|
||
|
|
878270503e
|
||
|
|
7c825b05d5 | ||
|
|
5401070321
|
||
|
|
75bf74967d | ||
|
|
eb04621d4e
|
||
|
|
38e74cc07f | ||
|
|
8aa4511839
|
||
|
|
9b2158bf97 | ||
|
|
fd81c88f32
|
||
|
|
63c441e6cd
|
||
|
|
b13122f1b0
|
No files matched your search
+11
-57
@@ -1,60 +1,14 @@
|
|||||||
version: 2.1
|
version: 2.1
|
||||||
|
setup: true
|
||||||
orbs:
|
orbs:
|
||||||
qqq-orb: kingsrook/qqq-orb@0.6.9
|
qqq-orb: kingsrook/qqq-orb@0.8
|
||||||
|
|
||||||
commands:
|
|
||||||
configure-package-install:
|
|
||||||
steps:
|
|
||||||
- run:
|
|
||||||
name: Keep package installation unattended
|
|
||||||
command: |
|
|
||||||
# These build machines are discarded after the job.
|
|
||||||
sudo install -d /etc/needrestart/conf.d
|
|
||||||
printf '%s\n' '$nrconf{restart} = "l";' | sudo tee /etc/needrestart/conf.d/99-qqq-ci.conf >/dev/null
|
|
||||||
|
|
||||||
workflows:
|
workflows:
|
||||||
test_only:
|
setup:
|
||||||
jobs:
|
jobs:
|
||||||
- qqq-orb/mvn_test_only:
|
- qqq-orb/generate_pipeline:
|
||||||
pre-steps:
|
orb_version: '0.8'
|
||||||
- configure-package-install
|
filters:
|
||||||
context: [qqq-maven-registry-credentials]
|
branches:
|
||||||
filters:
|
ignore: automation/qbit-rc-pending
|
||||||
branches:
|
tags:
|
||||||
ignore: /(develop|release\/.*)/
|
only: /^v\d+\.\d+\.\d+$/
|
||||||
|
|
||||||
publish_snapshot:
|
|
||||||
jobs:
|
|
||||||
- qqq-orb/mvn_publish:
|
|
||||||
pre-steps:
|
|
||||||
- configure-package-install
|
|
||||||
context: [qqq-maven-registry-credentials]
|
|
||||||
branch_type: snapshot
|
|
||||||
filters:
|
|
||||||
branches:
|
|
||||||
only: [develop]
|
|
||||||
|
|
||||||
publish_release_candidate:
|
|
||||||
jobs:
|
|
||||||
- qqq-orb/mvn_publish:
|
|
||||||
pre-steps:
|
|
||||||
- configure-package-install
|
|
||||||
context: [qqq-maven-registry-credentials]
|
|
||||||
branch_type: release_candidate
|
|
||||||
filters:
|
|
||||||
branches:
|
|
||||||
only: [/release\/.*/]
|
|
||||||
|
|
||||||
publish_release:
|
|
||||||
jobs:
|
|
||||||
- qqq-orb/mvn_publish:
|
|
||||||
pre-steps:
|
|
||||||
- configure-package-install
|
|
||||||
context: [qqq-maven-registry-credentials]
|
|
||||||
branch_type: release
|
|
||||||
filters:
|
|
||||||
branches:
|
|
||||||
ignore: /.*/
|
|
||||||
tags:
|
|
||||||
only: [/^v\d+\.\d+\.\d+$/]
|
|
||||||
@@ -0,0 +1,7 @@
|
|||||||
|
name: build
|
||||||
|
on:
|
||||||
|
push:
|
||||||
|
jobs:
|
||||||
|
maven:
|
||||||
|
uses: self:KofTwentyTwo/ci-workflows/.gitea/workflows/maven.yaml@v1
|
||||||
|
secrets: inherit
|
||||||
@@ -0,0 +1,23 @@
|
|||||||
|
name: CodeQL
|
||||||
|
'on':
|
||||||
|
push:
|
||||||
|
branches:
|
||||||
|
- main
|
||||||
|
- develop
|
||||||
|
- release/**
|
||||||
|
- hotfix/**
|
||||||
|
pull_request: {}
|
||||||
|
schedule:
|
||||||
|
- cron: 41 16 * * 0
|
||||||
|
workflow_dispatch: {}
|
||||||
|
permissions:
|
||||||
|
contents: read
|
||||||
|
actions: read
|
||||||
|
security-events: write
|
||||||
|
packages: read
|
||||||
|
concurrency:
|
||||||
|
group: codeql-${{ github.event.pull_request.number || github.ref }}
|
||||||
|
cancel-in-progress: ${{ github.event_name == 'pull_request' }}
|
||||||
|
jobs:
|
||||||
|
analyze:
|
||||||
|
uses: QRun-IO/qqq-orb/.github/workflows/qbit-codeql.yml@c288a8864fe9054f68410e1d78f5863aea4c4d22
|
||||||
@@ -1,25 +1,74 @@
|
|||||||
# Cut the next release candidate (QRun-IO/qqq docs/QRUN-QBIT-REPO-STANDARD.md, C6):
|
|
||||||
# run this workflow by hand, or merge a develop PR labelled `release:rc`. It merges develop into
|
|
||||||
# the release line and the release/* push runs CircleCI publish_release_candidate.
|
|
||||||
name: Release candidate
|
name: Release candidate
|
||||||
|
'on':
|
||||||
on:
|
|
||||||
workflow_dispatch:
|
workflow_dispatch:
|
||||||
inputs:
|
inputs:
|
||||||
dry_run:
|
dry_run:
|
||||||
description: Print the plan without merging or creating branches
|
description: Plan a cut without changing release state
|
||||||
type: boolean
|
type: boolean
|
||||||
default: false
|
default: false
|
||||||
|
finalize:
|
||||||
|
description: Retry finalization of an existing cut; do not cut another RC
|
||||||
|
type: boolean
|
||||||
|
default: false
|
||||||
|
recover:
|
||||||
|
description: Recover the same published RC after a deployment timeout; never
|
||||||
|
reupload
|
||||||
|
type: boolean
|
||||||
|
default: false
|
||||||
|
circle_build:
|
||||||
|
description: Original Circle publisher build number (recovery only)
|
||||||
|
type: string
|
||||||
|
default: ''
|
||||||
|
merge_sha:
|
||||||
|
description: Existing cut merge_sha (finalize retry only)
|
||||||
|
type: string
|
||||||
|
default: ''
|
||||||
|
bump_sha:
|
||||||
|
description: Existing cut bump_sha (finalize retry only)
|
||||||
|
type: string
|
||||||
|
default: ''
|
||||||
|
branch:
|
||||||
|
description: Existing cut branch (finalize retry only)
|
||||||
|
type: string
|
||||||
|
default: ''
|
||||||
|
version:
|
||||||
|
description: Existing cut version (finalize retry only)
|
||||||
|
type: string
|
||||||
|
default: ''
|
||||||
pull_request:
|
pull_request:
|
||||||
types: [closed]
|
types:
|
||||||
branches: [develop]
|
- closed
|
||||||
|
branches:
|
||||||
|
- develop
|
||||||
|
repository_dispatch:
|
||||||
|
types:
|
||||||
|
- qbit-rc-published
|
||||||
permissions:
|
permissions:
|
||||||
contents: write
|
contents: write
|
||||||
|
actions: read
|
||||||
|
checks: read
|
||||||
jobs:
|
jobs:
|
||||||
release-rc:
|
release-rc:
|
||||||
if: github.event_name == 'workflow_dispatch' || (github.event.pull_request.merged == true && contains(github.event.pull_request.labels.*.name, 'release:rc'))
|
if: github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch'
|
||||||
uses: QRun-IO/.github/.github/workflows/qbit-release-rc.yml@main
|
|| (github.event_name == 'pull_request' && github.event.pull_request.merged
|
||||||
|
== true && github.event.pull_request.base.ref == 'develop' && contains(github.event.pull_request.labels.*.name,
|
||||||
|
'release:rc'))
|
||||||
|
uses: QRun-IO/qqq-orb/.github/workflows/qbit-release-rc.yml@cc30f599e63c8e41629c99d23395bac1adb8bc44
|
||||||
with:
|
with:
|
||||||
dry_run: ${{ github.event_name == 'workflow_dispatch' && inputs.dry_run }}
|
dry_run: ${{ github.event_name == 'workflow_dispatch' && inputs.dry_run || false
|
||||||
|
}}
|
||||||
|
finalize: ${{ github.event_name == 'repository_dispatch' || (github.event_name
|
||||||
|
== 'workflow_dispatch' && inputs.finalize) }}
|
||||||
|
recover: ${{ github.event_name == 'workflow_dispatch' && inputs.recover || false
|
||||||
|
}}
|
||||||
|
circle_build: ${{ github.event_name == 'workflow_dispatch' && inputs.circle_build
|
||||||
|
|| '' }}
|
||||||
|
merge_sha: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.merge_sha
|
||||||
|
|| inputs.merge_sha || '' }}
|
||||||
|
bump_sha: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.bump_sha
|
||||||
|
|| inputs.bump_sha || '' }}
|
||||||
|
branch: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.branch
|
||||||
|
|| inputs.branch || '' }}
|
||||||
|
version: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.version
|
||||||
|
|| inputs.version || '' }}
|
||||||
|
secrets: inherit
|
||||||
+16
@@ -0,0 +1,16 @@
|
|||||||
|
profile: qbit
|
||||||
|
orb_version: '0.8'
|
||||||
|
default_branch: develop
|
||||||
|
java_version: '21'
|
||||||
|
browsers: true
|
||||||
|
api_checks: false
|
||||||
|
docs_only: false
|
||||||
|
release_on_main: false
|
||||||
|
release_tag_policy: stable
|
||||||
|
test_branch_policy: non-release
|
||||||
|
publish:
|
||||||
|
snapshot: true
|
||||||
|
release_candidate: true
|
||||||
|
release: true
|
||||||
|
asciidoc: false
|
||||||
|
release_completion: true
|
||||||
@@ -5,6 +5,8 @@ All notable changes to this project are documented here. The format follows Keep
|
|||||||
## [0.1.0] - Unreleased
|
## [0.1.0] - Unreleased
|
||||||
|
|
||||||
### Added
|
### 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.
|
- 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.
|
- 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.
|
- 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,20 @@ 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).
|
- 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.
|
- Conformance suite against MinIO and Garage (Testcontainers), service, index and action tests, and the index and actions on PostgreSQL.
|
||||||
|
|
||||||
|
### Fixed
|
||||||
|
- The service's own writes are no longer refused in a host that enforces permissions inside the core actions (`QInstance.enforcePermissionsInCoreActions`). `Db.asSystem` starts a new context, so each of its writes is an outermost action there and core asks `ObjectStoragePermissionChecker`, which refuses every write of a table nobody may write directly, to every session. The `object_storage_operation` row of every stored object was lost (logged as "Could not write a storage operation log row", the object itself was stored), and so was each other system write of those tables. `Db.insert`, `Db.update` and `Db.delete` now mark their input with core's per-call permission bypass (`withPermissionBypass`, reason `Db.OWN_WRITE_REASON`), so core does not ask about the service's own writes. The checker is unchanged in what it refuses: a write of those tables that does not carry a bypass is refused to a person, to `object_storage.admin` and to the core system user alike, so server code running as the system user cannot edit or delete operation log rows through an enforced action.
|
||||||
|
- An operation log row that could not be written is no longer only a warning. It is logged at ERROR, raises the `storage.operation_log.failed` alert (`StorageAlertSink`, severity `error`, dedupe key `bucket:<alias>`, resolved by the next row written for that bucket) and is counted in `OperationLog.failedWriteCount()` (per node, since start). A row the insert action hands back with a record error (a refused value, a record lock) is now detected as well; before, it was dropped without any log line from the qbit. The operation the row describes still succeeds.
|
||||||
|
|
||||||
|
### Removed
|
||||||
|
- The QRun-IO/qqq#1165 stand-ins: `SessionCredentialsClient` (session tokens now go through core `S3BackendMetaData.sessionToken` / `S3ClientFactory`, including for the presigner) and the file reading in `SecretRefs` (core `QMetaDataVariableInterpreter` resolves `${file.PATH}`). **Behavior change:** core reads `${file.PATH}` only inside its allowed roots, which are `/var/run/secrets` and `/run/secrets` by default. A host that mounts storage secrets elsewhere must set `qqq.fileReference.roots` or `QQQ_FILE_REFERENCE_ROOTS`. The qbit's default allow-list (`file./var/run/secrets/storage/*`) is already inside the default roots.
|
||||||
|
- None for the health migration; both Test connection record actions, including on-demand write tests, are retained.
|
||||||
|
|
||||||
### Changed
|
### Changed
|
||||||
|
- Multi-node deployments: the runtime caches are registered with QQQ `ClusterCaches`. Shared (cleared on every node): the bucket rows' presign limits (`qbit-object-storage.bucket-row-limits`, on a bucket row insert or update), the enabled header rules (`qbit-object-storage.header-rules`, on a header rule insert, update or delete), both through the new `ConfigRowChangeCustomizer`, and the capability statuses (`qbit-object-storage.capabilities`: a probe result recorded on one node clears that capability on the others, which reload it from the capability row; before, other nodes kept the old status until restart). The 15 s TTLs remain the safety net. Node-local, with reasons: the alias and connection row-id maps, the S3 client map, the process-permission map, the provider defaults, the running-scan guard and the per-pod presign rate limiter (`StorageBucket.PRESIGN_COUNTS`; per-node rate limits are accepted for now). `ObjectStorageClusterCachesTest` audits the package for unregistered static caches and covers remote and local invalidation.
|
||||||
|
- Builds against QQQ `4.1.0-SNAPSHOT` (`qqq.snapshot.version`) for `ClusterCaches` until the next QQQ RC ships; a release must pin that RC.
|
||||||
|
- Read the qbit metadata version from the Maven-filtered artifact version so release metadata matches the build and help bundle.
|
||||||
|
- 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.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.
|
- 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`.
|
- 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
|
## Requirements
|
||||||
|
|
||||||
- Java 21.
|
- 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.
|
- 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
|
## 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).
|
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
|
## 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.
|
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.
|
||||||
@@ -110,6 +125,7 @@ It is insert-only and inserts a **disabled** generic header rule, `default-mutab
|
|||||||
- **Previews** are short-lived (5 minute) presigned GETs, so the Next dashboard's `img-src` must allow the storage host (it allows `https:` by default). Video, audio and PDF previews need same-origin media (the dashboard's `media-src` and `frame-src` are `'self'`); a same-origin download route is the planned follow-up. Until then those files show the file card with download and open links.
|
- **Previews** are short-lived (5 minute) presigned GETs, so the Next dashboard's `img-src` must allow the storage host (it allows `https:` by default). Video, audio and PDF previews need same-origin media (the dashboard's `media-src` and `frame-src` are `'self'`); a same-origin download route is the planned follow-up. Until then those files show the file card with download and open links.
|
||||||
- **Scheduler.** Set `withSchedulerName(...)` to run the nightly scan, HEAD backlog, reference scan, usage snapshot and trash purge. Without it, run Rescan bucket by hand.
|
- **Scheduler.** Set `withSchedulerName(...)` to run the nightly scan, HEAD backlog, reference scan, usage snapshot and trash purge. Without it, run Rescan bucket by hand.
|
||||||
- **Alerts.** With the Alerts qbit installed, bind a `StorageAlertSink` adapter (`ObjectStorageRuntime.setAlertSink`); without it every condition is still visible on the home and in the tables.
|
- **Alerts.** With the Alerts qbit installed, bind a `StorageAlertSink` adapter (`ObjectStorageRuntime.setAlertSink`); without it every condition is still visible on the home and in the tables.
|
||||||
|
- **A missing audit row.** A storage operation succeeds even when its `object_storage_operation` row cannot be written, and that row is then the one condition no table shows. It is logged at ERROR ("Could not write a storage operation log row"), raised as the `storage.operation_log.failed` alert (severity `error`, dedupe key `bucket:<alias>`, resolved by the next row written for that bucket) and counted in `OperationLog.failedWriteCount()`, per node since it started. A host without an alert sink should export that count (a gauge, or a health check of its own) and alert when it is above zero.
|
||||||
|
|
||||||
## Conformance
|
## Conformance
|
||||||
|
|
||||||
|
|||||||
@@ -26,16 +26,21 @@ The public qbit configuration API is unchanged.
|
|||||||
|
|
||||||
## Remaining local code
|
## Remaining local code
|
||||||
|
|
||||||
- Core RC.5 has no session-token field or credentials-provider customization
|
- Core has no presigner factory. The local presigner keeps its separate public
|
||||||
in `S3BackendMetaData` / `S3ClientFactory.buildClient`. The narrow
|
endpoint and least-privilege keys; its credentials (including the session
|
||||||
`SessionCredentialsClient` bridge adds session credentials to requests while
|
token) come from core `S3ClientFactory.buildCredentialsProvider`.
|
||||||
retaining their overrides; core still creates the client and HTTP transport.
|
- `SecretRefs` keeps only the qbit's allow-list; resolution, including
|
||||||
Core needs session-token metadata (including enrichment and cache-key support)
|
`${file.PATH}`, is core `QMetaDataVariableInterpreter`.
|
||||||
and `AwsSessionCredentials` construction, or a caller-provided credentials hook.
|
|
||||||
Marker: `STAND-IN(core=QRun-IO/qqq#1165, removeBy=4.1.0-RC.7)`.
|
## Stand-ins removed (QRun-IO/qqq#1165)
|
||||||
- Core has no presigner factory. The local presigner retains its separate public
|
|
||||||
endpoint and least-privilege keys, along with existing session-token handling.
|
- `SessionCredentialsClient` is gone: core `S3BackendMetaData.sessionToken`
|
||||||
- `SecretRefs` file resolution is unchanged and carries the same RC.7 marker.
|
makes `S3ClientFactory` sign with `AwsSessionCredentials`.
|
||||||
|
- `SecretRefs` no longer reads files itself. Core reads `${file.PATH}`
|
||||||
|
read-only, and only inside its allowed roots: `/var/run/secrets` and
|
||||||
|
`/run/secrets` by default, or `qqq.fileReference.roots` /
|
||||||
|
`QQQ_FILE_REFERENCE_ROOTS` (comma-separated). The qbit's default allow-list
|
||||||
|
(`file./var/run/secrets/storage/*`) is inside the default roots.
|
||||||
|
|
||||||
## Resolved dependency diff
|
## Resolved dependency diff
|
||||||
|
|
||||||
|
|||||||
@@ -20,7 +20,7 @@
|
|||||||
<properties>
|
<properties>
|
||||||
<qbit.guardrails.enforce>true</qbit.guardrails.enforce>
|
<qbit.guardrails.enforce>true</qbit.guardrails.enforce>
|
||||||
<qbit.guardrails.strictUiApiRoutes>true</qbit.guardrails.strictUiApiRoutes>
|
<qbit.guardrails.strictUiApiRoutes>true</qbit.guardrails.strictUiApiRoutes>
|
||||||
<revision>0.1.0-SNAPSHOT</revision>
|
<revision>0.2.0-SNAPSHOT</revision>
|
||||||
<maven.compiler.source>21</maven.compiler.source>
|
<maven.compiler.source>21</maven.compiler.source>
|
||||||
<maven.compiler.target>21</maven.compiler.target>
|
<maven.compiler.target>21</maven.compiler.target>
|
||||||
<jacoco.version>0.8.15</jacoco.version>
|
<jacoco.version>0.8.15</jacoco.version>
|
||||||
@@ -29,7 +29,7 @@
|
|||||||
<parent>
|
<parent>
|
||||||
<groupId>com.kingsrook</groupId>
|
<groupId>com.kingsrook</groupId>
|
||||||
<artifactId>qbit-build-parent</artifactId>
|
<artifactId>qbit-build-parent</artifactId>
|
||||||
<version>2.1.0</version>
|
<version>2.2.1</version>
|
||||||
</parent>
|
</parent>
|
||||||
|
|
||||||
<groupId>com.kingsrook.qbits</groupId>
|
<groupId>com.kingsrook.qbits</groupId>
|
||||||
@@ -172,7 +172,35 @@
|
|||||||
</dependencies>
|
</dependencies>
|
||||||
|
|
||||||
<build>
|
<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>
|
||||||
|
<include>qbit-object-storage.properties</include>
|
||||||
|
</includes>
|
||||||
|
</resource>
|
||||||
|
<resource>
|
||||||
|
<directory>src/main/resources</directory>
|
||||||
|
<filtering>false</filtering>
|
||||||
|
<excludes>
|
||||||
|
<exclude>help/*.yaml</exclude>
|
||||||
|
<exclude>qbit-object-storage.properties</exclude>
|
||||||
|
</excludes>
|
||||||
|
</resource>
|
||||||
|
</resources>
|
||||||
<plugins>
|
<plugins>
|
||||||
|
<plugin>
|
||||||
|
<groupId>org.apache.maven.plugins</groupId>
|
||||||
|
<artifactId>maven-surefire-plugin</artifactId>
|
||||||
|
<configuration>
|
||||||
|
<systemPropertyVariables combine.children="append">
|
||||||
|
<qbit.artifactVersion>${project.version}</qbit.artifactVersion>
|
||||||
|
</systemPropertyVariables>
|
||||||
|
</configuration>
|
||||||
|
</plugin>
|
||||||
<plugin>
|
<plugin>
|
||||||
<groupId>org.apache.maven.plugins</groupId>
|
<groupId>org.apache.maven.plugins</groupId>
|
||||||
<artifactId>maven-checkstyle-plugin</artifactId>
|
<artifactId>maven-checkstyle-plugin</artifactId>
|
||||||
@@ -255,14 +283,14 @@
|
|||||||
<profiles>
|
<profiles>
|
||||||
<!-- Built against qqq 4.1 (QBitMetaDataProducer). Until qbit-build-parent -->
|
<!-- Built against qqq 4.1 (QBitMetaDataProducer). Until qbit-build-parent -->
|
||||||
<!-- imports the released 4.1 BOM, this profile is active by default and -->
|
<!-- imports the released 4.1 BOM, this profile is active by default and -->
|
||||||
<!-- imports qqq-bom-pom 4.1.0-RC.7 ahead of the parent's 4.0.0 BOM. -->
|
<!-- imports qqq-bom-pom 4.1.0-RC.15 ahead of the parent's 4.0.0 BOM. -->
|
||||||
<profile>
|
<profile>
|
||||||
<id>qqq-snapshot</id>
|
<id>qqq-snapshot</id>
|
||||||
<activation>
|
<activation>
|
||||||
<activeByDefault>true</activeByDefault>
|
<activeByDefault>true</activeByDefault>
|
||||||
</activation>
|
</activation>
|
||||||
<properties>
|
<properties>
|
||||||
<qqq.snapshot.version>4.1.0-RC.7</qqq.snapshot.version>
|
<qqq.snapshot.version>4.1.0-RC.25</qqq.snapshot.version>
|
||||||
</properties>
|
</properties>
|
||||||
<dependencyManagement>
|
<dependencyManagement>
|
||||||
<dependencies>
|
<dependencies>
|
||||||
|
|||||||
@@ -62,6 +62,7 @@ public class ObjectStorageQBitConfig implements QBitConfig
|
|||||||
private String appName = "objectStorageAdmin";
|
private String appName = "objectStorageAdmin";
|
||||||
private String appLabel = "Storage";
|
private String appLabel = "Storage";
|
||||||
private String parentAppName;
|
private String parentAppName;
|
||||||
|
private String tablePathPrefix;
|
||||||
private List<String> allowedSecretRefPatterns = new ArrayList<>(SecretRefs.DEFAULT_ALLOWED_PATTERNS);
|
private List<String> allowedSecretRefPatterns = new ArrayList<>(SecretRefs.DEFAULT_ALLOWED_PATTERNS);
|
||||||
private Integer bulkCap = 10_000;
|
private Integer bulkCap = 10_000;
|
||||||
private Boolean bucketScoping = true;
|
private Boolean bucketScoping = true;
|
||||||
@@ -425,6 +426,39 @@ public class ObjectStorageQBitConfig implements QBitConfig
|
|||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/********************************************************************************
|
||||||
|
** Getter for tablePathPrefix: the front of the links the Storage home makes to
|
||||||
|
** tables and records. Defaults to /app, the route every QQQ frontend serves
|
||||||
|
** tables at (/app/<tableName>).
|
||||||
|
*******************************************************************************/
|
||||||
|
public String getTablePathPrefix()
|
||||||
|
{
|
||||||
|
return (this.tablePathPrefix == null ? "/app" : this.tablePathPrefix);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/********************************************************************************
|
||||||
|
** Setter for tablePathPrefix.
|
||||||
|
*******************************************************************************/
|
||||||
|
public void setTablePathPrefix(String tablePathPrefix)
|
||||||
|
{
|
||||||
|
this.tablePathPrefix = tablePathPrefix;
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/********************************************************************************
|
||||||
|
** Fluent setter for tablePathPrefix.
|
||||||
|
*******************************************************************************/
|
||||||
|
public ObjectStorageQBitConfig withTablePathPrefix(String tablePathPrefix)
|
||||||
|
{
|
||||||
|
this.tablePathPrefix = tablePathPrefix;
|
||||||
|
return (this);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
/********************************************************************************
|
/********************************************************************************
|
||||||
** Getter for allowedSecretRefPatterns.
|
** Getter for allowedSecretRefPatterns.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
|
|||||||
@@ -21,9 +21,13 @@
|
|||||||
package com.kingsrook.qbits.objectstorage;
|
package com.kingsrook.qbits.objectstorage;
|
||||||
|
|
||||||
|
|
||||||
|
import java.io.IOException;
|
||||||
|
import java.io.InputStream;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
import java.util.Properties;
|
||||||
import com.kingsrook.qbits.objectstorage.app.StorageWidgets;
|
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.ConfigSync;
|
||||||
import com.kingsrook.qbits.objectstorage.internal.DefaultObjectHeaderPolicy;
|
import com.kingsrook.qbits.objectstorage.internal.DefaultObjectHeaderPolicy;
|
||||||
import com.kingsrook.qbits.objectstorage.internal.ObjectStorageRuntime;
|
import com.kingsrook.qbits.objectstorage.internal.ObjectStorageRuntime;
|
||||||
@@ -31,8 +35,11 @@ import com.kingsrook.qbits.objectstorage.model.ObjectStorageHelp;
|
|||||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||||
import com.kingsrook.qbits.objectstorage.processes.ObjectStorageProcesses;
|
import com.kingsrook.qbits.objectstorage.processes.ObjectStorageProcesses;
|
||||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
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.MetaDataProducerMultiOutput;
|
||||||
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
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.dashboard.QWidgetMetaData;
|
||||||
import com.kingsrook.qqq.backend.core.model.metadata.help.bundles.HelpBundles;
|
import com.kingsrook.qqq.backend.core.model.metadata.help.bundles.HelpBundles;
|
||||||
import com.kingsrook.qqq.backend.core.model.metadata.joins.QJoinMetaData;
|
import com.kingsrook.qqq.backend.core.model.metadata.joins.QJoinMetaData;
|
||||||
@@ -56,13 +63,41 @@ public class ObjectStorageQBitProducer implements QBitMetaDataProducer<ObjectSto
|
|||||||
{
|
{
|
||||||
public static final String GROUP_ID = "com.kingsrook.qbits";
|
public static final String GROUP_ID = "com.kingsrook.qbits";
|
||||||
public static final String ARTIFACT_ID = "qbit-object-storage";
|
public static final String ARTIFACT_ID = "qbit-object-storage";
|
||||||
public static final String VERSION = "0.1.0";
|
public static final String VERSION = loadVersion();
|
||||||
|
|
||||||
private ObjectStorageQBitConfig qBitConfig;
|
private ObjectStorageQBitConfig qBitConfig;
|
||||||
private String namespace;
|
private String namespace;
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Read the artifact version from the Maven-filtered resource.
|
||||||
|
*******************************************************************************/
|
||||||
|
private static String loadVersion()
|
||||||
|
{
|
||||||
|
try(InputStream inputStream = ObjectStorageQBitProducer.class.getClassLoader().getResourceAsStream("qbit-object-storage.properties"))
|
||||||
|
{
|
||||||
|
if(inputStream == null)
|
||||||
|
{
|
||||||
|
throw new IllegalStateException("Missing qbit-object-storage.properties; build with Maven to generate the artifact version");
|
||||||
|
}
|
||||||
|
Properties properties = new Properties();
|
||||||
|
properties.load(inputStream);
|
||||||
|
String version = properties.getProperty("version");
|
||||||
|
if(version == null || version.isBlank() || version.contains("${"))
|
||||||
|
{
|
||||||
|
throw new IllegalStateException("Missing or unfiltered version in qbit-object-storage.properties; build with Maven");
|
||||||
|
}
|
||||||
|
return (version);
|
||||||
|
}
|
||||||
|
catch(IOException e)
|
||||||
|
{
|
||||||
|
throw new IllegalStateException("Could not read artifact version from qbit-object-storage.properties", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Fluent setter for the config.
|
** Fluent setter for the config.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
@@ -188,6 +223,9 @@ public class ObjectStorageQBitProducer implements QBitMetaDataProducer<ObjectSto
|
|||||||
output.add(app);
|
output.add(app);
|
||||||
}
|
}
|
||||||
output.addSelfToInstance(qInstance);
|
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)
|
if(Boolean.TRUE.equals(qBitConfig.getAdminApp()) && qBitConfig.getParentAppName() != null && qInstance.getApp(qBitConfig.getParentAppName()) != null)
|
||||||
{
|
{
|
||||||
///////////////////////////////////////////////////////////////////////////////
|
///////////////////////////////////////////////////////////////////////////////
|
||||||
|
|||||||
@@ -46,6 +46,7 @@ import com.kingsrook.qbits.objectstorage.internal.PostPolicySigner;
|
|||||||
import com.kingsrook.qbits.objectstorage.internal.ProtectionRules;
|
import com.kingsrook.qbits.objectstorage.internal.ProtectionRules;
|
||||||
import com.kingsrook.qbits.objectstorage.internal.RateLimiter;
|
import com.kingsrook.qbits.objectstorage.internal.RateLimiter;
|
||||||
import com.kingsrook.qqq.backend.core.actions.permissions.CorePermissionEnforcer;
|
import com.kingsrook.qqq.backend.core.actions.permissions.CorePermissionEnforcer;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||||
import com.kingsrook.qqq.backend.core.context.QContext;
|
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||||
import com.kingsrook.qqq.backend.core.utils.StringUtils;
|
import com.kingsrook.qqq.backend.core.utils.StringUtils;
|
||||||
@@ -96,7 +97,8 @@ public final class StorageBucket
|
|||||||
{
|
{
|
||||||
private static final QLogger LOG = QLogger.getLogger(StorageBucket.class);
|
private static final QLogger LOG = QLogger.getLogger(StorageBucket.class);
|
||||||
|
|
||||||
private static final Map<String, long[]> PRESIGN_COUNTS = new ConcurrentHashMap<>();
|
private static final Map<String, long[]> PRESIGN_COUNTS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
|
||||||
|
"per-pod presign rate limiter; per-node rate limits are accepted for now (cluster HA spec, owner decision 4)");
|
||||||
|
|
||||||
private final StorageBucketConfig bucket;
|
private final StorageBucketConfig bucket;
|
||||||
private final ConnectionRuntime connection;
|
private final ConnectionRuntime connection;
|
||||||
|
|||||||
@@ -106,16 +106,12 @@ public final class StorageWidgets
|
|||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** The prefix of table and record links.
|
** The prefix of table and record links: /app, so a link is /app/<tableName>
|
||||||
|
** (the frontend serves every table, in any app, at that one flat route).
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public static String tablePathPrefix()
|
public static String tablePathPrefix()
|
||||||
{
|
{
|
||||||
if(!ObjectStorageRuntime.isInstalled())
|
return (ObjectStorageRuntime.isInstalled() ? ObjectStorageRuntime.getConfig().getTablePathPrefix() : "/app");
|
||||||
{
|
|
||||||
return ("/objectStorageAdmin");
|
|
||||||
}
|
|
||||||
ObjectStorageQBitConfig config = ObjectStorageRuntime.getConfig();
|
|
||||||
return (config.getParentAppName() == null ? "/" + config.getAppName() : "/" + config.getParentAppName() + "/" + config.getAppName());
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+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))));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -43,6 +43,7 @@ import com.kingsrook.qbits.objectstorage.internal.ConfigSync;
|
|||||||
import com.kingsrook.qbits.objectstorage.internal.Db;
|
import com.kingsrook.qbits.objectstorage.internal.Db;
|
||||||
import com.kingsrook.qbits.objectstorage.internal.ObjectStorageRuntime;
|
import com.kingsrook.qbits.objectstorage.internal.ObjectStorageRuntime;
|
||||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||||
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
|
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
|
||||||
@@ -78,7 +79,8 @@ public final class ObjectIndex
|
|||||||
|
|
||||||
public static final int PAGE_SIZE = 1000;
|
public static final int PAGE_SIZE = 1000;
|
||||||
|
|
||||||
private static final Set<String> RUNNING = java.util.concurrent.ConcurrentHashMap.newKeySet();
|
private static final Set<String> RUNNING = ClusterCaches.nodeLocal(java.util.concurrent.ConcurrentHashMap.newKeySet(),
|
||||||
|
"the buckets this node is scanning now (an in-flight guard, not cached data)");
|
||||||
|
|
||||||
/** Test seam: simulate a crash after this many pages of a scan (so resume can be tested). */
|
/** Test seam: simulate a crash after this many pages of a scan (so resume can be tested). */
|
||||||
static volatile Integer failAfterPages = null;
|
static volatile Integer failAfterPages = null;
|
||||||
|
|||||||
@@ -21,7 +21,9 @@
|
|||||||
package com.kingsrook.qbits.objectstorage.internal;
|
package com.kingsrook.qbits.objectstorage.internal;
|
||||||
|
|
||||||
|
|
||||||
|
import javax.net.ssl.SSLException;
|
||||||
import java.net.URI;
|
import java.net.URI;
|
||||||
|
import java.net.UnknownHostException;
|
||||||
import java.net.http.HttpClient;
|
import java.net.http.HttpClient;
|
||||||
import java.net.http.HttpRequest;
|
import java.net.http.HttpRequest;
|
||||||
import java.net.http.HttpResponse;
|
import java.net.http.HttpResponse;
|
||||||
@@ -35,6 +37,7 @@ import java.util.EnumMap;
|
|||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import java.util.UUID;
|
import java.util.UUID;
|
||||||
|
import java.util.function.Function;
|
||||||
import com.kingsrook.qbits.objectstorage.api.Capability;
|
import com.kingsrook.qbits.objectstorage.api.Capability;
|
||||||
import com.kingsrook.qbits.objectstorage.api.CapabilityStatus;
|
import com.kingsrook.qbits.objectstorage.api.CapabilityStatus;
|
||||||
import com.kingsrook.qbits.objectstorage.api.PresignedPost;
|
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.qbits.objectstorage.model.ObjectStorageTables;
|
||||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
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.core.sync.RequestBody;
|
||||||
|
import software.amazon.awssdk.services.s3.DelegatingS3Client;
|
||||||
import software.amazon.awssdk.services.s3.S3Client;
|
import software.amazon.awssdk.services.s3.S3Client;
|
||||||
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
|
import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadRequest;
|
||||||
import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
|
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.ObjectIdentifier;
|
||||||
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
|
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
|
||||||
import software.amazon.awssdk.services.s3.model.S3Exception;
|
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.model.UploadPartRequest;
|
||||||
import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest;
|
import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest;
|
||||||
import software.amazon.awssdk.services.s3.presigner.model.PutObjectPresignRequest;
|
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);
|
private static final QLogger LOG = QLogger.getLogger(CapabilityProbe.class);
|
||||||
|
|
||||||
public static final String PROBE_PREFIX = ".qqq-probe/";
|
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;
|
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 final List<String> problems = new ArrayList<>();
|
||||||
private Long clockSkewSeconds;
|
private Long clockSkewSeconds;
|
||||||
private Boolean presignHostReachable;
|
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)
|
public static Report probeConnection(String connectionName, boolean includeWrite)
|
||||||
{
|
{
|
||||||
details.clear();
|
details.get().clear();
|
||||||
Report report = new Report();
|
Report report = new Report();
|
||||||
ConnectionRuntime connection = ObjectStorageRuntime.connection(connectionName);
|
ConnectionRuntime connection = ObjectStorageRuntime.connection(connectionName);
|
||||||
StorageConnectionConfig config = connection.getConfig();
|
StorageConnectionConfig config = connection.getConfig();
|
||||||
@@ -230,7 +248,7 @@ public final class CapabilityProbe
|
|||||||
List<String> notes = new ArrayList<>();
|
List<String> notes = new ArrayList<>();
|
||||||
report.notes.put(bucket.getAlias(), notes);
|
report.notes.put(bucket.getAlias(), notes);
|
||||||
|
|
||||||
readProbes(connection, bucket, results, report);
|
readProbes(connection, bucket, results, report, null);
|
||||||
recordVersioningState(bucket);
|
recordVersioningState(bucket);
|
||||||
|
|
||||||
if(includeWrite)
|
if(includeWrite)
|
||||||
@@ -251,7 +269,7 @@ public final class CapabilityProbe
|
|||||||
}
|
}
|
||||||
for(Map.Entry<Capability, CapabilityStatus> e : results.entrySet())
|
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);
|
report.byBucket.get(writable.getAlias()).putAll(presign);
|
||||||
for(Map.Entry<Capability, CapabilityStatus> e : presign.entrySet())
|
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);
|
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)
|
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))
|
if(seen == null || !Db.available(ObjectStorageTables.BUCKET))
|
||||||
{
|
{
|
||||||
return;
|
return;
|
||||||
@@ -317,14 +444,14 @@ public final class CapabilityProbe
|
|||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** The read-only probes of one bucket.
|
** 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 name = name(bucket);
|
||||||
String alias = bucket.getAlias();
|
String alias = bucket.getAlias();
|
||||||
String sampleKey = null;
|
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());
|
ListObjectsV2Response r = client.listObjectsV2(ListObjectsV2Request.builder().bucket(name).maxKeys(1).build());
|
||||||
results.put(Capability.LIST_V2, CapabilityStatus.SUPPORTED);
|
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));
|
results.put(Capability.LIST_V2, classify(alias, Capability.LIST_V2, e));
|
||||||
report.problems.add(alias + ": listing failed (" + shortMessage(e) + ")");
|
report.problems.add(alias + ": listing failed (" + shortMessage(e) + ")");
|
||||||
|
report.problemKind = problemKind(e);
|
||||||
}
|
}
|
||||||
|
|
||||||
probe(results, alias, Capability.VERSIONING, () ->
|
probe(results, alias, Capability.VERSIONING, () ->
|
||||||
@@ -359,10 +487,10 @@ public final class CapabilityProbe
|
|||||||
String status = client.getBucketVersioning(b -> b.bucket(name)).statusAsString();
|
String status = client.getBucketVersioning(b -> b.bucket(name)).statusAsString();
|
||||||
if(status == null && ProviderDefaults.defaultStatus(connection.getConfig().getProviderType(), Capability.VERSIONING) == CapabilityStatus.UNSUPPORTED)
|
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);
|
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);
|
return (CapabilityStatus.SUPPORTED);
|
||||||
});
|
});
|
||||||
probe(results, alias, Capability.OBJECT_LOCK, () ->
|
probe(results, alias, Capability.OBJECT_LOCK, () ->
|
||||||
@@ -376,7 +504,7 @@ public final class CapabilityProbe
|
|||||||
{
|
{
|
||||||
if("ObjectLockConfigurationNotFoundError".equals(ErrorMapper.errorCode(e)))
|
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);
|
return (CapabilityStatus.UNSUPPORTED);
|
||||||
}
|
}
|
||||||
throw (e);
|
throw (e);
|
||||||
@@ -391,7 +519,7 @@ public final class CapabilityProbe
|
|||||||
if(key == null)
|
if(key == null)
|
||||||
{
|
{
|
||||||
results.put(Capability.OBJECT_TAGGING, CapabilityStatus.UNKNOWN);
|
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
|
else
|
||||||
{
|
{
|
||||||
@@ -505,7 +633,7 @@ public final class CapabilityProbe
|
|||||||
try
|
try
|
||||||
{
|
{
|
||||||
client.putObject(PutObjectRequest.builder().bucket(name).key(key).ifNoneMatch("*").build(), RequestBody.fromString("2"));
|
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);
|
return (CapabilityStatus.UNSUPPORTED);
|
||||||
}
|
}
|
||||||
catch(S3Exception e)
|
catch(S3Exception e)
|
||||||
@@ -564,7 +692,7 @@ public final class CapabilityProbe
|
|||||||
report.presignHostReachable = true;
|
report.presignHostReachable = true;
|
||||||
if(r.statusCode() == 200 && !"probe".equals(r.body()))
|
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 (CapabilityStatus.UNKNOWN);
|
||||||
}
|
}
|
||||||
return (httpStatus(alias, Capability.PRESIGNED_GET, r.statusCode(), r.body()));
|
return (httpStatus(alias, Capability.PRESIGNED_GET, r.statusCode(), r.body()));
|
||||||
@@ -618,7 +746,7 @@ public final class CapabilityProbe
|
|||||||
{
|
{
|
||||||
return (CapabilityStatus.SUPPORTED);
|
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)
|
if(status == 403)
|
||||||
{
|
{
|
||||||
return (CapabilityStatus.NOT_PERMITTED);
|
return (CapabilityStatus.NOT_PERMITTED);
|
||||||
@@ -669,7 +797,7 @@ public final class CapabilityProbe
|
|||||||
static CapabilityStatus classify(String alias, Capability capability, Exception e)
|
static CapabilityStatus classify(String alias, Capability capability, Exception e)
|
||||||
{
|
{
|
||||||
String detail = shortMessage(e);
|
String detail = shortMessage(e);
|
||||||
details.put(alias + ":" + capability, detail);
|
details.get().put(alias + ":" + capability, detail);
|
||||||
if(e instanceof S3Exception s3)
|
if(e instanceof S3Exception s3)
|
||||||
{
|
{
|
||||||
String code = ErrorMapper.errorCode(s3);
|
String code = ErrorMapper.errorCode(s3);
|
||||||
|
|||||||
@@ -30,6 +30,7 @@ import com.kingsrook.qbits.objectstorage.api.Capability;
|
|||||||
import com.kingsrook.qbits.objectstorage.api.CapabilityStatus;
|
import com.kingsrook.qbits.objectstorage.api.CapabilityStatus;
|
||||||
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
||||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||||
|
|
||||||
@@ -47,7 +48,11 @@ public final class CapabilityStore
|
|||||||
public static final Set<Capability> CONNECTION_SCOPE = EnumSet.of(Capability.LIST_V2, Capability.PRESIGNED_GET, Capability.PRESIGNED_PUT,
|
public static final Set<Capability> CONNECTION_SCOPE = EnumSet.of(Capability.LIST_V2, Capability.PRESIGNED_GET, Capability.PRESIGNED_PUT,
|
||||||
Capability.PRESIGNED_POST, Capability.USAGE_ADMIN_API, Capability.QUOTA_ADMIN_API);
|
Capability.PRESIGNED_POST, Capability.USAGE_ADMIN_API, Capability.QUOTA_ADMIN_API);
|
||||||
|
|
||||||
private static final Map<String, CapabilityStatus> CACHE = new ConcurrentHashMap<>();
|
/** Cluster cache name of the capability statuses (a probe result recorded on one node clears that key on every node). */
|
||||||
|
public static final String CACHE_NAME = "qbit-object-storage.capabilities";
|
||||||
|
|
||||||
|
private static final Map<String, CapabilityStatus> CACHE = ClusterCaches.shared(CACHE_NAME, new ConcurrentHashMap<>(),
|
||||||
|
CapabilityStore::clearCache, CapabilityStore::clearKey);
|
||||||
private static final Set<String> LOADED = ConcurrentHashMap.newKeySet();
|
private static final Set<String> LOADED = ConcurrentHashMap.newKeySet();
|
||||||
|
|
||||||
|
|
||||||
@@ -62,7 +67,7 @@ public final class CapabilityStore
|
|||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Drop everything cached.
|
** Drop everything cached on this node.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public static void clearCache()
|
public static void clearCache()
|
||||||
{
|
{
|
||||||
@@ -72,6 +77,28 @@ public final class CapabilityStore
|
|||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Drop one cached status on this node; the next read reloads the missing
|
||||||
|
** statuses from the capability rows (cached ones are kept).
|
||||||
|
*******************************************************************************/
|
||||||
|
static void clearKey(String key)
|
||||||
|
{
|
||||||
|
CACHE.remove(key);
|
||||||
|
LOADED.clear();
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Whether a status is cached on this node (tests).
|
||||||
|
*******************************************************************************/
|
||||||
|
static boolean isCached(StorageBucketConfig bucket, Capability capability)
|
||||||
|
{
|
||||||
|
return (CACHE.containsKey(scopeKey(bucket, capability)));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** The status of a capability for a bucket alias.
|
** The status of a capability for a bucket alias.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
@@ -112,10 +139,24 @@ public final class CapabilityStore
|
|||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Record one probe result (cache and, when the tables are there, the row).
|
** Record one probe result (cache and, when the tables are there, the row).
|
||||||
|
** The row is written first, then the key is cleared on every node (they
|
||||||
|
** reload it from the row), then cached here.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public static void record(StorageBucketConfig bucket, Capability capability, CapabilityStatus status, String detail)
|
public static void record(StorageBucketConfig bucket, Capability capability, CapabilityStatus status, String detail)
|
||||||
{
|
{
|
||||||
CACHE.put(scopeKey(bucket, capability), status);
|
String key = scopeKey(bucket, capability);
|
||||||
|
persist(bucket, capability, status, detail);
|
||||||
|
ClusterCaches.invalidateKey(CACHE_NAME, key);
|
||||||
|
CACHE.put(key, status);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Write one probe result to its capability row; never throws.
|
||||||
|
*******************************************************************************/
|
||||||
|
private static void persist(StorageBucketConfig bucket, Capability capability, CapabilityStatus status, String detail)
|
||||||
|
{
|
||||||
if(!Db.available(ObjectStorageTables.BUCKET_CAPABILITY))
|
if(!Db.available(ObjectStorageTables.BUCKET_CAPABILITY))
|
||||||
{
|
{
|
||||||
return;
|
return;
|
||||||
|
|||||||
+95
@@ -0,0 +1,95 @@
|
|||||||
|
/*
|
||||||
|
* 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.internal;
|
||||||
|
|
||||||
|
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Optional;
|
||||||
|
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||||
|
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||||
|
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Post insert, update and delete of the editable config rows that the runtime
|
||||||
|
** caches: a bucket row (presign limits, ConfigSync) or a header rule
|
||||||
|
** (DefaultObjectHeaderPolicy). Clears that cache on every node, so an admin's
|
||||||
|
** edit takes effect everywhere at once; the short TTLs remain the safety net.
|
||||||
|
*******************************************************************************/
|
||||||
|
public class ConfigRowChangeCustomizer implements TableCustomizerInterface
|
||||||
|
{
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** {@inheritDoc}
|
||||||
|
*******************************************************************************/
|
||||||
|
@Override
|
||||||
|
public List<QRecord> postInsert(InsertInput insertInput, List<QRecord> records) throws QException
|
||||||
|
{
|
||||||
|
invalidate(insertInput.getTableName());
|
||||||
|
return (records);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** {@inheritDoc}
|
||||||
|
*******************************************************************************/
|
||||||
|
@Override
|
||||||
|
public List<QRecord> postUpdate(UpdateInput updateInput, List<QRecord> records, Optional<List<QRecord>> oldRecordList) throws QException
|
||||||
|
{
|
||||||
|
invalidate(updateInput.getTableName());
|
||||||
|
return (records);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** {@inheritDoc}
|
||||||
|
*******************************************************************************/
|
||||||
|
@Override
|
||||||
|
public List<QRecord> postDelete(DeleteInput deleteInput, List<QRecord> records) throws QException
|
||||||
|
{
|
||||||
|
invalidate(deleteInput.getTableName());
|
||||||
|
return (records);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Clear the cache that holds the table's rows, on every node.
|
||||||
|
*******************************************************************************/
|
||||||
|
static void invalidate(String tableName)
|
||||||
|
{
|
||||||
|
if(ObjectStorageTables.BUCKET.equals(tableName))
|
||||||
|
{
|
||||||
|
ClusterCaches.invalidate(ConfigSync.ROW_LIMITS_CACHE_NAME);
|
||||||
|
}
|
||||||
|
else if(ObjectStorageTables.HEADER_RULE.equals(tableName))
|
||||||
|
{
|
||||||
|
ClusterCaches.invalidate(DefaultObjectHeaderPolicy.CACHE_NAME);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -30,6 +30,7 @@ import com.kingsrook.qbits.objectstorage.config.CredentialsMode;
|
|||||||
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
||||||
import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
||||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||||
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||||
@@ -48,10 +49,16 @@ public final class ConfigSync
|
|||||||
{
|
{
|
||||||
private static final QLogger LOG = QLogger.getLogger(ConfigSync.class);
|
private static final QLogger LOG = QLogger.getLogger(ConfigSync.class);
|
||||||
|
|
||||||
private static final Map<String, Integer> BUCKET_IDS = new ConcurrentHashMap<>();
|
/** Cluster cache name of the bucket rows' presign limits (cleared on every node when a bucket row changes). */
|
||||||
private static final Map<String, Integer> CONNECTION_IDS = new ConcurrentHashMap<>();
|
public static final String ROW_LIMITS_CACHE_NAME = "qbit-object-storage.bucket-row-limits";
|
||||||
|
|
||||||
private static final Map<String, RowLimits> ROW_LIMITS = new ConcurrentHashMap<>();
|
private static final Map<String, Integer> BUCKET_IDS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
|
||||||
|
"bucket alias to row id, written by the config sync; bucket rows cannot be inserted or deleted through the UI or API, so the ids are the same on every node");
|
||||||
|
private static final Map<String, Integer> CONNECTION_IDS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
|
||||||
|
"connection name to row id, written by the config sync; connection rows cannot be inserted or deleted through the UI or API, so the ids are the same on every node");
|
||||||
|
|
||||||
|
private static final Map<String, RowLimits> ROW_LIMITS = ClusterCaches.shared(ROW_LIMITS_CACHE_NAME, new ConcurrentHashMap<>(),
|
||||||
|
ConfigSync::clearRowLimits, alias -> ConfigSync.ROW_LIMITS.remove(alias));
|
||||||
private static final long ROW_LIMIT_CACHE_MILLIS = 15_000L;
|
private static final long ROW_LIMIT_CACHE_MILLIS = 15_000L;
|
||||||
|
|
||||||
private static volatile boolean synced = false;
|
private static volatile boolean synced = false;
|
||||||
@@ -313,7 +320,8 @@ public final class ConfigSync
|
|||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Forget the cached row limits (after a row edit in a test; they also expire).
|
** Forget the cached row limits on this node (a bucket row edit clears them on
|
||||||
|
** every node through ConfigRowChangeCustomizer; they also expire).
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public static void clearRowLimits()
|
public static void clearRowLimits()
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -22,7 +22,6 @@ package com.kingsrook.qbits.objectstorage.internal;
|
|||||||
|
|
||||||
|
|
||||||
import java.net.URI;
|
import java.net.URI;
|
||||||
import java.util.function.Function;
|
|
||||||
import com.kingsrook.qbits.objectstorage.config.AddressingStyle;
|
import com.kingsrook.qbits.objectstorage.config.AddressingStyle;
|
||||||
import com.kingsrook.qbits.objectstorage.config.CredentialsMode;
|
import com.kingsrook.qbits.objectstorage.config.CredentialsMode;
|
||||||
import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
||||||
@@ -30,17 +29,10 @@ import com.kingsrook.qqq.backend.core.utils.StringUtils;
|
|||||||
import com.kingsrook.qqq.backend.module.filesystem.s3.model.metadata.S3BackendMetaData;
|
import com.kingsrook.qqq.backend.module.filesystem.s3.model.metadata.S3BackendMetaData;
|
||||||
import com.kingsrook.qqq.backend.module.filesystem.s3.model.metadata.S3ChecksumMode;
|
import com.kingsrook.qqq.backend.module.filesystem.s3.model.metadata.S3ChecksumMode;
|
||||||
import com.kingsrook.qqq.backend.module.filesystem.s3.utils.S3ClientFactory;
|
import com.kingsrook.qqq.backend.module.filesystem.s3.utils.S3ClientFactory;
|
||||||
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
|
|
||||||
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
|
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
|
||||||
import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
|
|
||||||
import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
|
|
||||||
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
|
|
||||||
import software.amazon.awssdk.awscore.AwsRequestOverrideConfiguration;
|
|
||||||
import software.amazon.awssdk.regions.Region;
|
import software.amazon.awssdk.regions.Region;
|
||||||
import software.amazon.awssdk.services.s3.DelegatingS3Client;
|
|
||||||
import software.amazon.awssdk.services.s3.S3Client;
|
import software.amazon.awssdk.services.s3.S3Client;
|
||||||
import software.amazon.awssdk.services.s3.S3Configuration;
|
import software.amazon.awssdk.services.s3.S3Configuration;
|
||||||
import software.amazon.awssdk.services.s3.model.S3Request;
|
|
||||||
import software.amazon.awssdk.services.s3.presigner.S3Presigner;
|
import software.amazon.awssdk.services.s3.presigner.S3Presigner;
|
||||||
|
|
||||||
|
|
||||||
@@ -198,16 +190,7 @@ public class ConnectionRuntime
|
|||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
private S3Client buildClient()
|
private S3Client buildClient()
|
||||||
{
|
{
|
||||||
S3BackendMetaData backend = backendMetaData();
|
return (S3ClientFactory.buildClient(backendMetaData()));
|
||||||
String token = config.getCredentialsMode() == CredentialsMode.STATIC_REFS ? SecretRefs.resolve(config.getSessionTokenRef()) : null;
|
|
||||||
AwsCredentialsProvider sessionCredentials = null;
|
|
||||||
if(token != null && backend.getAccessKey() != null && backend.getSecretKey() != null)
|
|
||||||
{
|
|
||||||
sessionCredentials = StaticCredentialsProvider.create(AwsSessionCredentials.create(backend.getAccessKey(), backend.getSecretKey(), token));
|
|
||||||
}
|
|
||||||
|
|
||||||
S3Client coreClient = S3ClientFactory.buildClient(backend);
|
|
||||||
return (sessionCredentials == null ? coreClient : new SessionCredentialsClient(coreClient, sessionCredentials));
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -229,8 +212,7 @@ public class ConnectionRuntime
|
|||||||
.withSocketTimeoutMillis(config.getReadTimeoutMs());
|
.withSocketTimeoutMillis(config.getReadTimeoutMs());
|
||||||
if(config.getCredentialsMode() == CredentialsMode.STATIC_REFS)
|
if(config.getCredentialsMode() == CredentialsMode.STATIC_REFS)
|
||||||
{
|
{
|
||||||
backend.withAccessKey(SecretRefs.resolve(config.getAccessKeyRef()))
|
withStaticCredentials(backend, config.getAccessKeyRef(), config.getSecretKeyRef());
|
||||||
.withSecretKey(SecretRefs.resolve(config.getSecretKeyRef()));
|
|
||||||
}
|
}
|
||||||
return (backend);
|
return (backend);
|
||||||
}
|
}
|
||||||
@@ -270,63 +252,33 @@ public class ConnectionRuntime
|
|||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Static keys from the references, or the SDK default chain.
|
** Static keys (with the session token, when set) from the references, or the
|
||||||
|
** SDK default chain; core builds the provider.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
private AwsCredentialsProvider credentials(String accessRef, String secretRef)
|
private AwsCredentialsProvider credentials(String accessRef, String secretRef)
|
||||||
{
|
{
|
||||||
|
S3BackendMetaData credentials = new S3BackendMetaData();
|
||||||
if(config.getCredentialsMode() == CredentialsMode.STATIC_REFS)
|
if(config.getCredentialsMode() == CredentialsMode.STATIC_REFS)
|
||||||
{
|
{
|
||||||
String accessKey = SecretRefs.resolve(accessRef);
|
withStaticCredentials(credentials, accessRef, secretRef);
|
||||||
String secretKey = SecretRefs.resolve(secretRef);
|
|
||||||
if(accessKey != null && secretKey != null)
|
|
||||||
{
|
|
||||||
String token = SecretRefs.resolve(config.getSessionTokenRef());
|
|
||||||
return (StaticCredentialsProvider.create(token == null ? AwsBasicCredentials.create(accessKey, secretKey)
|
|
||||||
: AwsSessionCredentials.create(accessKey, secretKey, token)));
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
return (DefaultCredentialsProvider.builder().build());
|
return (S3ClientFactory.buildCredentialsProvider(credentials));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** STAND-IN(core=QRun-IO/qqq#1165, removeBy=4.1.0-RC.8)
|
** Resolve static keys onto core meta data. The session token applies only
|
||||||
** RC.7 S3BackendMetaData and S3ClientFactory still lack a session token or
|
** with both keys, as before (core rejects a token without keys).
|
||||||
** credentials-provider hook (verified against the released JAR).
|
|
||||||
** Override request credentials only; core still owns the client and transport.
|
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
private static class SessionCredentialsClient extends DelegatingS3Client
|
private void withStaticCredentials(S3BackendMetaData backend, String accessRef, String secretRef)
|
||||||
{
|
{
|
||||||
private final AwsCredentialsProvider credentialsProvider;
|
String accessKey = SecretRefs.resolve(accessRef);
|
||||||
|
String secretKey = SecretRefs.resolve(secretRef);
|
||||||
|
backend.withAccessKey(accessKey).withSecretKey(secretKey);
|
||||||
|
if(accessKey != null && secretKey != null)
|
||||||
/*******************************************************************************
|
|
||||||
** Wrap a core-built client with the connection's session credentials.
|
|
||||||
*******************************************************************************/
|
|
||||||
private SessionCredentialsClient(S3Client delegate, AwsCredentialsProvider credentialsProvider)
|
|
||||||
{
|
{
|
||||||
super(delegate);
|
backend.withSessionToken(SecretRefs.resolve(config.getSessionTokenRef()));
|
||||||
this.credentialsProvider = credentialsProvider;
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
|
||||||
** Preserve request overrides and the concrete request type while adding the
|
|
||||||
** session token to signing, including streaming and paginated operations.
|
|
||||||
*******************************************************************************/
|
|
||||||
@Override
|
|
||||||
@SuppressWarnings("unchecked") // toBuilder().build() retains the concrete request type T.
|
|
||||||
protected <T extends S3Request, ReturnT> ReturnT invokeOperation(T request, Function<T, ReturnT> operation)
|
|
||||||
{
|
|
||||||
AwsRequestOverrideConfiguration.Builder overrides = request.overrideConfiguration()
|
|
||||||
.map(AwsRequestOverrideConfiguration::toBuilder).orElseGet(AwsRequestOverrideConfiguration::builder);
|
|
||||||
T signedRequest = (T) request.toBuilder()
|
|
||||||
.overrideConfiguration(overrides.credentialsProvider(credentialsProvider).build())
|
|
||||||
.build();
|
|
||||||
return (operation.apply(signedRequest));
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -48,9 +48,21 @@ import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
|
|||||||
** Small helpers for reading and writing the qbit's own tables as the system
|
** Small helpers for reading and writing the qbit's own tables as the system
|
||||||
** user (the service writes its audit log and index regardless of the calling
|
** user (the service writes its audit log and index regardless of the calling
|
||||||
** user's per-bucket scope; the calling user's permissions were checked first).
|
** user's per-bucket scope; the calling user's permissions were checked first).
|
||||||
|
**
|
||||||
|
** Every write made here carries core's per-call permission bypass
|
||||||
|
** (AbstractActionInput.withPermissionBypass). That mark, not the session, is
|
||||||
|
** what lets the service write tables nobody may write directly: a host that
|
||||||
|
** enforces permissions inside the core actions asks ObjectStoragePermissionChecker
|
||||||
|
** about every outermost write that does not carry a bypass, and the checker
|
||||||
|
** refuses those tables to every session, the system user included.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public final class Db
|
public final class Db
|
||||||
{
|
{
|
||||||
|
/** The reason recorded on (and logged by core for) each write the service makes of its own tables. */
|
||||||
|
public static final String OWN_WRITE_REASON = "The object storage service writes its own tables (operation log, index, capabilities, usage); the caller's storage permission was checked before the storage call.";
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Something that may throw.
|
** Something that may throw.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
@@ -237,7 +249,9 @@ public final class Db
|
|||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public static List<QRecord> insert(String table, List<QRecord> records) throws QException
|
public static List<QRecord> insert(String table, List<QRecord> records) throws QException
|
||||||
{
|
{
|
||||||
return (new InsertAction().execute(new InsertInput(table).withRecords(records)).getRecords());
|
InsertInput input = new InsertInput(table).withRecords(records);
|
||||||
|
input.withPermissionBypass(OWN_WRITE_REASON);
|
||||||
|
return (new InsertAction().execute(input).getRecords());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -259,7 +273,9 @@ public final class Db
|
|||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public static List<QRecord> update(String table, List<QRecord> records) throws QException
|
public static List<QRecord> update(String table, List<QRecord> records) throws QException
|
||||||
{
|
{
|
||||||
return (new UpdateAction().execute(new UpdateInput(table).withRecords(records)).getRecords());
|
UpdateInput input = new UpdateInput(table).withRecords(records);
|
||||||
|
input.withPermissionBypass(OWN_WRITE_REASON);
|
||||||
|
return (new UpdateAction().execute(input).getRecords());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -271,7 +287,9 @@ public final class Db
|
|||||||
{
|
{
|
||||||
if(!ids.isEmpty())
|
if(!ids.isEmpty())
|
||||||
{
|
{
|
||||||
new DeleteAction().execute(new DeleteInput(table).withPrimaryKeys(ids));
|
DeleteInput input = new DeleteInput(table).withPrimaryKeys(ids);
|
||||||
|
input.withPermissionBypass(OWN_WRITE_REASON);
|
||||||
|
new DeleteAction().execute(input);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+23
-9
@@ -23,12 +23,14 @@ package com.kingsrook.qbits.objectstorage.internal;
|
|||||||
|
|
||||||
import java.util.Comparator;
|
import java.util.Comparator;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
import java.util.concurrent.atomic.AtomicReference;
|
||||||
import java.util.regex.Pattern;
|
import java.util.regex.Pattern;
|
||||||
import com.kingsrook.qbits.objectstorage.api.ObjectHeaderPolicy;
|
import com.kingsrook.qbits.objectstorage.api.ObjectHeaderPolicy;
|
||||||
import com.kingsrook.qbits.objectstorage.api.ObjectHeaders;
|
import com.kingsrook.qbits.objectstorage.api.ObjectHeaders;
|
||||||
import com.kingsrook.qbits.objectstorage.config.BucketRole;
|
import com.kingsrook.qbits.objectstorage.config.BucketRole;
|
||||||
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
||||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||||
import com.kingsrook.qqq.backend.core.utils.StringUtils;
|
import com.kingsrook.qqq.backend.core.utils.StringUtils;
|
||||||
@@ -44,19 +46,31 @@ public class DefaultObjectHeaderPolicy implements ObjectHeaderPolicy
|
|||||||
{
|
{
|
||||||
private static final QLogger LOG = QLogger.getLogger(DefaultObjectHeaderPolicy.class);
|
private static final QLogger LOG = QLogger.getLogger(DefaultObjectHeaderPolicy.class);
|
||||||
|
|
||||||
|
/** Cluster cache name of the enabled header rules (cleared on every node when a rule changes). */
|
||||||
|
public static final String CACHE_NAME = "qbit-object-storage.header-rules";
|
||||||
|
|
||||||
private static final long CACHE_MILLIS = 15_000L;
|
private static final long CACHE_MILLIS = 15_000L;
|
||||||
|
|
||||||
private static volatile List<QRecord> cachedRules;
|
private static final AtomicReference<CachedRules> CACHED = ClusterCaches.shared(CACHE_NAME, new AtomicReference<>(), DefaultObjectHeaderPolicy::clearCache, null);
|
||||||
private static volatile long cachedAt;
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Drop cached rules (after the rules change).
|
** The enabled rules and when they were read.
|
||||||
|
*******************************************************************************/
|
||||||
|
private record CachedRules(List<QRecord> rules, long readAt)
|
||||||
|
{
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Drop the cached rules on this node (a rule edit clears them on every node
|
||||||
|
** through ConfigRowChangeCustomizer).
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public static void clearCache()
|
public static void clearCache()
|
||||||
{
|
{
|
||||||
cachedRules = null;
|
CACHED.set(null);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -119,11 +133,12 @@ public class DefaultObjectHeaderPolicy implements ObjectHeaderPolicy
|
|||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
private static List<QRecord> rules()
|
private static List<QRecord> rules()
|
||||||
{
|
{
|
||||||
List<QRecord> rules = cachedRules;
|
CachedRules cached = CACHED.get();
|
||||||
if(rules != null && System.currentTimeMillis() - cachedAt < CACHE_MILLIS)
|
if(cached != null && System.currentTimeMillis() - cached.readAt() < CACHE_MILLIS)
|
||||||
{
|
{
|
||||||
return (rules);
|
return (cached.rules());
|
||||||
}
|
}
|
||||||
|
List<QRecord> rules;
|
||||||
if(!Db.available(ObjectStorageTables.HEADER_RULE))
|
if(!Db.available(ObjectStorageTables.HEADER_RULE))
|
||||||
{
|
{
|
||||||
return (List.of());
|
return (List.of());
|
||||||
@@ -133,8 +148,7 @@ public class DefaultObjectHeaderPolicy implements ObjectHeaderPolicy
|
|||||||
rules = Db.asSystem(() -> Db.query(ObjectStorageTables.HEADER_RULE, "enabled", true));
|
rules = Db.asSystem(() -> Db.query(ObjectStorageTables.HEADER_RULE, "enabled", true));
|
||||||
rules.sort(Comparator.comparing((QRecord r) -> r.getValueInteger("priority") == null ? 100 : r.getValueInteger("priority"))
|
rules.sort(Comparator.comparing((QRecord r) -> r.getValueInteger("priority") == null ? 100 : r.getValueInteger("priority"))
|
||||||
.thenComparing(r -> r.getValueInteger("id") == null ? 0 : r.getValueInteger("id")));
|
.thenComparing(r -> r.getValueInteger("id") == null ? 0 : r.getValueInteger("id")));
|
||||||
cachedRules = rules;
|
CACHED.set(new CachedRules(rules, System.currentTimeMillis()));
|
||||||
cachedAt = System.currentTimeMillis();
|
|
||||||
return (rules);
|
return (rules);
|
||||||
}
|
}
|
||||||
catch(Exception e)
|
catch(Exception e)
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ import com.kingsrook.qbits.objectstorage.api.StorageAlertSink;
|
|||||||
import com.kingsrook.qbits.objectstorage.api.StorageChangeListener;
|
import com.kingsrook.qbits.objectstorage.api.StorageChangeListener;
|
||||||
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
||||||
import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
@@ -43,9 +44,11 @@ import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
|||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public final class ObjectStorageRuntime
|
public final class ObjectStorageRuntime
|
||||||
{
|
{
|
||||||
private static final Map<String, ConnectionRuntime> CONNECTIONS = new ConcurrentHashMap<>();
|
private static final Map<String, ConnectionRuntime> CONNECTIONS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
|
||||||
|
"this node's S3 clients, built from the code config at first use; row overrides apply when a client is built, the same on every node");
|
||||||
private static final List<StorageChangeListener> LISTENERS = new CopyOnWriteArrayList<>();
|
private static final List<StorageChangeListener> LISTENERS = new CopyOnWriteArrayList<>();
|
||||||
private static final Map<String, String> PROCESS_PERMISSIONS = new ConcurrentHashMap<>();
|
private static final Map<String, String> PROCESS_PERMISSIONS = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
|
||||||
|
"built from the qbit config at startup, the same on every node");
|
||||||
|
|
||||||
private static volatile ObjectStorageQBitConfig config;
|
private static volatile ObjectStorageQBitConfig config;
|
||||||
private static volatile ObjectHeaderPolicy headerPolicy;
|
private static volatile ObjectHeaderPolicy headerPolicy;
|
||||||
@@ -96,6 +99,7 @@ public final class ObjectStorageRuntime
|
|||||||
clock = Clock.systemUTC();
|
clock = Clock.systemUTC();
|
||||||
indexHook = null;
|
indexHook = null;
|
||||||
CapabilityStore.clearCache();
|
CapabilityStore.clearCache();
|
||||||
|
OperationLog.reset();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -21,10 +21,17 @@
|
|||||||
package com.kingsrook.qbits.objectstorage.internal;
|
package com.kingsrook.qbits.objectstorage.internal;
|
||||||
|
|
||||||
|
|
||||||
|
import java.util.Map;
|
||||||
|
import java.util.Set;
|
||||||
|
import java.util.concurrent.ConcurrentHashMap;
|
||||||
|
import java.util.concurrent.atomic.AtomicLong;
|
||||||
import com.kingsrook.qbits.objectstorage.api.StorageCaller;
|
import com.kingsrook.qbits.objectstorage.api.StorageCaller;
|
||||||
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||||
|
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||||
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||||
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||||
|
import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
|
||||||
import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
|
import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
|
||||||
|
|
||||||
|
|
||||||
@@ -33,12 +40,22 @@ import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
|
|||||||
** every caller (UI action, app code, scheduled job) lands here. Presigned URLs
|
** every caller (UI action, app code, scheduled job) lands here. Presigned URLs
|
||||||
** are logged without the signature (key, method, expiry and issuer only),
|
** are logged without the signature (key, method, expiry and issuer only),
|
||||||
** because the URL itself is a bearer token. A logging failure never fails the
|
** because the URL itself is a bearer token. A logging failure never fails the
|
||||||
** operation it describes.
|
** operation it describes, so it is reported three ways instead: an ERROR log
|
||||||
|
** line, the storage.operation_log.failed alert (StorageAlertSink, resolved by
|
||||||
|
** the next row written for that bucket), and failedWriteCount().
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public final class OperationLog
|
public final class OperationLog
|
||||||
{
|
{
|
||||||
private static final QLogger LOG = QLogger.getLogger(OperationLog.class);
|
private static final QLogger LOG = QLogger.getLogger(OperationLog.class);
|
||||||
|
|
||||||
|
/** The alert raised (severity error, dedupe key "bucket:" + alias) when a row could not be written. */
|
||||||
|
public static final String ALERT_TYPE = "storage.operation_log.failed";
|
||||||
|
|
||||||
|
private static final AtomicLong FAILED_WRITES = new AtomicLong();
|
||||||
|
|
||||||
|
private static final Set<String> UNRESOLVED = ClusterCaches.nodeLocal(ConcurrentHashMap.newKeySet(),
|
||||||
|
"the buckets whose operation log alert this node raised and has not resolved (alert state, not cached data)");
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
@@ -183,6 +200,31 @@ public final class OperationLog
|
|||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** How many operation log rows this node could not write since it started (or
|
||||||
|
** since the runtime was reset). The operations themselves were carried out, so
|
||||||
|
** anything above zero means storage changes that the audit log does not show.
|
||||||
|
** A host that has no alert sink reads this (a metrics gauge, a health check of
|
||||||
|
** its own) to alert on a missing audit row.
|
||||||
|
*******************************************************************************/
|
||||||
|
public static long failedWriteCount()
|
||||||
|
{
|
||||||
|
return (FAILED_WRITES.get());
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Forget the failures counted and the alerts raised (ObjectStorageRuntime.reset).
|
||||||
|
*******************************************************************************/
|
||||||
|
static void reset()
|
||||||
|
{
|
||||||
|
FAILED_WRITES.set(0);
|
||||||
|
UNRESOLVED.clear();
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Insert the row (as the system user), never throwing.
|
** Insert the row (as the system user), never throwing.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
@@ -196,7 +238,7 @@ public final class OperationLog
|
|||||||
try
|
try
|
||||||
{
|
{
|
||||||
Integer bucketId = ConfigSync.bucketId(e.bucketAlias);
|
Integer bucketId = ConfigSync.bucketId(e.bucketAlias);
|
||||||
Db.asSystem(() -> Db.insert(ObjectStorageTables.OPERATION, new QRecord()
|
QRecord written = Db.asSystem(() -> Db.insert(ObjectStorageTables.OPERATION, new QRecord()
|
||||||
.withValue("occurredAt", ObjectStorageRuntime.clock().instant())
|
.withValue("occurredAt", ObjectStorageRuntime.clock().instant())
|
||||||
.withValue("actor", e.caller.describe())
|
.withValue("actor", e.caller.describe())
|
||||||
.withValue("action", e.action)
|
.withValue("action", e.action)
|
||||||
@@ -210,10 +252,66 @@ public final class OperationLog
|
|||||||
.withValue("providerRequestId", e.requestId)
|
.withValue("providerRequestId", e.requestId)
|
||||||
.withValue("processRunId", e.processRunId)
|
.withValue("processRunId", e.processRunId)
|
||||||
.withValue("detail", e.detail)));
|
.withValue("detail", e.detail)));
|
||||||
|
|
||||||
|
////////////////////////////////////////////////////////////////////////////////
|
||||||
|
// the insert action does not throw for a record it refused (a missing value, //
|
||||||
|
// a value too long, a record lock): it hands the record back with its errors //
|
||||||
|
////////////////////////////////////////////////////////////////////////////////
|
||||||
|
if(CollectionUtils.nullSafeHasContents(written.getErrors()))
|
||||||
|
{
|
||||||
|
throw (new QException("The operation log row was refused: " + written.getErrors()));
|
||||||
|
}
|
||||||
|
written(e);
|
||||||
}
|
}
|
||||||
catch(Exception ex)
|
catch(Exception ex)
|
||||||
{
|
{
|
||||||
LOG.warn("Could not write a storage operation log row", ex, logPair("action", e.action), logPair("key", e.key));
|
failed(e, ex);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** A row was written: resolve the alert of an earlier failure for its bucket.
|
||||||
|
*******************************************************************************/
|
||||||
|
private static void written(Entry e)
|
||||||
|
{
|
||||||
|
String dedupeKey = "bucket:" + e.bucketAlias;
|
||||||
|
if(UNRESOLVED.remove(dedupeKey))
|
||||||
|
{
|
||||||
|
try
|
||||||
|
{
|
||||||
|
ObjectStorageRuntime.alertSink().resolve(ALERT_TYPE, dedupeKey);
|
||||||
|
}
|
||||||
|
catch(RuntimeException sinkFailure)
|
||||||
|
{
|
||||||
|
LOG.warn("Could not resolve the storage operation log alert", sinkFailure, logPair("bucket", e.bucketAlias));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** A row could not be written: the operation happened and the audit log does
|
||||||
|
** not show it. Say so where an operator sees it, without failing the operation.
|
||||||
|
*******************************************************************************/
|
||||||
|
private static void failed(Entry e, Exception cause)
|
||||||
|
{
|
||||||
|
long count = FAILED_WRITES.incrementAndGet();
|
||||||
|
LOG.error("Could not write a storage operation log row: the operation is missing from the audit log", cause,
|
||||||
|
logPair("action", e.action), logPair("bucket", e.bucketAlias), logPair("key", e.key), logPair("result", e.result), logPair("failedWriteCount", count));
|
||||||
|
|
||||||
|
String dedupeKey = "bucket:" + e.bucketAlias;
|
||||||
|
UNRESOLVED.add(dedupeKey);
|
||||||
|
try
|
||||||
|
{
|
||||||
|
ObjectStorageRuntime.alertSink().raise(ALERT_TYPE, dedupeKey, "error", "A storage operation on " + e.bucketAlias + " was not recorded in the operation log",
|
||||||
|
Map.of("bucket", String.valueOf(e.bucketAlias), "action", String.valueOf(e.action), "count", count));
|
||||||
|
}
|
||||||
|
catch(RuntimeException sinkFailure)
|
||||||
|
{
|
||||||
|
LOG.warn("Could not raise the storage operation log alert", sinkFailure, logPair("bucket", e.bucketAlias));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -26,6 +26,7 @@ import java.util.Map;
|
|||||||
import com.kingsrook.qbits.objectstorage.api.Capability;
|
import com.kingsrook.qbits.objectstorage.api.Capability;
|
||||||
import com.kingsrook.qbits.objectstorage.api.CapabilityStatus;
|
import com.kingsrook.qbits.objectstorage.api.CapabilityStatus;
|
||||||
import com.kingsrook.qbits.objectstorage.config.ProviderType;
|
import com.kingsrook.qbits.objectstorage.config.ProviderType;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
|
||||||
import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.PARTIAL;
|
import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.PARTIAL;
|
||||||
import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.SUPPORTED;
|
import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.SUPPORTED;
|
||||||
import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.UNKNOWN;
|
import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.UNKNOWN;
|
||||||
@@ -39,7 +40,8 @@ import static com.kingsrook.qbits.objectstorage.api.CapabilityStatus.UNSUPPORTED
|
|||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public final class ProviderDefaults
|
public final class ProviderDefaults
|
||||||
{
|
{
|
||||||
private static final Map<ProviderType, Map<Capability, CapabilityStatus>> TABLE = new EnumMap<>(ProviderType.class);
|
private static final Map<ProviderType, Map<Capability, CapabilityStatus>> TABLE = ClusterCaches.nodeLocal(new EnumMap<>(ProviderType.class),
|
||||||
|
"provider defaults from code, the same on every node");
|
||||||
|
|
||||||
static
|
static
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -21,7 +21,10 @@
|
|||||||
package com.kingsrook.qbits.objectstorage.internal;
|
package com.kingsrook.qbits.objectstorage.internal;
|
||||||
|
|
||||||
|
|
||||||
|
import java.time.Duration;
|
||||||
|
import java.time.Instant;
|
||||||
import java.util.concurrent.Semaphore;
|
import java.util.concurrent.Semaphore;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
@@ -57,17 +60,52 @@ public class RateLimiter
|
|||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public Permit acquire()
|
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
|
try
|
||||||
{
|
{
|
||||||
concurrency.acquire();
|
if(deadline == null)
|
||||||
waitForToken();
|
{
|
||||||
|
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));
|
return (new Permit(concurrency));
|
||||||
}
|
}
|
||||||
catch(InterruptedException e)
|
catch(InterruptedException e)
|
||||||
{
|
{
|
||||||
|
if(acquired)
|
||||||
|
{
|
||||||
|
concurrency.release();
|
||||||
|
}
|
||||||
Thread.currentThread().interrupt();
|
Thread.currentThread().interrupt();
|
||||||
throw (new IllegalStateException("Interrupted while waiting for the storage rate limiter", e));
|
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.
|
** Block until a token is available.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
private void waitForToken() throws InterruptedException
|
private void waitForToken(Instant deadline) throws InterruptedException
|
||||||
{
|
{
|
||||||
while(true)
|
while(true)
|
||||||
{
|
{
|
||||||
|
if(deadline != null && !Instant.now().isBefore(deadline))
|
||||||
|
{
|
||||||
|
throw (new IllegalStateException("Storage probe deadline exceeded"));
|
||||||
|
}
|
||||||
long sleepMillis;
|
long sleepMillis;
|
||||||
synchronized(this)
|
synchronized(this)
|
||||||
{
|
{
|
||||||
@@ -93,7 +135,7 @@ public class RateLimiter
|
|||||||
throttledWaits++;
|
throttledWaits++;
|
||||||
sleepMillis = Math.max(1, (long) ((1.0 - tokens) / opsPerSecond * 1000));
|
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())));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -21,10 +21,6 @@
|
|||||||
package com.kingsrook.qbits.objectstorage.internal;
|
package com.kingsrook.qbits.objectstorage.internal;
|
||||||
|
|
||||||
|
|
||||||
import java.io.IOException;
|
|
||||||
import java.nio.charset.StandardCharsets;
|
|
||||||
import java.nio.file.Files;
|
|
||||||
import java.nio.file.Path;
|
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.regex.Pattern;
|
import java.util.regex.Pattern;
|
||||||
import com.kingsrook.qqq.backend.core.instances.QMetaDataVariableInterpreter;
|
import com.kingsrook.qqq.backend.core.instances.QMetaDataVariableInterpreter;
|
||||||
@@ -32,14 +28,12 @@ import com.kingsrook.qqq.backend.core.utils.StringUtils;
|
|||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Resolves secret references (${env.X}, ${prop.X}, ${file.PATH}) and enforces
|
** Enforces the qbit's secret-reference allow-list and resolves references
|
||||||
** the allow-list. Resolved values live only in memory, never in a record, a log
|
** (${env.X}, ${prop.X}, ${file.PATH}) through core QMetaDataVariableInterpreter.
|
||||||
** line or a process value.
|
** Core reads ${file.PATH} read-only from its allowed roots (default
|
||||||
**
|
** /var/run/secrets and /run/secrets; qqq.fileReference.roots or
|
||||||
** STAND-IN(core=QRun-IO/qqq#1165, removeBy=4.1.0-RC.8)
|
** QQQ_FILE_REFERENCE_ROOTS). Resolved values live only in memory, never in a
|
||||||
** RC.7 QMetaDataVariableInterpreter still supports env/prop, not file references.
|
** record, a log line or a process value.
|
||||||
** ${file.PATH} moves to QMetaDataVariableInterpreter in QQQ core
|
|
||||||
** (spec section 8.2); then this class delegates entirely.
|
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public final class SecretRefs
|
public final class SecretRefs
|
||||||
{
|
{
|
||||||
@@ -93,19 +87,7 @@ public final class SecretRefs
|
|||||||
{
|
{
|
||||||
return (null);
|
return (null);
|
||||||
}
|
}
|
||||||
String trimmed = ref.trim();
|
String value = new QMetaDataVariableInterpreter().interpret(ref.trim());
|
||||||
if(trimmed.startsWith("${file.") && trimmed.endsWith("}"))
|
|
||||||
{
|
|
||||||
try
|
|
||||||
{
|
|
||||||
return (Files.readString(Path.of(trimmed.substring("${file.".length(), trimmed.length() - 1)), StandardCharsets.UTF_8).trim());
|
|
||||||
}
|
|
||||||
catch(IOException e)
|
|
||||||
{
|
|
||||||
return (null);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
String value = new QMetaDataVariableInterpreter().interpret(trimmed);
|
|
||||||
return (value == null || value.startsWith("${") ? null : value);
|
return (value == null || value.startsWith("${") ? null : value);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -25,6 +25,7 @@ import java.util.ArrayList;
|
|||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig;
|
import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig;
|
||||||
|
import com.kingsrook.qbits.objectstorage.internal.ConfigRowChangeCustomizer;
|
||||||
import com.kingsrook.qbits.objectstorage.permissions.ObjectStoragePermissionChecker;
|
import com.kingsrook.qbits.objectstorage.permissions.ObjectStoragePermissionChecker;
|
||||||
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
|
import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
|
||||||
import com.kingsrook.qqq.backend.core.actions.dashboard.widgets.ChildRecordListRenderer;
|
import com.kingsrook.qqq.backend.core.actions.dashboard.widgets.ChildRecordListRenderer;
|
||||||
@@ -564,7 +565,9 @@ public final class ObjectStorageTables
|
|||||||
.withField(ro("orphanedCount", QFieldType.LONG))
|
.withField(ro("orphanedCount", QFieldType.LONG))
|
||||||
.withField(ro("orphanedBytes", QFieldType.LONG))
|
.withField(ro("orphanedBytes", QFieldType.LONG))
|
||||||
.withField(ro("missingCount", QFieldType.LONG))
|
.withField(ro("missingCount", QFieldType.LONG))
|
||||||
.withField(ro("lastReferenceScanDate", QFieldType.DATE_TIME)), KEY_TYPE_BUCKET, "id", false));
|
.withField(ro("lastReferenceScanDate", QFieldType.DATE_TIME)), KEY_TYPE_BUCKET, "id", false)
|
||||||
|
.withCustomizer(TableCustomizers.POST_INSERT_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class))
|
||||||
|
.withCustomizer(TableCustomizers.POST_UPDATE_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class)));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -701,7 +704,10 @@ public final class ObjectStorageTables
|
|||||||
.withField(str("cacheControl", 300))
|
.withField(str("cacheControl", 300))
|
||||||
.withField(str("contentDisposition", 300))
|
.withField(str("contentDisposition", 300))
|
||||||
.withField(new QFieldMetaData("priority", QFieldType.INTEGER).withDefaultValue(100))
|
.withField(new QFieldMetaData("priority", QFieldType.INTEGER).withDefaultValue(100))
|
||||||
.withField(new QFieldMetaData("enabled", QFieldType.BOOLEAN).withDefaultValue(true)), KEY_TYPE_BUCKET, "bucketId", true));
|
.withField(new QFieldMetaData("enabled", QFieldType.BOOLEAN).withDefaultValue(true)), KEY_TYPE_BUCKET, "bucketId", true)
|
||||||
|
.withCustomizer(TableCustomizers.POST_INSERT_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class))
|
||||||
|
.withCustomizer(TableCustomizers.POST_UPDATE_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class))
|
||||||
|
.withCustomizer(TableCustomizers.POST_DELETE_RECORD, new QCodeReference(ConfigRowChangeCustomizer.class)));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+9
-1
@@ -46,7 +46,9 @@ import com.kingsrook.qqq.backend.core.model.session.QSession;
|
|||||||
** - writes to the editable config tables (connection operational fields, header
|
** - writes to the editable config tables (connection operational fields, header
|
||||||
** rules, protection rules, bucket thresholds) need object_storage.admin;
|
** rules, protection rules, bucket thresholds) need object_storage.admin;
|
||||||
** observed and history tables are never written through the UI or API (they
|
** observed and history tables are never written through the UI or API (they
|
||||||
** change only through actions and the service);
|
** change only through actions and the service, whose own writes carry
|
||||||
|
** core's per-call permission bypass, see internal.Db; no session is let
|
||||||
|
** through, the core system user included);
|
||||||
** - each process needs the permission registered for it (default admin).
|
** - each process needs the permission registered for it (default admin).
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
public class ObjectStoragePermissionChecker implements CustomPermissionChecker
|
public class ObjectStoragePermissionChecker implements CustomPermissionChecker
|
||||||
@@ -65,6 +67,12 @@ public class ObjectStoragePermissionChecker implements CustomPermissionChecker
|
|||||||
String needed = required(actionInput, metaData);
|
String needed = required(actionInput, metaData);
|
||||||
if(needed == null)
|
if(needed == null)
|
||||||
{
|
{
|
||||||
|
////////////////////////////////////////////////////////////////////////////////////////
|
||||||
|
// nobody writes these tables directly, whatever the session: a person, the storage //
|
||||||
|
// administrator and the core system user are all refused. the service's own writes //
|
||||||
|
// (internal.Db) carry core's per-call permission bypass, so core does not ask about //
|
||||||
|
// them; a write that reaches this point did not come from the service //
|
||||||
|
////////////////////////////////////////////////////////////////////////////////////////
|
||||||
throw (new QPermissionDeniedException("This storage table changes only through storage actions."));
|
throw (new QPermissionDeniedException("This storage table changes only through storage actions."));
|
||||||
}
|
}
|
||||||
if(!has(needed))
|
if(!has(needed))
|
||||||
|
|||||||
+1
-1
@@ -227,7 +227,7 @@ public final class TestConnectionProcess
|
|||||||
}
|
}
|
||||||
if(report.getClockSkewSeconds() != null)
|
if(report.getClockSkewSeconds() != null)
|
||||||
{
|
{
|
||||||
boolean skewed = report.getClockSkewSeconds() > 900;
|
boolean skewed = report.getClockSkewSeconds() > CapabilityProbe.CLOCK_SKEW_THRESHOLD_SECONDS;
|
||||||
ok = ok && !skewed;
|
ok = ok && !skewed;
|
||||||
lines.add(new ProcessSummaryLine(skewed ? Status.ERROR : Status.INFO, "Clock skew against the store: " + report.getClockSkewSeconds() + " seconds"
|
lines.add(new ProcessSummaryLine(skewed ? Status.ERROR : Status.INFO, "Clock skew against the store: " + report.getClockSkewSeconds() + " seconds"
|
||||||
+ (skewed ? " (SigV4 fails beyond 15 minutes)" : "")));
|
+ (skewed ? " (SigV4 fails beyond 15 minutes)" : "")));
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
bundle: qbit-object-storage
|
bundle: qbit-object-storage
|
||||||
version: 0.1.0-SNAPSHOT
|
version: ${project.version}
|
||||||
layer: QBIT
|
layer: QBIT
|
||||||
entries:
|
entries:
|
||||||
- key: app:objectStorageAdmin;slot:help
|
- key: app:objectStorageAdmin;slot:help
|
||||||
|
|||||||
@@ -0,0 +1 @@
|
|||||||
|
version=${project.version}
|
||||||
@@ -59,7 +59,13 @@ class ObjectStorageHelpBundleTest
|
|||||||
|
|
||||||
assertThat(instance.getTable("helpContent")).isNull();
|
assertThat(instance.getTable("helpContent")).isNull();
|
||||||
assertThat(HelpBundles.of(instance).getBundles()).singleElement()
|
assertThat(HelpBundles.of(instance).getBundles()).singleElement()
|
||||||
.satisfies(bundle -> assertThat(bundle.getId()).isEqualTo("qbit-object-storage"));
|
.satisfies(bundle ->
|
||||||
|
{
|
||||||
|
assertThat(bundle.getId()).isEqualTo("qbit-object-storage");
|
||||||
|
assertThat(System.getProperty("qbit.artifactVersion")).isNotBlank();
|
||||||
|
assertThat(bundle.getVersion()).isEqualTo(System.getProperty("qbit.artifactVersion"))
|
||||||
|
.isEqualTo(ObjectStorageQBitProducer.VERSION);
|
||||||
|
});
|
||||||
HelpBundles.syncOnStartup(instance);
|
HelpBundles.syncOnStartup(instance);
|
||||||
Set<String> keys = HelpBundles.of(instance).getBundles().get(0).getEntries().stream()
|
Set<String> keys = HelpBundles.of(instance).getBundles().get(0).getEntries().stream()
|
||||||
.map(entry -> entry.getKey()).collect(Collectors.toSet());
|
.map(entry -> entry.getKey()).collect(Collectors.toSet());
|
||||||
|
|||||||
@@ -44,6 +44,10 @@ class ObjectStorageQBitProducerTest
|
|||||||
ObjectStorageQBitConfig config = new ObjectStorageQBitConfig().withBackendName("rdbms");
|
ObjectStorageQBitConfig config = new ObjectStorageQBitConfig().withBackendName("rdbms");
|
||||||
ObjectStorageQBitProducer producer = new ObjectStorageQBitProducer().withQBitConfig(config).withNamespace("object_storage");
|
ObjectStorageQBitProducer producer = new ObjectStorageQBitProducer().withQBitConfig(config).withNamespace("object_storage");
|
||||||
|
|
||||||
|
String artifactVersion = System.getProperty("qbit.artifactVersion");
|
||||||
|
assertThat(artifactVersion).isNotBlank();
|
||||||
|
assertThat(ObjectStorageQBitProducer.VERSION).isEqualTo(artifactVersion);
|
||||||
|
assertThat(producer.getQBitMetaData().getVersion()).isEqualTo(artifactVersion);
|
||||||
assertThat(producer.isEnabled()).isTrue();
|
assertThat(producer.isEnabled()).isTrue();
|
||||||
assertThat(producer.getQBitConfig()).isSameAs(config);
|
assertThat(producer.getQBitConfig()).isSameAs(config);
|
||||||
assertThat(producer.getNamespace()).isEqualTo("object_storage");
|
assertThat(producer.getNamespace()).isEqualTo("object_storage");
|
||||||
|
|||||||
+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");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -574,4 +574,40 @@ class MinioIndexTest
|
|||||||
.contains("Usage trend").contains("collecting since")
|
.contains("Usage trend").contains("collecting since")
|
||||||
.contains("Connections").contains("Never tested");
|
.contains("Connections").contains("Never tested");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Every link the home emits is a /app/<table>[...] route that the Next frontend
|
||||||
|
** resolves, never an /<appName>/<table> path, with data and without.
|
||||||
|
*******************************************************************************/
|
||||||
|
@Test
|
||||||
|
void everyHomeLinkUsesTheAppRoute() throws Exception
|
||||||
|
{
|
||||||
|
String prefix = com.kingsrook.qbits.objectstorage.app.StorageWidgets.tablePathPrefix();
|
||||||
|
assertThat(prefix).isEqualTo("/app");
|
||||||
|
java.util.List<String> hrefs = new java.util.ArrayList<>(hrefs(com.kingsrook.qbits.objectstorage.app.StorageHome.render(prefix, Instant.now())));
|
||||||
|
|
||||||
|
rawPut(target.indexBucket, "media/1/a.jpg", 2048);
|
||||||
|
rawPut(target.indexBucket, "media/9/orphan.jpg", 1024);
|
||||||
|
ObjectIndex.scan("indexed");
|
||||||
|
ObjectIndex.headBacklog("indexed", 10);
|
||||||
|
ReferenceScan.run("indexed");
|
||||||
|
ObjectStorage.bucket("indexed").put(PutRequest.of("media/1/c.jpg", new byte[3]));
|
||||||
|
hrefs.addAll(hrefs(com.kingsrook.qbits.objectstorage.app.StorageHome.render(prefix, Instant.now())));
|
||||||
|
|
||||||
|
assertThat(hrefs).isNotEmpty().allSatisfy(href -> assertThat(href).startsWith("/app/"));
|
||||||
|
assertThat(hrefs).noneMatch(href -> href.contains("objectStorageAdmin"));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
private static java.util.List<String> hrefs(String html)
|
||||||
|
{
|
||||||
|
java.util.List<String> hrefs = new java.util.ArrayList<>();
|
||||||
|
java.util.regex.Matcher m = java.util.regex.Pattern.compile("href=\"([^\"]*)\"").matcher(html);
|
||||||
|
while(m.find())
|
||||||
|
{
|
||||||
|
hrefs.add(m.group(1).replace("&", "&"));
|
||||||
|
}
|
||||||
|
return (hrefs);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
+17
-3
@@ -38,6 +38,7 @@ import com.kingsrook.qqq.backend.module.filesystem.s3.model.metadata.S3BackendMe
|
|||||||
import com.kingsrook.qqq.backend.module.filesystem.s3.model.metadata.S3ChecksumMode;
|
import com.kingsrook.qqq.backend.module.filesystem.s3.model.metadata.S3ChecksumMode;
|
||||||
import com.sun.net.httpserver.Headers;
|
import com.sun.net.httpserver.Headers;
|
||||||
import com.sun.net.httpserver.HttpServer;
|
import com.sun.net.httpserver.HttpServer;
|
||||||
|
import org.junit.jupiter.api.AfterEach;
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
import org.junit.jupiter.api.io.TempDir;
|
import org.junit.jupiter.api.io.TempDir;
|
||||||
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
|
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
|
||||||
@@ -58,9 +59,21 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
|||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
class ConnectionRuntimeTest
|
class ConnectionRuntimeTest
|
||||||
{
|
{
|
||||||
|
/*******************************************************************************
|
||||||
|
** Core reads ${file.PATH} only inside its allowed roots; reference() points
|
||||||
|
** them at each test's temporary directory.
|
||||||
|
*******************************************************************************/
|
||||||
|
@AfterEach
|
||||||
|
void afterEach()
|
||||||
|
{
|
||||||
|
System.clearProperty("qqq.fileReference.roots");
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Every core-supported setting is mapped after resolving references. Session
|
** Every core-supported setting is mapped after resolving references. Session
|
||||||
** tokens and presign settings have no core fields and are exercised below.
|
** tokens and presign settings are exercised below.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
@Test
|
@Test
|
||||||
void testBackendMetaDataCarriesEverySetting(@TempDir Path dir) throws Exception
|
void testBackendMetaDataCarriesEverySetting(@TempDir Path dir) throws Exception
|
||||||
@@ -199,8 +212,8 @@ class ConnectionRuntimeTest
|
|||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Session credentials sign streaming reads/writes and paginator requests.
|
** Session credentials (core sessionToken) sign streaming reads/writes and
|
||||||
** Existing request headers survive the bridge's credentials override.
|
** paginator requests; request overrides are untouched.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
@Test
|
@Test
|
||||||
void testSessionSigningAndRequestOverrides(@TempDir Path dir) throws Exception
|
void testSessionSigningAndRequestOverrides(@TempDir Path dir) throws Exception
|
||||||
@@ -313,6 +326,7 @@ class ConnectionRuntimeTest
|
|||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
private static String reference(Path dir, String name, String value) throws Exception
|
private static String reference(Path dir, String name, String value) throws Exception
|
||||||
{
|
{
|
||||||
|
System.setProperty("qqq.fileReference.roots", dir.toString());
|
||||||
Path file = dir.resolve(name);
|
Path file = dir.resolve(name);
|
||||||
Files.writeString(file, value);
|
Files.writeString(file, value);
|
||||||
return ("${file." + file + "}");
|
return ("${file." + file + "}");
|
||||||
|
|||||||
@@ -21,25 +21,50 @@
|
|||||||
package com.kingsrook.qbits.objectstorage.internal;
|
package com.kingsrook.qbits.objectstorage.internal;
|
||||||
|
|
||||||
|
|
||||||
|
import java.io.Serializable;
|
||||||
|
import java.util.List;
|
||||||
import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig;
|
import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig;
|
||||||
|
import com.kingsrook.qbits.objectstorage.api.StorageCaller;
|
||||||
|
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||||
|
import com.kingsrook.qbits.objectstorage.permissions.ObjectStoragePermissions;
|
||||||
import com.kingsrook.qbits.objectstorage.testsupport.StorageTestSupport;
|
import com.kingsrook.qbits.objectstorage.testsupport.StorageTestSupport;
|
||||||
|
import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
|
||||||
|
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
|
||||||
|
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
|
||||||
import com.kingsrook.qqq.backend.core.context.QContext;
|
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||||
|
import com.kingsrook.qqq.backend.core.exceptions.QException;
|
||||||
|
import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
||||||
import com.kingsrook.qqq.backend.core.model.metadata.security.QSecurityKeyType;
|
import com.kingsrook.qqq.backend.core.model.metadata.security.QSecurityKeyType;
|
||||||
import com.kingsrook.qqq.backend.core.model.session.QSession;
|
import com.kingsrook.qqq.backend.core.model.session.QSession;
|
||||||
import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
|
import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.session.QUser;
|
||||||
|
import org.assertj.core.api.ThrowableAssert.ThrowingCallable;
|
||||||
import org.junit.jupiter.api.AfterEach;
|
import org.junit.jupiter.api.AfterEach;
|
||||||
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.Test;
|
||||||
import static org.assertj.core.api.Assertions.assertThat;
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
|
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Db.asSystem runs work as the core system user. Token-based authentication
|
** Db.asSystem runs work as the core system user. Token-based authentication
|
||||||
** modules (OAuth2, Auth0) accept a QSystemUserSession without a token and
|
** modules (OAuth2, Auth0) accept a QSystemUserSession without a token and
|
||||||
** reject any other tokenless session, so the class matters, not just the
|
** reject any other tokenless session, so the class matters, not just the
|
||||||
** permissions.
|
** permissions. The system session is not what lets the service write its own
|
||||||
|
** tables, though: each Db write carries core's per-call permission bypass, and
|
||||||
|
** a write that does not is refused to every session.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
class DbSystemSessionTest
|
class DbSystemSessionTest
|
||||||
{
|
{
|
||||||
|
private static final String REFUSAL = "This storage table changes only through storage actions.";
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
/*******************************************************************************
|
/*******************************************************************************
|
||||||
** Reset the runtime and context after each test.
|
** Reset the runtime and context after each test.
|
||||||
*******************************************************************************/
|
*******************************************************************************/
|
||||||
@@ -69,4 +94,140 @@ class DbSystemSessionTest
|
|||||||
assertThat(inside.hasSecurityKeyValue("tenantAllAccess", true)).isTrue();
|
assertThat(inside.hasSecurityKeyValue("tenantAllAccess", true)).isTrue();
|
||||||
assertThat(QContext.getQSession()).isSameAs(caller);
|
assertThat(QContext.getQSession()).isSameAs(caller);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** A host that enforces permissions inside the core actions
|
||||||
|
** (QInstance.enforcePermissionsInCoreActions) still gets its operation log
|
||||||
|
** row, and the row names the person who acted: the caller is captured before
|
||||||
|
** the write moves to the system user. The other tables the service keeps
|
||||||
|
** (here the object index) are written, changed and removed the same way.
|
||||||
|
*******************************************************************************/
|
||||||
|
@Test
|
||||||
|
void testServiceWritesItsOwnTablesWhenCoreActionsEnforcePermissions() throws Exception
|
||||||
|
{
|
||||||
|
enforcingInstance();
|
||||||
|
|
||||||
|
QContext.setQSession(new QSession().withUser(new QUser().withIdReference("pat@example.com").withFullName("Pat Example")).withPermissions(ObjectStoragePermissions.WRITE));
|
||||||
|
new OperationLog.Entry(StorageCaller.user(), "PUT", "files").key("stored/by/the/service").write();
|
||||||
|
|
||||||
|
assertThat(operations()).as("the operation log row of a stored object").singleElement().satisfies(row ->
|
||||||
|
{
|
||||||
|
assertThat(row.getValueString("objectKey")).isEqualTo("stored/by/the/service");
|
||||||
|
assertThat(row.getValueString("actor")).as("the person who acted, not the system user that wrote the row").isEqualTo("pat@example.com");
|
||||||
|
});
|
||||||
|
assertThat(OperationLog.failedWriteCount()).isZero();
|
||||||
|
|
||||||
|
Serializable id = Db.asSystem(() -> Db.insert(ObjectStorageTables.OBJECT, new QRecord().withValue("objectKey", "a/b.txt").withValue("size", 1L))).getValue("id");
|
||||||
|
Db.asSystem(() -> Db.update(ObjectStorageTables.OBJECT, List.of(new QRecord().withValue("id", id).withValue("size", 2L))));
|
||||||
|
assertThat(Db.asSystem(() -> Db.first(ObjectStorageTables.OBJECT, "id", id)).getValueLong("size")).isEqualTo(2L);
|
||||||
|
Db.asSystem(() ->
|
||||||
|
{
|
||||||
|
Db.delete(ObjectStorageTables.OBJECT, List.of(id));
|
||||||
|
return (null);
|
||||||
|
});
|
||||||
|
assertThat(Db.asSystem(() -> Db.query(ObjectStorageTables.OBJECT))).isEmpty();
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Being the system user is not enough to write the operation log. Server code
|
||||||
|
** outside the service's Db (a host, another qbit) that runs as the core system
|
||||||
|
** user is refused the insert, the update and the delete, also when it borrows
|
||||||
|
** Db.asSystem for its session.
|
||||||
|
*******************************************************************************/
|
||||||
|
@Test
|
||||||
|
void testSystemSessionOutsideTheServiceCannotWriteTheOperationLog() throws Exception
|
||||||
|
{
|
||||||
|
QInstance instance = enforcingInstance();
|
||||||
|
new OperationLog.Entry(StorageCaller.user(), "PUT", "files").key("the/audited/key").write();
|
||||||
|
Serializable id = operations().get(0).getValue("id");
|
||||||
|
|
||||||
|
QContext.setQSession(new QSystemUserSession(instance));
|
||||||
|
assertRefused("insert as the system user", () -> new InsertAction().execute(new InsertInput(ObjectStorageTables.OPERATION).withRecords(List.of(new QRecord().withValue("action", "PUT")))));
|
||||||
|
assertRefused("update as the system user", () -> new UpdateAction().execute(new UpdateInput(ObjectStorageTables.OPERATION).withRecords(List.of(new QRecord().withValue("id", id).withValue("actor", "nobody")))));
|
||||||
|
assertRefused("delete as the system user", () -> new DeleteAction().execute(new DeleteInput(ObjectStorageTables.OPERATION).withPrimaryKey(id)));
|
||||||
|
|
||||||
|
QContext.setQSession(new QSession());
|
||||||
|
assertRefused("insert inside Db.asSystem, not through Db", () -> Db.asSystem(() -> new InsertAction().execute(new InsertInput(ObjectStorageTables.OPERATION).withRecords(List.of(new QRecord().withValue("action", "PUT"))))));
|
||||||
|
assertRefused("update inside Db.asSystem, not through Db", () -> Db.asSystem(() -> new UpdateAction().execute(new UpdateInput(ObjectStorageTables.OPERATION).withRecords(List.of(new QRecord().withValue("id", id).withValue("actor", "nobody"))))));
|
||||||
|
assertRefused("delete inside Db.asSystem, not through Db", () -> Db.asSystem(() -> new DeleteAction().execute(new DeleteInput(ObjectStorageTables.OPERATION).withPrimaryKey(id))));
|
||||||
|
|
||||||
|
assertUnchanged();
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** An operation row is not updated or deleted by an update or delete action
|
||||||
|
** that does not carry the bypass, whoever asks: a person, the storage
|
||||||
|
** administrator or the system user, by key or by filter.
|
||||||
|
*******************************************************************************/
|
||||||
|
@Test
|
||||||
|
void testOperationRowIsNeverUpdatedOrDeletedWithoutTheBypass() throws Exception
|
||||||
|
{
|
||||||
|
QInstance instance = enforcingInstance();
|
||||||
|
new OperationLog.Entry(StorageCaller.user(), "PUT", "files").key("the/audited/key").write();
|
||||||
|
Serializable id = operations().get(0).getValue("id");
|
||||||
|
|
||||||
|
List<QSession> sessions = List.of(new QSession(), new QSession().withPermissions(ObjectStoragePermissions.ADMIN), new QSystemUserSession(instance));
|
||||||
|
for(QSession session : sessions)
|
||||||
|
{
|
||||||
|
QContext.setQSession(session);
|
||||||
|
String who = session.getClass().getSimpleName() + (session.hasPermission(ObjectStoragePermissions.ADMIN) ? " (admin)" : "");
|
||||||
|
assertRefused("update by " + who, () -> new UpdateAction().execute(new UpdateInput(ObjectStorageTables.OPERATION).withRecords(List.of(new QRecord().withValue("id", id).withValue("result", "FAILED")))));
|
||||||
|
assertRefused("delete by key by " + who, () -> new DeleteAction().execute(new DeleteInput(ObjectStorageTables.OPERATION).withPrimaryKey(id)));
|
||||||
|
assertRefused("delete by filter by " + who, () -> new DeleteAction().execute(new DeleteInput(ObjectStorageTables.OPERATION).withQueryFilter(new QQueryFilter(Db.eq("action", "PUT")))));
|
||||||
|
}
|
||||||
|
|
||||||
|
assertUnchanged();
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** An instance that enforces permissions inside the core actions.
|
||||||
|
*******************************************************************************/
|
||||||
|
private static QInstance enforcingInstance() throws QException
|
||||||
|
{
|
||||||
|
ObjectStorageQBitConfig config = new ObjectStorageQBitConfig().withBackendName(StorageTestSupport.MEMORY).withBucketScoping(false);
|
||||||
|
return (StorageTestSupport.setUp(config, instance -> instance.setEnforcePermissionsInCoreActions(true)));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** The operation log, read by the service.
|
||||||
|
*******************************************************************************/
|
||||||
|
private static List<QRecord> operations() throws QException
|
||||||
|
{
|
||||||
|
return (Db.asSystem(() -> Db.query(ObjectStorageTables.OPERATION)));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** The qbit's checker refused the write.
|
||||||
|
*******************************************************************************/
|
||||||
|
private static void assertRefused(String what, ThrowingCallable write)
|
||||||
|
{
|
||||||
|
assertThatThrownBy(write).as(what).isInstanceOf(QPermissionDeniedException.class).hasMessage(REFUSAL);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** The one row the service wrote is still there, as it was written.
|
||||||
|
*******************************************************************************/
|
||||||
|
private static void assertUnchanged() throws QException
|
||||||
|
{
|
||||||
|
assertThat(operations()).singleElement().satisfies(row ->
|
||||||
|
{
|
||||||
|
assertThat(row.getValueString("objectKey")).isEqualTo("the/audited/key");
|
||||||
|
assertThat(row.getValueString("result")).isEqualTo("OK");
|
||||||
|
assertThat(row.getValueString("actor")).isEqualTo("system (no user)");
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
@@ -83,10 +83,23 @@ class InternalsTest
|
|||||||
assertThat(SecretRefs.resolve(null)).isNull();
|
assertThat(SecretRefs.resolve(null)).isNull();
|
||||||
assertThat(SecretRefs.resolveSetting("")).isNull();
|
assertThat(SecretRefs.resolveSetting("")).isNull();
|
||||||
|
|
||||||
|
//////////////////////////////////////////////////////////////////////
|
||||||
|
// core reads ${file.PATH} only inside its allowed roots (by default //
|
||||||
|
// /var/run/secrets and /run/secrets); point them at the temp dir. //
|
||||||
|
//////////////////////////////////////////////////////////////////////
|
||||||
Path file = dir.resolve("key");
|
Path file = dir.resolve("key");
|
||||||
Files.writeString(file, " from-file\n");
|
Files.writeString(file, " from-file\n");
|
||||||
assertThat(SecretRefs.resolve("${file." + file + "}")).isEqualTo("from-file");
|
assertThat(SecretRefs.resolve("${file." + file + "}")).as("outside the allowed roots").isNull();
|
||||||
assertThat(SecretRefs.resolve("${file." + dir.resolve("missing") + "}")).isNull();
|
System.setProperty("qqq.fileReference.roots", dir.toString());
|
||||||
|
try
|
||||||
|
{
|
||||||
|
assertThat(SecretRefs.resolve("${file." + file + "}")).isEqualTo("from-file");
|
||||||
|
assertThat(SecretRefs.resolve("${file." + dir.resolve("missing") + "}")).isNull();
|
||||||
|
}
|
||||||
|
finally
|
||||||
|
{
|
||||||
|
System.clearProperty("qqq.fileReference.roots");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
+240
@@ -0,0 +1,240 @@
|
|||||||
|
/*
|
||||||
|
* QQQ - Low-code Application Framework for Engineers.
|
||||||
|
* Copyright (C) 2021-2026. 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.internal;
|
||||||
|
|
||||||
|
|
||||||
|
import java.util.EnumSet;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.concurrent.CopyOnWriteArrayList;
|
||||||
|
import java.util.function.Consumer;
|
||||||
|
import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig;
|
||||||
|
import com.kingsrook.qbits.objectstorage.api.Capability;
|
||||||
|
import com.kingsrook.qbits.objectstorage.api.CapabilityStatus;
|
||||||
|
import com.kingsrook.qbits.objectstorage.config.BucketRole;
|
||||||
|
import com.kingsrook.qbits.objectstorage.config.ProviderType;
|
||||||
|
import com.kingsrook.qbits.objectstorage.config.StorageBucketConfig;
|
||||||
|
import com.kingsrook.qbits.objectstorage.config.StorageConnectionConfig;
|
||||||
|
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||||
|
import com.kingsrook.qbits.objectstorage.testsupport.StorageTestSupport;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterBus;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterBusInterface;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterBusMetaData;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterCacheAudit;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.ClusterEvent;
|
||||||
|
import com.kingsrook.qqq.backend.core.cluster.QNodeRoles;
|
||||||
|
import com.kingsrook.qqq.backend.core.context.QContext;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.data.QRecord;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
|
||||||
|
import org.junit.jupiter.api.AfterEach;
|
||||||
|
import org.junit.jupiter.api.BeforeEach;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** The Object Storage caches are correct on several nodes: every static cache
|
||||||
|
** is registered with ClusterCaches, a probe result recorded on one node clears
|
||||||
|
** that capability on the others, and an edited bucket row or header rule
|
||||||
|
** clears its cache on every node.
|
||||||
|
*******************************************************************************/
|
||||||
|
class ObjectStorageClusterCachesTest
|
||||||
|
{
|
||||||
|
/** the "wire": events every node's transport sees */
|
||||||
|
static final List<Consumer<ClusterEvent>> WIRE = new CopyOnWriteArrayList<>();
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** A transport over the static WIRE, so a "remote" publish reaches us.
|
||||||
|
*******************************************************************************/
|
||||||
|
public static class WireBus implements ClusterBusInterface
|
||||||
|
{
|
||||||
|
/*******************************************************************************
|
||||||
|
**
|
||||||
|
*******************************************************************************/
|
||||||
|
@Override
|
||||||
|
public void publish(ClusterEvent event)
|
||||||
|
{
|
||||||
|
WIRE.forEach(listener -> listener.accept(event));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
**
|
||||||
|
*******************************************************************************/
|
||||||
|
@Override
|
||||||
|
public AutoCloseable subscribe(Consumer<ClusterEvent> listener)
|
||||||
|
{
|
||||||
|
WIRE.add(listener);
|
||||||
|
return (() -> WIRE.remove(listener));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** A synced instance whose buckets never touch a real store, on the wire as
|
||||||
|
** node-a.
|
||||||
|
*******************************************************************************/
|
||||||
|
@BeforeEach
|
||||||
|
void setUp() throws Exception
|
||||||
|
{
|
||||||
|
System.setProperty("os.unit.ak", "ak");
|
||||||
|
System.setProperty("os.unit.sk", "sk");
|
||||||
|
StorageTestSupport.setUp(new ObjectStorageQBitConfig().withBackendName(StorageTestSupport.MEMORY).withBucketScoping(false).withOpenWhenAnonymous(true).withAdminApp(true)
|
||||||
|
.withAllowedSecretRefPatterns(List.of("prop.os.unit.*"))
|
||||||
|
.withConnection(new StorageConnectionConfig("main").withProviderType(ProviderType.MINIO).withEndpoint("http://localhost:1").withRegion("us-east-1")
|
||||||
|
.withAccessKeyRef("${prop.os.unit.ak}").withSecretKeyRef("${prop.os.unit.sk}"))
|
||||||
|
.withBucket(new StorageBucketConfig("media").withConnection("main").withBucketName("media-bucket").withRole(BucketRole.PRIVATE).withIndexed(false)),
|
||||||
|
instance -> instance.withSupplementalMetaData(new ClusterBusMetaData().withBus(new QCodeReference(WireBus.class))));
|
||||||
|
QNodeRoles.set(new QNodeRoles.Settings(EnumSet.allOf(QNodeRoles.Role.class), true, true, true, null, "node-a"));
|
||||||
|
ClusterBus.connect(QContext.getQInstance());
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
**
|
||||||
|
*******************************************************************************/
|
||||||
|
@AfterEach
|
||||||
|
void tearDown()
|
||||||
|
{
|
||||||
|
ClusterBus.close(QContext.getQInstance());
|
||||||
|
WIRE.clear();
|
||||||
|
QNodeRoles.set(null);
|
||||||
|
StorageTestSupport.tearDown();
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Every static Memoization in the QBit is registered (shared or node-local).
|
||||||
|
*******************************************************************************/
|
||||||
|
@Test
|
||||||
|
void testNoUnregisteredStaticCaches() throws Exception
|
||||||
|
{
|
||||||
|
assertThat(ClusterCacheAudit.unregisteredStaticCaches("com.kingsrook.qbits.objectstorage")).isEmpty();
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Another node's probe result clears the capability here; the next read
|
||||||
|
** reloads it from the capability row.
|
||||||
|
*******************************************************************************/
|
||||||
|
@Test
|
||||||
|
void testRemoteProbeResultClearsCapability() throws Exception
|
||||||
|
{
|
||||||
|
StorageBucketConfig bucket = ObjectStorageRuntime.bucketConfig("media");
|
||||||
|
CapabilityStore.record(bucket, Capability.VERSIONING, CapabilityStatus.UNSUPPORTED, "probe here");
|
||||||
|
assertThat(CapabilityStore.isCached(bucket, Capability.VERSIONING)).isTrue();
|
||||||
|
|
||||||
|
////////////////////////////////////////////////////////////
|
||||||
|
// node-b probes again and writes SUPPORTED to the row //
|
||||||
|
////////////////////////////////////////////////////////////
|
||||||
|
Integer bucketId = ConfigSync.bucketId("media");
|
||||||
|
Db.asSystem(() ->
|
||||||
|
{
|
||||||
|
QRecord row = Db.first(ObjectStorageTables.BUCKET_CAPABILITY, "bucketId", bucketId, "capability", Capability.VERSIONING.name());
|
||||||
|
Db.update(ObjectStorageTables.BUCKET_CAPABILITY, List.of(new QRecord().withValue("id", row.getValue("id")).withValue("status", CapabilityStatus.SUPPORTED.name())));
|
||||||
|
return (null);
|
||||||
|
});
|
||||||
|
ClusterEvent remote = new ClusterEvent(ClusterEvent.TYPE_MEMO_INVALIDATE, CapabilityStore.CACHE_NAME, "b:media:" + Capability.VERSIONING);
|
||||||
|
remote.setOriginNodeId("node-b");
|
||||||
|
WIRE.forEach(listener -> listener.accept(remote));
|
||||||
|
|
||||||
|
assertThat(CapabilityStore.isCached(bucket, Capability.VERSIONING)).isFalse();
|
||||||
|
assertThat(CapabilityStore.status("media", Capability.VERSIONING)).isEqualTo(CapabilityStatus.SUPPORTED);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** A probe result recorded here tells the other nodes, and stays cached here.
|
||||||
|
*******************************************************************************/
|
||||||
|
@Test
|
||||||
|
void testProbeResultPublishesInvalidation()
|
||||||
|
{
|
||||||
|
List<ClusterEvent> sent = sentEvents();
|
||||||
|
StorageBucketConfig bucket = ObjectStorageRuntime.bucketConfig("media");
|
||||||
|
CapabilityStore.record(bucket, Capability.VERSIONING, CapabilityStatus.SUPPORTED, "probe here");
|
||||||
|
|
||||||
|
assertThat(CapabilityStore.status("media", Capability.VERSIONING)).isEqualTo(CapabilityStatus.SUPPORTED);
|
||||||
|
assertThat(sent).anySatisfy(event -> assertInvalidation(event, CapabilityStore.CACHE_NAME, "b:media:VERSIONING"));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** An edited bucket row takes effect at once and tells the other nodes.
|
||||||
|
*******************************************************************************/
|
||||||
|
@Test
|
||||||
|
void testBucketRowEditInvalidatesRowLimitsEverywhere() throws Exception
|
||||||
|
{
|
||||||
|
StorageBucketConfig bucket = ObjectStorageRuntime.bucketConfig("media");
|
||||||
|
assertThat(ConfigSync.effectiveMaxPresignGetSeconds(bucket)).isEqualTo(3600);
|
||||||
|
|
||||||
|
List<ClusterEvent> sent = sentEvents();
|
||||||
|
Db.update(ObjectStorageTables.BUCKET, List.of(new QRecord().withValue("id", ConfigSync.bucketId("media")).withValue("maxPresignGetSeconds", 60)));
|
||||||
|
|
||||||
|
assertThat(ConfigSync.effectiveMaxPresignGetSeconds(bucket)).as("no wait for the 15 s cache").isEqualTo(60);
|
||||||
|
assertThat(sent).anySatisfy(event -> assertInvalidation(event, ConfigSync.ROW_LIMITS_CACHE_NAME, null));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** An edited header rule tells the other nodes.
|
||||||
|
*******************************************************************************/
|
||||||
|
@Test
|
||||||
|
void testHeaderRuleEditInvalidatesRulesEverywhere() throws Exception
|
||||||
|
{
|
||||||
|
List<ClusterEvent> sent = sentEvents();
|
||||||
|
StorageTestSupport.insert(ObjectStorageTables.HEADER_RULE, new QRecord().withValue("name", "all").withValue("keyGlob", "**").withValue("cacheControl", "no-store"));
|
||||||
|
assertThat(sent).anySatisfy(event -> assertInvalidation(event, DefaultObjectHeaderPolicy.CACHE_NAME, null));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Collect what this node puts on the wire.
|
||||||
|
*******************************************************************************/
|
||||||
|
private static List<ClusterEvent> sentEvents()
|
||||||
|
{
|
||||||
|
List<ClusterEvent> sent = new CopyOnWriteArrayList<>();
|
||||||
|
WIRE.add(sent::add);
|
||||||
|
return (sent);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** A memo invalidation from node-a for a cache (and key).
|
||||||
|
*******************************************************************************/
|
||||||
|
private static void assertInvalidation(ClusterEvent event, String name, String key)
|
||||||
|
{
|
||||||
|
assertThat(event.getType()).isEqualTo(ClusterEvent.TYPE_MEMO_INVALIDATE);
|
||||||
|
assertThat(event.getName()).isEqualTo(name);
|
||||||
|
assertThat(event.getKey()).isEqualTo(key);
|
||||||
|
assertThat(event.getOriginNodeId()).isEqualTo("node-a");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,125 @@
|
|||||||
|
/*
|
||||||
|
* 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.internal;
|
||||||
|
|
||||||
|
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.Map;
|
||||||
|
import com.kingsrook.qbits.objectstorage.ObjectStorageQBitConfig;
|
||||||
|
import com.kingsrook.qbits.objectstorage.api.StorageAlertSink;
|
||||||
|
import com.kingsrook.qbits.objectstorage.api.StorageCaller;
|
||||||
|
import com.kingsrook.qbits.objectstorage.model.ObjectStorageTables;
|
||||||
|
import com.kingsrook.qbits.objectstorage.testsupport.StorageTestSupport;
|
||||||
|
import com.kingsrook.qqq.backend.core.logging.CollectedLogMessage;
|
||||||
|
import com.kingsrook.qqq.backend.core.logging.QCollectingLogger;
|
||||||
|
import com.kingsrook.qqq.backend.core.logging.QLogger;
|
||||||
|
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
|
||||||
|
import org.apache.logging.log4j.Level;
|
||||||
|
import org.junit.jupiter.api.AfterEach;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** An operation log row that cannot be written does not fail the operation it
|
||||||
|
** describes, so an operator has to be able to see it: an ERROR log line, the
|
||||||
|
** storage.operation_log.failed alert and OperationLog.failedWriteCount().
|
||||||
|
*******************************************************************************/
|
||||||
|
class OperationLogTest
|
||||||
|
{
|
||||||
|
private final List<String> alerts = new ArrayList<>();
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Reset the runtime, the context and the logger after each test.
|
||||||
|
*******************************************************************************/
|
||||||
|
@AfterEach
|
||||||
|
void tearDown()
|
||||||
|
{
|
||||||
|
QLogger.deactivateCollectingLoggerForClass(OperationLog.class);
|
||||||
|
StorageTestSupport.tearDown();
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** A row the insert action refuses (it hands the record back with an error, it
|
||||||
|
** does not throw) and a row whose insert throws are both counted, logged at
|
||||||
|
** ERROR and raised as an alert; the caller is not failed. The next row written
|
||||||
|
** for the bucket resolves the alert and the count stays.
|
||||||
|
*******************************************************************************/
|
||||||
|
@Test
|
||||||
|
void testFailedRowIsAnErrorAnAlertAndACount() throws Exception
|
||||||
|
{
|
||||||
|
QInstance instance = StorageTestSupport.setUp(new ObjectStorageQBitConfig().withBackendName(StorageTestSupport.MEMORY).withBucketScoping(false));
|
||||||
|
ObjectStorageRuntime.setAlertSink(new StorageAlertSink()
|
||||||
|
{
|
||||||
|
/*******************************************************************************
|
||||||
|
** Record the alert.
|
||||||
|
*******************************************************************************/
|
||||||
|
@Override
|
||||||
|
public void raise(String alertType, String dedupeKey, String severity, String message, Map<String, Object> details)
|
||||||
|
{
|
||||||
|
alerts.add("raise " + alertType + " " + dedupeKey + " " + severity + " count=" + details.get("count") + " action=" + details.get("action"));
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
/*******************************************************************************
|
||||||
|
** Record the resolution.
|
||||||
|
*******************************************************************************/
|
||||||
|
@Override
|
||||||
|
public void resolve(String alertType, String dedupeKey)
|
||||||
|
{
|
||||||
|
alerts.add("resolve " + alertType + " " + dedupeKey);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
QCollectingLogger logger = QLogger.activateCollectingLoggerForClass(OperationLog.class);
|
||||||
|
|
||||||
|
new OperationLog.Entry(StorageCaller.user(), "PUT", "files").key("a").write();
|
||||||
|
assertThat(OperationLog.failedWriteCount()).isZero();
|
||||||
|
assertThat(alerts).isEmpty();
|
||||||
|
|
||||||
|
instance.getTable(ObjectStorageTables.OPERATION).getField("detail").setIsRequired(true);
|
||||||
|
new OperationLog.Entry(StorageCaller.user(), "PUT", "files").key("refused").write();
|
||||||
|
assertThat(OperationLog.failedWriteCount()).isEqualTo(1);
|
||||||
|
assertThat(alerts).containsExactly("raise storage.operation_log.failed bucket:files error count=1 action=PUT");
|
||||||
|
|
||||||
|
instance.getTable(ObjectStorageTables.OPERATION).getField("detail").setIsRequired(false);
|
||||||
|
String backendName = instance.getTable(ObjectStorageTables.OPERATION).getBackendName();
|
||||||
|
instance.getTable(ObjectStorageTables.OPERATION).setBackendName("no-such-backend");
|
||||||
|
new OperationLog.Entry(StorageCaller.user(), "DELETE", "files").key("thrown").write();
|
||||||
|
assertThat(OperationLog.failedWriteCount()).isEqualTo(2);
|
||||||
|
assertThat(alerts).last().isEqualTo("raise storage.operation_log.failed bucket:files error count=2 action=DELETE");
|
||||||
|
|
||||||
|
List<CollectedLogMessage> errors = logger.getCollectedMessages().stream().filter(message -> Level.ERROR.equals(message.getLevel())).toList();
|
||||||
|
assertThat(errors).hasSize(2).allSatisfy(message -> assertThat(message.getMessage()).contains("Could not write a storage operation log row"));
|
||||||
|
|
||||||
|
instance.getTable(ObjectStorageTables.OPERATION).setBackendName(backendName);
|
||||||
|
new OperationLog.Entry(StorageCaller.user(), "PUT", "files").key("b").write();
|
||||||
|
new OperationLog.Entry(StorageCaller.user(), "PUT", "files").key("c").write();
|
||||||
|
assertThat(alerts).hasSize(3).last().isEqualTo("resolve storage.operation_log.failed bucket:files");
|
||||||
|
assertThat(OperationLog.failedWriteCount()).as("the count is of this node since it started, a later row does not lower it").isEqualTo(2);
|
||||||
|
assertThat(Db.asSystem(() -> Db.query(ObjectStorageTables.OPERATION))).extracting(row -> row.getValueString("objectKey")).containsExactly("a", "b", "c");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -205,7 +205,7 @@ class TablesAndConfigTest
|
|||||||
new com.kingsrook.qbits.objectstorage.ObjectStorageQBitProducer().withQBitConfig(config).produce(instance);
|
new com.kingsrook.qbits.objectstorage.ObjectStorageQBitProducer().withQBitConfig(config).produce(instance);
|
||||||
assertThat(instance.getApp("objectStorageAdmin").getParentAppName()).isEqualTo("admin");
|
assertThat(instance.getApp("objectStorageAdmin").getParentAppName()).isEqualTo("admin");
|
||||||
assertThat(instance.getApp("admin").getChildren()).extracting(c -> c.getName()).contains("objectStorageAdmin");
|
assertThat(instance.getApp("admin").getChildren()).extracting(c -> c.getName()).contains("objectStorageAdmin");
|
||||||
assertThat(com.kingsrook.qbits.objectstorage.app.StorageWidgets.tablePathPrefix()).isEqualTo("/admin/objectStorageAdmin");
|
assertThat(com.kingsrook.qbits.objectstorage.app.StorageWidgets.tablePathPrefix()).isEqualTo("/app");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user