BB-168: separate transition processor

alexanderchan-scality · alexanderchan-scality · commit da6dca563a55 · 2022-05-06T16:01:36.000-07:00
diff --git a/extensions/lifecycle/LifecycleConfigValidator.js b/extensions/lifecycle/LifecycleConfigValidator.js
@@ -46,6 +46,13 @@ const joiSchema = joi.object({
         concurrency: joi.number().greater(0).default(10),
         probeServer: probeServerJoi.default(),
     },
+    transitionProcessor: {
+        auth: inheritedAuthJoi,
+        groupId: joi.string().required(),
+        retry: retryParamsJoi,
+        concurrency: joi.number().greater(0).default(10),
+        probeServer: probeServerJoi.default(),
+    },
 });
 
 function configValidator(backbeatConfig, extConfig) {
diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectExpirationProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectExpirationProcessor.js
@@ -0,0 +1,68 @@
+'LifecycleObjectExpirationProcessor'; // eslint-disable-line
+
+const Logger = require('werelogs').Logger;
+
+const LifecycleObjectProcessor = require('./LifecycleObjectProcessor');
+const LifecycleDeleteObjectTask =
+      require('../tasks/LifecycleDeleteObjectTask');
+
+class LifecycleObjectExpirationProcessor extends LifecycleObjectProcessor {
+
+    /**
+     * Constructor of LifecycleObjectProcessor
+     *
+     * @constructor
+     * @param {Object} zkConfig - zookeeper configuration object
+     * @param {String} zkConfig.connectionString - zookeeper connection string
+     *  as "host:port[/chroot]"
+     * @param {Object} kafkaConfig - kafka configuration object
+     * @param {string} kafkaConfig.hosts - list of kafka brokers
+     *   as "host:port[,host:port...]"
+     * @param {Object} [kafkaConfig.backlogMetrics] - param object to
+     * publish kafka topic metrics to zookeeper (see {@link
+     * BackbeatConsumer} constructor)
+     * @param {Object} lcConfig - lifecycle configuration object
+     * @param {String} lcConfig.auth - authentication info
+     * @param {String} lcConfig.objectTasksTopic - lifecycle object topic name
+     * @param {Object} lcConfig.objectProcessor - kafka consumer object
+     * @param {String} lcConfig.objectProcessor.groupId - kafka
+     * consumer group id
+     * @param {Number} [lcConfig.objectProcessor.concurrency] - number
+     *  of max allowed concurrent operations
+     * @param {Object} s3Config - S3 configuration
+     * @param {Object} s3Config.host - s3 endpoint host
+     * @param {Number} s3Config.port - s3 endpoint port
+     * @param {String} [transport="http"] - transport method ("http"
+     *  or "https")
+     */
+    constructor(zkConfig, kafkaConfig, lcConfig, s3Config, transport = 'http') {
+        super(zkConfig, kafkaConfig, lcConfig, s3Config, transport);
+        this._log = new Logger('Backbeat:Lifecycle:ObjectExpirationProcessor');
+    }
+
+    getProcessConfig(lcConfig) {
+        return lcConfig.objectProcessor;
+    }
+
+    getAuthConfig(lcConfig) {
+        if (lcConfig.objectProcessor.auth) {
+            return lcConfig.objectProcessor.auth;
+        }
+
+        return lcConfig.auth;
+    }
+
+    getTask(actionEntry) {
+        const actionType = actionEntry.getActionType();
+
+        if (actionType !== 'deleteObject' && actionType !== 'deleteMPU') {
+            this._log.warn(`skipped unsupported action ${actionType}`,
+                             actionEntry.getLogInfo());
+            return null;
+        }
+
+        return new LifecycleDeleteObjectTask(this);
+    }
+}
+
+module.exports = LifecycleObjectExpirationProcessor;
diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectProcessor.js
@@ -6,10 +6,6 @@ const https = require('https');
 const { EventEmitter } = require('events');
 const Logger = require('werelogs').Logger;
 
-const LifecycleDeleteObjectTask =
-      require('../tasks/LifecycleDeleteObjectTask');
-const LifecycleUpdateTransitionTask =
-      require('../tasks/LifecycleUpdateTransitionTask');
 const BackbeatConsumer = require('../../../lib/BackbeatConsumer');
 const BackbeatMetadataProxy = require('../../../lib/BackbeatMetadataProxy');
 const ActionQueueEntry = require('../../../lib/models/ActionQueueEntry');
@@ -48,10 +44,7 @@ class LifecycleObjectProcessor extends EventEmitter {
      * @param {Object} lcConfig - lifecycle configuration object
      * @param {String} lcConfig.auth - authentication info
      * @param {String} lcConfig.objectTasksTopic - lifecycle object topic name
-     * @param {Object} lcConfig.objectProcessor - kafka consumer object
-     * @param {String} lcConfig.objectProcessor.groupId - kafka
      * consumer group id
-     * @param {Number} [lcConfig.objectProcessor.concurrency] - number
      *  of max allowed concurrent operations
      * @param {Object} s3Config - S3 configuration
      * @param {Object} s3Config.host - s3 endpoint host
@@ -65,7 +58,8 @@ class LifecycleObjectProcessor extends EventEmitter {
         this._zkConfig = zkConfig;
         this._kafkaConfig = kafkaConfig;
         this._lcConfig = lcConfig;
-        this._authConfig = lcConfig.objectProcessor.auth || lcConfig.auth;
+        this._processConfig = this.getProcessConfig(this._lcConfig);
+        this._authConfig = this.getAuthConfig(this._lcConfig);
         this._s3Config = s3Config;
         this._transport = transport;
         this._consumer = null;
@@ -99,8 +93,8 @@ class LifecycleObjectProcessor extends EventEmitter {
                 backlogMetrics: this._kafkaConfig.backlogMetrics,
             },
             topic: this._lcConfig.objectTasksTopic,
-            groupId: this._lcConfig.objectProcessor.groupId,
-            concurrency: this._lcConfig.objectProcessor.concurrency,
+            groupId: this._processConfig.groupId,
+            concurrency: this._processConfig.concurrency,
             queueProcessor: this.processKafkaEntry.bind(this),
         });
         this._consumer.on('error', err => {
@@ -261,6 +255,32 @@ class LifecycleObjectProcessor extends EventEmitter {
         this._consumer.close(cb);
     }
 
+    /**
+     * Retrieve object processor config
+     * @return {object} - process config
+     */
+    getProcessConfig() {
+        throw new Error('LifecycleObjectProcessor.getProcessConfig not implemented');
+    }
+
+    /**
+     * Retrieve process auth config
+     * @return {object} - auth config
+     */
+    getAuthConfig() {
+        throw new Error('LifecycleObjectProcessor.getAuthConfig not implemented');
+    }
+
+    /**
+     * Retrieve object processor task action
+     * @param {ActionQueueEntry} actionEntry - lifecycle action entry
+     * @return {BackbeatTask|null} - backbeat task object
+     */
+    // eslint-disable-next-line
+    getTask(actionEntry) {
+        return null;
+    }
+
     /**
      * Proceed to the lifecycle action of an object given a kafka
      * object lifecycle queue entry
@@ -279,20 +299,12 @@ class LifecycleObjectProcessor extends EventEmitter {
         }
         this._log.debug('processing lifecycle object entry',
                           actionEntry.getLogInfo());
-        const actionType = actionEntry.getActionType();
-        let task;
-        if (actionType === 'deleteObject' ||
-            actionType === 'deleteMPU') {
-            task = new LifecycleDeleteObjectTask(this);
-        } else if (actionType === 'copyLocation' &&
-                   actionEntry.getContextAttribute('ruleType')
-                   === 'transition') {
-            task = new LifecycleUpdateTransitionTask(this);
-        } else {
-            this._log.warn(`skipped unsupported action ${actionType}`,
-                             actionEntry.getLogInfo());
+        const task = this.getTask(actionEntry);
+
+        if (task === null) {
             return process.nextTick(done);
         }
+
         return this.retryWrapper.retry({
             actionDesc: 'process lifecycle object entry',
             logFields: actionEntry.getLogInfo(),
@@ -306,6 +318,7 @@ class LifecycleObjectProcessor extends EventEmitter {
         return {
             s3Config: this._s3Config,
             lcConfig: this._lcConfig,
+            processConfig: this._processConfig,
             authConfig: this._authConfig,
             getS3Client: this._getS3Client.bind(this),
             getBackbeatClient: this._getBackbeatClient.bind(this),
@@ -317,7 +330,6 @@ class LifecycleObjectProcessor extends EventEmitter {
     isReady() {
         return this._consumer && this._consumer.isReady();
     }
-
 }
 
 module.exports = LifecycleObjectProcessor;
diff --git a/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js b/extensions/lifecycle/objectProcessor/LifecycleObjectTransitionProcessor.js
@@ -0,0 +1,69 @@
+'use strict'; // eslint-disable-line
+
+const Logger = require('werelogs').Logger;
+
+const LifecycleObjectProcessor = require('./LifecycleObjectProcessor');
+const LifecycleUpdateTransitionTask =
+      require('../tasks/LifecycleUpdateTransitionTask');
+
+class LifecycleObjectTransitionProcessor extends LifecycleObjectProcessor {
+
+    /**
+     * Constructor of LifecycleObjectProcessor
+     *
+     * @constructor
+     * @param {Object} zkConfig - zookeeper configuration object
+     * @param {String} zkConfig.connectionString - zookeeper connection string
+     *  as "host:port[/chroot]"
+     * @param {Object} kafkaConfig - kafka configuration object
+     * @param {string} kafkaConfig.hosts - list of kafka brokers
+     *   as "host:port[,host:port...]"
+     * @param {Object} [kafkaConfig.backlogMetrics] - param object to
+     * publish kafka topic metrics to zookeeper (see {@link
+     * BackbeatConsumer} constructor)
+     * @param {Object} lcConfig - lifecycle configuration object
+     * @param {String} lcConfig.auth - authentication info
+     * @param {String} lcConfig.objectTasksTopic - lifecycle object topic name
+     * @param {Object} lcConfig.transitionProcessor - kafka consumer object
+     * @param {String} lcConfig.transitionProcessor.groupId - kafka
+     * consumer group id
+     * @param {Number} [lcConfig.transitionProcessor.concurrency] - number
+     *  of max allowed concurrent operations
+     * @param {Object} s3Config - S3 configuration
+     * @param {Object} s3Config.host - s3 endpoint host
+     * @param {Number} s3Config.port - s3 endpoint port
+     * @param {String} [transport="http"] - transport method ("http"
+     *  or "https")
+     */
+    constructor(zkConfig, kafkaConfig, lcConfig, s3Config, transport = 'http') {
+        super(zkConfig, kafkaConfig, lcConfig, s3Config, transport);
+        this._log = new Logger('Backbeat:Lifecycle:ObjectTransitionProcessor');
+    }
+
+    getProcessConfig(lcConfig) {
+        return lcConfig.transitionProcessor;
+    }
+
+    getAuthConfig(lcConfig) {
+        if (lcConfig.transitionProcessor.auth) {
+            return lcConfig.transitionProcessor.auth;
+        }
+
+        return lcConfig.auth;
+    }
+
+    getTask(actionEntry) {
+        const actionType = actionEntry.getActionType();
+
+        if (actionType !== 'copyLocation' ||
+            actionEntry.getContextAttribute('ruleType') !== 'transition') {
+            this._log.warn(`skipped unsupported  action ${actionType}`,
+                             actionEntry.getLogInfo());
+            return null;
+        }
+
+        return new LifecycleUpdateTransitionTask(this);
+    }
+}
+
+module.exports = LifecycleObjectTransitionProcessor;
diff --git a/extensions/lifecycle/objectProcessor/task.js b/extensions/lifecycle/objectProcessor/task.js
@@ -12,7 +12,8 @@ const { sendSuccess, sendError } = require('arsenal').network.probe.Utils;
 const { ZenkoMetrics } = require('arsenal').metrics;
 
 const { initManagement } = require('../../../lib/management/index');
-const LifecycleObjectProcessor = require('./LifecycleObjectProcessor');
+const LifecycleObjectExpirationProcessor = require('./LifecycleObjectExpirationProcessor');
+const LifecycleObjectTransitionProcessor = require('./LifecycleObjectTransitionProcessor');
 const { startProbeServer } = require('../../../lib/util/probe');
 const config = require('../../../lib/Config');
 
@@ -24,8 +25,19 @@ const transport = config.transport;
 
 const logger = new werelogs.Logger('Backbeat:Lifecycle:Consumer');
 
-const objectProcessor = new LifecycleObjectProcessor(
-    zkConfig, kafkaConfig, lcConfig, s3Config, transport);
+let objectProcessor;
+
+switch (process.env.LIFECYCLE_OBJECT_PROCESSOR_TYPE) {
+    case 'transition':
+        objectProcessor = new LifecycleObjectTransitionProcessor(
+            zkConfig, kafkaConfig, lcConfig, s3Config, transport);
+        break;
+    case 'expiration': // fallthrough
+    default:
+        objectProcessor = new LifecycleObjectExpirationProcessor(
+            zkConfig, kafkaConfig, lcConfig, s3Config, transport);
+        break;
+}
 
 werelogs.configure({ level: config.log.logLevel,
     dump: config.log.dumpLevel });
diff --git a/package.json b/package.json
@@ -11,6 +11,7 @@
     "lifecycle_conductor": "node extensions/lifecycle/conductor/service.js",
     "lifecycle_bucket_processor": "node extensions/lifecycle/bucketProcessor/task.js",
     "lifecycle_object_processor": "node extensions/lifecycle/objectProcessor/task.js",
+    "lifecycle_object_transition_processor": "LIFECYCLE_OBJECT_PROCESSOR_TYPE=transition node extensions/lifecycle/objectProcessor/task.js",
     "mongo_queue_processor": "node extensions/mongoProcessor/mongoProcessorTask.js",
     "garbage_collector": "node extensions/gc/service.js",
     "test": "mocha --recursive tests/unit",
diff --git a/tests/functional/lifecycle/LifecycleConductor.spec.js b/tests/functional/lifecycle/LifecycleConductor.spec.js
@@ -315,7 +315,7 @@ describe('lifecycle conductor', function lifecycleConductor() {
             };
         }
 
-        describe(description, () => {
+        return describe(description, () => {
             beforeEach(done => {
                 bucketdListing = [];
 
diff --git a/tests/functional/lifecycle/LifecycleObjectProcessor.spec.js b/tests/functional/lifecycle/LifecycleObjectProcessor.spec.js
@@ -2,8 +2,8 @@
 
 const werelogs = require('werelogs');
 
-const LifecycleObjectProcessor = require(
-    '../../../extensions/lifecycle/objectProcessor/LifecycleObjectProcessor');
+const LifecycleObjectExpirationProcessor = require(
+    '../../../extensions/lifecycle/objectProcessor/LifecycleObjectExpirationProcessor');
 
 const { S3ClientMock } = require('../../utils/S3ClientMock');
 
@@ -23,7 +23,7 @@ describe('Lifecycle Object Processor', function lifecycleObjectProcessor() {
 
     function generateRetryTest(failures, message) {
         return function testRetries(done) {
-            const lop = new LifecycleObjectProcessor(
+            const lop = new LifecycleObjectExpirationProcessor(
                 zkConfig, kafkaConfig, lcConfig, s3Config);
 
             const s3Client = new S3ClientMock(failures);
diff --git a/tests/functional/lifecycle/configObjects.js b/tests/functional/lifecycle/configObjects.js
@@ -24,6 +24,9 @@ const lcConfig = {
     objectProcessor: {
         groupId: `object-processor-test-${Math.random()}`,
     },
+    transitionProcessor: {
+        groupId: `transition-processor-test-${Math.random()}`,
+    },
     bucketTasksTopic,
     objectTasksTopic,
     rules: {

Original file line number	Diff line number	Diff line change
`@@ -315,7 +315,7 @@ describe('lifecycle conductor', function lifecycleConductor() {`
`315`	`315`	`};`
`316`	`316`	`}`
`317`	`317`
`318`		`- describe(description, () => {`
	`318`	`+ return describe(description, () => {`
`319`	`319`	`beforeEach(done => {`
`320`	`320`	`bucketdListing = [];`
`321`	`321`