From 99c71b456df4639d878d7a0f5189dc298f9e2740 Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Wed, 16 Sep 2026 15:46:30 +0200 Subject: [PATCH 1/4] Set the backlog metrics interval through the generic override backlogMetricsIntervalSeconds was a dedicated CR field, added by zenko-operator#626 and reverted by #633 right after v1.8.16 in favour of the generic backbeat.configurationOverrides map. The two spellings never coexist in a released operator: v1.8.16 knows only the field, and every 1.8 release after it knows only the override. So the e2e config and the operator pin have to move together, and bumping the operator without this change breaks the whole suite on a rejected field. KAFKA_BACKLOG_METRICS_INTERVALS is backbeat's declared env alias for kafka.backlogMetrics.intervalS, so the value reaches the same config key with the same unit the removed field used. It is set on the four consumers whose lag the lifecycle conductor waits on. Issue: ZENKO-5359 --- .github/scripts/end2end/configs/zenko.yaml | 10 +++++++++- solution/deps.yaml | 2 +- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/.github/scripts/end2end/configs/zenko.yaml b/.github/scripts/end2end/configs/zenko.yaml index 1da2c39c64..ec038a62c7 100644 --- a/.github/scripts/end2end/configs/zenko.yaml +++ b/.github/scripts/end2end/configs/zenko.yaml @@ -25,7 +25,15 @@ spec: logLevel: debug backbeat: triggerExpirationsOneDayEarlierForTesting: ${EXPIRE_ONE_DAY_EARLIER} - backlogMetricsIntervalSeconds: 5 + configurationOverrides: + backbeat-lifecycle-producer: + KAFKA_BACKLOG_METRICS_INTERVALS: "5" + backbeat-lifecycle-bucket-processor: + KAFKA_BACKLOG_METRICS_INTERVALS: "5" + backbeat-lifecycle-object-processor: + KAFKA_BACKLOG_METRICS_INTERVALS: "5" + backbeat-replication-data-processor: + KAFKA_BACKLOG_METRICS_INTERVALS: "5" lifecycleConductor: cronRule: "${BACKBEAT_LCC_CRON_RULE}" lifecycleBucketProcessor: diff --git a/solution/deps.yaml b/solution/deps.yaml index 90572831bb..ca5f3e16a1 100644 --- a/solution/deps.yaml +++ b/solution/deps.yaml @@ -131,7 +131,7 @@ zenko-operator: sourceRegistry: ghcr.io/scality dashboard: zenko-operator/zenko-operator-dashboards image: zenko-operator - tag: v1.8.16 + tag: v1.8.17 envsubst: ZENKO_OPERATOR_TAG zookeeper: sourceRegistry: ghcr.io/adobe/zookeeper-operator From f7e9094d5261955c32e207b788697b8264df71cc Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Tue, 8 Sep 2026 16:12:21 +0200 Subject: [PATCH 2/4] Run the PRA scenario over a non-versioned bucket again The examples table carried `Non versioned` alone until 137e69dd91 swapped it for `Versioned`, in the same commit that added the cross-site restore steps. Nothing records why: the commit message says only "Complete PRA logic", the pull request body names the tests it was missing, its review discussed the zenkoversion, the locations and the scripts, and the ticket has no description at all. Nor does the code suggest a reason -- the restore helper that commit extracted keeps the version id optional, so it runs against a bucket that has none. Since the swap, nothing has restored or failed over an object whose bucket holds no version to carry its state, which is where a restore writes `x-amz-restore` onto a master with nothing beneath it. Both rows rather than the swap back: the version delete this scenario covers needs a bucket with versions to delete. Issue: ZENKO-5355 --- tests/functional/ctst/features/pra.feature | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/functional/ctst/features/pra.feature b/tests/functional/ctst/features/pra.feature index 0199b137b2..8925e8d152 100644 --- a/tests/functional/ctst/features/pra.feature +++ b/tests/functional/ctst/features/pra.feature @@ -89,4 +89,5 @@ Feature: PRA operations Examples: | versioningConfiguration | objectCount | objectSize | + | Non versioned | 2 | 100 | | Versioned | 2 | 100 | \ No newline at end of file From 63292e22e34c106d4aab6c3949eca4840cab44fa Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Mon, 7 Sep 2026 11:03:04 +0200 Subject: [PATCH 3/4] Cover D/R deletion and in-place overwrite in the PRA test Three paths the PRA scenario never touched: it deletes nothing, and its only bucket is versioned, so an object is never overwritten in place and never deleted without a version to remove. None of the deletions D/R has never replicated, nor the metadata a non-versioned overwrite has to carry, is visible to CI today. Removing a version and overwriting in place are steps of the nominal scenario, which already holds the state they need. Removing an unversioned object gets a scenario of its own: a failing step abandons the rest of its scenario, and this is the one whose outcome on a released deployment is not yet known. @PRA is assigned at most one pickle at a time, so scenarios cannot fight over the two clusters. Delete markers stay uncovered, as ZENKO-4886 records: a marker carries no placement, so the location the source selects on cannot match it. A suspended bucket gets a scenario of its own too. A put there writes a null version, which the delete addresses as a version even though the master may be the only document holding it, so what removes the object is the teardown of the placeholder left on that master -- a write the source makes in no other case. Issue: ZENKO-5355 --- tests/functional/ctst/common/common.ts | 37 ++++--- tests/functional/ctst/features/pra.feature | 47 +++++++- tests/functional/ctst/steps/pra.ts | 118 ++++++++++++++++++--- 3 files changed, 172 insertions(+), 30 deletions(-) diff --git a/tests/functional/ctst/common/common.ts b/tests/functional/ctst/common/common.ts index 6bc0bc81de..a4be2e70ea 100644 --- a/tests/functional/ctst/common/common.ts +++ b/tests/functional/ctst/common/common.ts @@ -129,11 +129,7 @@ Given('{int} additional accounts', async function (this: Zenko, count: number) { } }); -async function createBucket(world: Zenko, versioning: string, bucketName: string) { - world.resetCommand(); - world.addToSaved('bucketName', bucketName); - world.addCommandParameter({ bucket: bucketName }); - await S3.createBucket(world.getCommandParameters()); +async function setBucketVersioning(world: Zenko, versioning: string) { world.addToSaved('bucketVersioning', versioning); if (versioning !== 'Non versioned') { const versioningConfiguration = versioning === 'Versioned' ? 'Enabled' : 'Suspended'; @@ -142,18 +138,33 @@ async function createBucket(world: Zenko, versioning: string, bucketName: string } } +async function createBucket(world: Zenko, versioning: string, bucketName: string) { + world.resetCommand(); + world.addToSaved('bucketName', bucketName); + world.addCommandParameter({ bucket: bucketName }); + await S3.createBucket(world.getCommandParameters()); + await setBucketVersioning(world, versioning); +} + +export async function createUniqueBucket(world: Zenko, versioning: string, separator = '') { + const preName = world.getSaved('accountName') || + world.parameters.AccountName || Constants.ACCOUNT_NAME; + await createBucket(world, versioning, + `${preName}${separator}${Constants.BUCKET_NAME_TEST}${Utils.randomString()}`.toLocaleLowerCase()); +} + Given('a {string} bucket with dot', async function (this: Zenko, versioning: string) { - const preName = this.getSaved('accountName') || - this.parameters.AccountName || Constants.ACCOUNT_NAME; - await createBucket(this, versioning, - `${preName}.${Constants.BUCKET_NAME_TEST}${Utils.randomString()}`.toLocaleLowerCase()); + await createUniqueBucket(this, versioning, '.'); }); Given('a {string} bucket', async function (this: Zenko, versioning: string) { - const preName = this.getSaved('accountName') || - this.parameters.AccountName || Constants.ACCOUNT_NAME; - await createBucket(this, versioning, - `${preName}${Constants.BUCKET_NAME_TEST}${Utils.randomString()}`.toLocaleLowerCase()); + await createUniqueBucket(this, versioning); +}); + +When('the bucket versioning is set to {string}', async function (this: Zenko, versioning: string) { + this.resetCommand(); + this.addCommandParameter({ bucket: this.getSaved('bucketName') }); + await setBucketVersioning(this, versioning); }); Given('an existing bucket {string} {string} versioning, {string} ObjectLock {string} retention mode', async function diff --git a/tests/functional/ctst/features/pra.feature b/tests/functional/ctst/features/pra.feature index 8925e8d152..72aee0e34f 100644 --- a/tests/functional/ctst/features/pra.feature +++ b/tests/functional/ctst/features/pra.feature @@ -87,7 +87,52 @@ Feature: PRA operations Then object "obj3-1" should "" be "transitioned" and have the storage class "e2e-cold" on "DR" site And object "obj3-2" should "" be "transitioned" and have the storage class "e2e-cold" on "DR" site + # A version deleted on the primary site must be deleted on the DR site + Given 1 objects "objdel" of size bytes on "Primary" site + Then object "objdel-1" should "" be "transitioned" and have the storage class "e2e-cold" on "Primary" site + And object "objdel-1" should "" be "transitioned" and have the storage class "e2e-cold" on "DR" site + When i delete object "objdel-1" on "Primary" site + Then object "objdel-1" should "not" exist on "DR" site + Examples: | versioningConfiguration | objectCount | objectSize | | Non versioned | 2 | 100 | - | Versioned | 2 | 100 | \ No newline at end of file + | Versioned | 2 | 100 | + | Suspended | 2 | 100 | + + @2.6.0 + @PreMerge + @Dmf + @PRA + @ColdStorage + Scenario: PRA (null version in a versioned bucket) + Given a DR installed + Then the DR source should be in phase "Running" + And the DR sink should be in phase "Running" + Given a "Non versioned" bucket on "Primary" site + And 1 objects "objnull" of size 100 bytes on "Primary" site + When the bucket versioning is set to "Versioned" + And a transition workflow to "e2e-cold" location + Then object "objnull-1" should "" be "transitioned" and have the storage class "e2e-cold" on "Primary" site + Given access keys for the replicated account + Then object "objnull-1" should "" be "transitioned" and have the storage class "e2e-cold" on "DR" site + + @2.6.0 + @PreMerge + @Dmf + @PRA + @ColdStorage + Scenario: PRA (in-place overwrite) + Given a DR installed + Then the DR source should be in phase "Running" + And the DR sink should be in phase "Running" + Given a "Non versioned" bucket on "Primary" site + And a transition workflow to "e2e-cold" location + And 1 objects "objow" of size 100 bytes on "Primary" site + Then object "objow-1" should "" be "transitioned" and have the storage class "e2e-cold" on "Primary" site + Given access keys for the replicated account + Then object "objow-1" should "" be "transitioned" and have the storage class "e2e-cold" on "DR" site + When i overwrite object "objow-1" with 200 bytes on "Primary" site + Then object "objow-1" should "" be "transitioned" and have the storage class "e2e-cold" on "Primary" site + And object "objow-1" should "" be "transitioned" and have the storage class "e2e-cold" on "DR" site + And object "objow-1" should have the last written etag on "DR" site diff --git a/tests/functional/ctst/steps/pra.ts b/tests/functional/ctst/steps/pra.ts index dd3a5bfff5..31ad62a1b6 100644 --- a/tests/functional/ctst/steps/pra.ts +++ b/tests/functional/ctst/steps/pra.ts @@ -13,7 +13,8 @@ import { restoreObject, verifyObjectLocation, } from 'steps/utils/utils'; -import { CacheHelper, Constants, Identity, IdentityEnum, SuperAdmin, Utils } from 'cli-testing'; +import { CacheHelper, Constants, Identity, IdentityEnum, S3, SuperAdmin, Utils } from 'cli-testing'; +import { createUniqueBucket } from 'common/common'; import { safeJsonParse } from 'common/utils'; import { PrometheusDriver } from 'prometheus-query'; import assert from 'assert'; @@ -51,6 +52,26 @@ interface DrState { }; } +function useSiteIdentity(site: string) { + const accountName = Zenko.sites['source'].accountName; + Identity.useIdentity(IdentityEnum.ACCOUNT, site === 'DR' ? `${accountName}-replicated` : accountName); +} + +async function headObjectOnSite(world: Zenko, objectName: string, site: string) { + useSiteIdentity(site); + world.resetCommand(); + world.addCommandParameter({ bucket: world.getSaved('bucketName') }); + world.addCommandParameter({ key: objectName }); + + return S3.headObject(world.getCommandParameters()); +} + +function throwUnlessMissing(res: { err?: string | null }, objectName: string) { + if (res.err && !res.err.includes('NotFound')) { + throw new Error(`HeadObject error for "${objectName}": ${res.err}`); + } +} + async function installPRA(world: Zenko, sinkS3Endpoint = 'http://s3.zenko.local', timeout = '30m') { const kafkaExternalIpOption = world.parameters.KafkaExternalIps ? { kafkaExternalIps: world.parameters.KafkaExternalIps } : @@ -253,11 +274,7 @@ Then('object {string} should {string} be {string} and have the storage class {st storageClass: string, site: string) { this.resetCommand(); - if (site === 'DR') { - Identity.useIdentity(IdentityEnum.ACCOUNT, `${Zenko.sites['source'].accountName}-replicated`); - } else { - Identity.useIdentity(IdentityEnum.ACCOUNT, Zenko.sites['source'].accountName); - } + useSiteIdentity(site); try { await verifyObjectLocation.call(this, objName, objectTransitionStatus, storageClass); if (isVerb === 'not') { @@ -273,11 +290,7 @@ Then('object {string} should {string} be {string} and have the storage class {st When('the DATA_ACCESSOR user tries to perform PutObject on {string} site', { timeout: 5 * 60 * 1000 }, async function (this: Zenko, site: string) { - if (site === 'DR') { - Identity.useIdentity(IdentityEnum.ACCOUNT, `${Zenko.sites['source'].accountName}-replicated`); - } else { - Identity.useIdentity(IdentityEnum.ACCOUNT, Zenko.sites['source'].accountName); - } + useSiteIdentity(site); this.resetCommand(); this.addToSaved('accountName', Zenko.sites['source'].accountName); @@ -428,10 +441,83 @@ Given('access keys for the replicated account', { timeout: 360000 }, async () => When('i restore object {string} for {int} days on {string} site', async function (this: Zenko, objectName: string, days: number, site: string) { this.resetCommand(); - if (site === 'DR') { - Identity.useIdentity(IdentityEnum.ACCOUNT, `${Zenko.sites['source'].accountName}-replicated`); - } else { - Identity.useIdentity(IdentityEnum.ACCOUNT, Zenko.sites['source'].accountName); - } + useSiteIdentity(site); await restoreObject.call(this, objectName, days); }); + +Given('a {string} bucket on {string} site', async function (this: Zenko, versioning: string, site: string) { + useSiteIdentity(site); + await createUniqueBucket(this, versioning); +}); + +When('i delete object {string} on {string} site', + async function (this: Zenko, objectName: string, site: string) { + useSiteIdentity(site); + this.resetCommand(); + this.addCommandParameter({ bucket: this.getSaved('bucketName') }); + this.addCommandParameter({ key: objectName }); + const versionId = this.getLatestObjectVersion(objectName); + if (versionId) { + this.addCommandParameter({ versionId }); + } + const result = await S3.deleteObject(this.getCommandParameters()); + assert.ifError(result.err); + }); + +When('i overwrite object {string} with {int} bytes on {string} site', + async function (this: Zenko, objectName: string, sizeBytes: number, site: string) { + useSiteIdentity(site); + const result = await putObject(this, objectName, undefined, sizeBytes); + assert.ifError(result?.stderr || result?.err); + }); + +Then('object {string} should {string} exist on {string} site', { timeout: 360000 }, + async function (this: Zenko, objectName: string, isVerb: string, site: string) { + const shouldExist = isVerb !== 'not'; + const timeout = 300000; + const start = Date.now(); + + // the head carries no version id, so a master left behind by a deletion + // that did not replicate keeps answering it + while (Date.now() - start < timeout) { + const res = await headObjectOnSite(this, objectName, site); + throwUnlessMissing(res, objectName); + + const exists = !res.err; + if (exists === shouldExist) { + return; + } + + await Utils.sleep(1000); + } + + assert.fail(`object "${objectName}" should ${shouldExist ? '' : 'not '}exist on the ${site} site`); + }); + +Then('object {string} should have the last written etag on {string} site', { timeout: 360000 }, + async function (this: Zenko, objectName: string, site: string) { + const expected = this.getSaved('objectETag'); + assert(expected, 'no write saved an etag to compare against'); + + const timeout = 300000; + const start = Date.now(); + let seen; + + while (Date.now() - start < timeout) { + const res = await headObjectOnSite(this, objectName, site); + throwUnlessMissing(res, objectName); + if (!res.err) { + assert(res.stdout); + const parsed = safeJsonParse<{ ETag: string | undefined }>(res.stdout); + assert(parsed.ok); + seen = parsed.result?.ETag; + if (seen?.replace(/"/g, '') === expected.replace(/"/g, '')) { + return; + } + } + + await Utils.sleep(1000); + } + + assert.fail(`object "${objectName}" carries etag ${seen} on the ${site} site, expected ${expected}`); + }); From e535ca51519d153fc5e67e0861cca65cb3489450 Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Thu, 3 Sep 2026 19:35:54 +0200 Subject: [PATCH 4/4] WIP: point the solution at the D/R mongo-processor builds Not for merge. Exercises ZKOP-562, BB-811 and DRCTL-83 end to end: the @PRA CTST run replicates cold objects through the mongo-processor rather than the Kafka Connect mongo sink. - backbeat 5e8a9c0b (BB-811, the D/R processor mode, with an object rewritten in place taken whole, and BB-853 on top: the sink omits the metrics block, and the producer is a hard startup dependency without it) - zenko-operator cf7e60c4 (ZKOP-562, the sink deployment and the source pipeline that shapes the entries it reads, rebased on the 1.9 that carries ZKOP-588 -- so drctl no longer needs DRCTL-81's merge patch to get past the analytics section marshalling as a null) - drctl v1.1.0-preview.1 (carries DRCTL-83, the objects topic on both D/R CRs, and the merged DRCTL-79 that sets the backbeat image on the sink; its vendor tree carries an operator new enough that the analytics section marshals as absent rather than as a null the apiserver rejects) The retag prefix is 9.5.3 because that branch really is 9.5.2, and every operator gate on the backbeat tag sits either below 9.5.1 or at 9.6.0-preview.1, so it reads exactly as 9.5.1 did. drctl is pinned to a released tag rather than to the per-commit image its own CI pushes. --- solution/deps.yaml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/solution/deps.yaml b/solution/deps.yaml index ca5f3e16a1..e15f819cfd 100644 --- a/solution/deps.yaml +++ b/solution/deps.yaml @@ -2,11 +2,11 @@ # to sort keys, use the following command # yq eval 'sortKeys(.)' -i deps.yaml backbeat: - sourceRegistry: ghcr.io/scality + sourceRegistry: ghcr.io/scality/playground/delthas dashboard: backbeat/backbeat-dashboards image: backbeat policy: backbeat/backbeat-policies - tag: 9.5.1 + tag: 9.5.3-h5e8a9c0b94e7b667a8d82611454ff4796d647f5c envsubst: BACKBEAT_TAG busybox: image: busybox @@ -21,7 +21,7 @@ cloudserver: drctl: sourceRegistry: ghcr.io/scality image: zenko-drctl - tag: v1.0.6 + tag: v1.1.0-preview.1 envsubst: DRCTL_TAG toolName: zenko-drctl fubectl: @@ -131,7 +131,7 @@ zenko-operator: sourceRegistry: ghcr.io/scality dashboard: zenko-operator/zenko-operator-dashboards image: zenko-operator - tag: v1.8.17 + tag: cf7e60c45ce51a598ef5c39ea3a58057c86627ac envsubst: ZENKO_OPERATOR_TAG zookeeper: sourceRegistry: ghcr.io/adobe/zookeeper-operator