Metrics metadata (#422)

Author: xiazhvera

include/aws/mqtt/client.h

@@ -697,8 +697,6 @@ int aws_mqtt_client_connection_get_stats(
  * Sets IoT SDK metrics configuration for the connection.
  * These metrics will be appended to the username field during connection.
  *
- * NOTE: DO NOT USE METADATA. Metadata will not be set.
- *
  * \param connection The connection object
  * \param metrics The IoT SDK metrics configuration (pass NULL to disable metrics)
  * \returns AWS_OP_SUCCESS if successful, AWS_OP_ERR otherwise

include/aws/mqtt/mqtt.h

@@ -127,6 +127,12 @@ struct aws_mqtt_iot_metrics {
      * Library name string (SDK attribute)
      */
     struct aws_byte_cursor library_name;
+
+    /**
+     * Metadata entries, key value pair to set in metrics "Metadata" field
+     */
+    size_t metadata_count;
+    const struct aws_mqtt_metadata_entry *metadata_entries;
 };
 
 AWS_EXTERN_C_BEGIN

include/aws/mqtt/private/mqtt_iot_metrics.h

@@ -16,6 +16,8 @@ struct aws_mqtt_iot_metrics_storage {
 
     struct aws_byte_cursor library_name;
 
+    struct aws_array_list metadata_entries;
+
     struct aws_byte_buf storage;
 };
 
@@ -54,4 +56,17 @@ int aws_mqtt_append_sdk_metrics_to_username(
 AWS_MQTT_API
 int aws_mqtt_validate_iot_metrics(const struct aws_mqtt_iot_metrics *metrics);
 
+/**
+ * Checks if username should be included in the CONNECT packet.
+ * Returns true if either username is provided or metrics are configured.
+ *
+ * @param username The username cursor (can be NULL)
+ * @param metrics_storage The metrics storage (can be NULL)
+ * @return true if username should be included, false otherwise
+ */
+AWS_MQTT_API
+bool aws_mqtt_has_non_empty_username(
+    const struct aws_byte_cursor *username,
+    const struct aws_mqtt_iot_metrics_storage *metrics_storage);
+
 #endif /* AWS_MQTT_IOT_METRICS_H */

source/client.c

@@ -635,12 +635,13 @@ static void s_mqtt_client_init(
             &connect, topic_cur, connection->will.qos, connection->will.retain, payload_cur);
     }
 
-    if (connection->username || connection->metrics_storage) {
-        struct aws_byte_cursor username_cur;
-        AWS_ZERO_STRUCT(username_cur);
-        if (connection->username) {
-            username_cur = aws_byte_cursor_from_string(connection->username);
-        }
+    struct aws_byte_cursor username_cur;
+    AWS_ZERO_STRUCT(username_cur);
+    if (connection->username) {
+        username_cur = aws_byte_cursor_from_string(connection->username);
+    }
+
+    if (aws_mqtt_has_non_empty_username(&username_cur, connection->metrics_storage)) {
 
         /* Apply metrics to username if configured */
         if (connection->metrics_storage) {

source/mqtt_iot_metrics.c

@@ -94,7 +94,6 @@ int s_build_username_query(
     return AWS_OP_SUCCESS;
 }
 
-// TODO Future Work: we ignored the metadata field for now, will add them in future support
 int aws_mqtt_append_sdk_metrics_to_username(
     struct aws_allocator *allocator,
     const struct aws_byte_cursor *original_username,
@@ -133,10 +132,15 @@ int aws_mqtt_append_sdk_metrics_to_username(
     struct aws_byte_cursor question_mark_str = aws_byte_cursor_from_c_str("?");
     struct aws_byte_cursor sdk_str = aws_byte_cursor_from_c_str("SDK");
     struct aws_byte_cursor platform_str = aws_byte_cursor_from_c_str("Platform");
+    struct aws_byte_cursor metadata_str = aws_byte_cursor_from_c_str("Metadata");
 
     struct aws_array_list params_list;
     aws_array_list_init_dynamic(&params_list, allocator, DEFAULT_QUERY_PARAM_COUNT, sizeof(struct aws_uri_param));
 
+    /* Buffer to hold the metadata value string: (key1=value1;key2=value2;...) */
+    struct aws_byte_buf metadata_value_buf;
+    AWS_ZERO_STRUCT(metadata_value_buf);
+
     // Looking for any existing query in the original username
     if (local_original_username.len > 0) {
         struct aws_byte_cursor question_mark_find = local_original_username;
@@ -193,6 +197,66 @@ int aws_mqtt_append_sdk_metrics_to_username(
         aws_array_list_push_back(&params_list, &platform_params);
     }
 
+    /* Build metadata value string and add to params_list if metadata entries exist */
+    if (metrics->metadata_entries != NULL && metrics->metadata_count > 0) {
+        struct aws_byte_cursor key_value_delim = aws_byte_cursor_from_c_str("=");
+        struct aws_byte_cursor entry_delim = aws_byte_cursor_from_c_str(";");
+        struct aws_byte_cursor open_paren = aws_byte_cursor_from_c_str("(");
+        struct aws_byte_cursor close_paren = aws_byte_cursor_from_c_str(")");
+
+        /* Calculate size needed for metadata value: (key1=value1;key2=value2;...) */
+        size_t metadata_value_size = open_paren.len + close_paren.len;
+        for (size_t i = 0; i < metrics->metadata_count; ++i) {
+            const struct aws_mqtt_metadata_entry *entry = &metrics->metadata_entries[i];
+            metadata_value_size += entry->key.len + key_value_delim.len + entry->value.len;
+            if (i < metrics->metadata_count - 1) {
+                metadata_value_size += entry_delim.len; /* semicolon separator between entries */
+            }
+        }
+
+        /* Initialize buffer for metadata value */
+        if (aws_byte_buf_init(&metadata_value_buf, allocator, metadata_value_size)) {
+            goto cleanup;
+        }
+
+        /* Build metadata value: (key1=value1;key2=value2;...) */
+        if (aws_byte_buf_append(&metadata_value_buf, &open_paren)) {
+            goto cleanup;
+        }
+
+        for (size_t i = 0; i < metrics->metadata_count; ++i) {
+            const struct aws_mqtt_metadata_entry *entry = &metrics->metadata_entries[i];
+
+            if (aws_byte_buf_append(&metadata_value_buf, &entry->key)) {
+                goto cleanup;
+            }
+            if (aws_byte_buf_append(&metadata_value_buf, &key_value_delim)) {
+                goto cleanup;
+            }
+            if (aws_byte_buf_append(&metadata_value_buf, &entry->value)) {
+                goto cleanup;
+            }
+
+            /* Add semicolon separator between entries (not after the last one) */
+            if (i < metrics->metadata_count - 1) {
+                if (aws_byte_buf_append(&metadata_value_buf, &entry_delim)) {
+                    goto cleanup;
+                }
+            }
+        }
+
+        if (aws_byte_buf_append(&metadata_value_buf, &close_paren)) {
+            goto cleanup;
+        }
+
+        /* Add Metadata parameter to params_list */
+        struct aws_uri_param metadata_params = {
+            .key = metadata_str,
+            .value = aws_byte_cursor_from_buf(&metadata_value_buf),
+        };
+        aws_array_list_push_back(&params_list, &metadata_params);
+    }
+
     // Rebuild metrics string from params_list
     // First parse to get final username size
     size_t total_size = 0;
@@ -218,6 +282,7 @@ int aws_mqtt_append_sdk_metrics_to_username(
     result = AWS_OP_SUCCESS;
 
 cleanup:
+    aws_byte_buf_clean_up(&metadata_value_buf);
     aws_array_list_clean_up(&params_list);
 
     if (result == AWS_OP_ERR) {
@@ -238,6 +303,15 @@ size_t aws_mqtt_iot_metrics_compute_storage_size(const struct aws_mqtt_iot_metri
     size_t storage_size = 0;
     storage_size += metrics->library_name.len;
 
+    /* Add storage for metadata entries */
+    if (metrics->metadata_entries != NULL && metrics->metadata_count > 0) {
+        for (size_t i = 0; i < metrics->metadata_count; ++i) {
+            const struct aws_mqtt_metadata_entry *entry = &metrics->metadata_entries[i];
+            storage_size += entry->key.len;
+            storage_size += entry->value.len;
+        }
+    }
+
     return storage_size;
 }
 
@@ -270,13 +344,49 @@ struct aws_mqtt_iot_metrics_storage *aws_mqtt_iot_metrics_storage_new(
         storage_view->library_name = metrics_storage->library_name;
     }
 
+    /* Copy metadata entries */
+    if (metrics_options->metadata_entries != NULL && metrics_options->metadata_count > 0) {
+        if (aws_array_list_init_dynamic(
+                &metrics_storage->metadata_entries,
+                allocator,
+                metrics_options->metadata_count,
+                sizeof(struct aws_mqtt_metadata_entry))) {
+            goto cleanup_storage;
+        }
+
+        for (size_t i = 0; i < metrics_options->metadata_count; ++i) {
+            struct aws_mqtt_metadata_entry entry;
+            entry.key = metrics_options->metadata_entries[i].key;
+            entry.value = metrics_options->metadata_entries[i].value;
+
+            /* Copy key into storage buffer and update cursor */
+            if (entry.key.len > 0) {
+                if (aws_byte_buf_append_and_update(&metrics_storage->storage, &entry.key)) {
+                    goto cleanup_storage;
+                }
+            }
+
+            /* Copy value into storage buffer and update cursor */
+            if (entry.value.len > 0) {
+                if (aws_byte_buf_append_and_update(&metrics_storage->storage, &entry.value)) {
+                    goto cleanup_storage;
+                }
+            }
+
+            aws_array_list_push_back(&metrics_storage->metadata_entries, &entry);
+        }
+
+        /* Set storage_view to point to the metadata entries array */
+        storage_view->metadata_count = aws_array_list_length(&metrics_storage->metadata_entries);
+        storage_view->metadata_entries = metrics_storage->metadata_entries.data;
+    }
+
     return metrics_storage;
 
 cleanup_storage:
-    // TODO Future Work: add metadata entries once we implemented the metadata feature
-    // if (aws_array_list_is_valid(&metrics_storage->metadata_entries)) {
-    //     aws_array_list_clean_up(&metrics_storage->metadata_entries);
-    // }
+    if (aws_array_list_is_valid(&metrics_storage->metadata_entries)) {
+        aws_array_list_clean_up(&metrics_storage->metadata_entries);
+    }
 
     aws_byte_buf_clean_up(&metrics_storage->storage);
     aws_mem_release(allocator, metrics_storage);
@@ -288,6 +398,10 @@ void aws_mqtt_iot_metrics_storage_destroy(struct aws_mqtt_iot_metrics_storage *m
         return;
     }
 
+    if (aws_array_list_is_valid(&metrics_storage->metadata_entries)) {
+        aws_array_list_clean_up(&metrics_storage->metadata_entries);
+    }
+
     aws_byte_buf_clean_up(&metrics_storage->storage);
 
     aws_mem_release(metrics_storage->allocator, metrics_storage);
@@ -302,5 +416,23 @@ int aws_mqtt_validate_iot_metrics(const struct aws_mqtt_iot_metrics *metrics) {
         return AWS_OP_ERR;
     }
 
+    /* Validate metadata entries */
+    if (metrics->metadata_entries != NULL && metrics->metadata_count > 0) {
+        for (size_t i = 0; i < metrics->metadata_count; ++i) {
+            if (aws_mqtt_validate_utf8_text(metrics->metadata_entries[i].key)) {
+                return AWS_OP_ERR;
+            }
+            if (aws_mqtt_validate_utf8_text(metrics->metadata_entries[i].value)) {
+                return AWS_OP_ERR;
+            }
+        }
+    }
+
     return AWS_OP_SUCCESS;
 }
+
+bool aws_mqtt_has_non_empty_username(
+    const struct aws_byte_cursor *username,
+    const struct aws_mqtt_iot_metrics_storage *metrics_storage) {
+    return (username != NULL && username->len > 0) || metrics_storage != NULL;
+}

source/v5/mqtt5_options_storage.c

@@ -645,7 +645,9 @@ static size_t s_aws_mqtt5_packet_connect_compute_storage_size(
     size_t storage_size = 0;
 
     storage_size += view->client_id.len;
-    if (view->username != NULL) {
+
+    if (aws_mqtt_has_non_empty_username(view->username, options ? options->metrics_storage : NULL)) {
+
         if (options) {
             size_t username_size = 0;
             aws_mqtt_append_sdk_metrics_to_username(
@@ -699,17 +701,19 @@ int aws_mqtt5_packet_connect_storage_init(
         return AWS_OP_ERR;
     }
 
-    if (view->username != NULL) {
-        storage->username = *view->username;
+    if (aws_mqtt_has_non_empty_username(
+            view->username, client_options_storage ? client_options_storage->metrics_storage : NULL)) {
+        if (view->username) {
+            storage->username = *view->username;
+        }
         struct aws_byte_buf metrics_username_buf;
         AWS_ZERO_STRUCT(metrics_username_buf);
 
         /* Apply metrics to username if configured */
         if (client_options_storage) {
-            struct aws_byte_cursor username_cur = storage->username;
             if (aws_mqtt_append_sdk_metrics_to_username(
                     allocator,
-                    &username_cur,
+                    &storage->username,
                     client_options_storage->metrics_storage ? &client_options_storage->metrics_storage->storage_view
                                                             : NULL,
                     &metrics_username_buf,

tests/CMakeLists.txt

@@ -115,6 +115,7 @@ add_test_case(mqtt_websocket_failed_transform)
 # Connection state tests
 add_test_case(mqtt_connection_set_metrics_valid)
 add_test_case(mqtt_connection_set_metrics_null)
+add_test_case(mqtt_connection_set_metrics_with_null_username)
 add_test_case(mqtt_connection_set_metrics_invalid_utf8_library)
 add_test_case(mqtt_connection_set_metrics_modify_on_reconnect)
 
@@ -149,6 +150,11 @@ add_test_case(mqtt_append_sdk_metrics_special_chars)
 add_test_case(mqtt_append_sdk_metrics_long_strings)
 add_test_case(mqtt_append_sdk_metrics_invalid_utf8)
 add_test_case(mqtt_append_sdk_metrics_multiple_question_mark)
+add_test_case(mqtt_append_sdk_metrics_with_metadata)
+add_test_case(mqtt_append_sdk_metrics_with_metadata_invalid_utf8)
+add_test_case(mqtt_append_sdk_metrics_with_metadata_invalid_utf8_value)
+add_test_case(mqtt_iot_metrics_storage_with_metadata)
+add_test_case(mqtt_iot_metrics_storage_empty_metadata)
 
 # topic aliasing
 add_test_case(mqtt5_inbound_topic_alias_register_failure)
@@ -414,6 +420,7 @@ add_test_case(mqtt5_client_manual_pub_ack_disconnect_cancellation)
 # Mqtt5 Metrics tests
 add_test_case(mqtt5_client_set_metrics_valid)
 add_test_case(mqtt5_client_set_metrics_null)
+add_test_case(mqtt5_client_set_metrics_with_null_username)
 
 add_test_case(rate_limiter_token_bucket_init_invalid)
 add_test_case(rate_limiter_token_bucket_regeneration_integral)