open-source/qinglong
1
0
Fork 0
mirror of https://github.com/whyour/qinglong.git synced 2026-09-23 20:15:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(ql3): complete cluster log retention lifecycle

Browse Source
This commit is contained in:
whyour
2026-08-12 04:41:59 +08:00
parent 40628c843d
commit 5d3b40ce3b
20 changed files with 2026 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ql3-cluster-control/src/application-runtime/application.ts
+8
View File
@@ -13,6 +13,7 @@ import {
type ClusterControlAssemblyInput,
type ClusterControlRecoveryRuntimeOptions,
type ClusterRunCancellationConvergenceRuntimeOptions,
type ClusterRunAttemptLogRetentionRuntimeOptions,
type ClusterSchedulerRuntimeOptions,
type ClusterWorkerRuntimeDependencies,
} from './clusterControlRuntime';
@@ -39,6 +40,7 @@ export interface ClusterControlApplicationOptions {
readonly recovery?: ClusterControlRecoveryRuntimeOptions;
readonly scheduler?: ClusterSchedulerRuntimeOptions;
readonly cancellationConvergence?: ClusterRunCancellationConvergenceRuntimeOptions;
readonly logRetention?: ClusterRunAttemptLogRetentionRuntimeOptions;
readonly workerRuntime?: ClusterWorkerRuntimeDependencies;
readonly openDatabase: OpenPostgresDatabase;
readonly availability: ClusterControlAvailabilitySource;
@@ -88,6 +90,9 @@ function inactiveBootstrap(
...(options.cancellationConvergence === undefined
? {}
: { cancellationConvergence: options.cancellationConvergence }),
...(options.logRetention === undefined
? {}
: { logRetention: options.logRetention }),
...(options.workerRuntime === undefined
? {}
: { workerRuntime: options.workerRuntime }),
@@ -163,6 +168,9 @@ export async function startClusterControlApplication(
...(options.cancellationConvergence === undefined
? {}
: { cancellationConvergence: options.cancellationConvergence }),
...(options.logRetention === undefined
? {}
: { logRetention: options.logRetention }),
...(options.workerRuntime === undefined
? {}
: { workerRuntime: options.workerRuntime }),
packages/ql3-cluster-control/src/application-runtime/clusterControlRuntime.ts
+207 -1
View File
@@ -30,6 +30,17 @@ import {
type ClusterRunCancellationConvergenceCycleResult,
} from '@qinglong/runtime-core';
import type { ClusterRunCancellationRepository } from '@qinglong/runtime-core/cluster-run-cancellation';
import {
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_CLAIMS,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS,
MIN_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
} from '@qinglong/runtime-core/cluster-run-attempt-log-retention';
import {
MAX_RUN_ATTEMPT_LOG_RETENTION_MS,
MIN_RUN_ATTEMPT_LOG_RETENTION_MS,
type RunAttemptLogRetentionStateReader,
} from '@qinglong/runtime-core/run-attempt-log-retention';
import type { ProjectRunListReader } from '@qinglong/runtime-core/project-run-list';
import type { ClusterScheduleStore } from '@qinglong/runtime-core/cluster-scheduler';
import type { TaskDefinitionSource } from '@qinglong/runtime-core/task-definition';
@@ -60,6 +71,7 @@ import {
PostgresSecurityAuditRepository,
PostgresRunRepository,
PostgresWorkerExecutionAttestationRepository,
PostgresRunAttemptLogRetentionClaimRepository,
PostgresTaskDefinitionSource,
PostgresTaskExecutionRevisionSource,
PostgresTriggerSource,
@@ -104,6 +116,12 @@ import {
import { ClusterWorkflowSchedulerCoordinator } from '../scheduling/workflowScheduler';
import { ClusterRuntimeSchedulerCoordinator } from '../scheduling/runtimeScheduler';
import { ClusterRunCancellationConvergenceLifecycle } from '../run/runCancellationLifecycle';
import {
ClusterRunAttemptLogRetentionCoordinator,
ClusterRunAttemptLogRetentionLifecycle,
type ClusterRunAttemptLogRetirementStore,
type ClusterRunAttemptLogRetentionCycleSummary,
} from '../run/runAttemptLogRetentionLifecycle';
import type { TaskStartRepository } from '@qinglong/runtime-core/task-start';
import {
createClusterWorkerRuntimePort,
@@ -131,6 +149,7 @@ export interface ClusterControlAssemblyInput {
readonly evidence: ClusterControlReadinessEvidence;
readonly policies: ProjectPolicyRepository;
readonly runs: RunRepository & ProjectRunListReader;
readonly runAttemptLogRetention: RunAttemptLogRetentionStateReader;
readonly runCancellation: ClusterRunCancellationRepository;
readonly taskStart: TaskStartRepository;
readonly taskDefinitions: TaskDefinitionSource;
@@ -178,6 +197,24 @@ export interface ClusterRunCancellationConvergenceRuntimeOptions {
) => void | Promise<void>;
}
export interface ClusterRunAttemptLogRetentionRuntimeOptions {
readonly store: ClusterRunAttemptLogRetirementStore;
readonly ownerId?: string;
readonly retentionMs?: number;
readonly claimLimit?: number;
readonly leaseMs?: number;
readonly maximumCycleMs?: number;
readonly retryBaseMs?: number;
readonly retryMaximumMs?: number;
readonly maximumFailures?: number;
readonly intervalMs?: number;
readonly stopTimeoutMs?: number;
readonly onDiagnostic?: (
error: unknown,
summary?: Readonly<ClusterRunAttemptLogRetentionCycleSummary>,
) => void | Promise<void>;
}
export interface ClusterControlBootstrapOptions {
readonly enabled?: boolean;
readonly profile: DeploymentProfile;
@@ -185,6 +222,7 @@ export interface ClusterControlBootstrapOptions {
readonly recovery?: ClusterControlRecoveryRuntimeOptions;
readonly scheduler?: ClusterSchedulerRuntimeOptions;
readonly cancellationConvergence?: ClusterRunCancellationConvergenceRuntimeOptions;
readonly logRetention?: ClusterRunAttemptLogRetentionRuntimeOptions;
readonly workerRuntime?: ClusterWorkerRuntimeDependencies;
readonly openDatabase: OpenPostgresDatabase;
readonly create: (
@@ -223,6 +261,21 @@ interface PreparedCancellationConvergenceRuntime {
readonly onDiagnostic?: ClusterRunCancellationConvergenceRuntimeOptions['onDiagnostic'];
}
interface PreparedLogRetentionRuntime {
readonly store: ClusterRunAttemptLogRetirementStore;
readonly ownerId: string;
readonly retentionMs: number;
readonly claimLimit: number;
readonly leaseMs: number;
readonly maximumCycleMs: number;
readonly retryBaseMs: number;
readonly retryMaximumMs: number;
readonly maximumFailures: number;
readonly intervalMs: number;
readonly stopTimeoutMs: number;
readonly onDiagnostic?: ClusterRunAttemptLogRetentionRuntimeOptions['onDiagnostic'];
}
function boundedInteger(
name: string,
value: number | undefined,
@@ -439,6 +492,116 @@ function prepareCancellationConvergenceRuntime(
});
}
function prepareLogRetentionRuntime(
options: ClusterRunAttemptLogRetentionRuntimeOptions | undefined,
fallbackOwnerId: string,
): PreparedLogRetentionRuntime | undefined {
if (options === undefined) return undefined;
const allowedKeys = new Set([
'claimLimit',
'intervalMs',
'leaseMs',
'maximumCycleMs',
'maximumFailures',
'onDiagnostic',
'ownerId',
'retentionMs',
'retryBaseMs',
'retryMaximumMs',
'stopTimeoutMs',
'store',
]);
if (
!options ||
typeof options !== 'object' ||
Array.isArray(options) ||
Object.keys(options).some((key) => !allowedKeys.has(key)) ||
typeof options.store?.retire !== 'function' ||
(options.onDiagnostic !== undefined &&
typeof options.onDiagnostic !== 'function')
) {
throw new TypeError(
'Cluster Run Attempt log retention configuration is invalid',
);
}
const ownerId = options.ownerId ?? fallbackOwnerId;
if (!/^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/.test(ownerId)) {
throw new TypeError('Cluster Run Attempt log retention ownerId is invalid');
}
const leaseMs = boundedInteger(
'Cluster Run Attempt log retention lease',
options.leaseMs,
30_000,
MIN_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
);
const retryBaseMs = boundedInteger(
'Cluster Run Attempt log retention retry base',
options.retryBaseMs,
5_000,
0,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS,
);
return Object.freeze({
store: options.store,
ownerId,
retentionMs: boundedInteger(
'Cluster Run Attempt log retention duration',
options.retentionMs,
30 * 24 * 60 * 60_000,
MIN_RUN_ATTEMPT_LOG_RETENTION_MS,
MAX_RUN_ATTEMPT_LOG_RETENTION_MS,
),
claimLimit: boundedInteger(
'Cluster Run Attempt log retention claim limit',
options.claimLimit,
4,
1,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_CLAIMS,
),
leaseMs,
maximumCycleMs: boundedInteger(
'Cluster Run Attempt log retention cycle budget',
options.maximumCycleMs,
10_000,
100,
leaseMs - 500,
),
retryBaseMs,
retryMaximumMs: boundedInteger(
'Cluster Run Attempt log retention retry maximum',
options.retryMaximumMs,
60 * 60_000,
retryBaseMs,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS,
),
maximumFailures: boundedInteger(
'Cluster Run Attempt log retention failure limit',
options.maximumFailures,
8,
1,
32,
),
intervalMs: boundedInteger(
'Cluster Run Attempt log retention interval',
options.intervalMs,
60_000,
1_000,
24 * 60 * 60_000,
),
stopTimeoutMs: boundedInteger(
'Cluster Run Attempt log retention stop timeout',
options.stopTimeoutMs,
10_000,
100,
30_000,
),
...(options.onDiagnostic === undefined
? {}
: { onDiagnostic: options.onDiagnostic }),
});
}
function readinessEvidence(
report: PostgresSchemaReadinessReport,
): ClusterControlReadinessEvidence {
@@ -463,6 +626,7 @@ export async function bootstrapClusterControlRuntime(
let cancellationConvergenceRuntime:
| PreparedCancellationConvergenceRuntime
| undefined;
let logRetentionRuntime: PreparedLogRetentionRuntime | undefined;
let recoveryRegistry: ClusterControlRecoveryEvidenceRegistry | undefined;
if ((options.enabled ?? false) && options.profile === 'cluster-control') {
assertClusterControlApiCredentialPepper(options.apiCredentialPepper ?? '');
@@ -474,6 +638,10 @@ export async function bootstrapClusterControlRuntime(
cancellationConvergenceRuntime = prepareCancellationConvergenceRuntime(
options.cancellationConvergence,
);
logRetentionRuntime = prepareLogRetentionRuntime(
options.logRetention,
recoveryRuntime.ownerId,
);
}
let database: PostgresDatabaseResource | undefined;
let closePromise: Promise<void> | undefined;
@@ -576,6 +744,8 @@ export async function bootstrapClusterControlRuntime(
);
const schedules = new PostgresClusterScheduleRepository(database.pool);
const runs = new PostgresRunRepository(database.pool);
const runAttemptLogRetention =
new PostgresRunAttemptLogRetentionClaimRepository(database.pool);
const trustedToolStorage: ClusterTrustedToolStorage = Object.freeze({
invocationArtifacts: new PostgresToolInvocationArtifactRepository(
database.pool,
@@ -653,6 +823,32 @@ export async function bootstrapClusterControlRuntime(
}),
},
);
const logRetentionLifecycle =
logRetentionRuntime === undefined
? undefined
: new ClusterRunAttemptLogRetentionLifecycle(
new ClusterRunAttemptLogRetentionCoordinator(
runAttemptLogRetention,
logRetentionRuntime.store,
{
ownerId: logRetentionRuntime.ownerId,
retentionMs: logRetentionRuntime.retentionMs,
claimLimit: logRetentionRuntime.claimLimit,
leaseMs: logRetentionRuntime.leaseMs,
maximumCycleMs: logRetentionRuntime.maximumCycleMs,
retryBaseMs: logRetentionRuntime.retryBaseMs,
retryMaximumMs: logRetentionRuntime.retryMaximumMs,
maximumFailures: logRetentionRuntime.maximumFailures,
},
),
{
intervalMs: logRetentionRuntime.intervalMs,
stopTimeoutMs: logRetentionRuntime.stopTimeoutMs,
...(logRetentionRuntime.onDiagnostic === undefined
? {}
: { onDiagnostic: logRetentionRuntime.onDiagnostic }),
},
);
const runCancellation = new PostgresClusterRunCancellationRepository(
database.pool,
);
@@ -665,6 +861,7 @@ export async function bootstrapClusterControlRuntime(
),
policies: new PostgresProjectPolicyRepository(database.pool),
runs,
runAttemptLogRetention,
runCancellation,
taskStart,
taskDefinitions: new PostgresTaskDefinitionSource(database.pool),
@@ -739,6 +936,7 @@ export async function bootstrapClusterControlRuntime(
if (!(await application.startLifecycles())) return false;
schedulerLifecycle.start();
cancellationConvergenceLifecycle.start();
logRetentionLifecycle?.start();
return true;
},
installAdmission: () => application.installAdmission(),
@@ -746,14 +944,21 @@ export async function bootstrapClusterControlRuntime(
recoveryRegistry?.dispose();
let schedulerStatus: 'stopped' | 'timed_out' = 'stopped';
let cancellationStatus: 'stopped' | 'timed_out' = 'stopped';
let logRetentionStatus: 'stopped' | 'timed_out' = 'stopped';
let applicationStatus: ClusterControlStopResult = 'stopped';
let primaryError: unknown;
try {
logRetentionStatus =
(await logRetentionLifecycle?.stopAndDrain()) ?? 'stopped';
} catch (error) {
primaryError = error;
}
try {
cancellationStatus = (
await cancellationConvergenceLifecycle.stopAndDrain()
).status;
} catch (error) {
primaryError = error;
primaryError ??= error;
}
try {
schedulerStatus = (await schedulerLifecycle.stopAndDrain())
@@ -768,6 +973,7 @@ export async function bootstrapClusterControlRuntime(
}
if (primaryError) throw primaryError;
return cancellationStatus === 'timed_out' ||
logRetentionStatus === 'timed_out' ||
schedulerStatus === 'timed_out' ||
applicationStatus === 'timed_out'
? 'timed_out'
packages/ql3-cluster-control/src/application-runtime/productionApplication.ts
+1
View File
@@ -200,6 +200,7 @@ export function createProductionClusterControlApplicationStack(
createClusterControlRunAttemptLogReadRoute(
input.runs,
input.workerRuntime?.runAttemptLogRead,
input.runAttemptLogRetention,
),
createClusterControlRunCancellationRoute(
input.runCancellation,
packages/ql3-cluster-control/src/artifact/s3ArtifactStore.ts
+126 -8
View File
@@ -19,6 +19,12 @@ import {
type RunAttemptLogReadIdentity,
type RunAttemptLogReadRange,
} from '@qinglong/runtime-core/run-attempt-log-read';
import {
normalizeRunAttemptLogRetentionCandidate,
type RunAttemptLogRetentionCandidate,
type RunAttemptLogRetirementStore,
type RunAttemptLogRetirementStoreResult,
} from '@qinglong/runtime-core/run-attempt-log-retention';
import {
MAX_REMOTE_WORKER_ARTIFACT_BYTES,
REMOTE_WORKER_ARTIFACT_CONTENT_TYPE,
@@ -124,6 +130,7 @@ type ArtifactAuthority = Readonly<{
type StoredArtifactHead = Readonly<{
receipt: Readonly<RemoteWorkerArtifactReceipt>;
eTag?: string;
versionId?: string;
}>;
type NormalizedStorageCommand = ArtifactAuthority &
@@ -484,6 +491,18 @@ function canonicalETag(value: unknown): string {
return value;
}
function canonicalVersionId(value: unknown): string {
if (
typeof value !== 'string' ||
value.length < 1 ||
Buffer.byteLength(value, 'utf8') > 1024 ||
/[\u0000-\u001f\u007f]/.test(value)
) {
throw new S3ClusterRemoteWorkerArtifactStoreError('integrity_mismatch');
}
return value;
}
function assertRangeMetadata(
authority: ArtifactAuthority,
receipt: Readonly<RemoteWorkerArtifactReceipt>,
@@ -580,6 +599,20 @@ function isNotFound(error: unknown): boolean {
);
}
function isPreconditionFailed(error: unknown): boolean {
if (!error || typeof error !== 'object') return false;
const value = error as {
name?: unknown;
Code?: unknown;
$metadata?: { httpStatusCode?: unknown };
};
return (
value.name === 'PreconditionFailed' ||
value.Code === 'PreconditionFailed' ||
value.$metadata?.httpStatusCode === 412
);
}
function requestOptions(
signal?: AbortSignal,
): { abortSignal: AbortSignal } | undefined {
@@ -655,10 +688,11 @@ class ArtifactContentDigest {
/**
* Shared immutable S3 adapter. A unique temporary upload is checksummed first,
* then promoted by one destination-conditional server-side copy. Permanent
* objects are never overwritten or deleted by this adapter.
* objects are never overwritten and are retired only after an identity-checked
* HEAD followed by a VersionId- or ETag-fenced delete.
*/
export class S3ClusterRemoteWorkerArtifactStore
implements ClusterRemoteWorkerArtifactStore
implements ClusterRemoteWorkerArtifactStore, RunAttemptLogRetirementStore
{
private readonly options: PreparedOptions;
@@ -751,6 +785,75 @@ export class S3ClusterRemoteWorkerArtifactStore
}
}
async retire(
rawCandidate: Readonly<RunAttemptLogRetentionCandidate>,
signal?: AbortSignal,
): Promise<Readonly<RunAttemptLogRetirementStoreResult>> {
const candidate = normalizeRunAttemptLogRetentionCandidate(rawCandidate);
if (candidate.executorType !== 'remote_worker') {
throw new S3ClusterRemoteWorkerArtifactStoreError('integrity_mismatch');
}
const authority = Object.freeze({
projectId: candidate.projectId,
runId: candidate.runId,
attemptId: candidate.attemptId,
logArtifactId: candidate.logArtifactId,
});
const stored = await this.head(authority, signal);
if (!stored) {
return Object.freeze({
disposition: 'already_absent' as const,
byteLength: 0,
truncation: Object.freeze({ truncated: 'unknown' as const }),
});
}
const versionId =
stored.versionId === undefined
? undefined
: canonicalVersionId(stored.versionId);
const eTag =
versionId === undefined ? canonicalETag(stored.eTag) : undefined;
try {
await this.options.client.send(
new DeleteObjectCommand({
Bucket: this.options.bucket,
Key: finalObjectKey(this.options.prefix, authority),
...(versionId === undefined
? { IfMatch: eTag }
: { VersionId: versionId }),
...(this.options.expectedBucketOwner === undefined
? {}
: { ExpectedBucketOwner: this.options.expectedBucketOwner }),
}),
requestOptions(signal),
);
} catch (error) {
if (isNotFound(error)) {
return Object.freeze({
disposition: 'already_absent' as const,
byteLength: 0,
truncation: Object.freeze({ truncated: 'unknown' as const }),
});
}
if (isPreconditionFailed(error)) {
throw new S3ClusterRemoteWorkerArtifactStoreError(
'integrity_mismatch',
{ cause: error },
);
}
throw new S3ClusterRemoteWorkerArtifactStoreError('unavailable', {
cause: error,
});
}
return Object.freeze({
disposition: 'deleted' as const,
byteLength: stored.receipt.byteLength,
truncation: Object.freeze({
truncated: stored.receipt.truncated ?? ('unknown' as const),
}),
});
}
private async head(
authority: ArtifactAuthority,
signal?: AbortSignal,
@@ -771,6 +874,9 @@ export class S3ClusterRemoteWorkerArtifactStore
return Object.freeze({
receipt: parseStoredReceipt(authority, output),
...(output.ETag === undefined ? {} : { eTag: output.ETag }),
...(output.VersionId === undefined
? {}
: { versionId: output.VersionId }),
});
} catch (error) {
if (isNotFound(error)) return undefined;
@@ -811,7 +917,9 @@ export class S3ClusterRemoteWorkerArtifactStore
this.options.createTemporaryId,
);
const temporaryOwner = temporaryOwnershipDigest();
let temporaryOwned = false;
let temporaryDeleteAuthority:
| Readonly<{ readonly eTag: string; readonly versionId?: string }>
| undefined;
let result: Readonly<RemoteWorkerArtifactReceipt> | undefined;
let primaryError: unknown;
try {
@@ -837,21 +945,19 @@ export class S3ClusterRemoteWorkerArtifactStore
}),
requestOptions(signal),
);
temporaryOwned = true;
} catch (error) {
if (!digest.isComplete()) throw error;
} finally {
body.destroy();
}
const sha256 = digest.digest();
await this.assertTemporaryObject(
temporaryDeleteAuthority = await this.assertTemporaryObject(
temporaryKey,
temporaryOwner,
command.byteLength,
sha256,
signal,
);
temporaryOwned = true;
let copied = false;
try {
@@ -898,12 +1004,15 @@ export class S3ClusterRemoteWorkerArtifactStore
primaryError = error;
}
if (temporaryOwned) {
if (temporaryDeleteAuthority !== undefined) {
try {
await this.options.client.send(
new DeleteObjectCommand({
Bucket: this.options.bucket,
Key: temporaryKey,
...(temporaryDeleteAuthority.versionId === undefined
? { IfMatch: temporaryDeleteAuthority.eTag }
: { VersionId: temporaryDeleteAuthority.versionId }),
...(this.options.expectedBucketOwner === undefined
? {}
: { ExpectedBucketOwner: this.options.expectedBucketOwner }),
@@ -937,7 +1046,7 @@ export class S3ClusterRemoteWorkerArtifactStore
byteLength: number,
sha256: string,
signal?: AbortSignal,
): Promise<void> {
): Promise<Readonly<{ readonly eTag: string; readonly versionId?: string }>> {
let output;
try {
output = await this.options.client.send(
@@ -965,5 +1074,14 @@ export class S3ClusterRemoteWorkerArtifactStore
) {
throw new S3ClusterRemoteWorkerArtifactStoreError('integrity_mismatch');
}
const eTag = canonicalETag(output.ETag);
const versionId =
output.VersionId === undefined
? undefined
: canonicalVersionId(output.VersionId);
return Object.freeze({
eTag,
...(versionId === undefined ? {} : { versionId }),
});
}
}
packages/ql3-cluster-control/src/artifact/workerArtifactBinding.ts
+4 -4
View File
@@ -5,9 +5,11 @@ import {
} from './s3ArtifactStore';
import type { ClusterRemoteWorkerArtifactStore } from '../remote-execution/remoteWorkerCompletionService';
import type { ClusterWorkerArtifactS3Config } from '../worker-ingress/workerIngressConfig';
import type { ClusterRunAttemptLogRetirementStore } from '../run/runAttemptLogRetentionLifecycle';
export interface ClusterWorkerArtifactBinding {
readonly store: ClusterRemoteWorkerArtifactStore;
readonly store: ClusterRemoteWorkerArtifactStore &
ClusterRunAttemptLogRetirementStore;
close(): Promise<void>;
}
@@ -19,9 +21,7 @@ export function createClusterWorkerArtifactBinding(
}
const client = createS3ClusterRemoteWorkerArtifactClient({
region: config.region,
...(config.endpoint === undefined
? {}
: { endpoint: config.endpoint }),
...(config.endpoint === undefined ? {} : { endpoint: config.endpoint }),
forcePathStyle: config.forcePathStyle,
});
const store = new S3ClusterRemoteWorkerArtifactStore({
packages/ql3-cluster-control/src/production-process/config.ts
+101
View File
@@ -12,6 +12,16 @@ import {
} from '@qinglong/cluster-postgres/runtime';
import { ClusterControlAvailabilityFence } from '../database/availability';
import type { ClusterControlHttpSurfaceOptions } from '../transport/httpSurface';
import {
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_CLAIMS,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS,
MIN_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
} from '@qinglong/runtime-core/cluster-run-attempt-log-retention';
import {
MAX_RUN_ATTEMPT_LOG_RETENTION_MS,
MIN_RUN_ATTEMPT_LOG_RETENTION_MS,
} from '@qinglong/runtime-core/run-attempt-log-retention';
export type ClusterControlEnvironment = Readonly<
Record<string, string | undefined>
@@ -33,6 +43,20 @@ export interface EnabledClusterControlConfig {
readonly security: Readonly<{
apiCredentialPepper: string;
}>;
readonly logRetention:
| Readonly<{ readonly enabled: false }>
| Readonly<{
readonly enabled: true;
readonly retentionMs: number;
readonly claimLimit: number;
readonly leaseMs: number;
readonly maximumCycleMs: number;
readonly retryBaseMs: number;
readonly retryMaximumMs: number;
readonly maximumFailures: number;
readonly intervalMs: number;
readonly stopTimeoutMs: number;
}>;
}
export type ClusterControlConfig =
@@ -222,6 +246,82 @@ function apiCredentialPepper(environment: ClusterControlEnvironment): string {
return value;
}
function logRetentionConfig(
environment: ClusterControlEnvironment,
): EnabledClusterControlConfig['logRetention'] {
if (!booleanValue(environment, 'QL3_CLUSTER_LOG_RETENTION_ENABLED', true)) {
return Object.freeze({ enabled: false as const });
}
const leaseMs = integerValue(
environment,
'QL3_CLUSTER_LOG_RETENTION_LEASE_MS',
30_000,
MIN_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
);
const retryBaseMs = integerValue(
environment,
'QL3_CLUSTER_LOG_RETENTION_RETRY_BASE_MS',
5_000,
0,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS,
);
return Object.freeze({
enabled: true as const,
retentionMs: integerValue(
environment,
'QL3_CLUSTER_LOG_RETENTION_MS',
30 * 24 * 60 * 60_000,
MIN_RUN_ATTEMPT_LOG_RETENTION_MS,
MAX_RUN_ATTEMPT_LOG_RETENTION_MS,
),
claimLimit: integerValue(
environment,
'QL3_CLUSTER_LOG_RETENTION_CLAIM_LIMIT',
4,
1,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_CLAIMS,
),
leaseMs,
maximumCycleMs: integerValue(
environment,
'QL3_CLUSTER_LOG_RETENTION_CYCLE_BUDGET_MS',
10_000,
100,
leaseMs - 500,
),
retryBaseMs,
retryMaximumMs: integerValue(
environment,
'QL3_CLUSTER_LOG_RETENTION_RETRY_MAX_MS',
60 * 60_000,
retryBaseMs,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS,
),
maximumFailures: integerValue(
environment,
'QL3_CLUSTER_LOG_RETENTION_MAX_FAILURES',
8,
1,
32,
),
intervalMs: integerValue(
environment,
'QL3_CLUSTER_LOG_RETENTION_INTERVAL_MS',
60_000,
1_000,
24 * 60 * 60_000,
),
stopTimeoutMs: integerValue(
environment,
'QL3_CLUSTER_LOG_RETENTION_STOP_TIMEOUT_MS',
10_000,
100,
30_000,
),
});
}
/**
* Parses the profile gate before reading PostgreSQL configuration. A disabled
* cluster-control therefore does not touch its runtime credential source.
@@ -345,6 +445,7 @@ export function loadClusterControlConfig(
security: Object.freeze({
apiCredentialPepper: apiCredentialPepper(environment),
}),
logRetention: logRetentionConfig(environment),
};
return Object.freeze(config);
}
packages/ql3-cluster-control/src/production-process/processApplication.ts
+47 -23
View File
@@ -38,6 +38,7 @@ export interface ClusterControlProcessEvent {
scope:
| 'scheduler'
| 'cancellation-convergence'
| 'log-retention'
| 'database'
| 'worker-ingress';
name: string;
@@ -78,10 +79,7 @@ export class ClusterControlProcessError extends Error {
| 'QL3_CLUSTER_CONTROL_PROCESS_CONFIG_INVALID'
| 'QL3_CLUSTER_CONTROL_PROCESS_DISABLED';
constructor(
code: ClusterControlProcessError['code'],
message: string,
) {
constructor(code: ClusterControlProcessError['code'], message: string) {
super(message);
this.name = 'ClusterControlProcessError';
this.code = code;
@@ -103,10 +101,7 @@ function processConfiguration(environment: ClusterControlEnvironment): {
);
}
const replicaId = environment.QL3_CLUSTER_REPLICA_ID;
if (
typeof replicaId !== 'string' ||
!REPLICA_ID_PATTERN.test(replicaId)
) {
if (typeof replicaId !== 'string' || !REPLICA_ID_PATTERN.test(replicaId)) {
throw new ClusterControlProcessError(
'QL3_CLUSTER_CONTROL_PROCESS_CONFIG_INVALID',
'QL3_CLUSTER_REPLICA_ID must be a stable safe identifier',
@@ -205,9 +200,11 @@ export async function runProductionClusterControlProcess(
let resolveSignal:
| ((signal: ClusterControlProcessSignal) => void)
| undefined;
const requestedSignal = new Promise<ClusterControlProcessSignal>((resolve) => {
resolveSignal = resolve;
});
const requestedSignal = new Promise<ClusterControlProcessSignal>(
(resolve) => {
resolveSignal = resolve;
},
);
let acceptedSignal = false;
const unsubscribe = options.signals.subscribe((signal) => {
if (acceptedSignal) return;
@@ -230,6 +227,14 @@ export async function runProductionClusterControlProcess(
);
}
artifactBinding = await createBinding(workerIngress.artifact);
if (
config.logRetention.enabled &&
typeof artifactBinding?.store?.retire !== 'function'
) {
throw new TypeError(
'Cluster Worker Artifact binding has no log retirement capability',
);
}
if (
workerIngress.secret !== undefined &&
workerSecretProvider === undefined
@@ -274,15 +279,40 @@ export async function runProductionClusterControlProcess(
event(replicaId, {
level: 'error',
event: 'runtime_diagnostic',
diagnostic: diagnosticFact(
'cancellation-convergence',
error,
),
diagnostic: diagnosticFact('cancellation-convergence', error),
}),
),
).catch(() => undefined);
},
},
...(workerIngress !== undefined && config.logRetention.enabled
? {
logRetention: {
store: artifactBinding!.store,
ownerId: replicaId,
retentionMs: config.logRetention.retentionMs,
claimLimit: config.logRetention.claimLimit,
leaseMs: config.logRetention.leaseMs,
maximumCycleMs: config.logRetention.maximumCycleMs,
retryBaseMs: config.logRetention.retryBaseMs,
retryMaximumMs: config.logRetention.retryMaximumMs,
maximumFailures: config.logRetention.maximumFailures,
intervalMs: config.logRetention.intervalMs,
stopTimeoutMs: config.logRetention.stopTimeoutMs,
onDiagnostic(error: unknown) {
void Promise.resolve(
options.emit(
event(replicaId, {
level: 'error',
event: 'runtime_diagnostic',
diagnostic: diagnosticFact('log-retention', error),
}),
),
).catch(() => undefined);
},
},
}
: {}),
...(workerIngress === undefined
? {}
: {
@@ -298,10 +328,7 @@ export async function runProductionClusterControlProcess(
event(replicaId, {
level: 'error',
event: 'runtime_diagnostic',
diagnostic: diagnosticFact(
'worker-ingress',
error,
),
diagnostic: diagnosticFact('worker-ingress', error),
}),
),
).catch(() => undefined);
@@ -395,10 +422,7 @@ export async function runProductionClusterControlProcess(
unsubscribe();
resolveSignal = undefined;
let cleanupError: unknown;
if (
application?.status === 'active' &&
!applicationStopStarted
) {
if (application?.status === 'active' && !applicationStopStarted) {
try {
applicationStopStarted = true;
await application.stop();
packages/ql3-cluster-control/src/run/runAttemptLogRetentionLifecycle.ts
+490
View File
@@ -0,0 +1,490 @@
// Run owns bounded multi-replica Cluster log retirement and one lifecycle timer.
import {
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_CLAIMS,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS,
MIN_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
type ClusterRunAttemptLogRetentionClaim,
type ClusterRunAttemptLogRetentionClaimRepository,
type ClusterRunAttemptLogRetentionFailureCode,
} from '@qinglong/runtime-core/cluster-run-attempt-log-retention';
import {
createRunAttemptLogRetirementRecord,
MAX_RUN_ATTEMPT_LOG_RETENTION_MS,
MIN_RUN_ATTEMPT_LOG_RETENTION_MS,
type RunAttemptLogRetentionCandidate,
type RunAttemptLogRetirementStore,
type RunAttemptLogRetirementStoreResult,
} from '@qinglong/runtime-core/run-attempt-log-retention';
const OWNER_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/;
const MIN_SETTLEMENT_BUDGET_MS = 500;
export interface ClusterRunAttemptLogRetirementStore
extends RunAttemptLogRetirementStore {
retire(
candidate: Readonly<RunAttemptLogRetentionCandidate>,
signal?: AbortSignal,
): Promise<Readonly<RunAttemptLogRetirementStoreResult>>;
}
export interface ClusterRunAttemptLogRetentionCoordinatorOptions {
readonly ownerId: string;
readonly retentionMs: number;
readonly claimLimit: number;
readonly leaseMs: number;
readonly maximumCycleMs: number;
readonly retryBaseMs: number;
readonly retryMaximumMs: number;
readonly maximumFailures: number;
}
export interface ClusterRunAttemptLogRetentionCycleEntry {
readonly attemptId: string;
readonly outcome:
| 'deleted'
| 'already_absent'
| 'retry'
| 'manual'
| 'fenced';
}
export interface ClusterRunAttemptLogRetentionCycleSummary {
readonly status: 'complete' | 'saturated' | 'budget_exhausted';
readonly claimed: number;
readonly attempted: number;
readonly retired: number;
readonly alreadyAbsent: number;
readonly retried: number;
readonly manual: number;
readonly fenced: number;
readonly hasMore: boolean;
readonly entries: readonly Readonly<ClusterRunAttemptLogRetentionCycleEntry>[];
}
function integer(
name: string,
value: number,
minimum: number,
maximum: number,
): number {
if (!Number.isSafeInteger(value) || value < minimum || value > maximum) {
throw new RangeError(`${name} must be between ${minimum} and ${maximum}`);
}
return value;
}
function prepareOptions(
options: ClusterRunAttemptLogRetentionCoordinatorOptions,
): Readonly<ClusterRunAttemptLogRetentionCoordinatorOptions> {
const allowed = new Set([
'claimLimit',
'leaseMs',
'maximumCycleMs',
'maximumFailures',
'ownerId',
'retentionMs',
'retryBaseMs',
'retryMaximumMs',
]);
if (
!options ||
typeof options !== 'object' ||
Array.isArray(options) ||
Object.keys(options).some((key) => !allowed.has(key)) ||
!OWNER_PATTERN.test(options.ownerId)
) {
throw new TypeError(
'Cluster Run Attempt log retention options are invalid',
);
}
const leaseMs = integer(
'Cluster Run Attempt log retention lease',
options.leaseMs,
MIN_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_LEASE_MS,
);
const maximumCycleMs = integer(
'Cluster Run Attempt log retention cycle budget',
options.maximumCycleMs,
100,
leaseMs - MIN_SETTLEMENT_BUDGET_MS,
);
const retryBaseMs = integer(
'Cluster Run Attempt log retention retry base',
options.retryBaseMs,
0,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS,
);
const retryMaximumMs = integer(
'Cluster Run Attempt log retention retry maximum',
options.retryMaximumMs,
retryBaseMs,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_RETRY_DELAY_MS,
);
return Object.freeze({
ownerId: options.ownerId,
retentionMs: integer(
'Cluster Run Attempt log retention duration',
options.retentionMs,
MIN_RUN_ATTEMPT_LOG_RETENTION_MS,
MAX_RUN_ATTEMPT_LOG_RETENTION_MS,
),
claimLimit: integer(
'Cluster Run Attempt log retention claim limit',
options.claimLimit,
1,
MAX_CLUSTER_RUN_ATTEMPT_LOG_RETENTION_CLAIMS,
),
leaseMs,
maximumCycleMs,
retryBaseMs,
retryMaximumMs,
maximumFailures: integer(
'Cluster Run Attempt log retention failure limit',
options.maximumFailures,
1,
32,
),
});
}
function failureCode(error: unknown): ClusterRunAttemptLogRetentionFailureCode {
return (error as { readonly reason?: unknown })?.reason ===
'integrity_mismatch'
? 'artifact_integrity_mismatch'
: 'artifact_unavailable';
}
function retryDelay(
failureCount: number,
options: Readonly<ClusterRunAttemptLogRetentionCoordinatorOptions>,
): number {
const multiplier = 2 ** Math.min(failureCount, 30);
return Math.min(options.retryMaximumMs, options.retryBaseMs * multiplier);
}
function isAbortSignal(value: unknown): value is AbortSignal {
return (
!!value &&
typeof value === 'object' &&
typeof (value as AbortSignal).aborted === 'boolean' &&
typeof (value as AbortSignal).addEventListener === 'function' &&
typeof (value as AbortSignal).removeEventListener === 'function'
);
}
export class ClusterRunAttemptLogRetentionCoordinator {
private readonly options: Readonly<ClusterRunAttemptLogRetentionCoordinatorOptions>;
constructor(
private readonly repository: ClusterRunAttemptLogRetentionClaimRepository,
private readonly store: ClusterRunAttemptLogRetirementStore,
options: ClusterRunAttemptLogRetentionCoordinatorOptions,
) {
if (
!repository ||
typeof repository.claim !== 'function' ||
typeof repository.settle !== 'function' ||
!store ||
typeof store.retire !== 'function'
) {
throw new TypeError(
'Cluster Run Attempt log retention dependencies are invalid',
);
}
this.options = prepareOptions(options);
}
async runOnce(
externalSignal?: AbortSignal,
): Promise<Readonly<ClusterRunAttemptLogRetentionCycleSummary>> {
if (externalSignal !== undefined && !isAbortSignal(externalSignal)) {
throw new TypeError(
'Cluster Run Attempt log retention signal is invalid',
);
}
if (externalSignal?.aborted) throw externalSignal.reason;
const controller = new AbortController();
const forwardAbort = () => controller.abort(externalSignal?.reason);
externalSignal?.addEventListener('abort', forwardAbort, { once: true });
const timeout = setTimeout(
() =>
controller.abort(
new Error('Cluster log retention cycle budget expired'),
),
this.options.maximumCycleMs,
);
timeout.unref?.();
try {
const page = await this.repository.claim({
ownerId: this.options.ownerId,
retentionMs: this.options.retentionMs,
limit: this.options.claimLimit,
leaseMs: this.options.leaseMs,
});
const entries: ClusterRunAttemptLogRetentionCycleEntry[] = [];
let attempted = 0;
let retired = 0;
let alreadyAbsent = 0;
let retried = 0;
let manual = 0;
let fenced = 0;
for (const claim of page.claims) {
if (controller.signal.aborted) break;
attempted += 1;
let result: Readonly<RunAttemptLogRetirementStoreResult>;
try {
result = await this.store.retire(claim.candidate, controller.signal);
} catch (error) {
if (controller.signal.aborted) break;
const settled = await this.settleFailure(claim, failureCode(error));
if (settled === 'fenced') {
fenced += 1;
entries.push({
attemptId: claim.candidate.attemptId,
outcome: 'fenced',
});
} else if (claim.failureCount + 1 >= this.options.maximumFailures) {
manual += 1;
entries.push({
attemptId: claim.candidate.attemptId,
outcome: 'manual',
});
} else {
retried += 1;
entries.push({
attemptId: claim.candidate.attemptId,
outcome: 'retry',
});
}
continue;
}
if (controller.signal.aborted) break;
let record;
try {
record = createRunAttemptLogRetirementRecord({
...claim.candidate,
eligibleAtMs: claim.eligibleAtMs,
retiredAtMs: claim.observedAtMs,
...result,
});
} catch {
const settled = await this.settleFailure(
claim,
'retirement_record_unavailable',
);
if (settled === 'fenced') {
fenced += 1;
entries.push({
attemptId: claim.candidate.attemptId,
outcome: 'fenced',
});
} else if (claim.failureCount + 1 >= this.options.maximumFailures) {
manual += 1;
entries.push({
attemptId: claim.candidate.attemptId,
outcome: 'manual',
});
} else {
retried += 1;
entries.push({
attemptId: claim.candidate.attemptId,
outcome: 'retry',
});
}
continue;
}
const settled = await this.repository.settle(claim, {
status: 'retired',
record,
});
if (settled === 'fenced') {
fenced += 1;
entries.push({
attemptId: claim.candidate.attemptId,
outcome: 'fenced',
});
continue;
}
if (record.disposition === 'already_absent') alreadyAbsent += 1;
else retired += 1;
entries.push({
attemptId: claim.candidate.attemptId,
outcome: record.disposition,
});
}
const budgetExhausted = controller.signal.aborted;
return Object.freeze({
status: budgetExhausted
? ('budget_exhausted' as const)
: page.hasMore
? ('saturated' as const)
: ('complete' as const),
claimed: page.claims.length,
attempted,
retired,
alreadyAbsent,
retried,
manual,
fenced,
hasMore: page.hasMore,
entries: Object.freeze(entries.map((entry) => Object.freeze(entry))),
});
} finally {
clearTimeout(timeout);
externalSignal?.removeEventListener('abort', forwardAbort);
}
}
private settleFailure(
claim: Readonly<ClusterRunAttemptLogRetentionClaim>,
code: ClusterRunAttemptLogRetentionFailureCode,
): Promise<'settled' | 'fenced'> {
if (claim.failureCount + 1 >= this.options.maximumFailures) {
return this.repository.settle(claim, {
status: 'manual',
failureCode: code,
});
}
return this.repository.settle(claim, {
status: 'retry',
delayMs: retryDelay(claim.failureCount, this.options),
failureCode: code,
});
}
}
export interface ClusterRunAttemptLogRetentionLifecycleOptions {
readonly intervalMs: number;
readonly stopTimeoutMs: number;
readonly onDiagnostic?: (
error: unknown,
summary?: Readonly<ClusterRunAttemptLogRetentionCycleSummary>,
) => void | Promise<void>;
}
export class ClusterRunAttemptLogRetentionLifecycle {
private timer: NodeJS.Timeout | undefined;
private inFlight:
| Promise<Readonly<ClusterRunAttemptLogRetentionCycleSummary>>
| undefined;
private controller: AbortController | undefined;
private stopPromise: Promise<'stopped' | 'timed_out'> | undefined;
private running = false;
private stopping = false;
constructor(
private readonly coordinator: Pick<
ClusterRunAttemptLogRetentionCoordinator,
'runOnce'
>,
private readonly options: ClusterRunAttemptLogRetentionLifecycleOptions,
) {
if (
!coordinator ||
typeof coordinator.runOnce !== 'function' ||
!options ||
typeof options !== 'object' ||
Array.isArray(options) ||
!Number.isSafeInteger(options.intervalMs) ||
options.intervalMs < 1_000 ||
options.intervalMs > 24 * 60 * 60_000 ||
!Number.isSafeInteger(options.stopTimeoutMs) ||
options.stopTimeoutMs < 100 ||
options.stopTimeoutMs > 30_000 ||
(options.onDiagnostic !== undefined &&
typeof options.onDiagnostic !== 'function')
) {
throw new TypeError(
'Cluster Run Attempt log retention lifecycle options are invalid',
);
}
}
start(): 'started' {
if (!this.running && !this.stopping) {
this.running = true;
this.schedule();
}
return 'started';
}
runOnce(): Promise<Readonly<ClusterRunAttemptLogRetentionCycleSummary>> {
if (this.stopping) {
return Promise.reject(
new Error('Cluster Run Attempt log retention lifecycle is stopping'),
);
}
if (this.inFlight) return this.inFlight;
const controller = new AbortController();
this.controller = controller;
const work = this.coordinator.runOnce(controller.signal).finally(() => {
if (this.inFlight === work) {
this.inFlight = undefined;
this.controller = undefined;
}
});
this.inFlight = work;
return work;
}
stopAndDrain(): Promise<'stopped' | 'timed_out'> {
if (this.stopPromise) return this.stopPromise;
this.stopping = true;
this.running = false;
if (this.timer) clearTimeout(this.timer);
this.timer = undefined;
this.controller?.abort(
new Error('Cluster Run Attempt log retention lifecycle is stopping'),
);
this.stopPromise = (async () => {
const work = this.inFlight;
if (!work) return 'stopped' as const;
let timeout: NodeJS.Timeout | undefined;
try {
return await Promise.race([
work.then(
() => 'stopped' as const,
() => 'stopped' as const,
),
new Promise<'timed_out'>((resolve) => {
timeout = setTimeout(
() => resolve('timed_out' as const),
this.options.stopTimeoutMs,
);
timeout.unref?.();
}),
]);
} finally {
if (timeout) clearTimeout(timeout);
}
})();
return this.stopPromise;
}
private schedule(): void {
if (!this.running || this.timer) return;
this.timer = setTimeout(() => {
this.timer = undefined;
if (!this.running) return;
void this.runOnce()
.then((summary) => this.diagnostic(undefined, summary))
.catch((error) => this.diagnostic(error))
.finally(() => this.schedule());
}, this.options.intervalMs);
this.timer.unref?.();
}
private async diagnostic(
error: unknown,
summary?: Readonly<ClusterRunAttemptLogRetentionCycleSummary>,
): Promise<void> {
if (this.stopping) return;
try {
await this.options.onDiagnostic?.(error, summary);
} catch {
// Diagnostics cannot own or stop retention.
}
}
}