finished multi backend apis

SylvainSenechal · SylvainSenechal · commit d06be271e51d · 2025-10-13T11:19:29.000+02:00
diff --git a/extensions/replication/tasks/MultipleBackendTask.js b/extensions/replication/tasks/MultipleBackendTask.js
@@ -9,7 +9,12 @@ const ObjectQueueEntry = require('../../../lib/models/ObjectQueueEntry');
 const ReplicateObject = require('./ReplicateObject');
 const { 
     GetObjectCommand,
-    MultipleBackendPutObjectCommand
+    MultipleBackendPutObjectCommand,
+    MultipleBackendDeleteObjectCommand,
+    MultipleBackendInitiateMPUCommand,
+    MultipleBackendAbortMPUCommand,
+    MultipleBackendPutObjectTaggingCommand,
+    MultipleBackendDeleteObjectTaggingCommand
 } = require('../../../lib/clients/smithy/build/smithy/source/typescript-codegen');
 const { attachReqUids } = require('../../../lib/clients/utils');
 const getExtMetrics = require('../utils/getExtMetrics');
@@ -304,16 +309,20 @@ class MultipleBackendTask extends ReplicateObject {
             uploadId,
         });
         const doneOnce = jsutil.once(cb);
-        const destReq = this.backbeatSource.multipleBackendAbortMPU({
+        const command = new MultipleBackendAbortMPUCommand({
             Bucket: sourceEntry.getBucket(),
             Key: sourceEntry.getObjectKey(),
             StorageType: sourceEntry.getReplicationStorageType(),
             StorageClass: this.site,
             UploadId: uploadId,
+            RequestUids: log.getSerializedUids(),
         });
-        attachReqUids(destReq, log);
-        return destReq.send(err => {
-            if (err) {
+        
+        return this.backbeatSource.send(command)
+            .then(() => {
+                return doneOnce();
+            })
+            .catch(err => {
                 // eslint-disable-next-line no-param-reassign
                 err.origin = 'source';
                 log.error('an error occurred aborting multipart upload', {
@@ -324,9 +333,7 @@ class MultipleBackendTask extends ReplicateObject {
                     error: err.message,
                 });
                 return doneOnce(err);
-            }
-            return doneOnce();
-        });
+            });
     }
 
    /**
@@ -450,7 +457,7 @@ class MultipleBackendTask extends ReplicateObject {
             const uploadId = uuid().replace(/-/g, '');
             return setImmediate(() => cb(null, uploadId));
         }
-        const destReq = this.backbeatSource.multipleBackendInitiateMPU({
+        const command = new MultipleBackendInitiateMPUCommand({
             Bucket: sourceEntry.getBucket(),
             Key: sourceEntry.getObjectKey(),
             StorageType: sourceEntry.getReplicationStorageType(),
@@ -463,10 +470,14 @@ class MultipleBackendTask extends ReplicateObject {
                 sourceEntry.getContentDisposition() || undefined,
             ContentEncoding: sourceEntry.getContentEncoding() || undefined,
             Tags: JSON.stringify(sourceEntry.getTags()),
+            RequestUids: log.getSerializedUids(),
         });
-        attachReqUids(destReq, log);
-        return destReq.send((err, data) => {
-            if (err) {
+        
+        return this.backbeatSource.send(command)
+            .then(data => {
+                return cb(null, data.uploadId);
+            })
+            .catch(err => {
                 // eslint-disable-next-line no-param-reassign
                 err.origin = 'source';
                 log.error('an error occurred on initating MPU to S3', {
@@ -477,9 +488,7 @@ class MultipleBackendTask extends ReplicateObject {
                     error: err.message,
                 });
                 return cb(err);
-            }
-            return cb(null, data.uploadId);
-        });
+            });
     }
 
     /**
@@ -907,36 +916,38 @@ class MultipleBackendTask extends ReplicateObject {
         log.debug('replicating object tags', {
             entry: sourceEntry.getLogInfo(),
         });
-        const destReq = this.backbeatSource
-            .multipleBackendPutObjectTagging({
-                Bucket: sourceEntry.getBucket(),
-                Key: sourceEntry.getObjectKey(),
-                StorageType: sourceEntry.getReplicationStorageType(),
-                StorageClass: this.site,
-                DataStoreVersionId:
-                    sourceEntry.getReplicationSiteDataStoreVersionId(this.site),
-                Tags: JSON.stringify(sourceEntry.getTags()),
-                SourceBucket: sourceEntry.getBucket(),
-                SourceVersionId: sourceEntry.getVersionId(),
-                ReplicationEndpointSite: this.site,
-            });
-        attachReqUids(destReq, log);
+        const command = new MultipleBackendPutObjectTaggingCommand({
+            Bucket: sourceEntry.getBucket(),
+            Key: sourceEntry.getObjectKey(),
+            StorageType: sourceEntry.getReplicationStorageType(),
+            StorageClass: this.site,
+            DataStoreVersionId:
+                sourceEntry.getReplicationSiteDataStoreVersionId(this.site),
+            Tags: JSON.stringify(sourceEntry.getTags()),
+            SourceBucket: sourceEntry.getBucket(),
+            SourceVersionId: sourceEntry.getVersionId(),
+            ReplicationEndpointSite: this.site,
+            RequestUids: log.getSerializedUids(),
+        });
+        
         const writeStartTime = Date.now();
-        return destReq.send((err, data) => {
-            if (err) {
+        return this.backbeatSource.send(command)
+            .then(data => {
+                sourceEntry.setReplicationSiteDataStoreVersionId(this.site,
+                    data.versionId);
+                // TODO : review metadata metrics
+                this._publishMetadataWriteMetrics(JSON.stringify(command.input), writeStartTime);
+                return doneOnce();
+            })
+            .catch(err => {
                 log.error('an error occurred putting object tagging to S3', {
                     method: 'MultipleBackendTask._putObjectTaggingOnce',
                     entry: sourceEntry.getLogInfo(),
                     origin: 'target',
                     error: err.message,
                 });
                 return doneOnce(err);
-            }
-            sourceEntry.setReplicationSiteDataStoreVersionId(this.site,
-                data.versionId);
-            this._publishMetadataWriteMetrics(destReq.httpRequest.body, writeStartTime);
-            return doneOnce();
-        });
+            });
     }
 
     _deleteObjectTagging(sourceEntry, log, cb) {
@@ -961,7 +972,7 @@ class MultipleBackendTask extends ReplicateObject {
         log.debug('replicating delete object tagging', {
             entry: sourceEntry.getLogInfo(),
         });
-        const destReq = this.backbeatSource.multipleBackendDeleteObjectTagging({
+        const command = new MultipleBackendDeleteObjectTaggingCommand({
             Bucket: sourceEntry.getBucket(),
             Key: sourceEntry.getObjectKey(),
             StorageType: sourceEntry.getReplicationStorageType(),
@@ -971,11 +982,19 @@ class MultipleBackendTask extends ReplicateObject {
             SourceBucket: sourceEntry.getBucket(),
             SourceVersionId: sourceEntry.getVersionId(),
             ReplicationEndpointSite: this.site,
+            RequestUids: log.getSerializedUids(),
         });
-        attachReqUids(destReq, log);
+        
         const writeStartTime = Date.now();
-        return destReq.send((err, data) => {
-            if (err) {
+        return this.backbeatSource.send(command)
+            .then(data => {
+                sourceEntry.setReplicationSiteDataStoreVersionId(this.site,
+                    data.versionId);
+                // TODO : review metadata metrics
+                this._publishMetadataWriteMetrics(JSON.stringify(command.input), writeStartTime);
+                return doneOnce();
+            })
+            .catch(err => {
                 log.error('an error occurred on deleting object tagging', {
                     method: 'MultipleBackendTask._deleteObjectTaggingOnce',
                     entry: sourceEntry.getLogInfo(),
@@ -984,12 +1003,7 @@ class MultipleBackendTask extends ReplicateObject {
                     error: err.message,
                 });
                 return doneOnce(err);
-            }
-            sourceEntry.setReplicationSiteDataStoreVersionId(this.site,
-                data.versionId);
-            this._publishMetadataWriteMetrics(destReq.httpRequest.body, writeStartTime);
-            return doneOnce();
-        });
+            });
     }
 
     _putDeleteMarker(sourceEntry, log, cb) {
@@ -1038,16 +1052,22 @@ class MultipleBackendTask extends ReplicateObject {
      * @return {undefined}
      */
     _sendMultipleBackendDeleteObject(sourceEntry, log, doneOnce) {
-        const destReq = this.backbeatSource.multipleBackendDeleteObject({
+        const command = new MultipleBackendDeleteObjectCommand({
             Bucket: sourceEntry.getBucket(),
             Key: sourceEntry.getObjectKey(),
             StorageType: sourceEntry.getReplicationStorageType(),
             StorageClass: this.site,
+            RequestUids: log.getSerializedUids(),
         });
-        attachReqUids(destReq, log);
+        
         const writeStartTime = Date.now();
-        return destReq.send(err => {
-            if (err) {
+        return this.backbeatSource.send(command)
+            .then(() => {
+                // TODO : This metric is changed, needs double checking
+                this._publishMetadataWriteMetrics(JSON.stringify(command.input), writeStartTime);
+                return doneOnce();
+            })
+            .catch(err => {
                 // eslint-disable-next-line no-param-reassign
                 err.origin = 'source';
                 log.error('an error occurred on putting delete marker to S3', {
@@ -1058,10 +1078,7 @@ class MultipleBackendTask extends ReplicateObject {
                     error: err.message,
                 });
                 return doneOnce(err);
-            }
-            this._publishMetadataWriteMetrics(destReq.httpRequest.body, writeStartTime);
-            return doneOnce();
-        });
+            });
     }
 
     /**
diff --git a/extensions/replication/tasks/ReplicateObject.js b/extensions/replication/tasks/ReplicateObject.js
@@ -11,13 +11,7 @@ const BackbeatMetadataProxy = require('../../../lib/BackbeatMetadataProxy');
 const { 
     CloudserverClient,
     CloudserverClientConfig,
-    GetObjectCommand,
     PutDataCommand,
-    MultipleBackendPutObjectCommand,
-    MultipleBackendInitiateMPUCommand,
-    MultipleBackendPutMPUPartCommand,
-    MultipleBackendCompleteMPUCommand,
-    MultipleBackendAbortMPUCommand,
     BatchDeleteCommand
 } = require('../../../lib/clients/smithy/build/smithy/source/typescript-codegen');
 
diff --git a/lib/clients/ClientManager.js b/lib/clients/ClientManager.js
@@ -6,12 +6,6 @@ const { http: HttpAgent, https: HttpsAgent } = require('httpagent');
 const { 
     CloudserverClient,
     CloudserverClientConfig,
-    GetObjectCommand,
-    MultipleBackendPutObjectCommand,
-    MultipleBackendInitiateMPUCommand,
-    MultipleBackendPutMPUPartCommand,
-    MultipleBackendCompleteMPUCommand,
-    MultipleBackendAbortMPUCommand
 } = require('../../../lib/clients/smithy/build/smithy/source/typescript-codegen');
 
 // TODO: test inactive credential deletion
