add test for ingesting non versioned objets

Issue: ZENKO-4330

.github/scripts/end2end/configure-e2e.sh

@@ -89,6 +89,9 @@ kubectl run ${POD_NAME} \
   --env=RING_S3C_ENDPOINT=${RING_S3C_ENDPOINT} \
   --env=RING_S3C_BACKEND_SOURCE_LOCATION=${RING_S3C_BACKEND_SOURCE_LOCATION} \
   --env=RING_S3C_INGESTION_SRC_BUCKET_NAME=${RING_S3C_INGESTION_SRC_BUCKET_NAME} \
+  --env=RING_S3C_BACKEND_SOURCE_NON_VERSIONED_LOCATION=${RING_S3C_BACKEND_SOURCE_NON_VERSIONED_LOCATION} \
+  --env=RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME=${RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME} \
+  --env=RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE=${RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE} \
   --env=AWS_ACCESS_KEY=${AWS_ACCESS_KEY} \
   --env=AWS_SECRET_KEY=${AWS_SECRET_KEY} \
   --env=AWS_ENDPOINT=${AWS_ENDPOINT} \

.github/scripts/end2end/run-e2e-test.sh

@@ -106,6 +106,9 @@ run_e2e_test() {
         --env=RING_S3C_ENDPOINT=${RING_S3C_ENDPOINT} \
         --env=RING_S3C_BACKEND_SOURCE_LOCATION=${RING_S3C_BACKEND_SOURCE_LOCATION} \
         --env=RING_S3C_INGESTION_SRC_BUCKET_NAME=${RING_S3C_INGESTION_SRC_BUCKET_NAME} \
+        --env=RING_S3C_BACKEND_SOURCE_NON_VERSIONED_LOCATION=${RING_S3C_BACKEND_SOURCE_NON_VERSIONED_LOCATION} \
+        --env=RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME=${RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME} \
+        --env=RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE=${RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE} \
         --env=KEYCLOAK_TEST_USER=${KEYCLOAK_TEST_USER} \
         --env=KEYCLOAK_TEST_PASSWORD=${KEYCLOAK_TEST_PASSWORD} \
         --env=KEYCLOAK_TEST_HOST=${KEYCLOAK_TEST_HOST} \

.github/workflows/end2end.yaml

@@ -81,6 +81,9 @@ env:
   RING_S3C_ENDPOINT: http://s3c.local:8000
   RING_S3C_BACKEND_SOURCE_LOCATION: rings3cbackendingestion
   RING_S3C_INGESTION_SRC_BUCKET_NAME: ingestion-test-src-bucket-${{ github.run_id }}
+  RING_S3C_BACKEND_SOURCE_NON_VERSIONED_LOCATION: rings3cbackendingestionnonversioned
+  RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME: ingestion-test-src-non-versioned-bucket-${{ github.run_id }}
+  RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE: 2
   # CTST end2end tests
   NOTIF_DEST_NAME: "destination1"
   NOTIF_DEST_TOPIC: "destination-topic-1"

tests/zenko_tests/cleans3c.py

@@ -15,6 +15,7 @@ def get_env(key, default=None, error=False):
 RING_S3C_ACCESS_KEY = get_env('RING_S3C_ACCESS_KEY')
 RING_S3C_SECRET_KEY = get_env('RING_S3C_SECRET_KEY')
 RING_S3C_INGESTION_SRC_BUCKET_NAME = get_env('RING_S3C_INGESTION_SRC_BUCKET_NAME')
+RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME = get_env('RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME')
 RING_S3C_ENDPOINT = get_env('RING_S3C_ENDPOINT')
 ENABLE_RING_TESTS = get_env('ENABLE_RING_TESTS')
 VERIFY_CERTIFICATES = get_env('VERIFY_CERTIFICATES', False)
@@ -31,11 +32,12 @@ def bucket_safe_delete(bucketname):
     try:
         bucket = ring_s3c_client.Bucket(bucketname)
         _log.info('Deleting bucket %s' % bucket.name)
-        bucket.objects.all().delete()
+        bucket.object_versions.all().delete()
         bucket.delete()
     except Exception as exp:
         _log.info('Error deleting bucket %s - %s' % (bucket.name, str(exp)))
         raise exp
 
 _log.info('Removing S3C buckets...')
 bucket_safe_delete(RING_S3C_INGESTION_SRC_BUCKET_NAME)
+bucket_safe_delete(RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME)

tests/zenko_tests/create_buckets.py

@@ -25,13 +25,60 @@ def bucket_safe_create(bucket):
         _log.info('Error creating bucket %s - %s' % (bucket.name, str(exp)))
         raise exp
 
+def put_file(bucket, object_name, body):
+    """
+    Put an object in a bucket
+    """
+    try:
+        _log.info('Putting object %s' % object_name)
+        obj = bucket.Object(object_name)
+        obj.put(Body=body)
+    except Exception as exp:
+        _log.info('Error putting object %s - %s' % (bucket.name, object_name, str(exp)))
+        raise exp
+
+def put_singlepart_mpu(bucket, object_name, body):
+    """
+    Put an object in a bucket using a single part multipart upload
+    """
+    try:
+        _log.info('Putting object %s using multipart upload' % object_name)
+        mpu = bucket.meta.client.create_multipart_upload(
+            Bucket=bucket.name,
+            Key=object_name
+        )
+        response = bucket.meta.client.upload_part(
+            Body=body,
+            Bucket=bucket.name,
+            Key=object_name,
+            PartNumber=1,
+            UploadId=mpu['UploadId']
+        )
+        bucket.meta.client.complete_multipart_upload(
+            Bucket=bucket.name,
+            Key=object_name,
+            UploadId=mpu['UploadId'],
+            MultipartUpload={
+                'Parts': [{
+                    'PartNumber': 1,
+                    'ETag': response['ETag']
+                }]
+            }
+        )
+    except Exception as exp:
+        _log.error('Error in multipart upload for object %s - %s' % (object_name, str(exp)))
+        raise exp
+
 def create_ring_buckets():
     VERIFY_CERTIFICATES = get_env('VERIFY_CERTIFICATES', False)
     RING_S3C_ACCESS_KEY = get_env('RING_S3C_ACCESS_KEY')
     RING_S3C_SECRET_KEY = get_env('RING_S3C_SECRET_KEY')
-    RING_S3C_INGESTION_SRC_BUCKET_NAME = get_env('RING_S3C_INGESTION_SRC_BUCKET_NAME')
     RING_S3C_ENDPOINT = get_env('RING_S3C_ENDPOINT')
     ENABLE_RING_TESTS = get_env('ENABLE_RING_TESTS')
+    ## test bucket names
+    RING_S3C_INGESTION_SRC_BUCKET_NAME = get_env('RING_S3C_INGESTION_SRC_BUCKET_NAME')
+    RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME = get_env('RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME')
+    RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE = get_env('RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE')
 
     # Disable if Ring is not enabled
     if ENABLE_RING_TESTS == "false":
@@ -42,10 +89,28 @@ def create_ring_buckets():
     ring_s3c_client = s3c.resource('s3', endpoint_url=RING_S3C_ENDPOINT,
                       verify=VERIFY_CERTIFICATES)
 
+    versioned_bucket = ring_s3c_client.Bucket(RING_S3C_INGESTION_SRC_BUCKET_NAME)
+    non_versioned_bucket = ring_s3c_client.Bucket(RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME)
+
     ## Creating S3C buckets
     _log.info('Creating S3C buckets...')
-    bucket_safe_create(ring_s3c_client.Bucket(RING_S3C_INGESTION_SRC_BUCKET_NAME))
-    ring_s3c_client.Bucket(RING_S3C_INGESTION_SRC_BUCKET_NAME).Versioning().enable()
+    bucket_safe_create(versioned_bucket)
+    bucket_safe_create(non_versioned_bucket)
+
+    ## Adding non versioned objects before test execution to avoid
+    ## having to create the location mid tests which might cause flakiness.
+    ## A RING location can only be created if the bucket is versioned, and
+    ## once versioning is enabled it cannot be disabled.
+    _log.info('Putting non versioned objects...')
+    for i in range(int(RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE)):
+        put_file(non_versioned_bucket, 'simple-%d' % i, b'data')
+        put_file(non_versioned_bucket, 'zerobyte-%d' % i, b'')
+        put_singlepart_mpu(non_versioned_bucket, 'mpu-singlepart-%d' % i, b'mpudata')
+
+    ## Enabling versioning
+    _log.info('Enabling versioning on buckets...')
+    versioned_bucket.Versioning().enable()
+    non_versioned_bucket.Versioning().enable()
 
 def create_aws_buckets():
     VERIFY_CERTIFICATES = get_env('VERIFY_CERTIFICATES', False)
@@ -86,7 +151,6 @@ def create_azure_containers():
         except ResourceExistsError:
             _log.info('Container %s already exists!' % bucket_name)
 
-
 def create_azure_queues():
     AZURE_BACKEND_QUEUE_ENDPOINT = get_env("AZURE_BACKEND_QUEUE_ENDPOINT")
     AZURE_ACCOUNT_NAME = get_env("AZURE_ACCOUNT_NAME")

tests/zenko_tests/e2e-config.yaml.template

@@ -48,6 +48,15 @@ locations:
       secretKey: "${RING_S3C_SECRET_KEY}"
       bucketMatch: yes
       repoId: []
+  - name: "${RING_S3C_BACKEND_SOURCE_NON_VERSIONED_LOCATION}"
+    locationType: "location-scality-ring-s3-v1"
+    details:
+      bucketName: "${RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME}"
+      endpoint: "${RING_S3C_ENDPOINT}"
+      accessKey: "${RING_S3C_ACCESS_KEY}"
+      secretKey: "${RING_S3C_SECRET_KEY}"
+      bucketMatch: yes
+      repoId: []
   - name: "${GCP_BACKEND_DESTINATION_LOCATION}"
     locationType: "location-gcp-v1"
     details:

tests/zenko_tests/node_tests/backbeat/tests/ingestion/ingestionS3C.js

@@ -9,32 +9,44 @@ const scalityUtils = new IngestionUtility(scalityS3Client, ringS3Client);
 const ringS3CUtils = new IngestionUtility(ringS3Client);
 const ingestionSrcBucket = process.env.RING_S3C_INGESTION_SRC_BUCKET_NAME;
 const srcLocation = process.env.RING_S3C_BACKEND_SOURCE_LOCATION;
+const ingestionNonVersionedSrcBucket = process.env.RING_S3C_INGESTION_SRC_NON_VERSIONED_BUCKET_NAME;
+// Multiplying by 3 to account for the 3 types of objects: simple, zero-byte, and multipart
+const nonVersionedObjectCount = parseInt(process.env.RING_S3C_INGESTION_NON_VERSIONED_OBJECT_COUNT_PER_TYPE, 10) * 3;
+const srcNonVersionedLocation = process.env.RING_S3C_BACKEND_SOURCE_NON_VERSIONED_LOCATION;
 const location = srcLocation;
 let INGESTION_DEST_BUCKET;
 let KEY_PREFIX;
 let OBJ_KEY;
 
 describe('Ingesting existing data from RING S3C bucket', () => {
-    beforeEach(() => {
+    beforeEach(function () {
         INGESTION_DEST_BUCKET = `ingestion-${uuid()}`;
         KEY_PREFIX = `${ingestionSrcBucket}-${uuid()}`;
         OBJ_KEY = `${KEY_PREFIX}/object-to-ingest-${uuid()}`;
+        this.testState = {
+            skipCleanup: false,
+        };
     });
 
-    afterEach(done => async.series([
-        next => ringS3CUtils.deleteAllVersions(
-            ingestionSrcBucket,
-            null,
-            next,
-        ),
-        next => ringS3CUtils.putBucketVersioning(
-            ingestionSrcBucket,
-            'Enabled',
-            next,
-        ),
-        next => scalityUtils.waitUntilEmpty(INGESTION_DEST_BUCKET, next),
-        next => scalityUtils.deleteVersionedBucket(INGESTION_DEST_BUCKET, next),
-    ], done));
+    afterEach(function (done) {
+        if (this.testState.skipCleanup) {
+            return done();
+        }
+        return async.series([
+            next => ringS3CUtils.deleteAllVersions(
+                ingestionSrcBucket,
+                null,
+                next,
+            ),
+            next => ringS3CUtils.putBucketVersioning(
+                ingestionSrcBucket,
+                'Enabled',
+                next,
+            ),
+            next => scalityUtils.waitUntilEmpty(INGESTION_DEST_BUCKET, next),
+            next => scalityUtils.deleteVersionedBucket(INGESTION_DEST_BUCKET, next),
+        ], done);
+    });
 
     it('should ingest an object', done => async.waterfall([
         // object
@@ -356,4 +368,40 @@ describe('Ingesting existing data from RING S3C bucket', () => {
             next,
         ),
     ], done));
+
+    // The source bucket used in this test has non versioned objects of 3 types: simple, zero-byte, and multipart
+    // These non versioned objects were put before creating the location as the bucket has to be versioned
+    // for the location to be created, and once it's versioned can't be set to non versioned again. This is
+    // done to avoid having to create the location during the tests which might impact other tests running
+    // in parallel.
+    it('should ingest all non versioned objects', function (done) {
+        // skipping cleanup as deleting the ingestion bucket requires
+        // deleting all objects which will delete the source bucket objects
+        this.testState.skipCleanup = true;
+        async.waterfall([
+            next => scalityUtils.createIngestionBucket(
+                INGESTION_DEST_BUCKET,
+                srcNonVersionedLocation,
+                next,
+            ),
+            next => ringS3CUtils.s3.listObjectVersions({
+                Bucket: ingestionNonVersionedSrcBucket,
+            }, next),
+            (data, next) => {
+                assert.strictEqual(data.Versions.length, nonVersionedObjectCount);
+                assert.strictEqual(data.DeleteMarkers.length, 0);
+                async.forEach(data.Versions, (version, cb) => {
+                    scalityUtils.compareObjectsRINGS3C(
+                        ingestionNonVersionedSrcBucket,
+                        INGESTION_DEST_BUCKET,
+                        version.Key,
+                        'null',
+                        undefined,
+                        cb,
+                    );
+                }, next);
+            },
+        ], done);
+    });
 });
+