diff --git a/tests/functional/ctst/common/common.ts b/tests/functional/ctst/common/common.ts index 3a5f29c19..6bc0bc81d 100644 --- a/tests/functional/ctst/common/common.ts +++ b/tests/functional/ctst/common/common.ts @@ -60,6 +60,7 @@ export async function cleanS3Bucket( world.deleteKeyFromCommand('key'); world.deleteKeyFromCommand('versionId'); } + await S3.deleteBucketReplication(world.getCommandParameters()).catch(() => {}); await S3.deleteBucketLifecycle(world.getCommandParameters()); await S3.deleteBucket(world.getCommandParameters()); } @@ -274,7 +275,7 @@ Given('a transition workflow to {string} location', async function (this: Zenko, Given('a replication configuration to {string} location', async function (this: Zenko, replicationLocation: string) { - this.addToSaved('replicationLocation', replicationLocation); + this.addToSaved('replicationLocations', [replicationLocation]); await putBucketReplication.call(this, this.getSaved('bucketName'), replicationLocation); }); diff --git a/tests/functional/ctst/features/replication/crrCascade.feature b/tests/functional/ctst/features/replication/crrCascade.feature index c06c4f47b..b8f26936c 100644 --- a/tests/functional/ctst/features/replication/crrCascade.feature +++ b/tests/functional/ctst/features/replication/crrCascade.feature @@ -9,7 +9,7 @@ Feature: CRR Cascade Replication @ReplicationTest @CRRCascade Scenario: Cascade replication nominal case : A -> B -> C - Given cascade replication accounts are registered + Given CRR replication accounts are registered And a versioned bucket exists in location "crr-location-a" And a versioned bucket exists in location "crr-location-b" And a versioned bucket exists in location "crr-location-c" @@ -26,7 +26,7 @@ Feature: CRR Cascade Replication @ReplicationTest @CRRCascade Scenario: Cascade replication with loop : A -> B -> C -> A - Given cascade replication accounts are registered + Given CRR replication accounts are registered And a versioned bucket exists in location "crr-location-a" And a versioned bucket exists in location "crr-location-b" And a versioned bucket exists in location "crr-location-c" @@ -45,7 +45,7 @@ Feature: CRR Cascade Replication @ReplicationTest @CRRCascade Scenario: Multiple writes across a loop converge to the last write - Given cascade replication accounts are registered + Given CRR replication accounts are registered And a versioned bucket exists in location "crr-location-a" And a versioned bucket exists in location "crr-location-b" And a versioned bucket exists in location "crr-location-c" diff --git a/tests/functional/ctst/features/replication/crrReplicationS3utils.feature b/tests/functional/ctst/features/replication/crrReplicationS3utils.feature index e9d164406..a953809c2 100644 --- a/tests/functional/ctst/features/replication/crrReplicationS3utils.feature +++ b/tests/functional/ctst/features/replication/crrReplicationS3utils.feature @@ -20,7 +20,7 @@ Feature: Replication And a replication configuration to "awsbackendmismatch" location When the job to replicate existing objects with status "NEW" is executed Then the object replication should "succeed" within 300 seconds - And the replicated object should be the same as the source object + And the replicated object should match the source on every configured destination @2.12.0 @PreMerge @@ -35,4 +35,4 @@ Feature: Replication When the destination bucket on the location is created again And the job to replicate existing objects with status "FAILED" is executed Then the object replication should "succeed" within 300 seconds - And the replicated object should be the same as the source object \ No newline at end of file + And the replicated object should match the source on every configured destination \ No newline at end of file diff --git a/tests/functional/ctst/features/replication/replication.feature b/tests/functional/ctst/features/replication/replication.feature index 21f425ab5..d74b36c5a 100644 --- a/tests/functional/ctst/features/replication/replication.feature +++ b/tests/functional/ctst/features/replication/replication.feature @@ -1,22 +1,21 @@ -Feature: Live Object Replication - This feature tests live replication of objects across different sizes, - verifying that oplog location stripping does not affect replication - correctness. - - The objectRefreshSizeThresholdMB is set to 2 in the Zenko CR, meaning - objects >= 2 MiB have their location stripped from the MongoDB change - stream event. The replication consumer detects this and re-fetches - metadata from the source before replicating. +Feature: Bucket Replication + End-to-end coverage of Zenko's bucket replication, covering both + cloud destinations (AWS / GCP / Azure backends) and CRR loopback + destinations (replication between Zenko accounts). @2.14.0 @PreMerge @ReplicationTest + # Targets the location-stripping threshold: objectRefreshSizeThresholdMB + # is set to 2 in the Zenko CR, so objects >= 2 MiB have their location + # stripped from the MongoDB change stream event. The replication consumer + # detects this and re-fetches metadata from the source before replicating. Scenario Outline: Object of replicates with location stripping threshold Given an existing bucket "repl-strip-" "with" versioning, "without" ObjectLock "without" retention mode And a replication configuration to "awsbackendmismatch" location And 1 objects "repl-strip-obj" of size bytes Then the object replication should "succeed" within 300 seconds - And the replicated object should be the same as the source object + And the replicated object should match the source on every configured destination Examples: | sizeBytes | objectDescription | @@ -24,3 +23,148 @@ Feature: Live Object Replication | 1048576 | 1 MiB (below threshold) | | 2097152 | 2 MiB (at threshold, stripped) | | 4194304 | 4 MiB (above threshold, stripped) | + + @2.16.0 + @PreMerge + @ReplicationTest + # Multi-destination replication: a single replication configuration + # carries several rules targeting distinct destinations, optionally with + # overlapping prefixes resolved by Priority. The legacy V1 format and + # the non-standard comma-separated StorageClass form remain supported + # for backward compatibility and round-trip unchanged through + # getBucketReplication. + Scenario: V2 replication fans out to two cloud destinations + Given an existing bucket "multi-cloud-2" "with" versioning, "without" ObjectLock "without" retention mode + And a valid multi-destination replication configuration with rules: + | id | prefix | priority | location | + | rule-a | | 1 | awsbackendmismatch | + | rule-b | | 2 | awsbackend | + And 1 objects "multi-cloud-obj" of size 1024 bytes + Then the object replication should "succeed" within 300 seconds + And the replicated object should match the source on every configured destination + + @2.16.0 + @PreMerge + @ReplicationTest + Scenario: V2 replication fans out to a mix of cloud and CRR destinations + Given CRR replication accounts are registered + And a versioned bucket exists in location "crr-location-a" + And a versioned bucket exists in location "crr-location-b" + And an existing bucket "multi-fanout" "with" versioning, "without" ObjectLock "without" retention mode + And a valid multi-destination replication configuration with rules: + | id | prefix | priority | location | + | rule-a | | 1 | awsbackendmismatch | + | rule-b | | 2 | awsbackend | + | rule-c | | 3 | crr-location-a | + | rule-d | | 4 | crr-location-b | + And 1 objects "multi-fanout-obj" of size 1024 bytes + Then the object replication should "succeed" within 300 seconds + And the replicated object should match the source on every configured destination + + @2.16.0 + @PreMerge + @ReplicationTest + Scenario: CRR + cloud mixed destination replication + Given CRR replication accounts are registered + And a versioned bucket exists in location "crr-location-a" + And an existing bucket "multi-mixed" "with" versioning, "without" ObjectLock "without" retention mode + And a valid multi-destination replication configuration with rules: + | id | prefix | priority | location | + | rule-c | | 1 | crr-location-a | + | rule-x | | 2 | awsbackendmismatch | + And 1 objects "mixed-obj" of size 1024 bytes + Then the object replication should "succeed" within 300 seconds + And the replicated object should match the source on every configured destination + + @2.16.0 + @PreMerge + @ReplicationTest + Scenario: V2 rules with overlapping prefixes to different destinations all apply + Given an existing bucket "multi-overlap" "with" versioning, "without" ObjectLock "without" retention mode + And a valid multi-destination replication configuration with rules: + | id | prefix | priority | location | + | broad | | 1 | awsbackendmismatch | + | narrow | docs- | 2 | awsbackend | + And 1 objects "docs-report" of size 1024 bytes + Then the object replication should "succeed" within 300 seconds + And the replicated object should match the source on every configured destination + + @2.16.0 + @PreMerge + @ReplicationTest + Scenario: Disabled rule does not replicate + Given an existing bucket "multi-disabled" "with" versioning, "without" ObjectLock "without" retention mode + And a valid multi-destination replication configuration with rules: + | id | prefix | priority | status | location | + | active | | 1 | Enabled | awsbackendmismatch | + | off | | 2 | Disabled | awsbackend | + And 1 objects "disabled-obj" of size 1024 bytes + Then the object replication should "succeed" within 300 seconds + And the object should not be replicated to "awsbackend" + And the replicated object on "awsbackendmismatch" should match the source + + @2.16.0 + @PreMerge + @ReplicationTest + Scenario: Overlap on same destination without distinct priorities is rejected + Given an existing bucket "multi-bad-overlap" "with" versioning, "without" ObjectLock "without" retention mode + And an invalid multi-destination replication configuration with rules: + | id | prefix | priority | location | + | dup-a | | 1 | awsbackendmismatch | + | dup-b | docs-| 1 | awsbackendmismatch | + Then the replication configuration request should be rejected with "InvalidRequest" + + @2.16.0 + @PreMerge + @ReplicationTest + Scenario: Same destination overlap with distinct priorities is accepted + Given an existing bucket "multi-priority-dedup" "with" versioning, "without" ObjectLock "without" retention mode + And a valid multi-destination replication configuration with rules: + | id | prefix | priority | location | + | lo | | 1 | awsbackendmismatch | + | hi | docs-| 2 | awsbackendmismatch | + And 1 objects "docs-winner" of size 1024 bytes + Then the object replication should "succeed" within 300 seconds + And the replicated object on "awsbackendmismatch" should match the source + + @2.16.0 + @PreMerge + @ReplicationTest + Scenario: V1 configuration replicates and round-trips through getBucketReplication + Given an existing bucket "multi-v1-compat" "with" versioning, "without" ObjectLock "without" retention mode + And a valid multi-destination replication configuration with rules: + | id | prefix | location | + | v1-only | | awsbackendmismatch | + And 1 objects "v1-obj" of size 1024 bytes + Then the object replication should "succeed" within 300 seconds + And the replicated object on "awsbackendmismatch" should match the source + And getBucketReplication should return a "V1" configuration with 1 rules + + @2.16.0 + @PreMerge + @ReplicationTest + Scenario: Legacy comma-separated StorageClass replicates to multiple sites + Given an existing bucket "multi-legacy-comma" "with" versioning, "without" ObjectLock "without" retention mode + And a legacy comma-StorageClass replication configuration to locations "awsbackendmismatch,awsbackend" + And 1 objects "legacy-obj" of size 1024 bytes + Then the object replication should "succeed" within 300 seconds + And the replicated object should match the source on every configured destination + And getBucketReplication should return a "V1" configuration with 1 rules + And getBucketReplication rule 0 should have StorageClass "awsbackendmismatch,awsbackend" + + @2.16.0 + @PreMerge + @ReplicationTest + Scenario: Two CRR destinations replicate independently + Given CRR replication accounts are registered + And a versioned bucket exists in location "crr-location-a" + And a versioned bucket exists in location "crr-location-b" + And an existing bucket "multi-crr-pair" "with" versioning, "without" ObjectLock "without" retention mode + And a valid multi-destination replication configuration with rules: + | id | prefix | priority | location | + | crr-a | | 1 | crr-location-a | + | crr-b | | 2 | crr-location-b | + And 1 objects "crr-pair-obj" of size 1024 bytes + Then the object replication should "succeed" within 300 seconds + And the replicated object on "crr-location-a" should match the source + And the replicated object on "crr-location-b" should match the source diff --git a/tests/functional/ctst/steps/crrCascade.ts b/tests/functional/ctst/steps/crrCascade.ts index 4642f3ead..f29506df3 100644 --- a/tests/functional/ctst/steps/crrCascade.ts +++ b/tests/functional/ctst/steps/crrCascade.ts @@ -11,46 +11,19 @@ import assert from 'assert'; import { Identity, IdentityEnum, Utils } from 'cli-testing'; import Zenko from 'world/Zenko'; -interface CRRAccountInfo { +export interface CRRAccountInfo { AccessKeyId: string; SecretAccessKey: string; SessionToken?: string; AccountId: string; } -Given('cascade replication accounts are registered', function (this: Zenko) { - const roleName = process.env.CRR_ROLE_NAME; - assert.ok(roleName, 'CRR_ROLE_NAME must be set'); - - const locationNames = [ - process.env.CRR_LOCATION_A_NAME, - process.env.CRR_LOCATION_B_NAME, - process.env.CRR_LOCATION_C_NAME, - ]; - assert.ok(locationNames.every(Boolean), 'CRR_LOCATION_A/B/C_NAME must be set'); - - const infoByLocation: Record = {}; - for (const [i, loc] of locationNames.entries()) { - const varName = `CRR_INFO_${['A', 'B', 'C'][i]}`; - const raw = process.env[varName]; - assert.ok(raw, `${varName} must be set`); - const info = JSON.parse(raw) as CRRAccountInfo; - Identity.addIdentity(IdentityEnum.ACCOUNT, loc!, { - accessKeyId: info.AccessKeyId, - secretAccessKey: info.SecretAccessKey, - sessionToken: info.SessionToken, - }); - infoByLocation[loc!] = info; - } - - this.addToSaved('cascadeInfoByLocation', infoByLocation); - this.addToSaved('cascadeRoleName', roleName); - this.addToSaved('cascadeBuckets', {} as Record); -}); - Given('a versioned bucket exists in location {string}', async function (this: Zenko, location: string) { const bucket = `cascade-${Utils.randomString().toLowerCase()}`; Identity.useIdentity(IdentityEnum.ACCOUNT, location); + // Persist the identity so the default After-hook cleanup targets the + // same account that owns the buckets created here (and by later steps). + this.addToSaved('accountName', location); const client = this.createS3Client(); await client.send(new CreateBucketCommand({ Bucket: bucket })); await client.send(new PutBucketVersioningCommand({ diff --git a/tests/functional/ctst/steps/replication.ts b/tests/functional/ctst/steps/replication.ts index 7e0311355..48febdfb7 100644 --- a/tests/functional/ctst/steps/replication.ts +++ b/tests/functional/ctst/steps/replication.ts @@ -1,17 +1,298 @@ -import { Given, When, Then } from '@cucumber/cucumber'; -import Zenko from '../world/Zenko'; -import { createAndRunPod, getZenkoVersion } from 'steps/utils/kubernetes'; -import assert from 'assert'; -import { IdentityEnum, Identity, Utils } from 'cli-testing'; -import { - GetObjectCommand, - DeleteBucketCommand, +import { DataTable, Given, When, Then } from '@cucumber/cucumber'; +import { CreateBucketCommand, - PutBucketVersioningCommand + DeleteBucketCommand, + GetObjectCommand, + PutBucketVersioningCommand, + S3Client, } from '@aws-sdk/client-s3'; -import { getObject, headObject, getReplicationLocationConfig } from 'steps/utils/utils'; +import { IdentityEnum, Identity, S3, Utils } from 'cli-testing'; +import assert from 'assert'; +import Zenko from '../world/Zenko'; +import { createAndRunPod, getLocationConfigs, getZenkoVersion } from 'steps/utils/kubernetes'; +import { getObject, headObject, putBucketReplicationRaw } from 'steps/utils/utils'; import { safeJsonParse } from 'common/utils'; import { replicationLockTags } from 'common/hooks'; +import { CRRAccountInfo } from './crrCascade'; + +type ReplicationOutcome = 'succeed' | 'fail' | 'never happen'; + +type SourceObjectReplicationMeta = { + ReplicationStatus?: string; + LastModified?: string; + ETag?: string; + ContentLength?: number; + VersionId?: string; + Metadata?: Record; +}; + +async function getReplicationLocationConfig(world: Zenko, location: string): Promise<{ + destinationBucket: string; + locationType: string; + bucketMatch: boolean; + awsS3Client: S3Client; +}> { + const locationsConfigs = await getLocationConfigs(world); + if (!locationsConfigs[location]) { + throw new Error(`Unsupported replication location: '${location}'`); + } + return { + destinationBucket: locationsConfigs[location].details.bucketName, + locationType: locationsConfigs[location].type, + bucketMatch: locationsConfigs[location].details.bucketMatch, + awsS3Client: new S3Client({ + region: locationsConfigs[location].details.region, + endpoint: `https://${locationsConfigs[location].details.awsEndpoint}`, + credentials: { + accessKeyId: locationsConfigs[location].details.credentials.accessKey, + secretAccessKey: locationsConfigs[location].details.credentials.secretKey, + }, + tls: false, + maxAttempts: 1, + forcePathStyle: true, + }), + }; +} + +async function getBucketReplicationConfig( + this: Zenko, + srcBucket: string, +): Promise<{ ok: true; config: Record } | { ok: false; err: string }> { + this.resetCommand(); + this.addCommandParameter({ bucket: srcBucket }); + const res = await S3.getBucketReplication(this.getCommandParameters()); + if (res.err) { + return { ok: false, err: res.err }; + } + const parsed = safeJsonParse<{ ReplicationConfiguration: Record }>(res.stdout || '{}'); + if (!parsed.ok) { + return { ok: false, err: 'unparseable getBucketReplication response' }; + } + return { ok: true, config: parsed.result?.ReplicationConfiguration ?? {} }; +} + +function readPerBackendStatuses( + md: Record, + locations: string[], +): Record { + const out: Record = {}; + for (const loc of locations) { + out[loc] = md[`${loc}-replication-status`]; + } + return out; +} + +/** + * Poll the source object's HeadObject for replication outcome. + * + * - The top-level `ReplicationStatus` aggregates per-backend states (FAILED > + * PROCESSING/PENDING > COMPLETED in priority). + * - When `locations` is non-empty, each backend's per-destination status (read + * from `Metadata['-replication-status']`) must also match the + * expected outcome before the call returns successfully. + */ +async function pollReplicationOutcome( + world: Zenko, + objectName: string, + bucketName: string, + expected: ReplicationOutcome, + timeoutMs: number, + locations: string[] = [], +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + await new Promise(r => setTimeout(r, 3000)); + const res = await headObject(world, objectName, bucketName); + assert(res.stdout); + assert.strictEqual(res.statusCode, 200, `failed to headObject, ${res.statusCode}`); + const parsed = safeJsonParse(res.stdout || '{}'); + assert(parsed.ok); + const top = parsed.result?.ReplicationStatus; + const perBackend = readPerBackendStatuses(parsed.result?.Metadata || {}, locations); + const perBackendValues = Object.values(perBackend); + + switch (top) { + case 'PENDING': + case 'PROCESSING': + assert.notStrictEqual( + expected, 'never happen', + `replication status is ${top}, but expected to never happen`, + ); + continue; + case 'COMPLETED': + assert.strictEqual( + expected, 'succeed', + `replication is completed, but expected outcome was '${expected}'`, + ); + if (locations.length > 0) { + assert( + perBackendValues.every(s => s === 'COMPLETED'), + `top-level COMPLETED but per-backend statuses are ${JSON.stringify(perBackend)}`, + ); + } + return; + case 'FAILED': + assert.strictEqual( + expected, 'fail', + `replication is failed, but expected outcome was '${expected}'`, + ); + return; + case undefined: + assert.strictEqual( + expected, 'never happen', + `got undefined replication status for expected outcome '${expected}'`, + ); + return; + default: + throw new Error(`Unexpected replication status: ${top}`); + } + } + assert.fail(`Timeout: object '${objectName}' did not reach outcome '${expected}' in time`); +} + +/** + * For CRR cascade destinations the destination bucket lives on the cascade + * account and the S3 endpoint is the source Zenko itself. Build a + * destination-account-aware S3 client + look up the cascade-created bucket + * name from world state. + */ +function getCRRReplicaContext( + world: Zenko, + location: string, +): { destinationBucket: string; awsS3Client: S3Client } | undefined { + const cascadeInfo = world.getSaved>('cascadeInfoByLocation'); + const cascadeBuckets = world.getSaved>('cascadeBuckets'); + if (!cascadeInfo?.[location] || !cascadeBuckets?.[location]) { + return undefined; + } + const info = cascadeInfo[location]; + return { + destinationBucket: cascadeBuckets[location], + awsS3Client: new S3Client({ + region: 'us-east-1', + endpoint: 'http://s3.zenko.local', + credentials: { + accessKeyId: info.AccessKeyId, + secretAccessKey: info.SecretAccessKey, + sessionToken: info.SessionToken, + }, + tls: false, + maxAttempts: 1, + forcePathStyle: true, + }), + }; +} + +async function assertReplicaMatchesSource(world: Zenko, location: string): Promise { + const objectName = world.getSaved('objectName'); + const bucketSource = world.getSaved('bucketName'); + const crrCtx = getCRRReplicaContext(world, location); + // CRR loopback: the cascade-created destination bucket receives the + // object under the same key as the source — no source-bucket prefix. + const { destinationBucket, bucketMatch, awsS3Client } = crrCtx + ? { ...crrCtx, bucketMatch: true } + : await getReplicationLocationConfig(world, location); + + const key = bucketMatch ? objectName : `${bucketSource}/${objectName}`; + const replicaObj = await awsS3Client.send(new GetObjectCommand({ + Bucket: destinationBucket, + Key: key, + })); + + const sourceResponse = await getObject(world, objectName, bucketSource); + assert.strictEqual( + sourceResponse.statusCode, 200, + `failed to getObject, ${sourceResponse.statusCode}`, + ); + const sourceObj = safeJsonParse(sourceResponse.stdout || '{}'); + assert(sourceObj.ok); + assert.strictEqual(sourceObj.result?.ContentLength, replicaObj.ContentLength); + // CRR loopback only writes the per-destination status; cloud backends + // also stamp version-id / scal-version-id. + if (!crrCtx) { + assert.strictEqual( + sourceObj.result?.Metadata?.[`${location}-version-id`], + replicaObj.VersionId, + ); + assert.strictEqual( + sourceObj.result?.VersionId, + replicaObj.Metadata?.['scal-version-id'], + ); + assert.strictEqual(replicaObj.Metadata?.['scal-replication-status'], 'REPLICA'); + } + assert.strictEqual( + sourceObj.result?.Metadata?.[`${location}-replication-status`], + 'COMPLETED', + ); +} + +type RuleRow = { + id?: string; + prefix?: string; + priority?: string; + status?: string; + location: string; + bucket?: string; + account?: string; + deleteMarkerReplication?: string; +}; + +const DEFAULT_ROLE = 'arn:aws:iam::root:role/s3-replication-role'; + +function buildConfigFromRows(world: Zenko, srcBucket: string, rows: RuleRow[]): object { + // Optional cascade state, populated by the cascade-tests' "cascade + // replication accounts are registered" + per-location bucket steps. + // Absent for cloud-only scenarios — in which case the function emits + // the standard single-role config unchanged. + const cascadeInfoByLocation = world.getSaved>('cascadeInfoByLocation'); + const cascadeRoleName = world.getSaved('cascadeRoleName'); + const cascadeBuckets = world.getSaved>('cascadeBuckets'); + + // V2 format whenever any rule carries a Priority; V1 otherwise. + const isV2 = rows.some(r => r.priority !== undefined && r.priority !== ''); + // When any rule targets a CRR cascade location, emit the comma-separated + // source,destination role — each rule's Destination.Account substitutes + // the account-id portion of the destination role at runtime (ARSN-571). + const hasCRR = rows.some(r => cascadeInfoByLocation?.[r.location]); + const role = hasCRR && cascadeRoleName + ? `${DEFAULT_ROLE},arn:aws:iam::000000000000:role/${cascadeRoleName}` + : DEFAULT_ROLE; + const Rules = rows.map((row, i) => { + const crrInfo = cascadeInfoByLocation?.[row.location]; + // For CRR cascade destinations, the destination bucket was + // pre-created on the cascade account's side; use that bucket name. + const destBucket = crrInfo && cascadeBuckets?.[row.location] + ? cascadeBuckets[row.location] + : (row.bucket ?? srcBucket); + const Destination: Record = { + Bucket: `arn:aws:s3:::${destBucket}`, + StorageClass: row.location, + }; + if (row.account) { + Destination.Account = row.account; + } else if (crrInfo) { + Destination.Account = crrInfo.AccountId; + } + const rule: Record = { + ID: row.id || `rule-${i}`, + Status: row.status || 'Enabled', + Destination, + }; + if (isV2) { + rule.Filter = { Prefix: row.prefix ?? '' }; + if (row.priority !== undefined && row.priority !== '') { + rule.Priority = Number(row.priority); + } + } else { + rule.Prefix = row.prefix ?? ''; + } + if (row.deleteMarkerReplication) { + rule.DeleteMarkerReplication = { Status: row.deleteMarkerReplication }; + } + return rule; + }); + return { Role: role, Rules }; +} When('the job to replicate existing objects with status {string} is executed', { timeout: 600000 }, @@ -20,7 +301,7 @@ When('the job to replicate existing objects with status {string} is executed', sourceObjectStatus: string, ) { const sourceBucket = this.getSaved('bucketName'); - const replicationLocation = this.getSaved('replicationLocation'); + const replicationLocation = this.getSaved('replicationLocations')[0]; const { locationType } = await getReplicationLocationConfig(this, replicationLocation); const zenkoVersion = await getZenkoVersion(this); const s3utilsVersion = zenkoVersion.spec.versions.s3utils; @@ -59,122 +340,12 @@ When('the job to replicate existing objects with status {string} is executed', ] } }; - - await createAndRunPod(this, podManifest); - }); - -Then('the object replication should {string} within {int} seconds', { timeout: 600_000 }, - async function ( - this: Zenko, - expectedOutcome: 'succeed' | 'fail' | 'never happen', - replicationTimeout: number - ) { - const objectName = this.getSaved('objectName'); - const bucketSource = this.getSaved('bucketName'); - const startTime = Date.now(); - while (Date.now() - startTime < replicationTimeout * 1000) { - await new Promise(resolve => setTimeout(resolve, 3000)); - - const response = await headObject(this, objectName, bucketSource); - assert(response.stdout); - assert.strictEqual(response.statusCode, 200, `failed to headobject, ${response.statusCode}`); - const parsed = safeJsonParse<{ - ReplicationStatus?: string; - LastModified?: string; - ETag?: string; - ContentLength?: number; - VersionId?: string; - Metadata?: Record; - }>(response.stdout || '{}'); - assert(parsed.ok); - const replicationStatus = parsed.result?.ReplicationStatus; - - switch (replicationStatus) { - case 'PENDING': - case 'PROCESSING': - assert.notStrictEqual(expectedOutcome, 'never happen', - `replication status is ${replicationStatus}, but expected to never happen`); - continue; - case 'COMPLETED': - assert.strictEqual(expectedOutcome, 'succeed', - `replication is completed, but expected outcome was '${expectedOutcome}'`); - return; - case 'FAILED': - assert.strictEqual(expectedOutcome, 'fail', - `replication is failed, but expected outcome was '${expectedOutcome}'`); - return; - case undefined: - // If we don't expect a replication to happen, the replication status should remain undefined - assert.strictEqual(expectedOutcome, 'never happen', - `got undefined replication status for expected outcome '${expectedOutcome}'`); - return; - default: - throw new Error(`Unexpected replication status: ${replicationStatus}`); - } - } - assert.fail(`Timeout: Object '${objectName}' is still in pending/processing state after timeout`); - }); -Then( - 'the replicated object should be the same as the source object', - async function ( - this: Zenko, - ) { - const objectName = this.getSaved('objectName'); - const bucketSource = this.getSaved('bucketName'); - const replicationLocation = this.getSaved('replicationLocation'); - const { destinationBucket, bucketMatch, awsS3Client } = - await getReplicationLocationConfig(this, replicationLocation); - - // When bucketMatch is disabled on the destination bucket, - // replicated objects are named sourceBucket/objectName - let key = `${bucketSource}/${objectName}`; - if (bucketMatch) { - key = objectName; - } - - const command = new GetObjectCommand({ - Bucket: destinationBucket, - Key: key, - }); - const replicaObj = await awsS3Client.send(command); - const sourceResponse = await getObject(this, objectName, bucketSource); - assert.strictEqual(sourceResponse.statusCode, 200, `failed to getObject, ${sourceResponse.statusCode}`); - const sourceObj = safeJsonParse<{ - ReplicationStatus?: string; - LastModified?: string; - ETag?: string; - ContentLength?: number; - VersionId?: string; - Metadata?: Record; - }>(sourceResponse.stdout || '{}'); - assert(sourceObj.ok); - - assert.strictEqual(sourceObj.result?.ReplicationStatus, 'COMPLETED'); - assert.strictEqual( - sourceObj.result?.ContentLength, - replicaObj.ContentLength - ); - assert.strictEqual( - sourceObj.result?.Metadata?.[`${replicationLocation}-version-id`], - replicaObj.VersionId - ); - assert.strictEqual( - sourceObj.result?.Metadata?.[`${replicationLocation}-replication-status`], - 'COMPLETED' - ); - assert.strictEqual( - sourceObj.result?.VersionId, - replicaObj.Metadata?.['scal-version-id'] - ); - assert.strictEqual( - replicaObj.Metadata?.['scal-replication-status'], - 'REPLICA' - ); + await createAndRunPod(this, podManifest); }); Given('a deleted destination bucket on that location', async function (this: Zenko) { - const replicationLocation = this.getSaved('replicationLocation'); + const replicationLocation = this.getSaved('replicationLocations')[0]; const scenarioTags = this.getSaved('scenarioTags') || []; const lockTag = `@Lock${replicationLocation}`; const hasTestLock = scenarioTags.includes(lockTag); @@ -196,8 +367,8 @@ Given('a deleted destination bucket on that location', async function (this: Zen }); When('the destination bucket on the location is created again', async function (this: Zenko) { - const { destinationBucket, awsS3Client } = - await getReplicationLocationConfig(this, this.getSaved('replicationLocation')); + const { destinationBucket, awsS3Client } = + await getReplicationLocationConfig(this, this.getSaved('replicationLocations')[0]); const command = new CreateBucketCommand({ Bucket: destinationBucket, }); @@ -210,3 +381,181 @@ When('the destination bucket on the location is created again', async function ( }); await awsS3Client.send(versioningCommand); }); + +Given('CRR replication accounts are registered', function (this: Zenko) { + const roleName = process.env.CRR_ROLE_NAME; + assert.ok(roleName, 'CRR_ROLE_NAME must be set'); + + const locationNames = [ + process.env.CRR_LOCATION_A_NAME, + process.env.CRR_LOCATION_B_NAME, + process.env.CRR_LOCATION_C_NAME, + ]; + assert.ok(locationNames.every(Boolean), 'CRR_LOCATION_A/B/C_NAME must be set'); + + const infoByLocation: Record = {}; + for (const [i, loc] of locationNames.entries()) { + const varName = `CRR_INFO_${['A', 'B', 'C'][i]}`; + const raw = process.env[varName]; + assert.ok(raw, `${varName} must be set`); + const info = JSON.parse(raw) as CRRAccountInfo; + Identity.addIdentity(IdentityEnum.ACCOUNT, loc!, { + accessKeyId: info.AccessKeyId, + secretAccessKey: info.SecretAccessKey, + sessionToken: info.SessionToken, + }); + infoByLocation[loc!] = info; + } + + this.addToSaved('cascadeInfoByLocation', infoByLocation); + this.addToSaved('cascadeRoleName', roleName); + this.addToSaved('cascadeBuckets', {} as Record); +}); + +Given(/^an? (valid|invalid) multi-destination replication configuration with rules:$/, + async function (this: Zenko, validity: 'valid' | 'invalid', dataTable: DataTable) { + const rows = dataTable.hashes() as RuleRow[]; + const srcBucket = this.getSaved('bucketName'); + const config = buildConfigFromRows(this, srcBucket, rows); + this.addToSaved('lastReplicationConfig', config); + const res = await putBucketReplicationRaw.call(this, srcBucket, config); + if (validity === 'valid') { + assert(!res.err, `expected putBucketReplication to succeed, got: ${res.err}`); + const enabled = new Set(); + for (const r of rows) { + if ((r.status || 'Enabled') !== 'Enabled') continue; + for (const loc of r.location.split(',').map(s => s.trim())) { + enabled.add(loc); + } + } + this.addToSaved('replicationLocations', Array.from(enabled)); + } else { + this.addToSaved('lastPutReplicationError', res.err || ''); + } + }); + +Given('a legacy comma-StorageClass replication configuration to locations {string}', + async function (this: Zenko, locationsCsv: string) { + const srcBucket = this.getSaved('bucketName'); + const locations = locationsCsv.split(',').map(s => s.trim()); + this.addToSaved('replicationLocations', locations); + const config = { + Role: DEFAULT_ROLE, + Rules: [{ + ID: 'legacy-multi', + Prefix: '', + Status: 'Enabled', + Destination: { + Bucket: `arn:aws:s3:::${srcBucket}`, + StorageClass: locationsCsv, + }, + }], + }; + this.addToSaved('lastReplicationConfig', config); + const res = await putBucketReplicationRaw.call(this, srcBucket, config); + assert(!res.err, `expected putBucketReplication to succeed, got: ${res.err}`); + }); + +Then('the replication configuration request should be rejected with {string}', + function (this: Zenko, errorCode: string) { + const err = this.getSaved('lastPutReplicationError') ?? ''; + assert( + err.includes(errorCode), + `expected putBucketReplication error to contain '${errorCode}', got: ${err}`, + ); + }); + +Then('the object replication should {string} within {int} seconds', + { timeout: 600_000 }, + async function (this: Zenko, expectedOutcome: 'succeed' | 'fail' | 'never happen', timeoutSec: number) { + const objectName = this.getSaved('objectName'); + const srcBucket = this.getSaved('bucketName'); + const locations = this.getSaved('replicationLocations') || []; + assert(locations.length > 0, 'no replicationLocations saved on the scenario world'); + await pollReplicationOutcome( + this, + objectName, + srcBucket, + expectedOutcome, + timeoutSec * 1000, + locations, + ); + }); + +Then('the replicated object on {string} should match the source', + async function (this: Zenko, location: string) { + await assertReplicaMatchesSource(this, location); + }); + +Then('the replicated object should match the source on every configured destination', + async function (this: Zenko) { + const locations = this.getSaved('replicationLocations') || []; + assert(locations.length > 0, 'no replicationLocations saved on the scenario world'); + for (const location of locations) { + await assertReplicaMatchesSource(this, location); + } + }); + +Then('the object should not be replicated to {string}', + async function (this: Zenko, location: string) { + // Caller is expected to have first waited for the enabled destinations + // to complete, so backbeat has had its chance. We then assert that the + // disabled destination has no per-backend metadata stamp at all. + const objectName = this.getSaved('objectName'); + const srcBucket = this.getSaved('bucketName'); + const res = await headObject(this, objectName, srcBucket); + assert.strictEqual(res.statusCode, 200, `failed to headObject, ${res.statusCode}`); + const parsed = safeJsonParse(res.stdout || '{}'); + assert(parsed.ok); + const status = parsed.result?.Metadata?.[`${location}-replication-status`]; + assert.strictEqual( + status, undefined, + `object was unexpectedly replicated to '${location}', status=${status}`, + ); + }); + +Then('getBucketReplication should return a {string} configuration with {int} rules', + async function (this: Zenko, format: string, expectedCount: number) { + const srcBucket = this.getSaved('bucketName'); + const result = await getBucketReplicationConfig.call(this, srcBucket); + assert(result.ok, 'getBucketReplication failed'); + if (!result.ok) return; + const rules = (result.config.Rules as Array>) ?? []; + assert.strictEqual( + rules.length, expectedCount, + `expected ${expectedCount} rule(s), got ${rules.length}`, + ); + for (const rule of rules) { + if (format === 'V2') { + assert( + rule.Filter !== undefined, + `expected V2 rule to have Filter: ${JSON.stringify(rule)}`, + ); + assert( + rule.Prefix === undefined, + `V2 rule should not carry top-level Prefix: ${JSON.stringify(rule)}`, + ); + } else if (format === 'V1') { + assert( + rule.Prefix !== undefined, + `expected V1 rule to have top-level Prefix: ${JSON.stringify(rule)}`, + ); + assert( + rule.Filter === undefined, + `V1 rule should not have Filter: ${JSON.stringify(rule)}`, + ); + } else { + assert.fail(`unsupported format '${format}', expected 'V1' or 'V2'`); + } + } + }); + +Then('getBucketReplication rule {int} should have StorageClass {string}', + async function (this: Zenko, ruleIndex: number, expectedStorageClass: string) { + const srcBucket = this.getSaved('bucketName'); + const result = await getBucketReplicationConfig.call(this, srcBucket); + assert(result.ok, 'getBucketReplication failed'); + if (!result.ok) return; + const rules = (result.config.Rules as Array<{ Destination?: { StorageClass?: string } }>) ?? []; + assert.strictEqual(rules[ruleIndex]?.Destination?.StorageClass, expectedStorageClass); + }); diff --git a/tests/functional/ctst/steps/utils/utils.ts b/tests/functional/ctst/steps/utils/utils.ts index d19a73013..ed3ac344e 100644 --- a/tests/functional/ctst/steps/utils/utils.ts +++ b/tests/functional/ctst/steps/utils/utils.ts @@ -14,8 +14,6 @@ import { extractPropertyFromResults, s3FunctionExtraParams, safeJsonParse } from import Zenko from 'world/Zenko'; import assert from 'assert'; import constants from 'common/constants'; -import { getLocationConfigs } from './kubernetes'; -import { S3Client } from '@aws-sdk/client-s3'; import { pollDLQBuffer } from './kafka'; enum AuthorizationType { @@ -409,71 +407,6 @@ async function addTransitionWorkflow(this: Zenko, location: string, enabled = tr } } -async function getReplicationLocationConfig(world: Zenko, location: string): Promise<{ - destinationBucket: string; - locationType: string; - bucketMatch: boolean; - awsS3Client: S3Client; - }> { - const locationsConfigs = await getLocationConfigs(world); - if (!locationsConfigs[location]) { - throw new Error(`Unsupported replication location: '${location}'`); - } - return { - destinationBucket: locationsConfigs[location].details.bucketName, - locationType: locationsConfigs[location].type, - bucketMatch: locationsConfigs[location].details.bucketMatch, - awsS3Client: new S3Client({ - region: locationsConfigs[location].details.region, - endpoint: `https://${locationsConfigs[location].details.awsEndpoint}`, - credentials: { - accessKeyId: locationsConfigs[location].details.credentials.accessKey, - secretAccessKey: locationsConfigs[location].details.credentials.secretKey, - }, - tls: false, - maxAttempts: 1, - forcePathStyle: true, - }), - }; -} - -async function putBucketReplication( - this: Zenko, - srcBucket: string, - replicationLocation: string -) { - this.resetCommand(); - this.addCommandParameter({ bucket: srcBucket }); - this.addCommandParameter({ - replicationConfiguration: JSON.stringify({ - Role: 'arn:aws:iam::root:role/s3-replication-role', - Rules: [ - { - Prefix: '', - Destination: { - // eslint-disable-next-line max-len - // https://documentation.scality.com/Artesca/4.0.1/data_management/bucket_operations/replication_workflow/create_a_replication_workflow.html - Bucket: `arn:aws:s3:::${srcBucket}`, - StorageClass: replicationLocation, - }, - Status: 'Enabled', - }, - ], - }), - }); - - const commandParameters = this.getCommandParameters(); - const res = await S3.putBucketReplication(commandParameters); - if (res.err) { - this.logger.error('Failed to put bucket replication', { - srcBucket, - replicationLocation, - error: res.err, - }); - throw new Error(`Failed to put bucket replication, err : ${res.err}`); - } -} - // Polls for transition status and checks the DLQ buffer on every iteration for early failure. // Relies on bucket names being randomized so a DLQ entry from a previous run cannot match. async function verifyObjectLocation(this: Zenko, objectName: string, @@ -554,6 +487,47 @@ async function restoreObject(this: Zenko, objectName: string, days: number) { this.setResult(result); } +async function putBucketReplicationRaw( + this: Zenko, + srcBucket: string, + replicationConfiguration: object, +): Promise { + this.resetCommand(); + this.addCommandParameter({ bucket: srcBucket }); + this.addCommandParameter({ + replicationConfiguration: JSON.stringify(replicationConfiguration), + }); + return S3.putBucketReplication(this.getCommandParameters()); +} + +async function putBucketReplication( + this: Zenko, + srcBucket: string, + replicationLocation: string, +) { + const res = await putBucketReplicationRaw.call(this, srcBucket, { + Role: 'arn:aws:iam::root:role/s3-replication-role', + Rules: [ + { + Prefix: '', + Destination: { + Bucket: `arn:aws:s3:::${srcBucket}`, + StorageClass: replicationLocation, + }, + Status: 'Enabled', + }, + ], + }); + if (res.err) { + this.logger.error('Failed to put bucket replication', { + srcBucket, + replicationLocation, + error: res.err, + }); + throw new Error(`Failed to put bucket replication, err : ${res.err}`); + } +} + /** * @param {Zenko} this world object * @param {string} objectName object name @@ -601,6 +575,6 @@ export { getObjectNameWithBackendFlakiness, restoreObject, addTransitionWorkflow, - getReplicationLocationConfig, putBucketReplication, + putBucketReplicationRaw, };