Extend PRA coverage to deletions, overwrites and non-versioned buckets

.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:

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.16
+  tag: cf7e60c45ce51a598ef5c39ea3a58057c86627ac
   envsubst: ZENKO_OPERATOR_TAG
 zookeeper:
   sourceRegistry: ghcr.io/adobe/zookeeper-operator

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<string>('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<string>('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<string>('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<string>('bucketName') });
+    await setBucketVersioning(this, versioning);
 });
 
 Given('an existing bucket {string} {string} versioning, {string} ObjectLock {string} retention mode', async function

tests/functional/ctst/features/pra.feature

@@ -87,6 +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 <objectSize> 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 |
-    |               Versioned |           2 |        100 |
+    |           Non versioned |           2 |        100 |
+    |               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

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<string>('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<string>('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<string>('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}`);
+    });