metrics: extend support for Tag metrics

include/fluent-bit/flb_config.h

@@ -32,6 +32,7 @@
 #include <monkey/mk_core.h>
 
 struct flb_router;
+struct flb_hash_table;
 
 #define FLB_CONFIG_FLUSH_SECS   1
 #define FLB_CONFIG_HTTP_LISTEN  "0.0.0.0"
@@ -199,6 +200,17 @@ struct flb_config {
      */
     struct mk_list cmetrics;
 
+    /*
+     * Optional telemetry metrics with user-controlled cardinality.
+     */
+    int telemetry_metrics_logs_tag_records;
+    int telemetry_metrics_logs_tag_records_max_series;
+    int telemetry_metrics_logs_tag_records_max_tag_length;
+    size_t telemetry_metrics_logs_tag_records_series_count;
+    struct flb_hash_table *telemetry_metrics_logs_tag_records_ht;
+    pthread_mutex_t telemetry_metrics_logs_tag_records_lock;
+    int telemetry_metrics_logs_tag_records_lock_inited;
+
     /* HTTP Server */
 #ifdef FLB_HAVE_HTTP_SERVER
     int http_server;                /* HTTP Server running    */

include/fluent-bit/flb_input.h

@@ -263,6 +263,7 @@ struct flb_input_instance {
     char *tag;                           /* Input tag for routing        */
     int tag_len;
     int tag_default;                     /* is it using the default tag? */
+    int telemetry_metrics_logs_tag_records;   /* override service tag records */
 
     /* By default all input instances are 'routable' */
     int routable;
@@ -403,6 +404,8 @@ struct flb_input_instance {
     struct cmt *cmt;                     /* parent context              */
     struct cmt_counter *cmt_bytes;       /* metric: input_bytes_total   */
     struct cmt_counter *cmt_records;     /* metric: input_records_total */
+    struct cmt_counter *cmt_logs_tag_records;
+    struct cmt_counter *cmt_logs_tag_records_untracked;
 
     /* is the input instance overlimit ?: 1 or 0 */
     struct cmt_gauge   *cmt_storage_overlimit;

src/config_format/flb_cf_yaml.c

@@ -1966,6 +1966,19 @@ static int consume_event(struct flb_cf *conf, struct local_ctx *ctx,
                 return YAML_FAILURE;
             }
             break;
+        case YAML_MAPPING_START_EVENT:
+            if (state->section == SECTION_ENV) {
+                flb_error("nested maps are not allowed in env section");
+                yaml_error_event(ctx, state, event);
+                return YAML_FAILURE;
+            }
+
+            state = state_push_variant(ctx, state, 1);
+            if (state == NULL) {
+                flb_error("unable to allocate state");
+                return YAML_FAILURE;
+            }
+            break;
         default:
             yaml_error_event(ctx, state, event);
             return YAML_FAILURE;
@@ -2238,6 +2251,21 @@ static int consume_event(struct flb_cf *conf, struct local_ctx *ctx,
                 break;
             }
 
+            if (state->section == SECTION_INPUT &&
+                strcmp(state->key, "telemetry") == 0) {
+                /*
+                 * Input telemetry is consumed structurally at load time. Other
+                 * nested input maps keep the legacy group behavior.
+                 */
+                state = state_push_variant(ctx, state, 1);
+
+                if (state == NULL) {
+                    flb_error("unable to allocate state");
+                    return YAML_FAILURE;
+                }
+                break;
+            }
+
             state = state_push(ctx, STATE_GROUP_KEY);
 
             if (state == NULL) {
@@ -2404,6 +2432,21 @@ static int consume_event(struct flb_cf *conf, struct local_ctx *ctx,
                 break;
             }
 
+            if (state->state == STATE_SECTION_VAL) {
+                if (flb_cf_section_property_add_variant(conf,
+                                                        state->cf_section->properties,
+                                                        state->key,
+                                                        flb_sds_len(state->key),
+                                                        variant) == NULL) {
+                    flb_error("unable to insert section variant");
+                    return YAML_FAILURE;
+                }
+
+                state = state_pop(ctx);
+
+                break;
+            }
+
             if (state->variant->type == CFL_VARIANT_KVLIST && state->variant_kvlist_key == NULL) {
                 flb_error("invalid state, should have a variant key");
                 cfl_variant_destroy(variant);