bucket notifications - facilitate notif conf to override connection

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

@@ -484,7 +484,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,38 @@ 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 is, by index:
+                //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/test/unit_tests/jest_tests/test_nc_account_cli.test.js

@@ -643,6 +643,20 @@ describe('manage nsfs cli account flow', () => {
             expect(JSON.parse(res_list).response.reply.map(item => item.name))
                 .toEqual(expect.arrayContaining([name]));
         });
+
+        it('cli account add - default connection', async function() {
+            const action = ACTIONS.ADD;
+            const { type, name, new_buckets_path, uid, gid } = defaults;
+            const default_connection = 'defcon';
+            const account_options = { config_root, name, new_buckets_path, uid, gid, default_connection };
+            await fs_utils.create_fresh_path(new_buckets_path);
+            await fs_utils.file_must_exist(new_buckets_path);
+            await set_path_permissions_and_owner(new_buckets_path, account_options, 0o700);
+            await exec_manage_cli(type, action, account_options);
+            const account = await config_fs.get_account_by_name(name, config_fs_account_options);
+            assert_account(account, account_options, false);
+            expect(account.default_connection).toBe(default_connection);
+        });
     });
 
     describe('cli update account', () => {

src/test/unit_tests/test_notifications.js

@@ -36,7 +36,7 @@ const http_connect_path = path.join(tmp_connect, http_connect_filename);
 //content of connect file, will be written to a file in before()
 const http_connect = {
     agent_request_object: {"host": "localhost", "port": 9998, "timeout": 1500},
-    request_options_object: {"auth": "amit:passw", "timeout": 1500},
+    request_options_object: {"auth": "amit:passw", "timeout": 1500, "path": "/default"},
     notification_protocol: 'http',
     name: 'http_notif'
 };
@@ -240,7 +240,7 @@ mocha.describe('notifications', function() {
 });
 
 const step_wait = 100;
-async function notify_await_result({bucket_name, event_name, etag, key, timeout = undefined}) {
+async function notify_await_result({bucket_name, event_name, etag, key, url = "/default", timeout = undefined}) {
 
     //remember expected result here so server could compare it to actual result later
     expected_bucket = bucket_name;

src/util/notifications_util.js

@@ -186,8 +186,16 @@ class Notificator {
         }
     }
 
-    async parse_connect_file(connect_filename, decrypt = false) {
+    async parse_connect_file(connection_name, decrypt = false) {
         let connect;
+        let connect_filename = connection_name;
+        let kafka_topic_from_connection_name;
+        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 > 1 && connection_parts[2];
+        }
+
         if (this.nc_config_fs) {
             connect = await this.nc_config_fs.get_connection_by_name(connect_filename);
         } else {
@@ -203,6 +211,11 @@ class Notificator {
                 connect.request_options_object.auth, connect.master_key_id);
         }
         load_files(connect);
+
+        //use the kafka topic, if it was present in connection_name
+        if (kafka_topic_from_connection_name) {
+            connect.topic = kafka_topic_from_connection_name;
+        }
         return connect;
     }
 }
@@ -362,6 +375,7 @@ async function test_notifications(notifs, nc_config_dir, req) {
         let notif_failure;
         try {
             connect = await notificator.parse_connect_file(notif.topic[0]);
+            dbg.log0(`effective connect for notif ${notif.id[0]} is`, connect);
             connection = get_connection(connect);
             await connection.connect();
             await connection.promise_notify(compose_notification_test(req), async (notif_cb, err_cb, err) => {