Compare commits

..
Author SHA1 Message Date
CircleCI 036cd45f32 Bump version to 0.1.0-RC.10 [skip ci] 2026-10-16 16:31:49 +00:00
github-actions[bot] eaa6c9b93a Merge develop into release/0.1 for the next 0.1.0 RC 2026-10-16 15:57:53 +00:00
james.maes 825944fab4 Merge pull request #32 from QRun-IO/feature/backups-setup-attention-consistent
fix(app): Backups Setup and Needs attention agree
2026-10-16 10:19:49 -05:00
james.maesandClaude Opus 5.5 b25f4ff244 fix(app): Backups Setup and Needs attention derive from one list
The home said "1 item needs attention" in Setup while Needs attention said
"Nothing needs attention" (unresolved storage aliases). Every Setup item that
needs setup is now also listed under Needs attention, from the same item list.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-16 09:08:20 -05:00
james.maes b58c0edbd5 Merge pull request #31 from QRun-IO/ci/release-rc-secrets-inherit
ci(release-rc): pass secrets so the shared workflow can sign the RC tag
2026-10-16 08:04:37 -05:00
james.maesandClaude Opus 5.5 17aad52cdf ci(release-rc): pass secrets to the shared workflow so it can sign the RC tag
QRun-IO/.github#2 adds a tag-and-release job that signs vX with the
QRUN_TAG_SIGNING_KEY secret after Maven Central confirms the artifact.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 16:20:00 -05:00
CircleCI 33cb1f7a17 Bump version to 0.1.0-RC.9 [skip ci] 2026-10-15 19:24:05 +00:00
james.maesandClaude Opus 5.5 773629d253 Merge develop into release/0.1 for 0.1.0-RC.9 (cluster caches, QQQ RC.14)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 14:20:21 -05:00
james.maes 1b37582233 Merge pull request #30 from QRun-IO/feature/cluster-caches
Register backups caches with ClusterCaches (multi-node)
2026-10-15 14:11:22 -05:00
james.maesandClaude Opus 5.5 2f1964ddd1 build: QQQ 4.1.0-RC.14 (ships ClusterCaches)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 14:05:06 -05:00
james.maesandClaude Opus 5.5 e57ab62193 feat(cluster): register backups caches with ClusterCaches
The cached Prometheus text is shared as qbit-backups.metrics and
cleared on every node when a run finishes or a plan is saved or
deleted (the plan customizer now also runs post insert/update/delete);
the 60 s TTL stays as the safety net. SlaChecker's manifest-head and
bucket-audit throttles and the process-permission map are node-local.

Builds against QQQ 4.1.0-SNAPSHOT (qqq.snapshot.version) until the QQQ
RC with ClusterCaches is published; a release must pin that RC.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 12:11:11 -05:00
james.maes e263f3bc8b Merge pull request #29 from QRun-IO/fix/producer-version-from-build
fix: qbit metadata version comes from the build
2026-10-15 01:42:56 -05:00
james.maesandClaude Opus 5.5 ab969bf99b fix: the qbit metadata version comes from the build, not a hardcoded constant
The producer read a literal version, so released artifacts reported a stale
or -SNAPSHOT version. qbit-backups.properties (version=${project.version}) is
Maven-filtered and the producer loads it; tests assert producer == bundle ==
artifact version.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-15 01:27:59 -05:00
james.maes b17664db8c Merge pull request #28 from QRun-IO/fix/help-bundle-version
fix(help): help bundle version comes from the build, not a hardcoded SNAPSHOT
2026-10-14 22:37:20 -05:00
CircleCI 9855521637 Bump version to 0.1.0-RC.8 [skip ci] 2026-10-15 03:36:37 +00:00
james.maesandClaude Opus 5.5 99f74b4b5f Merge develop into release/0.1 for 0.1.0-RC.8 (#23 health checks)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 22:32:35 -05:00
james.maes eb413b6e15 Merge pull request #23 from QRun-IO/feature/health-checks
feat(health): register backups SLA and destination checks with core central health
2026-10-14 22:05:12 -05:00
james.maes 1e17dada29 Merge remote-tracking branch 'origin/develop' into feature/health-checks
# Conflicts:
#	CHANGELOG.md
2026-10-14 22:02:35 -05:00
james.maesandClaude Opus 5.5 991f310ccc fix(help): the help bundle carries the artifact version (Maven-filtered) instead of a hardcoded SNAPSHOT
On release branches the bundle's version and the producer's differed, so the
bundle test that asserts they match failed, and released bundles recorded
SNAPSHOT as their provenance. help/*.yaml is now filtered:
version: ${project.version}.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 22:01:33 -05:00
CircleCI fa9c4c1774 Bump version to 0.1.0-RC.7 [skip ci] 2026-10-15 02:21:57 +00:00
james.maesandClaude Opus 5.5 c8ba79f55b Merge develop into release/0.1 for 0.1.0-RC.7 (#26 QQQ RC.8, #27 help bundle)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 21:18:12 -05:00
james.maes 0e798e3079 Merge pull request #26 from QRun-IO/feature/job-progress
feat(progress): backup phases, counts, bytes and log lines (durable job progress)
2026-10-14 21:15:10 -05:00
james.maesandClaude Opus 5.5 9a2a1dd88e Merge origin/develop into feature/health-checks; pin QQQ 4.1.0-RC.8
RC.8 ships the central health kernel (QRun-IO/qqq#1207).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 21:14:59 -05:00
james.maes 20dbcc223c Merge remote-tracking branch 'origin/develop' into feature/job-progress
# Conflicts:
#	CHANGELOG.md
#	README.md
#	pom.xml
#	src/main/java/com/kingsrook/qbits/backups/processes/BackupsProcesses.java
2026-10-14 21:09:15 -05:00
james.maes ab4c578ba8 Merge pull request #27 from QRun-IO/feature/help-bundle
feat(help): qbit-backups help bundle via QQQ help bundles, with coverage gate
2026-10-14 20:29:38 -05:00
CircleCI 7830226982 Bump version to 0.1.0-RC.6 [skip ci] 2026-10-14 22:44:27 +00:00
james.maesandClaude Opus 5.5 b1f258a3d5 Merge develop into release/0.1 for 0.1.0-RC.6 (#24)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 17:41:40 -05:00
james.maesandClaude Opus 5.5 b06c1fd8e4 feat(progress): report backup phases, counts, bytes and log lines as a durable job
Run backup now, restore and verify always run as durable jobs and report
their work through QQQ JobProgress: prepare, tables (N of M, bytes),
media (N of M, copied vs already stored) and upload phases, one log line
per table and artifact (keys relative to the run, never credentials or
row values), warnings and the failure. Restore and verify honour the job
page check for pause and cancel.

Needs QQQ 4.1.0-RC.8 (QRun-IO/qqq#1212).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 17:32:53 -05:00
CircleCI 3df24cda5a Bump version to 0.1.0-RC.5 [skip ci] 2026-10-14 22:22:38 +00:00
james.maesandClaude Opus 5.5 987c20899b Merge develop into release/0.1 for 0.1.0-RC.5 (#25 setup card, default plan, scheduled verify)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 17:19:54 -05:00
james.maesandClaude Opus 5.5 f9767ede83 chore: drop optional qqq-middleware-health dependency; docs for backups health checks
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 14:55:03 -05:00
james.maesandClaude Opus 5.5 8d56613f7d feat(health): register backups SLA and destination checks; drop BackupsHealthIndicator
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 14:55:03 -05:00
james.maesandClaude Opus 5.5 0f9178dc14 feat(health): read-only destination probe mode
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 14:55:02 -05:00
CircleCI 14bb6fa1b4 Bump version to 0.1.0-RC.4 [skip ci] 2026-10-14 12:26:09 +00:00
james.maesandClaude Opus 5.5 b1ff70d7d3 Merge develop into release/0.1 for 0.1.0-RC.4 (#21 schema fingerprint, keyset pager, guardrails)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 07:21:32 -05:00
CircleCI cda334f2fe Bump version to 0.1.0-RC.3 [skip ci] 2026-10-14 00:53:05 +00:00
james.maesandClaude Opus 5.5 590b132fb6 Merge develop into release/0.1 for 0.1.0-RC.3 (core alerts, renamed alert types, #20)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-13 19:50:34 -05:00
CircleCI f0a8c5000e Bump version to 0.1.0-RC.2 [skip ci] 2026-10-13 16:11:19 +00:00
james.maesandClaude Opus 5.5 8f3f9ec60a Merge develop into release/0.1 for 0.1.0-RC.2
qbit-object-storage 0.1.0-RC.2 (AWS SDK v2 2.41.34, GHSA-443w-3rq3-5m5h; #17).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-13 11:08:28 -05:00
CircleCI cdaf218201 Bump version to 0.1.0-RC.1 [skip ci] 2026-10-13 00:26:41 +00:00
41 changed files with 2216 additions and 372 deletions

No files matched your search

+1
View File
@@ -21,5 +21,6 @@ jobs:
release-rc:
if: github.event_name == 'workflow_dispatch' || (github.event.pull_request.merged == true && contains(github.event.pull_request.labels.*.name, 'release:rc'))
uses: QRun-IO/.github/.github/workflows/qbit-release-rc.yml@main
secrets: inherit
with:
dry_run: ${{ github.event_name == 'workflow_dispatch' && inputs.dry_run }}
+12
View File
@@ -5,6 +5,10 @@ All notable changes to this project are documented here. The format follows Keep
## [0.1.0] - Unreleased
### Added
- Central `backups.sla` and per-active-plan `backups.destination.<planId>` health checks, with bounded read-only destination diagnostics and secret-free status details.
- Always-durable backup, resume, scheduled backup, restore and verify processes with conservative `NONE` idempotency and stable backup identities on explicit job retry.
- Weighted backup phases, table/file counts, stored bytes, safe per-table/artifact logs, warnings, failures and resume notices through QQQ `JobProgress`.
- Cooperative job cancellation during backup pages, media copying, restore pages and verification scans; local media-reference spooling keeps phase separation from growing heap usage across tables.
- Complete host-neutral in-app help bundle with live references and a help coverage gate. Hosts can override entries with a HOST bundle.
- Guided empty states: help bundle `help/qbit-backups.yaml` (app and all nine tables, with "How rows get created") applied by `BackupsHelp`, and a Setup card on the Backups home (plan, scope rules, storage, schedule, verification, prune bucket).
- "Create default plan" table action on Backup plans: adds "Nightly full" for a storage bucket alias; insert-only.
@@ -12,6 +16,11 @@ All notable changes to this project are documented here. The format follows Keep
- Repository scaffold: `BackupsQBitConfig` and `BackupsQBitProducer` (table prefix `backups_`, permissions `backups.*`), build, CI and repository standard files.
### Changed
- Multi-node deployments: the cached Prometheus text (`BackupsMetrics`) is registered with QQQ `ClusterCaches` as `qbit-backups.metrics` and cleared on every node when a run finishes or a plan is inserted, updated or deleted; the 60 s TTL remains the safety net. `SlaChecker`'s manifest-head and bucket-audit throttles and `BackupsRuntime`'s process-permission map are registered as node-local, with reasons. `BackupsClusterCachesTest` audits the package for unregistered static caches and covers a remote 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 artifact version so release builds report their released version.
- Target QQQ `4.1.0-RC.8` (job progress).
- Persist controlled backup failure messages instead of untrusted provider exception text.
- Pin QQQ to released 4.1.0-RC.7 and replace the temporary help loader with core HelpBundles registration.
- Saving a plan with Use read replica on is refused (it was never read); the not-yet-available settings (read replica, native dump, incremental, PARQUET, ZSTD) say so in their field help; scheduledJobId is documented as unused.
- Build against the released QQQ BOM 4.1.0-RC.5; replace reflective alerts binding with core QAlerts and real sink integration tests.
@@ -21,3 +30,6 @@ All notable changes to this project are documented here. The format follows Keep
- New manifests identify `qqq-table-schema-v1` and carry core schema descriptors; restore and content verification still accept pre-RC.6 manifests with the original fingerprint algorithm.
- Preserve primary-key ordering, exclusive resume cursors, bounded pages and caller-owned transactions; reject non-orderable primary keys and scope filters with skip/limit.
- Require the core `backups.restore` permission during Review when using a core restore writer, including anonymous-mode hosts; preserve supplied timestamps through `RestoreWriteFlag`.
### Removed
- The legacy `BackupsHealthIndicator` and its optional `qqq-middleware-health` dependency; central health registration replaces manual indicator configuration.
+46 -1
View File
@@ -10,7 +10,7 @@
## Requirements
- Java 21.
- QQQ `4.1.0-RC.7` or later.
- QQQ `4.1.0-RC.8` or later.
## Install
@@ -43,6 +43,11 @@ Follow the QQQ [adoption checklist](https://github.com/QRun-IO/qqq/blob/develop/
See [the RC.6 migration notes](docs/RC6-MIGRATION.md) for fingerprint compatibility,
paging guarantees and restore permissions.
Health checks use the core health API released in QQQ `4.1.0-RC.8`.
The producer registers `backups.sla` and a read-only destination check for each
active plan (`backups.destination.<planId>`). With dependencies cached locally,
run `./mvnw -o verify`; no enforcer skip flags are required.
## In-app help
The producer registers [help/qbit-backups.yaml](src/main/resources/help/qbit-backups.yaml)
@@ -57,6 +62,26 @@ checks empty-state, widget and possible-value-source coverage. Run it with
Always use the Maven wrapper (Maven 3.9.11); Maven 3.10 cannot read the parent POM.
## Durable jobs and progress
Run backup now, Resume, scheduled backups, both Restore actions, and Verify backup
are always durable jobs. Hosts must configure QQQ durable jobs and its worker.
Review remains a preview; execution runs in the background and reports through
the job page and jobs tray.
Backups report Resolve scope, Back up tables, Copy media files, and Write catalog
and manifest. Separate verification jobs report a single Verify phase.
The job includes table/file counts, stored bytes, per-table
rows and duration, artifact names, warnings and failures. Logs exclude row values,
source file keys, storage credentials and endpoints. Provider errors use controlled
failure messages; detailed exceptions stay in server diagnostics.
These processes use `NONE` idempotency: a worker interruption requires review before
retry because object writes and catalog updates are not atomic. Explicitly retrying
a backup job reuses its run and resumes completed parts; it does not create another
backup. Snapshot runs that cannot resume must be started as a new backup. See
[the progress and retry contract](docs/DESIGN.md#job-progress-implementation) for limits and cancellation.
## License
Apache License 2.0. See [LICENSE](LICENSE) and [NOTICE](NOTICE).
@@ -75,3 +100,23 @@ The backup SLA checker still owns expected intervals and grace periods.
RC.5 requires hyphenated alert names: `backups.run.missing-in-storage` and
`backups.plan.owner-invalid` replace the former underscore spellings, including
their deduplication keys. Update host routing rules that used those spellings.
## Health checks
The producer registers these central checks under component `backups`, category `STORAGE`:
| Name | Kind | Interval | Timeout |
| --- | --- | --- | --- |
| `backups.sla` | DEPENDENCY | 900s (15 minutes) | 5000ms |
| `backups.destination.<planId>` | DEPENDENCY | 86400s (24 hours) | 30000ms |
The SLA check reads persisted `slaStatus`: critical active plans that are LATE,
FAILING or MISSING_IN_STORAGE yield DOWN; non-critical failures yield DEGRADED.
Only UNVERIFIED/UNKNOWN plans yield UNKNOWN; no active plans yield UP. Details
contain counts per SLA status. The scheduled `backupCheckSla` still computes SLA state.
Destination checks are supplied for active plans and reuse `DestinationProbe` in
read-only list/head mode, including on unlocked buckets. They never write, delete,
or resolve prune credentials. The Test destination action retains its write tests.
Storage calls are bounded by the central runner, with deadline checks between calls.
Override intervals in `health_check` and disable checks through `health_check.enabled`.
+15
View File
@@ -1,3 +1,18 @@
# QBit Backups: design
The design for this qbit is tracked in [QRun-IO/qqq#1091](https://github.com/QRun-IO/qqq/pull/1091). This repository implements it; the implementation is in progress.
## Job progress implementation
Backup, resume, scheduled backup, restore, and verify execution use always-durable
jobs with `NONE` idempotency against the QQQ job-progress API targeted for RC.8.
Backup phases are `prepare`, `tables`, `media`, and `upload`, weighted 2, 75, 18,
and 5. Verification runs separately with a single `verify` phase.
Progress reports table/file counts, stored bytes, and safe artifact names without
row data, source keys, credentials, or endpoints. Cancellation is cooperative at
page and media-copy boundaries. Media references spool locally between phases.
Object writes and catalog updates are not atomic, so worker interruptions require
review before retry. Explicit backup retries reuse the run and resume completed
parts; snapshot runs that cannot resume require a new backup.
+25 -10
View File
@@ -20,7 +20,7 @@
<properties>
<qbit.guardrails.enforce>true</qbit.guardrails.enforce>
<qbit.guardrails.strictUiApiRoutes>true</qbit.guardrails.strictUiApiRoutes>
<revision>0.1.0-SNAPSHOT</revision>
<revision>0.1.0-RC.10</revision>
<maven.compiler.source>21</maven.compiler.source>
<maven.compiler.target>21</maven.compiler.target>
<jacoco.version>0.8.15</jacoco.version>
@@ -74,13 +74,6 @@
<artifactId>qqq-backend-core</artifactId>
</dependency>
<!-- optional: the BackupsHealthIndicator (apps without qqq-middleware-health never load it) -->
<dependency>
<groupId>com.kingsrook.qqq</groupId>
<artifactId>qqq-middleware-health</artifactId>
<optional>true</optional>
</dependency>
<!-- optional: ESB events (apps without qqq-esb never load it) -->
<dependency>
<groupId>com.kingsrook.qqq</groupId>
@@ -131,11 +124,33 @@
</dependencies>
<build>
<resources>
<!-- help bundles carry the artifact version (QQQ help bundle provenance) -->
<resource>
<directory>src/main/resources</directory>
<filtering>true</filtering>
<includes>
<include>help/*.yaml</include>
<include>qbit-backups.properties</include>
</includes>
</resource>
<resource>
<directory>src/main/resources</directory>
<filtering>false</filtering>
<excludes>
<exclude>help/*.yaml</exclude>
<exclude>qbit-backups.properties</exclude>
</excludes>
</resource>
</resources>
<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>
<!-- the MinIO test container default credentials, resolved through ${env.X} refs like production -->
<environmentVariables>
<STORAGE_TEST_ACCESS_KEY>minioadmin</STORAGE_TEST_ACCESS_KEY>
@@ -194,14 +209,14 @@
<profiles>
<!-- Built against qqq 4.1 (QBitMetaDataProducer). Until qbit-build-parent -->
<!-- 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.8 ahead of the parent's 4.0.0 BOM. -->
<profile>
<id>qqq-snapshot</id>
<activation>
<activeByDefault>true</activeByDefault>
</activation>
<properties>
<qqq.snapshot.version>4.1.0-RC.7</qqq.snapshot.version>
<qqq.snapshot.version>4.1.0-RC.14</qqq.snapshot.version>
</properties>
<dependencyManagement>
<dependencies>
@@ -17,6 +17,17 @@
package com.kingsrook.qbits.backups;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import com.kingsrook.qbits.backups.health.BackupsDestinationHealthChecks;
import com.kingsrook.qbits.backups.health.BackupsSlaHealthCheck;
import com.kingsrook.qqq.backend.core.health.HealthCategory;
import com.kingsrook.qqq.backend.core.health.HealthCheckKind;
import com.kingsrook.qqq.backend.core.health.HealthCheckMetaData;
import com.kingsrook.qqq.backend.core.health.HealthCheckSupplierMetaData;
import com.kingsrook.qqq.backend.core.health.HealthChecksMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
import com.kingsrook.qqq.backend.core.model.metadata.help.bundles.HelpBundles;
import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaDataProducer;
@@ -30,13 +41,41 @@ public class BackupsQBitProducer implements QBitMetaDataProducer<BackupsQBitConf
{
public static final String GROUP_ID = "com.kingsrook.qbits";
public static final String ARTIFACT_ID = "qbit-backups";
public static final String VERSION = "0.1.0";
public static final String VERSION = loadVersion();
private BackupsQBitConfig qBitConfig;
private String namespace;
/*******************************************************************************
** Read the artifact version from the Maven-filtered resource.
*******************************************************************************/
private static String loadVersion()
{
try(InputStream inputStream = BackupsQBitProducer.class.getClassLoader().getResourceAsStream("qbit-backups.properties"))
{
if(inputStream == null)
{
throw (new IllegalStateException("Missing qbit-backups.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-backups.properties; build with Maven"));
}
return (version);
}
catch(IOException e)
{
throw (new IllegalStateException("Could not read artifact version from qbit-backups.properties", e));
}
}
/*******************************************************************************
** Fluent setter for the config.
*******************************************************************************/
@@ -164,6 +203,13 @@ public class BackupsQBitProducer implements QBitMetaDataProducer<BackupsQBitConf
// Registration is metadata-only; core guards startup sync when helpContent is absent.
HelpBundles.register(qInstance, "help/qbit-backups.yaml");
HealthChecksMetaData.ofOrWithNew(qInstance)
.withCheck(new HealthCheckMetaData().withName(BackupsSlaHealthCheck.NAME).withLabel("Backup SLA")
.withComponent("backups").withKind(HealthCheckKind.DEPENDENCY).withCategory(HealthCategory.STORAGE)
.withIntervalSeconds(900).withCheck(new QCodeReference(BackupsSlaHealthCheck.class)))
.withSupplier(new HealthCheckSupplierMetaData().withNamePrefix(BackupsDestinationHealthChecks.PREFIX)
.withComponent("backups").withSupplier(new QCodeReference(BackupsDestinationHealthChecks.class)));
// A parent app only lists the apps that are its children, so attach ourselves (for example Admin, then Backups).
com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData parent = qBitConfig.getParentAppName() == null ? null : qInstance.getApp(qBitConfig.getParentAppName());
if(parent != null && (parent.getChildren() == null || parent.getChildren().stream().noneMatch(child -> app.getName().equals(child.getName()))))
@@ -25,6 +25,7 @@ import com.kingsrook.qbits.backups.alerts.BackupAlertSink;
import com.kingsrook.qbits.backups.alerts.NoOpBackupAlertSink;
import com.kingsrook.qbits.backups.storage.BackupStorage;
import com.kingsrook.qbits.backups.storage.BackupStorageException;
import com.kingsrook.qqq.backend.core.cluster.ClusterCaches;
/*******************************************************************************
@@ -34,7 +35,8 @@ import com.kingsrook.qbits.backups.storage.BackupStorageException;
*******************************************************************************/
public final class BackupsRuntime
{
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 BackupsQBitConfig config;
private static volatile Clock clock = Clock.systemUTC();
@@ -69,10 +69,11 @@ public final class BackupsHomeView
public static String render(BackupsHomeData data, String pathPrefix)
{
StringBuilder html = new StringBuilder("<div class=\"backups-home\" style=\"" + ROOT + "\">");
List<BackupsSetupStatus.Item> setupItems = BackupsSetupStatus.load(data.plans());
planStatus(html, data, pathPrefix);
setup(html, data, pathPrefix);
setup(html, setupItems, pathPrefix);
html.append("<div style=\"").append(GRID).append("\">");
needsAttention(html, data, pathPrefix);
needsAttention(html, data, setupItems, pathPrefix);
latestTestRestore(html, data, pathPrefix);
recentActivity(html, data, pathPrefix);
trend(html, data);
@@ -144,9 +145,8 @@ public final class BackupsHomeView
** The Setup card: what a working backup needs, whether each part is in place,
** and the next step, linked to its table.
*******************************************************************************/
private static void setup(StringBuilder html, BackupsHomeData data, String prefix)
private static void setup(StringBuilder html, List<BackupsSetupStatus.Item> items, String prefix)
{
List<BackupsSetupStatus.Item> items = BackupsSetupStatus.load(data.plans());
open(html, "Setup");
long todo = items.stream().filter(i -> i.state() == BackupsSetupStatus.State.TODO).count();
html.append("<p style=\"margin:0 0 .6rem;font-size:.9rem;\">")
@@ -185,23 +185,27 @@ public final class BackupsHomeView
/*******************************************************************************
** Needs attention.
** Needs attention: every Setup item that needs setup (the same list the Setup
** card counts), plus plans in a problem state and recent failed runs, so the two
** cards never disagree.
*******************************************************************************/
private static void needsAttention(StringBuilder html, BackupsHomeData data, String prefix)
private static void needsAttention(StringBuilder html, BackupsHomeData data, List<BackupsSetupStatus.Item> setupItems, String prefix)
{
open(html, "Needs attention");
List<QRecord> plans = data.plansNeedingAttention();
if(data.plans().isEmpty())
List<BackupsSetupStatus.Item> setupTodo = setupItems.stream().filter(i -> i.state() == BackupsSetupStatus.State.TODO).toList();
List<QRecord> plans = data.plansNeedingAttention();
if(setupTodo.isEmpty() && plans.isEmpty() && data.failedRuns().isEmpty())
{
empty(html, "Nothing to check yet: no plans.");
}
else if(plans.isEmpty() && data.failedRuns().isEmpty())
{
empty(html, "Nothing needs attention.");
empty(html, data.plans().isEmpty() ? "Nothing to check yet: no plans." : "Nothing needs attention.");
}
else
{
html.append("<ul style=\"").append(ITEMS).append("\">");
for(BackupsSetupStatus.Item item : setupTodo)
{
html.append("<li>Setup: <a style=\"").append(LINK).append("\" href=\"").append(esc(prefix + "/" + item.tableName())).append("\">").append(esc(item.label())).append("</a>: ")
.append(esc(item.detail())).append("</li>");
}
for(QRecord plan : plans)
{
html.append("<li>").append(recordLink(prefix, BackupsTables.PLAN, plan.getValue("id"), plan.getValueString("name")))
@@ -0,0 +1,95 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.backups.health;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.plan.DestinationProbe;
import com.kingsrook.qbits.backups.run.BackupCatalog;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.health.HealthCheck;
import com.kingsrook.qqq.backend.core.health.HealthCheckInput;
import com.kingsrook.qqq.backend.core.health.HealthCheckOutcome;
import com.kingsrook.qqq.backend.core.health.HealthDetails;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
/*******************************************************************************
** Reuse the destination diagnostic in read-only mode; never publish its text.
*******************************************************************************/
public class BackupsDestinationHealthCheck implements HealthCheck
{
private final Integer planId;
/*******************************************************************************
** Retain only the id; destination and active state are read on each run.
*******************************************************************************/
public BackupsDestinationHealthCheck(Integer planId)
{
this.planId = planId;
}
/*******************************************************************************
** A removed or disabled plan can remain registered until supplier refresh.
*******************************************************************************/
@Override
public HealthCheckOutcome run(HealthCheckInput input)
{
boolean ownsContext = QContext.getQInstance() == null;
if(ownsContext)
{
QContext.init(input.getQInstance(), new QSystemUserSession(input.getQInstance()));
}
try
{
if(BackupsSlaHealthCheck.expired(input))
{
return (HealthCheckOutcome.down("Backup destination check timed out"));
}
QRecord plan = new BackupCatalog().get(BackupsTables.PLAN, planId).orElse(null);
if(plan == null || !Boolean.TRUE.equals(plan.getValueBoolean("isActive")))
{
return (HealthCheckOutcome.unknown("Backup plan is no longer active"));
}
DestinationProbe.Report report = DestinationProbe.probeReadOnly(plan, input.getDeadline());
if(BackupsSlaHealthCheck.expired(input))
{
return (HealthCheckOutcome.down("Backup destination check timed out"));
}
return ((report.ok() ? HealthCheckOutcome.up("Backup destination is readable") : HealthCheckOutcome.down("Backup destination probe failed"))
.withDetails(new HealthDetails().withValue("planId", planId).withValue("readOnly", true)
.withValue("checks", report.checks().size()).withValue("failedChecks", report.checks().stream().filter(check -> !check.ok()).count())));
}
catch(QException | RuntimeException e)
{
return (HealthCheckOutcome.down("Backup destination could not be checked"));
}
finally
{
if(ownsContext)
{
QContext.clear();
}
}
}
}
@@ -0,0 +1,77 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.backups.health;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.run.BackupCatalog;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.health.HealthCategory;
import com.kingsrook.qqq.backend.core.health.HealthCheckKind;
import com.kingsrook.qqq.backend.core.health.HealthCheckMetaData;
import com.kingsrook.qqq.backend.core.health.HealthCheckSupplier;
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.session.QSystemUserSession;
/*******************************************************************************
** Discover one daily, read-only destination check per active backup plan.
*******************************************************************************/
public class BackupsDestinationHealthChecks implements HealthCheckSupplier
{
public static final String PREFIX = "backups.destination";
/*******************************************************************************
** Resolve from current catalog data, without opening any storage connections.
*******************************************************************************/
@Override
public List<HealthCheckMetaData> supply(QInstance qInstance) throws QException
{
boolean ownsContext = QContext.getQInstance() == null;
if(ownsContext)
{
QContext.init(qInstance, new QSystemUserSession(qInstance));
}
try
{
List<HealthCheckMetaData> checks = new ArrayList<>();
for(QRecord plan : new BackupCatalog().where(BackupsTables.PLAN, "isActive", true))
{
Integer planId = plan.getValueInteger("id");
checks.add(new HealthCheckMetaData().withName(PREFIX + "." + planId.toString().toLowerCase(Locale.ROOT))
.withLabel("Backup destination " + planId).withComponent("backups")
.withKind(HealthCheckKind.DEPENDENCY).withCategory(HealthCategory.STORAGE)
.withIntervalSeconds(86400).withTimeoutMs(30000).withCheckInstance(new BackupsDestinationHealthCheck(planId)));
}
return (checks);
}
finally
{
if(ownsContext)
{
QContext.clear();
}
}
}
}
@@ -1,133 +0,0 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.backups.health;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.run.BackupCatalog;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
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.session.QSession;
import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
import com.kingsrook.qqq.middleware.health.HealthIndicator;
import com.kingsrook.qqq.middleware.health.model.metadata.HealthCheckResult;
import com.kingsrook.qqq.middleware.health.model.metadata.HealthStatus;
/*******************************************************************************
** The Backups health indicator (design spec section 10.4), for
** qqq-middleware-health (an optional dependency: an app that does not have that
** module never loads this class). It reports DOWN if any active plan marked
** critical is LATE, FAILING or MISSING_IN_STORAGE, and UP with details otherwise.
** It reads the slaStatus the SLA checker maintains; it never touches storage, so
** a health call is cheap.
**
** It is OFF the liveness path by design. Add it to the indicators of the
** instance's health check (the readiness and monitoring view), and do not point a
** Kubernetes liveness probe at a check that includes it: a backup problem must
** never restart pods.
*******************************************************************************/
public class BackupsHealthIndicator implements HealthIndicator
{
public static final String NAME = "backups";
private static final Set<String> CRITICAL_STATUSES = Set.of("LATE", "FAILING", "MISSING_IN_STORAGE");
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public String getName()
{
return (NAME);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public HealthCheckResult check(QInstance qInstance) throws QException
{
long started = System.currentTimeMillis();
boolean ownsContext = QContext.getQInstance() == null;
if(ownsContext)
{
QContext.init(qInstance, systemSession(qInstance));
}
try
{
List<QRecord> plans = new BackupCatalog().query(BackupsTables.PLAN, new QQueryFilter(new QFilterCriteria("isActive", QCriteriaOperator.EQUALS, true)));
List<String> criticalProblems = new ArrayList<>();
Map<String, Integer> byStatus = new LinkedHashMap<>();
for(QRecord plan : plans)
{
String status = plan.getValueString("slaStatus") == null ? "UNKNOWN" : plan.getValueString("slaStatus");
byStatus.merge(status, 1, Integer::sum);
if(Boolean.TRUE.equals(plan.getValueBoolean("critical")) && CRITICAL_STATUSES.contains(status))
{
criticalProblems.add(plan.getValueString("planKey") + ": " + status);
}
}
Map<String, Object> details = new LinkedHashMap<>();
details.put("activePlans", plans.size());
details.put("plansByStatus", byStatus);
if(!criticalProblems.isEmpty())
{
details.put("criticalPlansWithProblems", criticalProblems);
}
return (new HealthCheckResult()
.withStatus(criticalProblems.isEmpty() ? HealthStatus.UP : HealthStatus.DOWN)
.withDurationMs(System.currentTimeMillis() - started)
.withDetails(details));
}
finally
{
if(ownsContext)
{
QContext.clear();
}
}
}
/*******************************************************************************
** The session for a health request that has no QQQ context. It must be the
** core system user: token-based authentication modules (OAuth2, Auth0)
** reject any other tokenless session, which failed the check on such hosts.
*******************************************************************************/
static QSession systemSession(QInstance qInstance)
{
return (new QSystemUserSession(qInstance));
}
}
@@ -0,0 +1,132 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.backups.health;
import java.time.Instant;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.run.BackupCatalog;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.health.HealthCheck;
import com.kingsrook.qqq.backend.core.health.HealthCheckInput;
import com.kingsrook.qqq.backend.core.health.HealthCheckOutcome;
import com.kingsrook.qqq.backend.core.health.HealthDetails;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
/*******************************************************************************
** Read persisted SLA state using the former indicator's critical-plan rule.
** SLA calculation and storage verification remain the scheduled checker's job.
*******************************************************************************/
public class BackupsSlaHealthCheck implements HealthCheck
{
public static final String NAME = "backups.sla";
private static final Set<String> BAD_STATUSES = Set.of("LATE", "FAILING", "MISSING_IN_STORAGE");
private static final List<String> STATUSES = List.of("OK", "LATE", "FAILING", "MISSING_IN_STORAGE", "UNVERIFIED", "ABANDONED", "OWNER_INVALID", "UNKNOWN");
/*******************************************************************************
** Counts use a fixed vocabulary so stored strings cannot become detail keys.
*******************************************************************************/
@Override
public HealthCheckOutcome run(HealthCheckInput input)
{
boolean ownsContext = QContext.getQInstance() == null;
if(ownsContext)
{
QContext.init(input.getQInstance(), new QSystemUserSession(input.getQInstance()));
}
try
{
if(expired(input))
{
return (HealthCheckOutcome.down("Backup SLA check timed out"));
}
List<QRecord> plans = new BackupCatalog().where(BackupsTables.PLAN, "isActive", true);
Map<String, Integer> counts = new LinkedHashMap<>();
STATUSES.forEach(status -> counts.put(status, 0));
boolean criticalBad = false;
boolean ordinaryBad = false;
for(QRecord plan : plans)
{
String status = plan.getValueString("slaStatus");
status = status != null && STATUSES.contains(status) ? status : "UNKNOWN";
counts.merge(status, 1, Integer::sum);
if(BAD_STATUSES.contains(status))
{
if(Boolean.TRUE.equals(plan.getValueBoolean("critical")))
{
criticalBad = true;
}
else
{
ordinaryBad = true;
}
}
}
if(expired(input))
{
return (HealthCheckOutcome.down("Backup SLA check timed out"));
}
HealthDetails details = new HealthDetails().withValue("activePlans", plans.size());
counts.forEach((status, count) -> details.withValue("slaStatus." + status, count));
HealthCheckOutcome outcome = HealthCheckOutcome.up("Backup SLA has no failures");
if(criticalBad)
{
outcome = HealthCheckOutcome.down("Critical backup plans have SLA failures");
}
else if(ordinaryBad)
{
outcome = HealthCheckOutcome.degraded("Backup plans have SLA failures");
}
else if(!plans.isEmpty() && counts.get("UNVERIFIED") + counts.get("UNKNOWN") == plans.size())
{
outcome = HealthCheckOutcome.unknown("Backup SLA is not verified yet");
}
return (outcome.withDetails(details));
}
catch(QException | RuntimeException e)
{
return (HealthCheckOutcome.down("Backup SLA state could not be read"));
}
finally
{
if(ownsContext)
{
QContext.clear();
}
}
}
/*******************************************************************************
** Stop between catalog operations when the runner's budget is exhausted.
*******************************************************************************/
static boolean expired(HealthCheckInput input)
{
return (Thread.currentThread().isInterrupted() || !Instant.now().isBefore(input.getDeadline()));
}
}
@@ -20,10 +20,12 @@ package com.kingsrook.qbits.backups.metrics;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
import com.kingsrook.qbits.backups.BackupsQBitConfig;
import com.kingsrook.qbits.backups.BackupsRuntime;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.run.BackupCatalog;
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.query.QCriteriaOperator;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
@@ -56,10 +58,21 @@ public final class BackupsMetrics
{
public static final String ENDPOINT_PATH = "/metrics/backups";
/** Cluster cache name of the rendered text: cleared on every node when a run finishes or a plan changes. */
public static final String CACHE_NAME = "qbit-backups.metrics";
private static final Duration CACHE = Duration.ofSeconds(60);
private static String cachedText;
private static Instant cachedAt;
private static final AtomicReference<Rendered> CACHED = ClusterCaches.shared(CACHE_NAME, new AtomicReference<>(), BackupsMetrics::resetCache, null);
/*******************************************************************************
** The rendered text and when it was rendered.
*******************************************************************************/
private record Rendered(String text, Instant at)
{
}
@@ -84,12 +97,21 @@ public final class BackupsMetrics
/*******************************************************************************
** Forget the cached text (tests).
** Forget the cached text on this node.
*******************************************************************************/
public static synchronized void resetCache()
public static void resetCache()
{
cachedText = null;
cachedAt = null;
CACHED.set(null);
}
/*******************************************************************************
** Forget the cached text on every node (the backup data changed).
*******************************************************************************/
public static void invalidate()
{
ClusterCaches.invalidate(CACHE_NAME);
}
@@ -99,13 +121,14 @@ public final class BackupsMetrics
*******************************************************************************/
public static synchronized String renderCached() throws QException
{
Instant now = BackupsRuntime.getClock().instant();
if(cachedText == null || cachedAt == null || cachedAt.plus(CACHE).isBefore(now))
Instant now = BackupsRuntime.getClock().instant();
Rendered cached = CACHED.get();
if(cached == null || cached.at().plus(CACHE).isBefore(now))
{
cachedText = render();
cachedAt = now;
cached = new Rendered(render(), now);
CACHED.set(cached);
}
return (cachedText);
return (cached.text());
}
@@ -413,15 +413,22 @@ public final class BackupsTables
QCodeReference validation = new QCodeReference(BackupPlanTableCustomizer.class);
table.withCustomizer(TableCustomizers.PRE_INSERT_RECORD, validation);
table.withCustomizer(TableCustomizers.PRE_UPDATE_RECORD, validation);
/////////////////////////////////////////////////////////////////////////////
// after a save or delete, the plan customizer clears the cached metrics on //
// every node //
/////////////////////////////////////////////////////////////////////////////
QCodeReference post = validation;
if(StringUtils.hasContent(config.getSchedulerName()))
{
QCodeReferenceWithProperties sync = MultiCustomizer.of(
post = MultiCustomizer.of(
BaseSyncToScheduledJobTableCustomizer.makeCodeReference(table, new BackupPlanSyncToScheduledJobProcess()),
BaseSyncToScheduledJobTableCustomizer.makeCodeReference(table, new BackupPlanVerifySyncToScheduledJobProcess()));
table.withCustomizer(TableCustomizers.POST_INSERT_RECORD, sync);
table.withCustomizer(TableCustomizers.POST_UPDATE_RECORD, sync);
table.withCustomizer(TableCustomizers.POST_DELETE_RECORD, sync);
BaseSyncToScheduledJobTableCustomizer.makeCodeReference(table, new BackupPlanVerifySyncToScheduledJobProcess()),
validation);
}
table.withCustomizer(TableCustomizers.POST_INSERT_RECORD, post);
table.withCustomizer(TableCustomizers.POST_UPDATE_RECORD, post);
table.withCustomizer(TableCustomizers.POST_DELETE_RECORD, post);
return (table);
}
@@ -22,6 +22,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import com.kingsrook.qbits.backups.metrics.BackupsMetrics;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.run.BackupCatalog;
import com.kingsrook.qbits.backups.run.BackupRunFactory;
@@ -31,6 +32,9 @@ import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterfa
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.AbstractActionInput;
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;
import com.kingsrook.qqq.backend.core.model.session.QSession;
import com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage;
@@ -43,7 +47,8 @@ import com.kingsrook.qqq.backend.core.model.statusmessages.QWarningMessage;
** configuration (the processes write the last run and SLA fields) is not
** validated again.
**
** The saving user becomes the plan owner when none is given.
** The saving user becomes the plan owner when none is given. After an insert,
** update or delete, the cached Prometheus text is cleared on every node.
*******************************************************************************/
public class BackupPlanTableCustomizer implements TableCustomizerInterface
{
@@ -122,6 +127,42 @@ public class BackupPlanTableCustomizer implements TableCustomizerInterface
/*******************************************************************************
** A saved plan changes the metrics on every node.
*******************************************************************************/
@Override
public List<QRecord> postInsert(InsertInput insertInput, List<QRecord> records) throws QException
{
BackupsMetrics.invalidate();
return (records);
}
/*******************************************************************************
** A saved plan changes the metrics on every node.
*******************************************************************************/
@Override
public List<QRecord> postUpdate(UpdateInput updateInput, List<QRecord> records, Optional<List<QRecord>> oldRecordList) throws QException
{
BackupsMetrics.invalidate();
return (records);
}
/*******************************************************************************
** A deleted plan changes the metrics on every node.
*******************************************************************************/
@Override
public List<QRecord> postDelete(DeleteInput deleteInput, List<QRecord> records) throws QException
{
BackupsMetrics.invalidate();
return (records);
}
/*******************************************************************************
** The scope rules to resolve: the ones sent with the plan (associated rows), or
** the saved ones.
@@ -20,6 +20,7 @@ package com.kingsrook.qbits.backups.plan;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
@@ -81,6 +82,28 @@ public final class DestinationProbe
** Probe the destination of a plan.
*******************************************************************************/
public static Report probe(QRecord plan)
{
return (probe(plan, false, null));
}
/*******************************************************************************
** Health monitoring only lists and heads; it never resolves prune credentials
** or probes capabilities. The runner bounds in-flight storage calls; check the
** deadline between calls because BackupStorage has no per-call timeout API.
*******************************************************************************/
public static Report probeReadOnly(QRecord plan, Instant deadline)
{
return (probe(plan, true, deadline));
}
/*******************************************************************************
** Share storage resolution and the existing locked-bucket read path.
*******************************************************************************/
private static Report probe(QRecord plan, boolean readOnly, Instant deadline)
{
List<Check> checks = new ArrayList<>();
List<String> warnings = new ArrayList<>();
@@ -89,15 +112,23 @@ public final class DestinationProbe
BackupStorage storage;
try
{
requireTime(deadline);
storage = BackupsRuntime.storageFor(alias);
requireTime(deadline);
}
catch(RuntimeException e)
{
checks.add(new Check("Open the backup bucket " + alias, false, e.getMessage()));
return (new Report(false, checks, List.of(), warnings));
return (new Report(readOnly, checks, List.of(), warnings));
}
checks.add(new Check("Open the backup bucket " + alias, true, "found"));
if(readOnly)
{
headOnly(plan, storage, BackupPruner.planPrefix(plan), checks, deadline);
return (new Report(true, checks, List.of(), warnings));
}
BackupStorage deleteSide = storage;
String pruneAlias = plan.getValueString("pruneBucketAlias");
if(pruneAlias == null || pruneAlias.isBlank())
@@ -133,7 +164,7 @@ public final class DestinationProbe
String prefix = BackupPruner.planPrefix(plan);
if(locked)
{
headOnly(plan, storage, prefix, checks);
headOnly(plan, storage, prefix, checks, null);
}
else
{
@@ -147,10 +178,11 @@ public final class DestinationProbe
/*******************************************************************************
** The probe for a locked bucket: list the prefix, and head the newest manifest.
*******************************************************************************/
private static void headOnly(QRecord plan, BackupStorage storage, String prefix, List<Check> checks)
private static void headOnly(QRecord plan, BackupStorage storage, String prefix, List<Check> checks, Instant deadline)
{
checks.add(attempt("Write and delete a probe object", () -> "skipped: object lock is on, a probe object could not be deleted again"));
checks.add(attempt("List the plan prefix", () ->
checks.add(new Check("Write and delete a probe object", true,
deadline == null ? "skipped: object lock is on, a probe object could not be deleted again" : "skipped: read-only probe"));
checks.add(attempt("List the plan prefix", deadline, () ->
{
List<StoredObjectInfo> objects = storage.list(prefix);
return (objects.size() + " objects under " + prefix);
@@ -162,10 +194,11 @@ public final class DestinationProbe
checks.add(new Check("Head the newest manifest", true, "there is no successful run yet"));
return;
}
checks.add(attempt("Head the newest manifest", () ->
checks.add(attempt("Head the newest manifest", deadline, () ->
{
QRecord run = new com.kingsrook.qbits.backups.run.BackupCatalog().get(com.kingsrook.qbits.backups.model.BackupsTables.RUN, lastSuccess).orElseThrow();
String key = run.getValueString("manifestKey");
requireTime(deadline);
if(key == null || storage.head(key).isEmpty())
{
throw (new IllegalStateException("the manifest of the newest successful run is not in the bucket"));
@@ -238,6 +271,9 @@ public final class DestinationProbe
@FunctionalInterface
private interface Attempt
{
/*******************************************************************************
** Execute one diagnostic operation.
*******************************************************************************/
String run() throws Exception;
}
@@ -247,14 +283,40 @@ public final class DestinationProbe
**
*******************************************************************************/
private static Check attempt(String name, Attempt attempt)
{
return (attempt(name, null, attempt));
}
/*******************************************************************************
** Never begin another storage operation after the health budget expires.
*******************************************************************************/
private static Check attempt(String name, Instant deadline, Attempt attempt)
{
try
{
return (new Check(name, true, attempt.run()));
requireTime(deadline);
String detail = attempt.run();
requireTime(deadline);
return (new Check(name, true, detail));
}
catch(Exception e)
{
return (new Check(name, false, e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage()));
}
}
/*******************************************************************************
** Existing on-demand diagnostics have no deadline; health probes always do.
*******************************************************************************/
private static void requireTime(Instant deadline)
{
if(deadline != null && (Thread.currentThread().isInterrupted() || !Instant.now().isBefore(deadline)))
{
throw (new IllegalStateException("Destination probe timed out"));
}
}
}
@@ -30,6 +30,11 @@ import com.kingsrook.qbits.backups.run.BackupRunFactory;
import com.kingsrook.qbits.backups.run.BackupRunResult;
import com.kingsrook.qbits.backups.scope.ScopeResolver;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.jobs.DurableProcessMetaData;
import com.kingsrook.qqq.backend.core.jobs.model.JobIdempotency;
import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
import com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException;
import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
@@ -71,7 +76,7 @@ public final class BackupsProcesses
{
List<QProcessMetaData> processes = new ArrayList<>();
processes.add(permission(etl(RUN_PLAN, "Run backup now", BackupsTables.PLAN, RunPlanTransform.class, RunPlanLoad.class,
"This is a preview of what the backup will include. Nothing is written until you continue."), BackupsPermissions.RUN));
"This is a preview of what the backup will include. Nothing is written until you continue.").withSupplementalMetaData(durability()), BackupsPermissions.RUN));
processes.add(DefaultPlanProcess.process());
processes.addAll(VerifyProcesses.processes());
processes.addAll(RestoreProcesses.processes());
@@ -82,6 +87,40 @@ public final class BackupsProcesses
/*******************************************************************************
** External writes cannot commit with the job chunk. Require review before retry
** instead of claiming that replay is naturally idempotent.
*******************************************************************************/
public static DurableProcessMetaData durability()
{
return (new DurableProcessMetaData().withEligible(true).withAlwaysDurable(true)
.withChunkSize(1).withIdempotency(JobIdempotency.NONE));
}
/*******************************************************************************
** A durable worker must see failures/cancellation instead of a successful ETL
** summary. Inline callers retain the existing result-screen behavior.
*******************************************************************************/
public static void requireSuccessfulJob(BackupRunResult result) throws QException
{
if(JobChunkScope.current().isEmpty())
{
return;
}
if("CANCELLED".equals(result.getStatus()))
{
throw (new JobInterruptedException(JobInterruptedException.Reason.CANCEL));
}
if(!result.isUsable())
{
throw (new QException(result.getErrorMessage() == null ? "Backup did not complete" : result.getErrorMessage()));
}
}
/*******************************************************************************
** Record the permission a process needs.
*******************************************************************************/
@@ -229,8 +268,9 @@ public final class BackupsProcesses
{
for(QRecord plan : input.getRecords())
{
BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "MANUAL");
BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "MANUAL", JobProgress.of(input), null);
results.add(result);
requireSuccessfulJob(result);
switch(result.getStatus())
{
case "SUCCEEDED" -> succeeded.incrementCountAndAddPrimaryKey(result.getRunId());
@@ -33,6 +33,8 @@ import com.kingsrook.qbits.backups.restore.RestoreRequest;
import com.kingsrook.qbits.backups.restore.RestoreSummaries;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.exceptions.QUserFacingException;
import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
@@ -107,7 +109,7 @@ public final class RestoreProcesses
process.setLabel(label);
process.setTableName(table);
process.setPermissionRules(BackupsTables.permissionRules());
return (process);
return (process.withSupplementalMetaData(BackupsProcesses.durability()));
}
@@ -241,13 +243,38 @@ public final class RestoreProcesses
@Override
public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
{
JobProgress progress = JobProgress.of(input);
progress.phases(JobPhase.of("restore", "Restore backups", 100)).phase("restore");
Map<Integer, Set<String>> sources = sources(input.getRecords());
progress.total(sources.size(), "backups").done(0);
int done = 0;
RestoreEngine engine = new RestoreEngine();
for(Map.Entry<Integer, Set<String>> source : sources(input.getRecords()).entrySet())
try
{
RestoreRequest request = RestoreRequest.fromInput(input, source.getKey(), new ArrayList<>(source.getValue()));
outcomes.add(engine.execute(request));
for(Map.Entry<Integer, Set<String>> source : sources.entrySet())
{
progress.message("Restoring backup " + source.getKey());
RestoreRequest request = RestoreRequest.fromInput(input, source.getKey(), new ArrayList<>(source.getValue()));
outcomes.add(engine.execute(request));
progress.info("Restored backup " + source.getKey());
progress.done(++done);
}
progress.getSnapshot().getPhases().getFirst().setStatus(JobPhase.Status.DONE);
output.getRecords().addAll(input.getRecords());
}
catch(QException | RuntimeException e)
{
progress.getSnapshot().getPhases().getFirst().setStatus(JobPhase.Status.FAILED);
if(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.find(e) == null)
{
progress.error("Restore failed (" + e.getClass().getSimpleName() + ")");
}
throw (e);
}
finally
{
progress.flush();
}
output.getRecords().addAll(input.getRecords());
}
@@ -70,7 +70,7 @@ public final class ResumeRunProcess
public static QProcessMetaData process()
{
return (BackupsProcesses.permission(BackupsProcesses.etl(NAME, "Resume", BackupsTables.RUN, Transform.class, Load.class,
"This is a preview of where each selected run will carry on. Nothing is written until you continue."), BackupsPermissions.RUN));
"This is a preview of where each selected run will carry on. Nothing is written until you continue.").withSupplementalMetaData(BackupsProcesses.durability()), BackupsPermissions.RUN));
}
@@ -167,7 +167,8 @@ public final class ResumeRunProcess
{
for(QRecord run : input.getRecords())
{
BackupRunResult result = new BackupRunCoordinator().resume(run);
BackupRunResult result = new BackupRunCoordinator().resume(run, com.kingsrook.qqq.backend.core.jobs.progress.JobProgress.of(input));
BackupsProcesses.requireSuccessfulJob(result);
switch(result.getStatus())
{
case "SUCCEEDED" -> succeeded.incrementCountAndAddPrimaryKey(run.getValue("id"));
@@ -30,6 +30,9 @@ import com.kingsrook.qbits.backups.verify.VerificationOutcome;
import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
@@ -85,7 +88,7 @@ public final class VerifyProcesses
{
List<QProcessMetaData> processes = new ArrayList<>();
processes.add(BackupsProcesses.permission(BackupsProcesses.etl(VERIFY, "Verify backup", BackupsTables.RUN, VerifyTransform.class, VerifyLoad.class,
"This checks that the backup's objects are all in storage, then reads every byte back, decrypts and decodes it, and compares counts and hashes with the manifest. Nothing is changed in the backup."),
"This checks that the backup's objects are all in storage, then reads every byte back, decrypts and decodes it, and compares counts and hashes with the manifest. Nothing is changed in the backup.").withSupplementalMetaData(BackupsProcesses.durability()),
BackupsPermissions.RUN));
processes.add(checkSlaProcess(BackupsRuntime.getConfig()));
return (processes);
@@ -220,12 +223,31 @@ public final class VerifyProcesses
@Override
public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
{
JobProgress progress = JobProgress.of(input);
progress.phases(JobPhase.of("verify", "Verify", 100)).phase("verify").total(input.getRecords().size(), "backups").done(0);
int done = 0;
String level = input.getValueString(FIELD_VERIFY_LEVEL) == null ? BackupVerifier.CONTENT : input.getValueString(FIELD_VERIFY_LEVEL);
String user = QContext.getQSession() == null || QContext.getQSession().getUser() == null ? "system" : QContext.getQSession().getUser().getIdReference();
for(QRecord run : input.getRecords())
{
JobChunkScope.checkBetweenPages();
progress.message("Verifying backup " + run.getValueInteger("id"));
List<VerificationOutcome> outcomes = new BackupVerifier().verifyThrough(run.getValueInteger("id"), level, "MANUAL", user);
VerificationOutcome last = outcomes.get(outcomes.size() - 1);
for(VerificationOutcome outcome : outcomes)
{
progress.info("Verification " + outcome.level() + ": " + outcome.status());
}
if(!last.passed())
{
progress.getSnapshot().getPhases().getFirst().setStatus(JobPhase.Status.FAILED);
progress.error("Backup verification " + last.status());
if(JobChunkScope.current().isPresent())
{
throw (new QException("Backup verification " + last.status()));
}
}
progress.done(++done);
switch(last.status())
{
case "PASSED" -> passed.incrementCountAndAddPrimaryKey(run.getValue("id"));
@@ -248,6 +270,8 @@ public final class VerifyProcesses
}
output.getRecords().add(run);
}
progress.getSnapshot().getPhases().getFirst().setStatus(failed.getCount() > 0 || errored.getCount() > 0 ? JobPhase.Status.FAILED : JobPhase.Status.DONE);
progress.flush();
}
@@ -49,6 +49,8 @@ import com.kingsrook.qqq.backend.core.actions.audits.AuditAction;
import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
import com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
@@ -373,6 +375,15 @@ public class RestoreEngine
}
catch(QException | RuntimeException e)
{
JobInterruptedException interrupted = JobInterruptedException.find(e);
if(interrupted != null)
{
if(interrupted.getReason() == JobInterruptedException.Reason.CANCEL)
{
catalog.update(BackupsTables.RESTORE_RUN, restoreRunId, Map.of("status", "CANCELLED", "endDate", BackupsRuntime.getClock().instant()));
}
throw (interrupted);
}
LOG.warn("Restore failed", e, logPair("restoreRunId", restoreRunId));
String message = e.getClass().getSimpleName() + ": " + e.getMessage();
try
@@ -447,6 +458,7 @@ public class RestoreEngine
}
BackupRunRequest backup = new BackupRunRequest()
.withProgress(new com.kingsrook.qqq.backend.core.jobs.progress.JobProgress((com.kingsrook.qqq.backend.core.actions.async.AsyncJobCallback) null))
.withSettings(settings)
.withScope(new ScopeResolver(QContext.getQInstance()).resolve(settings, rules))
.withTrigger("PRE_RESTORE")
@@ -507,10 +519,11 @@ public class RestoreEngine
** index. A conflicting file (different content at the target) is only replaced
** by a REPLACE restore.
*******************************************************************************/
private void restoreFiles(RestoreRequest request, Plan plan, RestorePass pass, RestoreOutcome outcome, Integer restoreRunId)
private void restoreFiles(RestoreRequest request, Plan plan, RestorePass pass, RestoreOutcome outcome, Integer restoreRunId) throws QException
{
for(RestorePass.FileWork work : pass.getFileWork())
{
JobChunkScope.checkBetweenPages();
if(work.conflict() && !RestoreRequest.MODE_REPLACE.equals(request.getMode()))
{
continue;
@@ -46,6 +46,7 @@ import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
@@ -200,6 +201,7 @@ public class RestorePass
for(String name : order)
{
JobChunkScope.checkBetweenPages();
restoreTable(name);
}
retryPass();
@@ -566,6 +568,7 @@ public class RestorePass
*******************************************************************************/
private void processPage(TableContext c, List<QRecord> page) throws QException
{
JobChunkScope.checkBetweenPages();
RestoreOutcome.TableOutcome o = c.outcome;
o.addRows(page.size());
@@ -17,16 +17,21 @@
package com.kingsrook.qbits.backups.run;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStream;
import java.io.Serializable;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Instant;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.UUID;
import java.util.function.Function;
@@ -42,11 +47,17 @@ import com.kingsrook.qbits.backups.files.FileReference;
import com.kingsrook.qbits.backups.files.FileReferenceProvider;
import com.kingsrook.qbits.backups.manifest.BackupManifest;
import com.kingsrook.qbits.backups.manifest.TableFingerprints;
import com.kingsrook.qbits.backups.metrics.BackupsMetrics;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.scope.ScopeResolver;
import com.kingsrook.qbits.backups.storage.BackupStorage;
import com.kingsrook.qbits.backups.storage.WriteOptions;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
import com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException;
import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
@@ -85,13 +96,19 @@ public class BackupEngine
/*******************************************************************************
** Run a backup. Never throws for a backup failure: the result carries the
** status and message (and the run row is FAILED).
** status and message (and the run row is FAILED). Cooperative job interrupts
** propagate as QException so the durable worker handles cancellation/lease loss.
*******************************************************************************/
public BackupRunResult execute(BackupRunRequest request)
public BackupRunResult execute(BackupRunRequest request) throws QException
{
JobProgress progress = request.getProgress();
if(!"prepare".equals(progress.getSnapshot().getCurrentPhase()))
{
prepare(progress);
}
BackupRunResult result = new BackupRunResult();
Instant started = request.getClock().instant();
String runKey = request.getResume() != null ? request.getResume().runKey() : RUN_KEY_FORMAT.format(started) + "-" + UUID.randomUUID().toString().substring(0, 4);
String runKey = request.getResume() != null ? request.getResume().runKey() : request.getRunKey() != null ? request.getRunKey() : RUN_KEY_FORMAT.format(started) + "-" + UUID.randomUUID().toString().substring(0, 4);
String planPrefix = request.getSettings().string("prefix", "qqq-backups");
String instancePrefix = trimSlashes(planPrefix) + "/" + request.getInstanceName() + "/";
String runPrefix = request.getResume() != null ? request.getResume().runPrefix() : instancePrefix + request.getSettings().planKey() + "/" + DATE_PATH.format(started) + "/" + runKey + "/";
@@ -139,27 +156,52 @@ public class BackupEngine
catch(Exception e)
{
result.setStatus("FAILED");
result.setErrorMessage("Could not create the run record: " + e.getMessage());
result.setErrorMessage("Could not create the run record (" + e.getClass().getSimpleName() + ")");
failProgress(progress, "Could not create the run record (" + e.getClass().getSimpleName() + ")");
return (result);
}
try
{
checkJob();
if(request.getResume() != null)
{
long complete = request.getResume().tables().values().stream().filter(ResumeState.TableState::complete).count();
progress.info("Resuming run " + result.getRunId() + ": " + complete + " tables already done");
}
doRun(request, result, runPrefix, instancePrefix, started);
completePhase(progress, JobPhase.Status.DONE);
}
catch(CancelledException e)
{
finish(request, result, "CANCELLED", "Cancelled", started);
progress.info("Backup cancelled");
}
catch(TimedOutException e)
{
finish(request, result, "TIMED_OUT", e.getMessage(), started);
failProgress(progress, e.getMessage());
}
catch(Exception e)
{
JobInterruptedException interrupted = JobInterruptedException.find(e);
if(interrupted != null)
{
if(interrupted.getReason() != JobInterruptedException.Reason.LEASE_LOST)
{
finish(request, result, interrupted.getReason() == JobInterruptedException.Reason.CANCEL ? "CANCELLED" : "ABANDONED", "Job interrupted", started);
}
throw (interrupted);
}
LOG.warn("Backup run failed", e, logPair("runKey", runKey));
// Provider and reader messages can contain credentials or row values.
failProgress(progress, scrub(e));
finish(request, result, "FAILED", scrub(e), started);
}
finally
{
progress.flush();
}
return (result);
}
@@ -170,6 +212,7 @@ public class BackupEngine
*******************************************************************************/
private void doRun(BackupRunRequest request, BackupRunResult result, String runPrefix, String instancePrefix, Instant started) throws Exception
{
JobProgress progress = request.getProgress();
QInstance qInstance = QContext.getQInstance();
BackupStorage storage = request.getStorage();
var settings = request.getSettings();
@@ -188,6 +231,7 @@ public class BackupEngine
throw (new IllegalStateException("The sensitive field policy INCLUDE_ENCRYPTED_ONLY requires client-side encryption."));
}
String consistency = consistencyFor(settings.string("consistency", "SNAPSHOT_PREFERRED"), result);
result.getWarnings().forEach(progress::warn);
catalog.update(BackupsTables.RUN, result.getRunId(), Map.of("consistencyAchieved", result.getConsistencyAchieved()));
//////////////////////////////////////////////////////////////////////////////////////
@@ -230,10 +274,15 @@ public class BackupEngine
long fileBytes = 0;
long fileBytesRef = 0;
try
progress.phase("tables").total(request.getScope().tables().size(), "tables").done(0).bytesDone(0);
try(PendingFiles pendingFiles = new PendingFiles())
{
for(ScopeResolver.ResolvedTable resolved : request.getScope().tables())
{
checkJob();
long tableStarted = System.nanoTime();
long tableBytesBefore = bytesStored;
progress.message("Backing up " + resolved.name() + " (" + (done + 1) + " of " + request.getScope().tables().size() + ")");
QTableMetaData table = qInstance.getTable(resolved.name());
String backendType = qInstance.getBackend(table.getBackendName()) == null ? null : qInstance.getBackend(table.getBackendName()).getBackendType();
@@ -242,6 +291,7 @@ public class BackupEngine
ResumeState.TableState prior = request.getResume() == null ? null : request.getResume().table(table.getName());
long[] tableWritten = { 0 };
List<TableArtifactWriter> writers = new ArrayList<>();
String tablePrefix = runPrefix + "tables/" + table.getName() + "/";
Map<TableArtifactWriter, Integer> lastArtifactIds = new LinkedHashMap<>();
@@ -277,6 +327,9 @@ public class BackupEngine
{
try
{
tableWritten[0] += part.bytesStored();
progress.bytesDone(tableBytesBefore + tableWritten[0]);
progress.info("Wrote " + part.key().substring(runPrefix.length()));
lastArtifactIds.put(writer, catalog.insertArtifact(result.getRunId(), "TABLE_DATA", table.getName(), part.partNumber(), part.key().substring(runPrefix.length()),
part.format(), part.compression(), part.encrypted(), part.rows(), part.bytesRaw(), part.bytesStored(), part.sha256Stored(), part.sha256Raw(), null,
part.firstKey(), part.lastKey(), TableFingerprints.of(table, backendType)));
@@ -296,6 +349,8 @@ public class BackupEngine
if(canContinue)
{
TableArtifactWriter writer = writers.get(0);
tableWritten[0] = prior.parts().stream().mapToLong(TableArtifactWriter.PartResult::bytesStored).sum();
progress.bytesDone(tableBytesBefore + tableWritten[0]);
writer.resumeFrom(prior.parts());
seedRowsHash(storage, runPrefix, prior.parts(), writer);
lastArtifactIds.put(writer, prior.artifactIds().get(prior.artifactIds().size() - 1));
@@ -368,8 +423,31 @@ public class BackupEngine
}
rows += counter[0];
for(FileReference ref : references)
pendingFiles.write(references);
Map<String, String> keyRange = new LinkedHashMap<>();
keyRange.put("min", allResults.isEmpty() ? null : allResults.get(0).firstKey());
keyRange.put("max", allResults.isEmpty() ? null : allResults.get(allResults.size() - 1).lastKey());
manifestTables.add(new BackupManifest.TableEntry(table.getName(), resolved.role().name(), table.getBackendName(), table.getPrimaryKeyField(),
fingerprint, TableFingerprints.CORE_ALGORITHM, TableFingerprints.descriptor(table, backendType), resolved.filterJson(), excluded, counter[0], rowsHash, keyRange, parts, consistency));
done++;
catalogProgress(request, result, done, rows, bytesRaw, bytesStored);
progress.done(done).bytesDone(bytesStored);
progress.info(String.format(Locale.US, "%s: %,d rows, %.1f MB in %.1f s", table.getName(), counter[0],
(bytesStored - tableBytesBefore) / 1_000_000.0, (System.nanoTime() - tableStarted) / 1_000_000_000.0));
if(request.getCancelRequested().getAsBoolean())
{
throw (new CancelledException());
}
}
progress.phase("media").total(pendingFiles.count, "files").done(0).bytesDone(bytesStored);
try(BufferedReader pending = pendingFiles.openReader())
{
String line;
while((line = pending.readLine()) != null)
{
FileReference ref = MAPPER.readValue(line, FileReference.class);
checkJob();
if(indexWriter == null)
{
indexWriter = new FileIndexWriter(storage, runPrefix + "files/index-00001.jsonl.gz", result.getRunKey(), fileLockMode, fileRetainUntil);
@@ -388,24 +466,16 @@ public class BackupEngine
case FileCopier.MISSING -> missing++;
default -> fileBytesRef += Math.max(0, entry.size());
}
progress.done(referenced).bytesDone(bytesStored + fileBytes)
.message("Media: " + copied + " copied, " + deduplicated + " already stored (" + referenced + " of " + pendingFiles.count + ")");
if(request.getCancelRequested().getAsBoolean())
{
throw (new CancelledException());
}
}
Map<String, String> keyRange = new LinkedHashMap<>();
keyRange.put("min", allResults.isEmpty() ? null : allResults.get(0).firstKey());
keyRange.put("max", allResults.isEmpty() ? null : allResults.get(allResults.size() - 1).lastKey());
manifestTables.add(new BackupManifest.TableEntry(table.getName(), resolved.role().name(), table.getBackendName(), table.getPrimaryKeyField(),
fingerprint, TableFingerprints.CORE_ALGORITHM, TableFingerprints.descriptor(table, backendType), resolved.filterJson(), excluded, counter[0], rowsHash, keyRange, parts, consistency));
done++;
catalogProgress(result, done, rows, bytesRaw, bytesStored);
if(request.getCancelRequested().getAsBoolean())
{
throw (new CancelledException());
}
}
progress.info("Media: " + referenced + " files, " + copied + " copied, " + deduplicated + " already stored, " + missing + " missing");
}
finally
{
@@ -415,10 +485,13 @@ public class BackupEngine
}
}
checkJob();
progress.phase("upload").message("Writing catalog and manifest");
List<String> indexKeys = new ArrayList<>();
if(indexWriter != null)
{
indexKeys.add("files/index-00001.jsonl.gz");
progress.info("Wrote files/index-00001.jsonl.gz");
checksums.add(indexWriter.sha256Stored() + " files/index-00001.jsonl.gz");
catalog.insertArtifact(result.getRunId(), "FILE_OBJECTS_INDEX", null, 1, "files/index-00001.jsonl.gz", "JSONL", "GZIP", request.getTransform().encrypts(),
indexWriter.rows(), indexWriter.bytesRaw(), indexWriter.bytesStored(), indexWriter.sha256Stored(), indexWriter.sha256Raw(), null, null, null, null);
@@ -431,7 +504,9 @@ public class BackupEngine
{
throw (new IllegalStateException(missing + " referenced files are missing at the source"));
}
result.getWarnings().add(missing + " referenced files were missing at the source");
String warning = missing + " referenced files were missing at the source";
result.getWarnings().add(warning);
progress.warn(warning);
}
/////////////////////////////////////////////////////////
@@ -439,6 +514,7 @@ public class BackupEngine
/////////////////////////////////////////////////////////
byte[] checksumBytes = (String.join("\n", checksums) + "\n").getBytes(StandardCharsets.UTF_8);
putBytes(storage, runPrefix + "checksums.sha256", checksumBytes, result.getRunKey(), lockMode, retainUntil);
progress.info("Wrote checksums.sha256");
catalog.insertArtifact(result.getRunId(), "CHECKSUMS", null, 1, "checksums.sha256", "TEXT", "NONE", false, checksums.size(), checksumBytes.length,
checksumBytes.length, HashingOutputStream.sha256Hex(checksumBytes), HashingOutputStream.sha256Hex(checksumBytes), null, null, null, null);
@@ -463,6 +539,7 @@ public class BackupEngine
{
throw (new IllegalStateException("The manifest was written but is not visible in the bucket"));
}
progress.info("Wrote manifest.json");
catalog.insertArtifact(result.getRunId(), "MANIFEST", null, 1, "manifest.json", "JSON", "NONE", false, manifestTables.size(), manifestBytes.length,
manifestBytes.length, HashingOutputStream.sha256Hex(manifestBytes), HashingOutputStream.sha256Hex(manifestBytes), null, null, null, null);
@@ -501,6 +578,58 @@ public class BackupEngine
/*******************************************************************************
** Declare the work before scope resolution. No byte estimate requires a scan.
*******************************************************************************/
public static void prepare(JobProgress progress)
{
progress.phases(JobPhase.of("prepare", "Resolve scope", 2), JobPhase.of("tables", "Back up tables", 75),
JobPhase.of("media", "Copy media files", 18), JobPhase.of("upload", "Write catalog and manifest", 5));
progress.phase("prepare").message("Resolving scope");
}
/*******************************************************************************
** Complete or fail the current phase; the core has no phase-finish method.
*******************************************************************************/
private static void completePhase(JobProgress progress, JobPhase.Status status)
{
progress.getSnapshot().getPhases().stream().filter(p -> p.getStatus() == JobPhase.Status.RUNNING).forEach(p -> p.setStatus(status));
progress.flush();
}
/*******************************************************************************
** Log only a controlled failure description, never provider or row text.
*******************************************************************************/
private static void failProgress(JobProgress progress, String message)
{
completePhase(progress, JobPhase.Status.FAILED);
progress.error(message);
}
/*******************************************************************************
** Bridge checked job interrupts through the row consumer, preserving the cause
** so execute can propagate it back to the durable worker.
*******************************************************************************/
private static void checkJob()
{
try
{
JobChunkScope.checkBetweenPages();
}
catch(QException e)
{
throw (new IllegalStateException(e));
}
}
/*******************************************************************************
** Work out the consistency achieved. Until the core consistent-read hook ships
** (QQQ core PR feature/consistent-read-transactions, next 4.1.x RC), no backend
@@ -563,6 +692,7 @@ public class BackupEngine
*******************************************************************************/
private void heartbeat(BackupRunRequest request, BackupRunResult result, Instant deadline, String tableName, long rowsSoFar)
{
checkJob();
Instant now = request.getClock().instant();
if(request.getCancelRequested().getAsBoolean())
{
@@ -579,6 +709,7 @@ public class BackupEngine
catch(Exception e)
{
LOG.warn("Could not record a heartbeat", e);
request.getProgress().warn("Could not record a heartbeat");
}
}
@@ -587,7 +718,7 @@ public class BackupEngine
/*******************************************************************************
**
*******************************************************************************/
private void catalogProgress(BackupRunResult result, int done, long rows, long bytesRaw, long bytesStored)
private void catalogProgress(BackupRunRequest request, BackupRunResult result, int done, long rows, long bytesRaw, long bytesStored)
{
try
{
@@ -601,6 +732,7 @@ public class BackupEngine
catch(Exception e)
{
LOG.warn("Could not record progress", e);
request.getProgress().warn("Could not record progress");
}
}
@@ -618,6 +750,7 @@ public class BackupEngine
catch(Exception e)
{
LOG.warn("Could not finish the run record", e);
request.getProgress().warn("Could not finish the run record");
result.setStatus(status);
result.setErrorMessage(message);
}
@@ -642,6 +775,7 @@ public class BackupEngine
result.setStatus(status);
result.setErrorMessage(message);
signal(request, result, status, message);
BackupsMetrics.invalidate();
}
@@ -694,13 +828,12 @@ public class BackupEngine
/*******************************************************************************
** A message safe to store: the exception type and message, never row values (the
** engine never puts values in messages), truncated.
** Provider exceptions may contain credentials or row values. Persist only the
** failure type; detailed diagnostics stay in the server log.
*******************************************************************************/
private static String scrub(Exception e)
{
String message = e.getClass().getSimpleName() + ": " + e.getMessage();
return (message.length() > 2000 ? message.substring(0, 2000) : message);
return ("Backup failed (" + e.getClass().getSimpleName() + "); see server logs for details.");
}
@@ -769,6 +902,11 @@ public class BackupEngine
{
throw (new StopReadingException());
}
checkJob();
if(request.getCancelRequested().getAsBoolean())
{
throw (new CancelledException());
}
for(FileReferenceProvider provider : request.getFileProviders())
{
references.addAll(provider.referencesIn(table, record));
@@ -854,6 +992,75 @@ public class BackupEngine
/***************************************************************************
** Spool references locally between table export and media copy. Keeping the
** whole run's references in heap would make memory grow with every table.
** The temporary file is removed on success, failure and cancellation.
***************************************************************************/
private static class PendingFiles implements AutoCloseable
{
private final Path path;
private final BufferedWriter writer;
private long count;
/*******************************************************************************
** Create a private temporary spool, outside the stored backup artifacts.
*******************************************************************************/
PendingFiles() throws IOException
{
path = Files.createTempFile("qqq-backup-references-", ".jsonl");
writer = Files.newBufferedWriter(path, StandardCharsets.UTF_8);
}
/*******************************************************************************
** Append this table's references without retaining earlier tables in memory.
*******************************************************************************/
void write(List<FileReference> references) throws IOException
{
for(FileReference reference : references)
{
writer.write(MAPPER.writeValueAsString(reference));
writer.newLine();
count++;
}
}
/*******************************************************************************
** Finish the spool and stream it back during the media phase.
*******************************************************************************/
BufferedReader openReader() throws IOException
{
writer.close();
return (Files.newBufferedReader(path, StandardCharsets.UTF_8));
}
/*******************************************************************************
** Always release the local references, including on exceptions.
*******************************************************************************/
@Override
public void close() throws IOException
{
try
{
writer.close();
}
finally
{
Files.deleteIfExists(path);
}
}
}
/***************************************************************************
** The per-run file index, written as it goes: one JSON line per reference.
***************************************************************************/
@@ -30,6 +30,8 @@ import com.kingsrook.qbits.backups.BackupsRuntime;
import com.kingsrook.qbits.backups.alerts.BackupAlertSink;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
@@ -85,7 +87,28 @@ public class BackupRunCoordinator
*******************************************************************************/
public BackupRunResult runPlan(QRecord plan, String trigger, java.util.function.Consumer<BackupRunRequest> customizer) throws QException
{
BackupRunRequest request = BackupRunFactory.forPlan(plan, trigger);
return (runPlan(plan, trigger, JobProgress.current(), customizer));
}
/*******************************************************************************
** Share the step reporter and reuse a durable job's run on an explicit retry.
** NONE idempotency prevents automatic replay of non-atomic external writes.
*******************************************************************************/
public BackupRunResult runPlan(QRecord plan, String trigger, JobProgress progress, java.util.function.Consumer<BackupRunRequest> customizer) throws QException
{
BackupEngine.prepare(progress);
BackupRunRequest request;
try
{
request = BackupRunFactory.forPlan(plan, trigger).withProgress(progress);
}
catch(QException | RuntimeException e)
{
progress.error("Could not resolve backup scope (" + e.getClass().getSimpleName() + ")");
throw (new QException("Could not resolve backup scope", e));
}
if(customizer != null)
{
customizer.accept(request);
@@ -97,21 +120,72 @@ public class BackupRunCoordinator
BackupRunResult result = new BackupRunResult();
result.setStatus("FAILED");
result.setErrorMessage(refusal);
progress.error(refusal);
return (result);
}
Claim claim = claim(plan.getValueInteger("id"), null, trigger, request.getRequestedBy(), true);
JobChunkScope.ChunkIdentity identity = JobChunkScope.current().map(JobChunkScope::getIdentity).orElse(null);
Integer existingRunId = null;
if(identity != null)
{
String key = "job-" + identity.jobId() + "-plan-" + plan.getValueInteger("id");
request.withRunKey(key);
List<QRecord> previous = catalog.where(BackupsTables.RUN, "runKey", key);
if(!previous.isEmpty())
{
QRecord run = previous.get(0);
existingRunId = run.getValueInteger("id");
if(run.getValueString("manifestKey") != null)
{
progress.info("Run " + existingRunId + " already has a manifest; nothing was rewritten. Use Verify backup to recheck it.");
BackupRunResult result = new BackupRunResult();
result.setRunId(existingRunId);
result.setRunKey(key);
result.setRunPrefix(run.getValueString("runPrefix"));
result.setManifestKey(run.getValueString("manifestKey"));
result.setStatus(run.getValueString("status"));
return (result);
}
if(run.getValueString("runPrefix") != null)
{
// Only the same durable job can recover its interrupted RUNNING row.
if("RUNNING".equals(run.getValueString("status")))
{
catalog.update(BackupsTables.RUN, existingRunId, Map.of("status", "ABANDONED"));
run.setValue("status", "ABANDONED");
}
String why = BackupResumePlanner.whyNotResumable(run, plan);
if(why != null)
{
progress.error(why);
throw (new QException(why));
}
request.withResume(BackupResumePlanner.plan(run));
}
}
}
Claim claim = claim(plan.getValueInteger("id"), existingRunId, trigger, request.getRequestedBy(), identity == null, request.getRunKey());
try
{
if(claim.outcome != Outcome.STARTED)
{
progress.warn(claim.message);
if(identity != null)
{
if(claim.outcome == Outcome.QUEUED)
{
catalog.update(BackupsTables.RUN, claim.runId, Map.of("status", "FAILED", "endDate", now(), "errorMessage", claim.message));
}
progress.error("Backup could not start: " + claim.message);
throw (new QException("Backup could not start: " + claim.message));
}
return (claim.toResult());
}
return (execute(request, claim.runId));
}
finally
{
if(claim.outcome == Outcome.STARTED)
if(claim.outcome == Outcome.STARTED && JobChunkScope.current().isEmpty())
{
drainQueue();
}
@@ -133,7 +207,10 @@ public class BackupRunCoordinator
}
finally
{
drainQueue();
if(JobChunkScope.current().isEmpty())
{
drainQueue();
}
}
}
@@ -144,6 +221,16 @@ public class BackupRunCoordinator
** message) when it cannot be resumed, or when a run of the plan is live.
*******************************************************************************/
public BackupRunResult resume(QRecord run) throws QException
{
return (resume(run, JobProgress.current()));
}
/*******************************************************************************
** Resume using the caller's reporter (including an inline async callback).
*******************************************************************************/
public BackupRunResult resume(QRecord run, JobProgress progress) throws QException
{
QRecord plan = run.getValueInteger("backupPlanId") == null ? null : catalog.get(BackupsTables.PLAN, run.getValueInteger("backupPlanId")).orElse(null);
String why = BackupResumePlanner.whyNotResumable(run, plan);
@@ -156,7 +243,7 @@ public class BackupRunCoordinator
return (refused);
}
BackupRunRequest request = BackupRunFactory.forPlan(plan, "MANUAL");
BackupRunRequest request = BackupRunFactory.forPlan(plan, "MANUAL").withProgress(progress);
Claim claim = claim(plan.getValueInteger("id"), run.getValueInteger("id"), run.getValueString("trigger"), request.getRequestedBy(), false);
if(claim.outcome != Outcome.STARTED)
{
@@ -173,7 +260,10 @@ public class BackupRunCoordinator
}
finally
{
drainQueue();
if(JobChunkScope.current().isEmpty())
{
drainQueue();
}
}
}
@@ -238,7 +328,7 @@ public class BackupRunCoordinator
/*******************************************************************************
** Run the engine on a claimed run row, then return its result.
*******************************************************************************/
private BackupRunResult execute(BackupRunRequest request, Integer runId)
private BackupRunResult execute(BackupRunRequest request, Integer runId) throws QException
{
request.withExistingRunId(runId).withCancelRequested(() -> catalog.isCancelRequested(runId));
return (new BackupEngine().execute(request));
@@ -250,6 +340,16 @@ public class BackupRunCoordinator
** The lease: insert (or reuse) the run row and decide.
*******************************************************************************/
private Claim claim(Integer planId, Integer existingRunId, String trigger, String requestedBy, boolean queueIfFull) throws QException
{
return (claim(planId, existingRunId, trigger, requestedBy, queueIfFull, null));
}
/*******************************************************************************
** Persist the job identity in the initial claim, before any external writes.
*******************************************************************************/
private Claim claim(Integer planId, Integer existingRunId, String trigger, String requestedBy, boolean queueIfFull, String runKey) throws QException
{
synchronized(LEASE_LOCK)
{
@@ -261,7 +361,7 @@ public class BackupRunCoordinator
{
Map<String, Serializable> values = new LinkedHashMap<>();
values.put("backupPlanId", planId);
values.put("runKey", "queued-" + UUID.randomUUID().toString().substring(0, 8));
values.put("runKey", runKey != null ? runKey : "queued-" + UUID.randomUUID().toString().substring(0, 8));
values.put("trigger", trigger);
values.put("requestedByUserId", requestedBy);
values.put("status", QUEUED);
@@ -26,6 +26,7 @@ import com.kingsrook.qbits.backups.files.FileReferenceProvider;
import com.kingsrook.qbits.backups.scope.PlanSettings;
import com.kingsrook.qbits.backups.scope.ScopeResolver;
import com.kingsrook.qbits.backups.storage.BackupStorage;
import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
/*******************************************************************************
@@ -54,6 +55,50 @@ public class BackupRunRequest
private java.util.function.Function<String, BackupStorage> sourceResolver;
private Integer existingRunId;
private ResumeState resume;
private JobProgress progress = JobProgress.current();
private String runKey;
/*******************************************************************************
** The reporter shared by the process, coordinator and engine.
*******************************************************************************/
public JobProgress getProgress()
{
return (progress);
}
/*******************************************************************************
** Supply the process reporter, including its inline async callback.
*******************************************************************************/
public BackupRunRequest withProgress(JobProgress progress)
{
this.progress = progress;
return (this);
}
/*******************************************************************************
** A stable identity for an explicitly retried durable job.
*******************************************************************************/
public String getRunKey()
{
return (runKey);
}
/*******************************************************************************
** Set the durable job's backup identity before creating any artifacts.
*******************************************************************************/
public BackupRunRequest withRunKey(String runKey)
{
this.runKey = runKey;
return (this);
}
@@ -80,7 +80,7 @@ public class BackupScheduledRunProcess implements BackendStep
.withInputData(new QFunctionInputMetaData().withRecordListMetaData(new QRecordListMetaData().withTableName(BackupsTables.PLAN)))
.withCode(new QCodeReference(BackupScheduledRunProcess.class)));
process.setPermissionRules(BackupsTables.permissionRules());
return (BackupsProcesses.permission(process, BackupsPermissions.RUN));
return (BackupsProcesses.permission(process.withSupplementalMetaData(BackupsProcesses.durability()), BackupsPermissions.RUN));
}
@@ -147,11 +147,12 @@ public class BackupScheduledRunProcess implements BackendStep
}
BackupsRuntime.getAlertSink().resolve("backups.plan.owner-invalid", "backups.plan.owner-invalid:" + plan.getValueString("planKey"));
BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "SCHEDULED");
BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "SCHEDULED", com.kingsrook.qqq.backend.core.jobs.progress.JobProgress.of(input), null);
output.addValue("status", result.getStatus());
output.addValue("runId", result.getRunId());
if(!"SUCCEEDED".equals(result.getStatus()) && !"SUCCEEDED_WITH_WARNINGS".equals(result.getStatus()))
{
BackupsProcesses.requireSuccessfulJob(result);
LOG.info("A scheduled backup did not succeed", logPair("plan", plan.getValueString("planKey")), logPair("status", result.getStatus()), logPair("message", result.getErrorMessage()));
}
}
@@ -35,6 +35,7 @@ import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.run.BackupCatalog;
import com.kingsrook.qbits.backups.storage.BackupStorage;
import com.kingsrook.qbits.backups.storage.StoredObjectInfo;
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.model.actions.tables.query.QCriteriaOperator;
@@ -95,8 +96,11 @@ public class SlaChecker
private static final Set<String> FAILED_STATUSES = Set.of("FAILED", "TIMED_OUT", "ABANDONED");
private static final Set<String> FINISHED_STATUSES = Set.of("SUCCEEDED", "SUCCEEDED_WITH_WARNINGS", "FAILED", "TIMED_OUT", "ABANDONED");
private static final Map<String, Instant> MANIFEST_OK_AT = new ConcurrentHashMap<>();
private static final Map<String, Instant> LAST_AUDIT_AT = new ConcurrentHashMap<>();
private static final Map<String, Instant> MANIFEST_OK_AT = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
"throttles this node's storage HEAD of a manifest to once per MANIFEST_CACHE; it caches storage state, not editable data, "
+ "and a pruned run is skipped by its prunedDate, so each node is as fresh as a single node");
private static final Map<String, Instant> LAST_AUDIT_AT = ClusterCaches.nodeLocal(new ConcurrentHashMap<>(),
"throttles this node's bucket audit to once per BUCKET_AUDIT_INTERVAL; another node auditing again is only repeated work");
private final BackupCatalog catalog = new BackupCatalog();
@@ -47,6 +47,8 @@ import com.kingsrook.qbits.backups.storage.BackupStorageException;
import com.kingsrook.qbits.backups.storage.StoredObjectInfo;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
import com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
@@ -153,6 +155,10 @@ public class BackupVerifier
}
status = findings.status();
}
catch(JobInterruptedException e)
{
throw (e);
}
catch(VerifyException | BackupStorageException | QException e)
{
status = "ERROR";
@@ -188,7 +194,7 @@ public class BackupVerifier
/*******************************************************************************
** STORAGE level: head every object, including every file-store object.
*******************************************************************************/
void storageLevel(RunBundle bundle, VerifyFindings findings)
void storageLevel(RunBundle bundle, VerifyFindings findings) throws QException
{
BackupManifest manifest = bundle.manifest();
BackupStorage storage = bundle.storage();
@@ -199,6 +205,7 @@ public class BackupVerifier
for(BackupManifest.TableEntry table : manifest.tables())
{
JobChunkScope.checkBetweenPages();
for(BackupManifest.Part part : table.parts() == null ? List.<BackupManifest.Part>of() : table.parts())
{
findings.objectsChecked++;
@@ -231,6 +238,7 @@ public class BackupVerifier
int missingFiles = 0;
for(RunBundle.FileIndexEntry entry : bundle.distinctStoredFiles(findings))
{
JobChunkScope.checkBetweenPages();
findings.objectsChecked++;
findings.filesChecked++;
Optional<StoredObjectInfo> head = storage.head(entry.casKey());
@@ -262,6 +270,7 @@ public class BackupVerifier
for(BackupManifest.TableEntry entry : manifest.tables())
{
JobChunkScope.checkBetweenPages();
QTableMetaData table = qInstance.getTable(entry.name());
JsonlRowCodec codec = null;
boolean roundTrip = false;
@@ -291,7 +300,7 @@ public class BackupVerifier
/*******************************************************************************
** All parts of one table.
*******************************************************************************/
private void verifyTable(RunBundle bundle, BackupManifest.TableEntry entry, JsonlRowCodec codec, boolean roundTrip, VerifyFindings findings)
private void verifyTable(RunBundle bundle, BackupManifest.TableEntry entry, JsonlRowCodec codec, boolean roundTrip, VerifyFindings findings) throws QException
{
MessageDigest rowsDigest = HashingOutputStream.newSha256();
long rows = 0;
@@ -339,6 +348,11 @@ public class BackupVerifier
}
catch(QException e)
{
JobInterruptedException interrupted = JobInterruptedException.find(e);
if(interrupted != null)
{
throw (interrupted);
}
findings.mismatch("The part " + part.key() + " could not be read: " + e.getMessage());
}
}
@@ -451,6 +465,7 @@ public class BackupVerifier
for(RunBundle.FileIndexEntry entry : bundle.distinctStoredFiles(findings))
{
JobChunkScope.checkBetweenPages();
QRecord fileObject = catalog.where(BackupsTables.FILE_OBJECT, "sha256", entry.sha256()).stream().filter(r -> r.getValue("deletedDate") == null).findFirst().orElse(null);
if(fileObject != null && "PASSED".equals(fileObject.getValueString("lastVerifyStatus"))
&& fileObject.getValueInstant("lastVerifiedDate") != null && fileObject.getValueInstant("lastVerifiedDate").isAfter(cutoff))
@@ -29,6 +29,7 @@ import java.util.zip.GZIPInputStream;
import com.kingsrook.qbits.backups.codec.HashingOutputStream;
import com.kingsrook.qbits.backups.manifest.BackupManifest;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
/*******************************************************************************
@@ -80,6 +81,7 @@ final class PartScanner
*******************************************************************************/
static Scan scan(RunBundle bundle, BackupManifest.Part part, LineConsumer consumer) throws IOException, QException
{
JobChunkScope.checkBetweenPages();
MessageDigest storedDigest = HashingOutputStream.newSha256();
MessageDigest rawDigest = HashingOutputStream.newSha256();
long lines = 0;
@@ -102,6 +104,10 @@ final class PartScanner
String line;
while((line = reader.readLine()) != null)
{
if(lines % 1000 == 0)
{
JobChunkScope.checkBetweenPages();
}
lines++;
consumer.accept(line);
}
+1 -1
View File
@@ -1,5 +1,5 @@
bundle: qbit-backups
version: 0.1.0-SNAPSHOT
version: ${project.version}
layer: QBIT
entries:
- key: table:backups_plan;field:name
@@ -0,0 +1 @@
version=${project.version}
@@ -0,0 +1,164 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.backups;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;
import com.kingsrook.qbits.backups.metrics.BackupsMetrics;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.run.BackupCatalog;
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.QInstance;
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 Backups caches are correct on several nodes: every static cache is
** registered with ClusterCaches, and the cached metrics are cleared on every
** node when a plan changes.
*******************************************************************************/
class BackupsClusterCachesTest
{
/** 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 backed-up test instance, connected to the wire as node-a.
*******************************************************************************/
@BeforeEach
void setUp() throws Exception
{
VerifyTestSupport.setUp(1);
QInstance instance = QContext.getQInstance();
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(instance);
}
/*******************************************************************************
**
*******************************************************************************/
@AfterEach
void afterEach()
{
ClusterBus.close(QContext.getQInstance());
WIRE.clear();
QNodeRoles.set(null);
}
/*******************************************************************************
** Every static Memoization in the QBit is registered (shared or node-local).
*******************************************************************************/
@Test
void testNoUnregisteredStaticCaches() throws Exception
{
assertThat(ClusterCacheAudit.unregisteredStaticCaches("com.kingsrook.qbits.backups")).isEmpty();
}
/*******************************************************************************
** Another node's invalidation clears this node's cached metrics.
*******************************************************************************/
@Test
void testRemoteInvalidationClearsMetricsCache() throws Exception
{
String cached = BackupsMetrics.renderCached();
assertThat(BackupsMetrics.renderCached()).isSameAs(cached);
ClusterEvent remote = new ClusterEvent(ClusterEvent.TYPE_MEMO_INVALIDATE, BackupsMetrics.CACHE_NAME, null);
remote.setOriginNodeId("node-b");
WIRE.forEach(listener -> listener.accept(remote));
assertThat(BackupsMetrics.renderCached()).isNotSameAs(cached);
}
/*******************************************************************************
** A plan edit here clears the metrics here and tells the other nodes.
*******************************************************************************/
@Test
void testPlanEditPublishesMetricsInvalidation() throws Exception
{
QRecord plan = VerifyTestSupport.plan(Map.of());
String cached = BackupsMetrics.renderCached();
List<ClusterEvent> sent = new CopyOnWriteArrayList<>();
WIRE.add(sent::add);
new BackupCatalog().update(BackupsTables.PLAN, plan.getValueInteger("id"), Map.of("slaHours", 48));
assertThat(BackupsMetrics.renderCached()).isNotSameAs(cached);
assertThat(sent).anySatisfy(event ->
{
assertThat(event.getType()).isEqualTo(ClusterEvent.TYPE_MEMO_INVALIDATE);
assertThat(event.getName()).isEqualTo(BackupsMetrics.CACHE_NAME);
assertThat(event.getOriginNodeId()).isEqualTo("node-a");
});
}
}
@@ -18,6 +18,7 @@ package com.kingsrook.qbits.backups;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.metadata.help.bundles.HelpBundles;
import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitConfigValidationException;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
@@ -40,12 +41,16 @@ class BackupsQBitProducerTest
BackupsQBitConfig config = new BackupsQBitConfig().withBackendName("rdbms");
BackupsQBitProducer producer = new BackupsQBitProducer().withQBitConfig(config).withNamespace("backups");
String artifactVersion = System.getProperty("qbit.artifactVersion");
assertThat(artifactVersion).isNotBlank();
assertThat(producer.getQBitMetaData().getVersion()).isEqualTo(artifactVersion).isEqualTo(BackupsQBitProducer.VERSION);
assertThat(producer.isEnabled()).isTrue();
assertThat(producer.getQBitConfig()).isSameAs(config);
assertThat(producer.getNamespace()).isEqualTo("backups");
assertThat(config.getDefaultBackendNameForTables()).isEqualTo("rdbms");
producer.produce(qInstance);
assertThat(HelpBundles.of(qInstance).getBundles().get(0).getVersion()).isEqualTo(artifactVersion).isEqualTo(BackupsQBitProducer.VERSION);
assertThat(qInstance.getQBits()).hasSize(1);
assertThat(qInstance.getQBits().values().iterator().next().getArtifactId()).isEqualTo("qbit-backups");
}
@@ -79,4 +84,26 @@ class BackupsQBitProducerTest
assertThat(BackupsQBitConfig.PERMISSION_BASE_NAME).isEqualTo("backups");
}
/*******************************************************************************
** Long operations always enqueue, with no unsafe automatic external replay.
*******************************************************************************/
@Test
void testDurableProcesses() throws Exception
{
BackupsTestUtils.init();
for(String name : java.util.List.of("backupRunPlan", "backupRunScheduledPlan", "backupResumeRun", "backupRestore", "backupRestoreTable", "backupVerify"))
{
var process = com.kingsrook.qqq.backend.core.context.QContext.getQInstance().getProcess(name);
assertThat(process).as(name).isNotNull();
var durability = com.kingsrook.qqq.backend.core.jobs.DurableProcessMetaData.of(process);
assertThat(durability).as(name).isNotNull();
assertThat(durability.getAlwaysDurable()).isTrue();
assertThat(durability.getEligible()).isTrue();
assertThat(durability.getIdempotency()).isEqualTo(com.kingsrook.qqq.backend.core.jobs.model.JobIdempotency.NONE);
assertThat(durability.getChunkSize()).isEqualTo(1);
}
}
}
@@ -88,7 +88,7 @@ class BackupsHomeTest
void testEmptyStates()
{
String html = BackupsHomeView.render(new BackupsHomeData(List.of(), List.of(), List.of(), List.of(), List.of(), Instant.now()), "/app");
assertThat(html).contains("No backup plans yet").contains("Nothing to check yet").contains("No test restore yet")
assertThat(html).contains("No backup plans yet").contains("Setup: <a").contains("No plan: nothing is backed up.").doesNotContain("Nothing needs attention").contains("No test restore yet")
.contains("Nothing has run yet").contains("Trends appear after the first successful run").contains("Retention starts once runs exist");
assertThat(BackupsHomeView.renderHealthCard(new BackupsHomeData(List.of(), List.of(), List.of(), List.of(), List.of(), Instant.now()), "/app")).contains("No backup plans yet");
}
@@ -121,6 +121,45 @@ class BackupsHomeTest
/*******************************************************************************
** The Setup card and the Needs attention card never contradict each other: a plan
** whose storage aliases do not resolve is a Setup fault, and it is listed under
** Needs attention too, with the same count.
*******************************************************************************/
@Test
void testUnresolvedAliasAppearsInSetupAndNeedsAttention() throws Exception
{
BackupsRuntime.getConfig().withStorageResolver(alias -> BackupsTestUtils.BACKUPS);
new BackupCatalog().insert(BackupsTables.PLAN, new LinkedHashMap<>(Map.of("name", "Offsite", "planKey", "offsite", "isActive", true, "scopeType", "INSTANCE",
"storageBucketAlias", "backupsAws", "pruneBucketAlias", "backupsAwsPrune")));
//////////////////////////////////////////////////////////////////////////////
// the aliases resolved when the plan was saved; now the bucket is gone //
//////////////////////////////////////////////////////////////////////////////
BackupsRuntime.getConfig().withStorageResolver(alias -> "backups".equals(alias) ? BackupsTestUtils.BACKUPS : null);
BackupsHomeData data = BackupsHomeData.load();
List<BackupsSetupStatus.Item> setupTodo = BackupsSetupStatus.load(data.plans()).stream().filter(i -> i.state() == BackupsSetupStatus.State.TODO).toList();
assertThat(setupTodo).extracting(BackupsSetupStatus.Item::label).contains("Storage");
String html = BackupsHomeView.render(data, "/app");
String attention = html.substring(html.indexOf(">Needs attention<"));
attention = attention.substring(0, attention.indexOf("All runs"));
assertThat(attention).doesNotContain("Nothing needs attention").contains("backupsAws").contains("backupsAwsPrune");
///////////////////////////////////////////////////////////////////////////////
// every Setup TODO is listed under Needs attention (counts agree: 2 and 2) //
///////////////////////////////////////////////////////////////////////////////
assertThat(html).contains(setupTodo.size() + " items need attention.");
for(BackupsSetupStatus.Item item : setupTodo)
{
assertThat(attention).contains("Setup: <a").contains(item.label());
}
assertThat(attention.split("<li>", -1).length - 1).isEqualTo(setupTodo.size() + data.plansNeedingAttention().size() + data.failedRuns().size());
}
/*******************************************************************************
** The links follow the configured table path prefix.
*******************************************************************************/
@@ -0,0 +1,495 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.backups.health;
import java.io.InputStream;
import java.io.OutputStream;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import com.kingsrook.qbits.backups.BackupsRuntime;
import com.kingsrook.qbits.backups.BackupsTestUtils;
import com.kingsrook.qbits.backups.VerifyTestSupport;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.plan.DestinationProbe;
import com.kingsrook.qbits.backups.run.BackupCatalog;
import com.kingsrook.qbits.backups.storage.InMemoryBackupStorage;
import com.kingsrook.qbits.backups.storage.StoredObjectInfo;
import com.kingsrook.qbits.backups.storage.WriteOptions;
import com.kingsrook.qqq.backend.core.actions.customizers.QCodeLoader;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.health.HealthCategory;
import com.kingsrook.qqq.backend.core.health.HealthCheck;
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 com.kingsrook.qqq.backend.core.model.session.QSession;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/*******************************************************************************
** Central registration, migrated SLA rules, read-only probes and secret safety.
*******************************************************************************/
class BackupsHealthChecksTest
{
private static final String SECRET = "S3NT1NEL-secret";
private final BackupCatalog catalog = new BackupCatalog();
private ReadOnlyStorage storage;
/*******************************************************************************
** All probes use memory storage; the resolver represents credential injection.
*******************************************************************************/
@BeforeEach
void setUp() throws Exception
{
VerifyTestSupport.setUp(3);
storage = new ReadOnlyStorage(SECRET);
BackupsRuntime.getConfig().withStorageResolver(alias ->
{
assertThat(alias).as("never resolve prune credentials").isNotEqualTo("prune");
return (storage);
});
}
/*******************************************************************************
** Use the actual producer registry, including its reflective supplier loading.
*******************************************************************************/
@Test
void testRegistrationAndDynamicPlans() throws Exception
{
HealthChecksMetaData registry = HealthChecksMetaData.ofOrWithNew(QContext.getQInstance());
assertThat(registry.getChecks()).extracting(HealthCheckMetaData::getName).containsExactly("backups.sla");
assertThat(registry.getSuppliers()).singleElement().satisfies(supplier ->
{
assertThat(supplier.getNamePrefix()).isEqualTo("backups.destination");
assertThat(supplier.getComponent()).isEqualTo("backups");
});
QRecord active = VerifyTestSupport.plan(Map.of());
VerifyTestSupport.plan(Map.of("name", "Off", "planKey", "off", "isActive", false));
List<HealthCheckMetaData> checks = registry.resolveAll(QContext.getQInstance());
assertThat(checks).hasSize(2).allSatisfy(check ->
{
assertThat(check.getComponent()).isEqualTo("backups");
assertThat(check.getKind()).isEqualTo(HealthCheckKind.DEPENDENCY);
assertThat(check.getCategory()).isEqualTo(HealthCategory.STORAGE);
assertThat(check.getIncludeInReadiness()).isFalse();
assertThat(check.getName()).matches("[a-z0-9]+([._-][a-z0-9]+)*");
});
assertThat(checks.get(0).getIntervalSeconds()).isEqualTo(900);
assertThat(checks.get(0).getTimeoutMs()).isEqualTo(5000);
assertThat(QCodeLoader.getAdHoc(HealthCheck.class, checks.get(0).getCheck()).run(input()).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
HealthCheckMetaData destination = checks.get(1);
assertThat(destination.getName()).isEqualTo("backups.destination." + active.getValueInteger("id"));
assertThat(destination.getIntervalSeconds()).isEqualTo(86400);
assertThat(destination.getTimeoutMs()).isEqualTo(30000);
assertThat(destination.getCheckInstance().run(input()).getStatus()).isEqualTo(HealthStatus.UP);
catalog.update(BackupsTables.PLAN, active.getValueInteger("id"), Map.of("isActive", false));
assertThat(destination.getCheckInstance().run(input()).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
assertThat(registry.resolveAll(QContext.getQInstance())).hasSize(1);
VerifyTestSupport.plan(Map.of("name", "New", "planKey", "new"));
assertThat(registry.resolveAll(QContext.getQInstance())).hasSize(2);
}
/*******************************************************************************
** Carry over the indicator's empty, critical, ordinary and inactive cases.
*******************************************************************************/
@Test
void testSlaUpDownAndDegraded() throws Exception
{
BackupsSlaHealthCheck check = new BackupsSlaHealthCheck();
HealthCheckOutcome empty = check.run(input());
assertThat(empty.getStatus()).isEqualTo(HealthStatus.UP);
assertThat(empty.getDetails().getValues()).containsEntry("activePlans", 0);
QRecord critical = VerifyTestSupport.plan(Map.of("critical", true));
QRecord ordinary = VerifyTestSupport.plan(Map.of("name", "Other", "planKey", "other"));
QRecord inactive = VerifyTestSupport.plan(Map.of("name", "Off", "planKey", "off", "isActive", false, "critical", true));
setStatus(inactive, "LATE");
setStatus(critical, "OK");
setStatus(ordinary, "OK");
assertThat(check.run(input()).getStatus()).isEqualTo(HealthStatus.UP);
for(String status : List.of("LATE", "FAILING", "MISSING_IN_STORAGE"))
{
setStatus(ordinary, status);
assertThat(check.run(input()).getStatus()).as(status).isEqualTo(HealthStatus.DEGRADED);
setStatus(critical, status);
HealthCheckOutcome down = check.run(input());
assertThat(down.getStatus()).as(status).isEqualTo(HealthStatus.DOWN);
assertThat(down.getDetails().getValues()).containsEntry("activePlans", 2).containsEntry("slaStatus." + status, 2);
assertClean(down);
setStatus(critical, "OK");
}
}
/*******************************************************************************
** Unknown or unverified plans do not become a false UP or a critical DOWN.
*******************************************************************************/
@Test
void testSlaUnknownAndMixedStates() throws Exception
{
QRecord plan = VerifyTestSupport.plan(Map.of("critical", true));
BackupsSlaHealthCheck check = new BackupsSlaHealthCheck();
assertThat(check.run(input()).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
for(String status : List.of("UNVERIFIED", "UNKNOWN"))
{
setStatus(plan, status);
assertThat(check.run(input()).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
}
QRecord other = VerifyTestSupport.plan(Map.of("name", "Other", "planKey", "other"));
setStatus(other, "UNVERIFIED");
assertThat(check.run(input()).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
setStatus(other, "OK");
assertThat(check.run(input()).getStatus()).isEqualTo(HealthStatus.UP);
}
/*******************************************************************************
** Preserve the old indicator's no-context behavior and existing sessions.
*******************************************************************************/
@Test
void testWithoutContext() throws Exception
{
QRecord plan = VerifyTestSupport.plan(Map.of());
QInstance instance = QContext.getQInstance();
QSession session = QContext.getQSession();
new BackupsSlaHealthCheck().run(input());
assertThat(QContext.getQSession()).isSameAs(session);
QContext.clear();
try
{
HealthCheckInput input = new HealthCheckInput().withQInstance(instance).withDeadline(Instant.now().plusSeconds(5));
assertThat(new BackupsSlaHealthCheck().run(input).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
assertThat(QContext.getQInstance()).isNull();
assertThat(new BackupsDestinationHealthChecks().supply(instance)).hasSize(1);
assertThat(QContext.getQInstance()).isNull();
BackupsRuntime.getConfig().withStorageResolver(alias ->
{
assertThat(QContext.getQSession()).isInstanceOf(com.kingsrook.qqq.backend.core.model.session.QSystemUserSession.class);
return (storage);
});
assertThat(destination(plan).run(input).getStatus()).isEqualTo(HealthStatus.UP);
assertThat(QContext.getQInstance()).isNull();
}
finally
{
QContext.init(instance, session);
}
}
/*******************************************************************************
** Locked and unlocked plans only list/head; neither reads object contents.
*******************************************************************************/
@Test
void testDestinationReadOnlyAndMissingManifest() throws Exception
{
for(String lockMode : List.of("NONE", "GOVERNANCE"))
{
BackupsRuntime.getConfig().withStorageResolver(alias -> BackupsTestUtils.BACKUPS);
QRecord plan = VerifyTestSupport.plan(Map.of("name", lockMode, "planKey", lockMode.toLowerCase(), "objectLockMode", lockMode, "objectLockDays", 3, "pruneBucketAlias", "prune"));
BackupsRuntime.getConfig().withStorageResolver(alias ->
{
assertThat(alias).as("never resolve prune credentials").isNotEqualTo("prune");
return (storage);
});
Integer runId = catalog.insert(BackupsTables.RUN, Map.of("manifestKey", "manifest.json"));
catalog.update(BackupsTables.PLAN, plan.getValueInteger("id"), Map.of("lastSuccessRunId", runId));
storage.putBytes("manifest.json", new byte[] { 1 });
assertThat(destination(plan).run(input()).getStatus()).isEqualTo(HealthStatus.UP);
storage.clear();
assertThat(destination(plan).run(input()).getStatus()).isEqualTo(HealthStatus.DOWN);
}
assertThat(storage.lists).isEqualTo(4);
assertThat(storage.heads).isEqualTo(4);
}
/*******************************************************************************
** A deleted plan is UNKNOWN and does not touch storage.
*******************************************************************************/
@Test
void testDeletedPlan() throws Exception
{
QRecord plan = VerifyTestSupport.plan(Map.of());
catalog.deleteByIds(BackupsTables.PLAN, List.of(plan.getValueInteger("id")));
assertThat(destination(plan).run(input()).getStatus()).isEqualTo(HealthStatus.UNKNOWN);
assertThat(storage.lists).isZero();
}
/*******************************************************************************
** No work starts with an expired deadline, including direct probe calls.
*******************************************************************************/
@Test
void testExpiredDeadline() throws Exception
{
QRecord plan = VerifyTestSupport.plan(Map.of());
HealthCheckInput expired = input().withDeadline(Instant.now().minusSeconds(1));
assertThat(new BackupsSlaHealthCheck().run(expired).getStatus()).isEqualTo(HealthStatus.DOWN);
assertThat(destination(plan).run(expired).getStatus()).isEqualTo(HealthStatus.DOWN);
assertThat(DestinationProbe.probeReadOnly(plan, expired.getDeadline()).ok()).isFalse();
assertThat(storage.lists).isZero();
assertThat(storage.heads).isZero();
}
/*******************************************************************************
** Runner cancellation after list prevents the following manifest head.
*******************************************************************************/
@Test
void testCancellationBetweenStorageCalls() throws Exception
{
QRecord plan = VerifyTestSupport.plan(Map.of());
Integer runId = catalog.insert(BackupsTables.RUN, Map.of("manifestKey", "manifest.json"));
catalog.update(BackupsTables.PLAN, plan.getValueInteger("id"), Map.of("lastSuccessRunId", runId));
storage.interruptAfterList = true;
try
{
assertThat(destination(plan).run(input()).getStatus()).isEqualTo(HealthStatus.DOWN);
assertThat(storage.lists).isEqualTo(1);
assertThat(storage.heads).isZero();
}
finally
{
Thread.interrupted();
}
}
/*******************************************************************************
** Poison configuration, persisted state and provider errors. sanitize is
** package-private in core; assert raw outcomes already contain no secrets.
*******************************************************************************/
@Test
void testNoSecretsInSuccessAndFailureOutcomes() throws Exception
{
QRecord plan = VerifyTestSupport.plan(Map.of("name", SECRET, "storageBucketAlias", SECRET, "critical", true));
Integer runId = catalog.insert(BackupsTables.RUN, Map.of("manifestKey", SECRET));
catalog.update(BackupsTables.PLAN, plan.getValueInteger("id"), Map.of("lastSuccessRunId", runId));
storage.putBytes(SECRET, new byte[] { 1 });
for(String status : List.of("OK", "LATE", "UNVERIFIED", SECRET))
{
setStatus(plan, status);
assertClean(new BackupsSlaHealthCheck().run(input()));
assertClean(destination(plan).run(input()));
}
storage.failList = true;
HealthCheckOutcome failure = destination(plan).run(input());
assertThat(failure.getStatus()).isEqualTo(HealthStatus.DOWN);
assertClean(failure);
storage.failList = false;
storage.failHead = true;
failure = destination(plan).run(input());
assertThat(failure.getStatus()).isEqualTo(HealthStatus.DOWN);
assertClean(failure);
BackupsRuntime.getConfig().withStorageResolver(alias ->
{
throw (new IllegalStateException(SECRET));
});
failure = destination(plan).run(input());
assertThat(failure.getStatus()).isEqualTo(HealthStatus.DOWN);
assertClean(failure);
assertClean(new BackupsSlaHealthCheck().run(input()));
}
/*******************************************************************************
** Catalog failures are DOWN with fixed messages, even if metadata is poisoned.
*******************************************************************************/
@Test
void testCatalogFailureDoesNotExposeSecrets()
{
QContext.getQInstance().getTable(BackupsTables.PLAN).setBackendName(SECRET);
HealthCheckOutcome sla = new BackupsSlaHealthCheck().run(input());
HealthCheckOutcome destination = new BackupsDestinationHealthCheck(1).run(input());
assertThat(sla.getStatus()).isEqualTo(HealthStatus.DOWN);
assertThat(destination.getStatus()).isEqualTo(HealthStatus.DOWN);
assertClean(sla);
assertClean(destination);
}
/*******************************************************************************
** Update persisted SLA state without running the scheduled calculator.
*******************************************************************************/
private void setStatus(QRecord plan, String status) throws Exception
{
catalog.update(BackupsTables.PLAN, plan.getValueInteger("id"), Map.of("slaStatus", status));
}
/*******************************************************************************
** Fresh bounded input for each call.
*******************************************************************************/
private HealthCheckInput input()
{
return (new HealthCheckInput().withQInstance(QContext.getQInstance()).withDeadline(Instant.now().plusSeconds(5)));
}
/*******************************************************************************
** Construct the same per-plan check as the supplier.
*******************************************************************************/
private BackupsDestinationHealthCheck destination(QRecord plan)
{
return (new BackupsDestinationHealthCheck(plan.getValueInteger("id")));
}
/*******************************************************************************
** Validate the raw public message and flattened detail representation.
*******************************************************************************/
private void assertClean(HealthCheckOutcome outcome)
{
assertThat(outcome.getMessage()).doesNotContain(SECRET, "\n", "\r");
assertThat(outcome.getDetails().toJson()).doesNotContain(SECRET);
}
/*******************************************************************************
** Existing memory fake with read failures and assertions against side effects.
*******************************************************************************/
private static class ReadOnlyStorage extends InMemoryBackupStorage
{
private final String credential;
private boolean failList;
private boolean failHead;
private boolean interruptAfterList;
private int lists;
private int heads;
/*******************************************************************************
** Mimic the credential held by a host-provided storage adapter.
*******************************************************************************/
ReadOnlyStorage(String credential)
{
this.credential = credential;
}
/*******************************************************************************
** Record list calls and inject credential-bearing failures or cancellation.
*******************************************************************************/
@Override
public List<StoredObjectInfo> list(String prefix)
{
lists++;
if(failList)
{
throw (new IllegalStateException(credential));
}
if(interruptAfterList)
{
Thread.currentThread().interrupt();
}
return (super.list(prefix));
}
/*******************************************************************************
** Record head calls and inject credential-bearing failures.
*******************************************************************************/
@Override
public Optional<StoredObjectInfo> head(String key)
{
heads++;
if(failHead)
{
throw (new IllegalStateException(credential));
}
return (super.head(key));
}
/*******************************************************************************
** Health probes must not write, even to an unlocked bucket.
*******************************************************************************/
@Override
public OutputStream openWrite(String key, WriteOptions options)
{
throw (new AssertionError("Health probe wrote an object"));
}
/*******************************************************************************
** Health probes must not fetch backup contents.
*******************************************************************************/
@Override
public InputStream openRead(String key)
{
throw (new AssertionError("Health probe read object contents"));
}
/*******************************************************************************
** Health probes must not delete.
*******************************************************************************/
@Override
public DeleteResult deletePermanent(String key)
{
throw (new AssertionError("Health probe deleted an object"));
}
/*******************************************************************************
** Capability diagnostics can be active probes; health must not invoke them.
*******************************************************************************/
@Override
public Set<StorageCapability> capabilities()
{
throw (new AssertionError("Health probe requested capabilities"));
}
}
}
@@ -1,132 +0,0 @@
/*
* Copyright 2026 Kingsrook, LLC
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.kingsrook.qbits.backups.health;
import java.util.Map;
import com.kingsrook.qbits.backups.VerifyTestSupport;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.run.BackupCatalog;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.middleware.health.model.metadata.HealthCheckResult;
import com.kingsrook.qqq.middleware.health.model.metadata.HealthStatus;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/*******************************************************************************
** The health indicator: DOWN only when a critical, active plan is LATE, FAILING or
** MISSING_IN_STORAGE.
*******************************************************************************/
class BackupsHealthIndicatorTest
{
/*******************************************************************************
**
*******************************************************************************/
@BeforeEach
void setUp() throws Exception
{
VerifyTestSupport.setUp(3);
}
/*******************************************************************************
**
*******************************************************************************/
private void setStatus(QRecord plan, String status) throws Exception
{
new BackupCatalog().update(BackupsTables.PLAN, plan.getValueInteger("id"), Map.of("slaStatus", status));
}
/*******************************************************************************
** UP with details when nothing critical is wrong, including a non-critical plan
** that is LATE; DOWN when a critical plan is LATE, FAILING or MISSING_IN_STORAGE.
*******************************************************************************/
@Test
void testUpAndDown() throws Exception
{
BackupsHealthIndicator indicator = new BackupsHealthIndicator();
assertThat(indicator.getName()).isEqualTo("backups");
HealthCheckResult empty = indicator.check(QContext.getQInstance());
assertThat(empty.getStatus()).isEqualTo(HealthStatus.UP);
assertThat(empty.getDetails()).containsEntry("activePlans", 0);
QRecord critical = VerifyTestSupport.plan(Map.of("critical", true));
QRecord ordinary = VerifyTestSupport.plan(Map.of("name", "Other", "planKey", "other"));
VerifyTestSupport.plan(Map.of("name", "Off", "planKey", "off", "isActive", false, "critical", true));
setStatus(critical, "OK");
setStatus(ordinary, "LATE");
HealthCheckResult up = indicator.check(QContext.getQInstance());
assertThat(up.getStatus()).isEqualTo(HealthStatus.UP);
assertThat(up.getDetails()).containsEntry("activePlans", 2).doesNotContainKey("criticalPlansWithProblems");
assertThat(up.getDurationMs()).isNotNull();
for(String status : new String[] { "LATE", "FAILING", "MISSING_IN_STORAGE" })
{
setStatus(critical, status);
HealthCheckResult down = indicator.check(QContext.getQInstance());
assertThat(down.getStatus()).as(status).isEqualTo(HealthStatus.DOWN);
assertThat(down.getDetails().get("criticalPlansWithProblems").toString()).contains("nightly: " + status);
}
setStatus(critical, "UNVERIFIED");
assertThat(indicator.check(QContext.getQInstance()).getStatus()).as("not verified is not down").isEqualTo(HealthStatus.UP);
}
/*******************************************************************************
** On a thread with no QQQ context (a health request) the indicator makes its own
** and cleans up.
*******************************************************************************/
@Test
void testWorksWithoutAContext() throws Exception
{
QInstance qInstance = QContext.getQInstance();
QContext.clear();
try
{
assertThat(new BackupsHealthIndicator().check(qInstance).getStatus()).isEqualTo(HealthStatus.UP);
assertThat(QContext.getQInstance()).isNull();
}
finally
{
QContext.init(qInstance, new com.kingsrook.qqq.backend.core.model.session.QSession());
}
}
/*******************************************************************************
** The indicator's own context runs as the core system user, which token-based
** authentication modules accept without a token.
*******************************************************************************/
@Test
void testOwnContextUsesSystemUserSession()
{
assertThat(BackupsHealthIndicator.systemSession(QContext.getQInstance()))
.isInstanceOf(com.kingsrook.qqq.backend.core.model.session.QSystemUserSession.class);
}
}
@@ -87,10 +87,15 @@ class BackupsMetricsTest
String empty = BackupsMetrics.renderCached();
assertThat(empty).doesNotContain("{instance_name");
VerifyTestSupport.backup(VerifyTestSupport.plan(Map.of()));
VerifyTestSupport.clock.advance(Duration.ofSeconds(30));
assertThat(BackupsMetrics.renderCached()).as("cached").isEqualTo(empty);
assertThat(BackupsMetrics.renderCached()).as("cached").isSameAs(empty);
VerifyTestSupport.clock.advance(Duration.ofSeconds(31));
assertThat(BackupsMetrics.renderCached()).as("expired").isNotSameAs(empty);
/////////////////////////////////////////////////////////////
// a saved plan and a finished run clear the cache at once //
/////////////////////////////////////////////////////////////
VerifyTestSupport.backup(VerifyTestSupport.plan(Map.of()));
assertThat(BackupsMetrics.renderCached()).contains("plan=\"nightly\"");
assertThat(BackupsMetrics.escape("a\"b\\c\nd")).isEqualTo("a\\\"b\\\\c\\nd");
@@ -205,4 +205,24 @@ class RestoreSafetyTest
List.of(new com.kingsrook.qqq.backend.core.model.data.QRecord().withValue("id", 99)), request, null)).hasMessageContaining("backups.restore");
}
/*******************************************************************************
** Job-page cancellation reaches the restore worker and its catalog status.
*******************************************************************************/
@Test
void testDurableCancellation() throws Exception
{
BackupRunResult run = RestoreTestSupport.backup(Map.of());
RestoreRequest request = RestoreTestSupport.request(run, RestoreRequest.MODE_UPSERT).withPreRestoreBackup(false);
try(var scope = com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.open(null, () ->
{
throw (new com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.Reason.CANCEL));
}))
{
assertThatThrownBy(() -> new RestoreEngine().execute(request)).isInstanceOf(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.class);
}
assertThat(new BackupCatalog().where(BackupsTables.RESTORE_RUN, "status", "CANCELLED")).hasSize(1);
}
}
@@ -20,6 +20,7 @@ package com.kingsrook.qbits.backups.run;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.zip.GZIPInputStream;
@@ -28,11 +29,18 @@ import com.kingsrook.qbits.backups.manifest.BackupManifest;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.scope.PlanSettings;
import com.kingsrook.qbits.backups.scope.ScopeResolver;
import com.kingsrook.qqq.backend.core.actions.async.AsyncJobCallback;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
import com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException;
import com.kingsrook.qqq.backend.core.jobs.progress.JobLogLevel;
import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/*******************************************************************************
@@ -187,7 +195,9 @@ class BackupEngineTest
assertThat(gunzip(BackupsTestUtils.BACKUPS.getBytes(refOnly.getRunPrefix() + "files/index-00001.jsonl.gz"))).contains("REFERENCE_ONLY");
BackupsTestUtils.MEDIA.removeIgnoringLock("media/photo-1.jpg");
BackupRunResult missing = run(new BackupCatalog().where(BackupsTables.PLAN, "planKey", "nightly").get(0));
RecordingProgress progress = new RecordingProgress();
BackupRunResult missing = new BackupEngine().execute(BackupRunFactory.forPlan(new BackupCatalog().where(BackupsTables.PLAN, "planKey", "nightly").get(0), "MANUAL").withProgress(progress));
assertThat(progress.logs).anyMatch(l -> l.startsWith("WARN:") && l.contains("missing"));
assertThat(missing.getFilesMissing()).isGreaterThan(0);
assertThat(missing.getWarnings()).anyMatch(w -> w.contains("missing"));
assertThat(missing.getStatus()).isEqualTo("SUCCEEDED_WITH_WARNINGS");
@@ -208,7 +218,7 @@ class BackupEngineTest
BackupRunResult result = run(plan(Map.of()));
assertThat(result.getStatus()).isEqualTo("FAILED");
assertThat(result.isUsable()).isFalse();
assertThat(result.getErrorMessage()).contains("injected write failure");
assertThat(result.getErrorMessage()).contains("Backup failed");
assertThat(BackupsTestUtils.BACKUPS.list(result.getRunPrefix() + "manifest.json")).isEmpty();
assertThat(new BackupCatalog().get(BackupsTables.RUN, result.getRunId()).orElseThrow().getValueString("status")).isEqualTo("FAILED");
}
@@ -281,4 +291,215 @@ class BackupEngineTest
assertThat(scope.tables()).isEmpty();
assertThat(scope.excluded()).extracting(ScopeResolver.ExcludedTable::reason).containsExactly("BACKUPS");
}
/*******************************************************************************
** A real plan reports work in order, relative artifact names and no row data.
*******************************************************************************/
@Test
void testJobProgress() throws Exception
{
RecordingProgress progress = new RecordingProgress();
BackupRunResult result = new BackupRunCoordinator().runPlan(plan(Map.of("prefix", "private-prefix")), "MANUAL", progress, null);
assertThat(result.isUsable()).isTrue();
assertThat(progress.phases).containsExactly("prepare", "tables", "media", "upload");
assertThat(progress.getSnapshot().getPhases()).extracting(JobPhase::getWeight).containsExactly(2.0, 75.0, 18.0, 5.0);
assertThat(progress.getSnapshot().getPhases()).extracting(JobPhase::getStatus)
.containsExactly(JobPhase.Status.DONE, JobPhase.Status.DONE, JobPhase.Status.DONE, JobPhase.Status.DONE);
assertThat(progress.counts).contains("tables:0/2 tables", "tables:1/2 tables", "tables:2/2 tables", "media:0/4 files", "media:4/4 files");
assertThat(progress.bytes).isSorted();
assertThat(progress.bytes.getLast()).isGreaterThan(0);
assertThat(progress.getSnapshot().getBytesTotal()).isNull();
assertThat(progress.messages).contains("Backing up author (1 of 2)", "Backing up blogPost (2 of 2)");
assertThat(progress.logs).anyMatch(l -> l.matches("INFO:author: 3 rows, .* MB in .* s"));
assertThat(progress.logs).anyMatch(l -> l.matches("INFO:blogPost: 7 rows, .* MB in .* s"));
assertThat(progress.logs).contains("INFO:Media: 4 files, 3 copied, 1 already stored, 0 missing", "INFO:Wrote checksums.sha256", "INFO:Wrote manifest.json", "INFO:Wrote files/index-00001.jsonl.gz");
assertThat(progress.logs).anyMatch(l -> l.startsWith("INFO:Wrote tables/blogPost/"));
for(String warning : result.getWarnings())
{
assertThat(progress.logs).contains("WARN:" + warning);
}
assertThat(String.join("\n", progress.logs)).doesNotContain("private-prefix", "do-not-leak", "Body of", "media/photo-");
}
/*******************************************************************************
** Repeated backups report deduplicated media and complete all four phases.
*******************************************************************************/
@Test
void testProgressWithDeduplication() throws Exception
{
QRecord plan = plan(Map.of());
run(plan);
RecordingProgress progress = new RecordingProgress();
BackupRunResult result = new BackupEngine().execute(BackupRunFactory.forPlan(plan, "MANUAL").withProgress(progress));
assertThat(result.isUsable()).isTrue();
assertThat(progress.phases).containsExactly("prepare", "tables", "media", "upload");
assertThat(progress.getSnapshot().getPhases()).allMatch(p -> p.getStatus() == JobPhase.Status.DONE);
assertThat(progress.logs).contains("INFO:Media: 4 files, 0 copied, 4 already stored, 0 missing");
}
/*******************************************************************************
** Arbitrary provider failures must not put credentials or row values in jobs.
*******************************************************************************/
@Test
void testProgressFailureDoesNotLeakProviderText() throws Exception
{
RecordingProgress progress = new RecordingProgress();
BackupRunRequest request = BackupRunFactory.forPlan(plan(Map.of()), "MANUAL").withProgress(progress)
.withReader((table, filter, start, size, transaction, sink) ->
{
throw (new com.kingsrook.qqq.backend.core.exceptions.QException("https://storage-user:storage-password@host/?secret=storage-secret accessKey=storage-access row=private-row"));
});
BackupRunResult result = new BackupEngine().execute(request);
assertThat(result.getStatus()).isEqualTo("FAILED");
assertThat(progress.logs).contains("ERROR:" + result.getErrorMessage());
assertThat(progress.getSnapshot().getPhases()).anyMatch(p -> p.getStatus() == JobPhase.Status.FAILED);
assertThat(String.join("\n", progress.logs) + result.getErrorMessage()).doesNotContain("storage-user", "storage-password", "storage-secret", "storage-access", "private-row", "https://");
}
/*******************************************************************************
** The job page's cancel check must escape the engine to the durable worker.
*******************************************************************************/
@Test
void testJobCancellationBetweenPages() throws Exception
{
BackupRunRequest request = BackupRunFactory.forPlan(plan(Map.of("pageSize", 1)), "MANUAL");
int[] checks = { 0 };
try(JobChunkScope ignored = JobChunkScope.open(null, () ->
{
if(++checks[0] == 3)
{
throw (new JobInterruptedException(JobInterruptedException.Reason.CANCEL));
}
}))
{
assertThatThrownBy(() -> new BackupEngine().execute(request)).isInstanceOf(JobInterruptedException.class);
}
assertThat(new BackupCatalog().where(BackupsTables.RUN, "status", "CANCELLED")).hasSize(1);
}
/*******************************************************************************
** Storage providers can include endpoint user-info and raw keys in exceptions.
*******************************************************************************/
@Test
void testStorageFailureDoesNotLeakCredentials() throws Exception
{
RecordingProgress progress = new RecordingProgress();
var storage = new com.kingsrook.qbits.backups.storage.InMemoryBackupStorage()
{
/*******************************************************************************
** Deliberately unsafe provider diagnostics must stay out of job output.
*******************************************************************************/
@Override
public java.io.OutputStream openWrite(String key, com.kingsrook.qbits.backups.storage.WriteOptions options)
{
throw (new com.kingsrook.qbits.backups.storage.BackupStorageException(com.kingsrook.qbits.backups.storage.BackupStorageException.Reason.ACCESS_DENIED,
"https://endpoint-user:endpoint-password@storage accessKey=access-secret secretKey=secret-secret"));
}
};
BackupRunResult result = new BackupEngine().execute(BackupRunFactory.forPlan(plan(Map.of()), "MANUAL").withProgress(progress).withStorage(storage));
assertThat(result.getStatus()).isEqualTo("FAILED");
assertThat(progress.logs).contains("ERROR:" + result.getErrorMessage());
assertThat(String.join("\n", progress.logs) + result.getErrorMessage()).doesNotContain("endpoint-user", "endpoint-password", "access-secret", "secret-secret", "https://");
}
/*******************************************************************************
** Failed backups must fail the durable chunk, not just add a summary line.
*******************************************************************************/
@Test
void testDurableLoadPropagatesFailure() throws Exception
{
var input = new com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput();
input.setRecords(List.of(plan(Map.of())));
BackupsTestUtils.BACKUPS.withFailWritesFor(key -> key.contains("/tables/"));
try(JobChunkScope scope = JobChunkScope.open(null).withProgress(new RecordingProgress()))
{
assertThatThrownBy(() -> new com.kingsrook.qbits.backups.processes.BackupsProcesses.RunPlanLoad().runOnePage(input,
new com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput()))
.isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QException.class).hasMessageContaining("Backup failed");
}
}
/*******************************************************************************
** Record values at call time: JobProgress's live snapshot is mutable.
*******************************************************************************/
static class RecordingProgress extends JobProgress
{
final List<String> phases = new ArrayList<>();
final List<String> counts = new ArrayList<>();
final List<Long> bytes = new ArrayList<>();
final List<String> logs = new ArrayList<>();
final List<String> messages = new ArrayList<>();
/*******************************************************************************
** Use the core's safe standalone destination.
*******************************************************************************/
RecordingProgress()
{
super((AsyncJobCallback) null);
}
/*******************************************************************************
** Record transitions.
*******************************************************************************/
@Override
public JobProgress phase(String name)
{
phases.add(name);
return (super.phase(name));
}
/*******************************************************************************
** Record counts and their units.
*******************************************************************************/
@Override
public JobProgress done(long done)
{
counts.add(getSnapshot().getCurrentPhase() + ":" + done + "/" + getSnapshot().getTotal() + " " + getSnapshot().getUnitLabel());
return (super.done(done));
}
/*******************************************************************************
** Record byte counts.
*******************************************************************************/
@Override
public JobProgress bytesDone(long done)
{
bytes.add(done);
return (super.bytesDone(done));
}
/*******************************************************************************
** Capture text before core redaction as well as exercising core publication.
*******************************************************************************/
@Override
public JobProgress log(JobLogLevel level, String message)
{
logs.add(level + ":" + message);
return (super.log(level, message));
}
/*******************************************************************************
** Record live messages.
*******************************************************************************/
@Override
public JobProgress message(String message)
{
messages.add(message);
return (super.message(message));
}
}
}
@@ -24,10 +24,10 @@ import java.util.List;
import java.util.Map;
import com.kingsrook.qbits.backups.BackupsRuntime;
import com.kingsrook.qbits.backups.BackupsTestUtils;
import com.kingsrook.qbits.backups.LifecycleTestSupport;
import com.kingsrook.qbits.backups.LifecycleTestSupport.EtlSummaries;
import com.kingsrook.qbits.backups.LifecycleTestSupport.MutableClock;
import com.kingsrook.qbits.backups.LifecycleTestSupport.RecordingAlertSink;
import com.kingsrook.qbits.backups.LifecycleTestSupport;
import com.kingsrook.qbits.backups.manifest.BackupManifest;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.processes.CancelRunProcess;
@@ -296,7 +296,7 @@ class BackupRunCoordinatorTest
QRecord plan = LifecycleTestSupport.plan(perTableTinyParts);
BackupRunResult failed = new BackupRunCoordinator().runPlan(plan, "MANUAL", request -> request.withReader(failingAfter(request.getReader(), "blogPost", 4)));
assertThat(failed.getStatus()).isEqualTo("FAILED");
assertThat(failed.getErrorMessage()).contains("injected failure in blogPost");
assertThat(failed.getErrorMessage()).contains("Backup failed");
assertThat(BackupsTestUtils.BACKUPS.list(failed.getRunPrefix() + "manifest.json")).isEmpty();
QRecord failedRow = catalog.get(BackupsTables.RUN, failed.getRunId()).orElseThrow();
@@ -384,4 +384,59 @@ class BackupRunCoordinatorTest
assertThat(manifest.table("blogPost").parts()).hasSize(7);
assertThat(catalog.where(BackupsTables.ARTIFACT, "backupRunId", failed.getRunId()).stream().filter(a -> "blogPost".equals(a.getValueString("tableName")))).hasSize(7);
}
/*******************************************************************************
** Explicit retry of a durable chunk must reuse its backup and completed parts.
*******************************************************************************/
@Test
void testDurableRetryResumesSameRun() throws Exception
{
QRecord plan = LifecycleTestSupport.plan(Map.of());
BackupEngineTest.RecordingProgress progress = new BackupEngineTest.RecordingProgress();
var identity = new com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.ChunkIdentity(42, 8, 0, "backupRunPlan");
try(var scope = com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.open(null, null, identity).withProgress(progress))
{
BackupsTestUtils.BACKUPS.withFailWritesFor(key -> key.contains("/tables/blogPost/"));
BackupRunResult failed = new BackupRunCoordinator().runPlan(plan, "MANUAL");
assertThat(failed.getStatus()).isEqualTo("FAILED");
BackupsTestUtils.BACKUPS.withFailWritesFor(key -> false);
// A worker death can leave RUNNING rather than FAILED; identity, not age,
// ties this recovery to the same durable job.
catalog.update(BackupsTables.RUN, failed.getRunId(), Map.of("status", "RUNNING"));
BackupRunResult resumed = new BackupRunCoordinator().runPlan(plan, "MANUAL");
assertThat(resumed.isUsable()).isTrue();
assertThat(resumed.getRunId()).isEqualTo(failed.getRunId());
assertThat(resumed.getRunPrefix()).isEqualTo(failed.getRunPrefix());
assertThat(progress.logs).contains("INFO:Resuming run " + failed.getRunId() + ": 1 tables already done");
assertThat(catalog.where(BackupsTables.RUN, "backupPlanId", plan.getValueInteger("id"))).hasSize(1);
int artifacts = catalog.where(BackupsTables.ARTIFACT, "backupRunId", failed.getRunId()).size();
BackupRunResult again = new BackupRunCoordinator().runPlan(plan, "MANUAL");
assertThat(again.getRunId()).isEqualTo(failed.getRunId());
assertThat(catalog.where(BackupsTables.ARTIFACT, "backupRunId", failed.getRunId())).hasSize(artifacts);
}
}
/*******************************************************************************
** A durable job refused for capacity must not escape into the inline queue.
*******************************************************************************/
@Test
void testDurableCapacityRefusalDoesNotQueue() throws Exception
{
QRecord occupied = LifecycleTestSupport.plan(Map.of());
liveRun(occupied.getValueInteger("id"), "RUNNING", clock.instant());
QRecord requested = LifecycleTestSupport.plan(Map.of("planKey", "second", "name", "Second"));
var identity = new com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.ChunkIdentity(43, 9, 0, "backupRunPlan");
try(var scope = com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.open(null, null, identity))
{
org.assertj.core.api.Assertions.assertThatThrownBy(() -> new BackupRunCoordinator().runPlan(requested, "MANUAL"))
.isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QException.class).hasMessageContaining("could not start");
}
assertThat(catalog.queuedRuns()).isEmpty();
assertThat(catalog.where(BackupsTables.RUN, "runKey", "job-43-plan-" + requested.getValueInteger("id")))
.singleElement().satisfies(run -> assertThat(run.getValueString("status")).isEqualTo("FAILED"));
}
}
@@ -591,4 +591,23 @@ class BackupVerifierTest
}
return (bytes.toByteArray());
}
/*******************************************************************************
** A verification must propagate a job interrupt, not report it as corruption.
*******************************************************************************/
@Test
void testDurableCancellation() throws Exception
{
BackupRunResult run = backup(plan(Map.of()));
try(var scope = com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.open(null, () ->
{
throw (new com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.Reason.CANCEL));
}))
{
org.assertj.core.api.Assertions.assertThatThrownBy(() -> new BackupVerifier().verify(run.getRunId(), BackupVerifier.CONTENT, "MANUAL", "test"))
.isInstanceOf(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.class);
}
}
}