bucket notifications - allow kafka topic in TopicArn, default connection

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

Author: alphaprinz

docs/NooBaaNonContainerized/NooBaaCLI.md

@@ -75,7 +75,7 @@ The `account add` command is used to create a new account with customizable opti
 ```sh
 noobaa-cli account add --name <account_name> --uid <uid> --gid <gid> [--user]
 [--new_buckets_path][--access_key][--secret_key][--fs_backend]
-[--allow_bucket_creation][--force_md5_etag][--anonymous][--from_file][--iam_operate_on_root_account]
+[--allow_bucket_creation][--force_md5_etag][--anonymous][--from_file][--iam_operate_on_root_account][--default_connection]
 ```
 #### Flags -
 - `name` (Required)
@@ -135,6 +135,10 @@ noobaa-cli account add --name <account_name> --uid <uid> --gid <gid> [--user]
     - Type: Boolean
     - Description: Specifies if the account allowed to create root accounts using the IAM API (the default behavior is to create of IAM accounts). See - [IAM - Root Accounts Manager](./../design/iam.md#root-accounts-manager).
 
+- `default_connection`
+    - Type: String
+    - Description: A default account for Kafka external servers. See bucket-notifications.md.
+
 ### Update Account
 
 The `account update` command is used to update an existing account with customizable options.
@@ -143,7 +147,7 @@ The `account update` command is used to update an existing account with customiz
 ```sh
 noobaa-cli account update --name <account_name> [--new_name][--uid][--gid][--user]
 [--new_buckets_path][--access_key][--secret_key][--regenerate][--fs_backend]
-[--allow_bucket_creation][--force_md5_etag][--anonymous][--iam_operate_on_root_account]
+[--allow_bucket_creation][--force_md5_etag][--anonymous][--iam_operate_on_root_account][--default_connection]
 ```
 #### Flags -
 - `name` (Required)
@@ -207,6 +211,10 @@ noobaa-cli account update --name <account_name> [--new_name][--uid][--gid][--use
     - Type: Boolean
     - Description: Specifies if the account allowed to create root accounts using the IAM API (the default behavior is to create of IAM accounts). See - [IAM - Root Accounts Manager](./../design/iam.md#root-accounts-manager).
 
+- `default_connection`
+    - Type: String
+    - Description: A default account for Kafka external servers. See bucket-notifications.md.
+
 ### Account Status
 
 The `account status` command is used to print the status of the account.

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

@@ -10,13 +10,37 @@ const notif_util = require('../../../util/notifications_util');
 async function put_bucket_notification(req) {
 
     let topic_configuration = req.body.NotificationConfiguration?.TopicConfiguration;
+    const default_connection = req.object_sdk.requesting_account.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;
+            //handle Kafka's topic synax, if present
+            if (conf.Topic && conf.Topic.length > 0 && conf.Topic[0].startsWith('kafka:::topic/')) {
+                //kafka_topic_parts[0] = 'kafka:::topic'
+                //kafka_topic_parts[1] = connection, optional
+                //kafka_topic_parts[2] = Kafka topic, mandatory
+                const kafka_topic_parts = conf.Topic[0].split('/');
+                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]
+                    });
+                }
+                //connection is optionally kafka_topic_parts[1], default to account's default_connection
+                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];
+                //write the full Topic string with the connection
+                conf.topic = ['kafka:::topic/' + connection + "/" + 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) {