Merge pull request #8801 from alphaprinz/notif_backports

[Backport int 5.18] bucket notification - backports

Author: nimrod-becker

config.js

@@ -716,7 +716,7 @@ config.NOTIFICATION_LOG_NS = 'notification_logging';
 config.NOTIFICATION_LOG_DIR = process.env.NOTIFICATION_LOG_DIR;
 config.NOTIFICATION_BATCH = process.env.BATCH || 10;
 config.NOTIFICATION_REQ_PER_SPACE_CHECK = process.env.NOTIFICATION_REQ_PER_SPACE_CHECK || 0;
-config.NOTIFICATION_SPACE_CHECK_THRESHOLD = parseInt(process.env.NOTIFICATION_SPACE_CHECK_THRESHOLD, 10) || 0.1;
+config.NOTIFICATION_SPACE_CHECK_THRESHOLD = parseFloat(process.env.NOTIFICATION_SPACE_CHECK_THRESHOLD) || 0.2;
 
 ///////////////////////////
 //      KEY ROTATOR      //

docs/NooBaaNonContainerized/ConfigFileCustomizations.md

@@ -449,6 +449,18 @@ Warning: After setting this configuration, NooBaa will skip schema validations a
     "LOG_TO_STDERR_ENABLED": false
     3. systemctl restart noobaa
     ```
+### 30. Notification log directory
+* <u>Key</u> `NOTIFICATION_LOG_DIR`
+* <u>Type</u> String
+* <u>Default</u> empty
+* <u>Description</u> Path to directory that will hold pending notifications to be sent,
+* <u>Steps</u>
+    ```
+    1. Open /path/to/config_dir/config.json file.
+    2. Set the config key -
+    Example:
+    "NOTIFICATION_LOG_DIR": "/etc/notif"
+    3. systemctl restart noobaa
 
 ### 31. Prometheus HTTP enable flag -
 * <u>Key</u>: `ALLOW_HTTP_METRICS`  

docs/NooBaaNonContainerized/Health.md

@@ -42,6 +42,8 @@ For more details about NooBaa RPM installation, see - [Getting Started](./Gettin
     - checks if there is ongoing upgrade
     - returns error if there is no ongoing upgrade, but the config directory phase is locked
     - returns message if there is no ongoing upgrade and the config directory is unlocked
+  - `Bucket event notifications connections health`
+    - Sends out a test notification for each connection.
 
 * Health CLI requires root permissions.
 
@@ -53,7 +55,7 @@ The `health` command is used to analyze NooBaa health with customizable options.
 
 ```sh
 noobaa-cli diagnose health [--deployment_type][--https_port]
-[--all_account_details][--all_bucket_details][--config_root][--debug]
+[--all_account_details][--all_bucket_details][--all_connection_details][--notif_storage_threshold][--config_root][--debug]
 ```
 ### Flags -
 
@@ -75,7 +77,17 @@ noobaa-cli diagnose health [--deployment_type][--https_port]
 - `all_bucket_details`
     - Type: Boolean
     - Default: false
-    - Description: Indicates if health output should contain valid buckets.  
+    - Description: Indicates if health output should contain valid buckets.
+
+- `all_connection_details`
+    - Type: Boolean
+    - Default: false
+    - Description: Indicates if health output should contain connection test result.
+
+- `notif_storage_threshold`
+    - Type: Boolean
+    - Default: false
+    - Description: Whether health ouput should check if notification storage FS is below threshold.
 
 - `config_root`
     - Type: String
@@ -143,6 +155,10 @@ The output of the Health CLI is a JSON object containing the following propertie
     2. The account doesn't have RW access to its `new_buckets_path` or having invalid config JSON file.
     3. The account has an invalid config JSON file.
 
+- `invalid connections`:
+  - Type: [{ "name": "connection_name", "config_path": "/connection/file/path", "code": String }]
+  - Description: Array of connections that failed test notification.
+
 - `valid_accounts`
   - Type: [{ "name": account_name, "storage_path": "/path/to/accounts/new_buckets_path" }]
   - Description: Array of all the valid accounts. If the all_account_details flag is set to true, valid_accounts will be included in the Health response.
@@ -151,6 +167,10 @@ The output of the Health CLI is a JSON object containing the following propertie
   - Type: [{ "name": bucket_name, "storage_path": "/path/to/bucket/path" }]
   - Description: Array of all the valid buckets. If the all_bucket_details flag is set to true, valid_buckets will be included in the Health response.
 
+- `valid_connections`:
+  - Type: [{ "name": "connection_name", "config_path": "/connection/file/path" }]
+  - Description: Array of all connections to which test notification was send successfully.
+
 - `error_type`
   - Type: String
   - Enum: 'PERSISTENT' | 'TEMPORARY'
@@ -239,6 +259,26 @@ Output:
       ],
       "error_type": "PERSISTENT"
     },
+    "connectoins_status": {
+      "invalid_connections": [
+        {
+          "name": "notif_invalid",
+          "config_path": "/etc/noobaa.conf.d/connections/notif_invalid.json",
+          "code": "ECONNREFUSED"
+        }
+      ],
+      "valid_connections": [
+        {
+          "name": "notif_valid",
+          "config_path": "/etc/noobaa.conf.d/connections/notif_valid.json"
+        }
+      ]
+    },
+    "notif_storage_threshold_details": {
+      "threshold": 0.2,
+      "ratio": 0.9,
+      "result": "above threshold"
+    }
     "config_directory": {
       "phase": "CONFIG_DIR_UNLOCKED",
       "config_dir_version": "1.0.0",

docs/NooBaaNonContainerized/NooBaaCLI.md

@@ -569,6 +569,16 @@ noobaa-cli connection add --from_file
    - Type: Object
    - Description: An object given as options to node http(s) request. If "auth" field is specified, it's value is encrypted.
 
+- `kafka_options_object`
+   - Type: Object
+   - Description: An object given as options to kafka client.
+     Options are listed in https://github.com/edenhill/librdkafka/blob/v2.8.0/CONFIGURATION.md.
+     Specifically, 'metadata.broker.list' is used to specify the external kafka server.
+
+- `topic`
+   - Type: String
+   - Description - Topic for kafka messages.
+
 - `from_file`
     - Type: String
     - Description: Path to a JSON file which includes connection properties. When using `from_file` flag the connection details must only appear inside the options JSON file. See example below.

docs/bucket-notifications.md

@@ -0,0 +1,106 @@
+# Bucket Notifications
+
+## Bucket Notification Configuration
+
+Bucket's notifications can be configured with the s3api operation [put-bucket-notification-configuration](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/s3api/put-bucket-notification-configuration.html).
+Specify notifications under the "TopicConfigurations" field, which is an array of jsons, one for each notification.
+A notification json has these fields:
+- Id: Mandatory. A unique string identifying the notification configuration.
+- Events: Optional. An array of [events](https://docs.aws.amazon.com/AmazonS3/latest/userguide/notification-how-to-event-types-and-destinations.html) for which the notification is relevant.
+          If not specified, the notification is relevant for all events.
+- TopicArn: The connection file (see below). (To specify a Kafka target topic, see "Kafka Connection Fields" below).
+
+Example for a bucket's notification configuration, in a file:
+{
+    "TopicConfigurations": [
+        {
+	    "Id": "created_from_s3op",
+            "TopicArn": "secret-name/connect.json",
+            "Events": [
+                "s3:ObjectCreated:*"
+            ]
+        }
+    ]
+}
+
+
+## Connection File
+A connection file contains some fields that specify the target notification server.
+The connection file name is specified in TopicArn field of the [notification configuration](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/s3api/put-bucket-notification-configuration.html)
+
+-In a containerized environment, the operator will mount the secrets as file in the core pod (see operator doc for more info).
+The mount path is `/etc/notif_connect/<secret-name>/<secret failename>`, and the notification reference the file by the `<secret-name>/<secret filename>` path in TopicArn field.
+For example, if seceret was created with:
+`kubectl create secret generic notif-secret --from_file connect.json`
+Then TopicArn should be `notif-secret/connect.json`.
+
+-In a non-containerized system, you must create the relevant file using the 'connections' CRUD cli.
+See the NC cli doc for more info.
+Connect file contains a single json with the fields specified below.
+
+### Common Connection Fields
+- name: A string identifying the connection (mandatory).
+- notification_protocol: One of: http, https, kafka (mandatory).
+
+### Http(s) Connection Fields
+- agent_request_object: Value is a JSON that is passed to to nodejs' http(s) agent (mandatory).
+Any field supported by nodejs' http(s) agent can be used, specifically 'host' and 'port'.
+A full list of options is [here](https://nodejs.org/docs/latest-v22.x/api/http.html#new-agentoptions).
+Notable fields include:
+    - host: hostname (or ip) of external server to receive the notifications
+    - port: http(s) port on which the external server listens
+    - timeout: connection timeout in milliseconds.
+    - local_file_ca - path to CA pem file that signed server's TLS cert (if CA needs to be customized)
+    - rejectUnauthorized - set to true to accept self-signed certs (useful for testing, do not use in production).
+    - local_file_cert: path to client's private TLS certificate PEM file.
+    - local_file_key: path to client's private TLS key PEM file.
+
+- request_options_object: Value is a JSON that is passed to nodejs' http(s) request (optional).
+Any field supported by nodejs' http(s) request option can be used, specifically:
+-- path: used to specify the url path
+-- auth: used for http simple auth. Value for 'auth' is of the syntax: <name>:<passowrd>.
+
+A full list of options is [here](https://nodejs.org/docs/latest-v22.x/api/http.html#httprequesturl-options-callback).
+
+### Kafka Connection Fields
+- metadata.broker.list: A CSV list of Kafka brokers (mandatory).
+- topic: A topic for the Kafka message (mandatory).
+
+## Event Types
+S3 spec lists several events and "sub events".
+
+The list of supported events are:
+
+- s3:ObjectCreated:*
+- s3:ObjectCreated:Put
+- s3:ObjectCreated:Post
+- s3:ObjectCreated:Copy
+- s3:ObjectCreated:CompleteMultipartUpload
+- s3:ObjectRemoved:*
+- s3:ObjectRemoved:Delete
+- s3:ObjectRemoved:DeleteMarkerCreated
+- s3:ObjectRestore:*
+- s3:ObjectRestore:Post
+- s3:ObjectRestore:Completed
+- s3:ObjectRestore:Delete
+- s3:ObjectTagging:*
+- s3:ObjectTagging:Put
+- s3:ObjectTagging:Delete
+- s3:LifecycleExpiration:*
+- s3:LifecycleExpiration:Delete
+- s3:LifecycleExpiration:DeleteMarkerCreated
+
+## Event Processing and Failure Handling
+Once NooBaa finds an event with a relevant notification configuration, the notification
+is written to a persistent file.
+Location of persistent files is determined by-
+- For containerized, the pvc specified in NooBaa Bucket Notification spec (see Operator docs for more info).
+- For NC, the env variable NOTIFICATION_LOG_DIR (see NC docs for more info).
+
+Files are processed by-
+- For containerized, files are contantly being processed in the background of the core pod.
+- For NC, files are processed by running manage_nsfs with 'notification' action.
+
+If a notification fails to be sent to the external server, it will be re-written to a persistent file (assuming the
+notification is still configured).
+Once this new file is processed, NooBaa will try to re-send the failed notification.

src/cmd/manage_nsfs.js

@@ -758,40 +758,48 @@ async function notification_management() {
 async function connection_management(action, user_input) {
     manage_nsfs_validations.validate_connection_args(user_input, action);
 
+    //don't reply the internal '_' field.
+    delete user_input._;
+
     let response = {};
     let data;
 
-    switch (action) {
-        case ACTIONS.ADD:
-            data = await notifications_util.add_connect_file(user_input, config_fs);
-            response = { code: ManageCLIResponse.ConnectionCreated, detail: data };
-            break;
-        case ACTIONS.DELETE:
-            await config_fs.delete_connection_config_file(user_input.name);
-            response = { code: ManageCLIResponse.ConnectionDeleted, detail: {name: user_input.name} };
-            break;
-        case ACTIONS.UPDATE:
-            await notifications_util.update_connect_file(user_input.name, user_input.key,
-                user_input.value, user_input.remove_key, config_fs);
-            response = { code: ManageCLIResponse.ConnectionUpdated, detail: {name: user_input.name} };
-            break;
-        case ACTIONS.STATUS:
-            data = await new notifications_util.Notificator({
-                fs_context: config_fs.fs_context,
-                connect_files_dir: config_fs.connections_dir_path,
-                nc_config_fs: config_fs,
-            }).parse_connect_file(user_input.name, user_input.decrypt);
-            response = { code: ManageCLIResponse.ConnectionStatus, detail: data };
-            break;
-        case ACTIONS.LIST:
-            data = await list_connections();
-            response = { code: ManageCLIResponse.ConnectionList, detail: data };
-            break;
-        default:
-            throw_cli_error(ManageCLIError.InvalidAction);
+    try {
+        switch (action) {
+            case ACTIONS.ADD:
+                data = await notifications_util.add_connect_file(user_input, config_fs);
+                response = { code: ManageCLIResponse.ConnectionCreated, detail: data };
+                break;
+            case ACTIONS.DELETE:
+                await config_fs.delete_connection_config_file(user_input.name);
+                response = { code: ManageCLIResponse.ConnectionDeleted, detail: {name: user_input.name} };
+                break;
+            case ACTIONS.UPDATE:
+                await notifications_util.update_connect_file(user_input.name, user_input.key,
+                    user_input.value, user_input.remove_key, config_fs);
+                response = { code: ManageCLIResponse.ConnectionUpdated, detail: {name: user_input.name} };
+                break;
+            case ACTIONS.STATUS:
+                data = await new notifications_util.Notificator({
+                    fs_context: config_fs.fs_context,
+                    connect_files_dir: config_fs.connections_dir_path,
+                    nc_config_fs: config_fs,
+                }).parse_connect_file(user_input.name, user_input.decrypt);
+                response = { code: ManageCLIResponse.ConnectionStatus, detail: data };
+                break;
+            case ACTIONS.LIST:
+                data = await list_connections();
+                response = { code: ManageCLIResponse.ConnectionList, detail: data };
+                break;
+            default:
+                throw_cli_error(ManageCLIError.InvalidAction);
+        }
+        write_stdout_response(response.code, response.detail, response.event_arg);
+    } catch (err) {
+        if (err.code === 'EEXIST') throw_cli_error(ManageCLIError.ConnectionAlreadyExists, user_input.name);
+        if (err.code === 'ENOENT') throw_cli_error(ManageCLIError.NoSuchConnection, user_input.name);
+        throw err;
     }
-
-    write_stdout_response(response.code, response.detail, response.event_arg);
 }
 
 ////////////////////

src/endpoint/s3/ops/s3_put_object.js

@@ -65,6 +65,8 @@ async function put_object(req, res) {
     }
     s3_utils.set_encryption_response_headers(req, res, reply.encryption);
 
+    res.size_for_notif = size || reply.size;
+
     if (copy_source) {
         // TODO: This needs to be checked regarding copy between diff namespaces
         // In that case we do not have the copy_source property and just read and upload the stream

src/endpoint/s3/s3_bucket_logging.js

@@ -6,7 +6,7 @@ const http_utils = require('../../util/http_utils');
 const dgram = require('node:dgram');
 const { Buffer } = require('node:buffer');
 const config = require('../../../config');
-const {compose_notification_req, check_notif_relevant, check_free_space} = require('../../util/notifications_util');
+const {compose_notification_req, check_notif_relevant, check_free_space_if_needed} = require('../../util/notifications_util');
 
 async function send_bucket_op_logs(req, res) {
     if (req.params && req.params.bucket &&
@@ -44,7 +44,7 @@ async function send_bucket_op_logs(req, res) {
                         buffer: JSON.stringify(notif)
                     });
 
-                    check_free_space(req);
+                    check_free_space_if_needed(req);
                 }
             }
         }