bucket notifications - allow kafka topic in TopicArn, default connection

Signed-off-by: Amit Prinz Setter <[email protected]>

Author: alphaprinz

src/cmd/manage_nsfs.js

@@ -422,7 +422,8 @@ async function fetch_account_data(action, user_input) {
             gid: user_input.user ? undefined : user_input.gid,
             new_buckets_path: user_input.new_buckets_path,
             fs_backend: user_input.fs_backend ? String(user_input.fs_backend) : config.NSFS_NC_STORAGE_BACKEND
-        }
+        },
+        default_connection: user_input.default_connection === undefined ? undefined : String(user_input.default_connection)
     };
     if (action === ACTIONS.UPDATE || action === ACTIONS.DELETE) {
         // @ts-ignore

src/endpoint/s3/ops/s3_put_bucket_notification.js

@@ -11,12 +11,38 @@ async function put_bucket_notification(req) {
 
     let topic_configuration = req.body.NotificationConfiguration?.TopicConfiguration;
 
+    console.log("AAAAAAAAAAAAAAAAAAAAAA req.object_sdk.requesting_account = ", req.object_sdk.requesting_account);
+
+    const default_connection = req.object_sdk.requesting_account.default_connection;
+    console.log("AAAAAAAAAAAAAAAAAAAAAAAAAAA2 default_connection = ", default_connection);
+
     //adapt to db shcema
     if (topic_configuration) {
         for (const conf of topic_configuration) {
             conf.id = conf.Id;
             conf.event = conf.Event;
             conf.topic = conf.Topic;
+            //if Topic is not specified, default to account's default connection
+            if (conf.Topic && conf.Topic.length > 0 && conf.Topic[0].startsWith('kafka:::topic/')) {
+                const kafka_topic_parts = conf.Topic[0].split('/');
+                console.log("AAAAAAAAAAAAAAAAAAAAAAAA5 kafka_topic_parts", kafka_topic_parts);
+                if (kafka_topic_parts.length != 3) {
+                    throw new S3Error({
+                        code: 'InvalidArgument',
+                        message: "kafka:::topic is invalid. Must be of syntax: kafka:::topic:/connection/topic",
+                        http_code: 400,
+                        detail: conf.Topic[0]
+                    });
+                }
+                let connection = default_connection;
+                if (typeof kafka_topic_parts[1] === 'string' && kafka_topic_parts[1].length > 0) {
+                    connection = kafka_topic_parts[1];
+                }
+                const topic = kafka_topic_parts[2];
+                conf.topic = ['kafka:::topic/' + connection + "/" + topic];
+            }
+            console.log("AAAAAAAAAAAAAAAAAAAA3 conf.Topic = ", conf.Topic);
+            console.log("AAAAAAAAAAAAAAA4 conf.topic = ", conf.topic);
             delete conf.Id;
             delete conf.Event;
             delete conf.Topic;

src/manage_nsfs/manage_nsfs_constants.js

@@ -46,8 +46,8 @@ const FROM_FILE = 'from_file';
 const ANONYMOUS = 'anonymous';
 
 const VALID_OPTIONS_ACCOUNT = {
-    'add': new Set(['name', 'uid', 'gid', 'supplemental_groups', 'new_buckets_path', 'user', 'access_key', 'secret_key', 'fs_backend', 'allow_bucket_creation', 'force_md5_etag', 'iam_operate_on_root_account', FROM_FILE, ...CLI_MUTUAL_OPTIONS]),
-    'update': new Set(['name', 'uid', 'gid', 'supplemental_groups', 'new_buckets_path', 'user', 'access_key', 'secret_key', 'fs_backend', 'allow_bucket_creation', 'force_md5_etag', 'iam_operate_on_root_account', 'new_name', 'regenerate', ...CLI_MUTUAL_OPTIONS]),
+    'add': new Set(['name', 'uid', 'gid', 'supplemental_groups', 'new_buckets_path', 'user', 'access_key', 'secret_key', 'fs_backend', 'allow_bucket_creation', 'force_md5_etag', 'iam_operate_on_root_account', 'default_connection', FROM_FILE, ...CLI_MUTUAL_OPTIONS]),
+    'update': new Set(['name', 'uid', 'gid', 'supplemental_groups', 'new_buckets_path', 'user', 'access_key', 'secret_key', 'fs_backend', 'allow_bucket_creation', 'force_md5_etag', 'iam_operate_on_root_account', 'new_name', 'regenerate', 'default_connection', ...CLI_MUTUAL_OPTIONS]),
     'delete': new Set(['name', ...CLI_MUTUAL_OPTIONS]),
     'list': new Set(['wide', 'show_secrets', 'gid', 'uid', 'user', 'name', 'access_key', ...CLI_MUTUAL_OPTIONS]),
     'status': new Set(['name', 'access_key', 'show_secrets', ...CLI_MUTUAL_OPTIONS]),
@@ -142,6 +142,7 @@ const OPTION_TYPE = {
     ips: 'string',
     force: 'boolean',
     anonymous: 'boolean',
+    default_connection: 'string',
     // health options
     deployment_type: 'string',
     all_account_details: 'boolean',

src/server/system_services/schemas/nsfs_account_schema.js

@@ -101,5 +101,8 @@ module.exports = {
                 }
             }]
         },
+        default_connection: {
+            type: 'string'
+        }
     }
 };

src/util/notifications_util.js

@@ -193,7 +193,7 @@ class Notificator {
         if (connection_name.startsWith("kafka:::topic/")) {
             const connection_parts = connection_name.split('/');
             connect_filename = connection_parts[1];
-            kafka_topic_from_connection_name = connection_parts.length > 2 && connection_parts[3];
+            kafka_topic_from_connection_name = connection_parts.length > 1 && connection_parts[2];
         }
 
         if (this.nc_config_fs) {