out_stackdriver: add static labels defined in configuration (#5176)

A feature that mirrors the out_loki exporter functionality
to set to set static labels during configuration. This as
a list of comma separate key=value pairs, e.g.
label_a=value_a,label_b=value_b. The record_accesor wasn't
implemented since the functionality can be recreated using
label_keys.

The expected behavior is that elements set with labels and
label_keys are combined to set the LogEntry Labels in the
LogExplorer.

Signed-off-by: Francisco Valente <[email protected]>

Author: franciscovalentecastro

plugins/out_stackdriver/stackdriver.c

@@ -25,6 +25,7 @@
 #include <fluent-bit/flb_oauth2.h>
 #include <fluent-bit/flb_regex.h>
 #include <fluent-bit/flb_pthread.h>
+#include <fluent-bit/flb_kv.h>
 
 #include <msgpack.h>
 
@@ -911,12 +912,12 @@ static int process_local_resource_id(struct flb_stackdriver *ctx,
 }
 
 /*
- * parse_labels
+ * get_payload_labels
  * - Iterate throught the original payload (obj) and find out the entry that matches
  *   the labels_key
  * - Used to convert all labels under labels_key to root-level `labels` field
  */
-static msgpack_object *parse_labels(struct flb_stackdriver *ctx, msgpack_object *obj)
+static msgpack_object *get_payload_labels(struct flb_stackdriver *ctx, msgpack_object *obj)
 {
     int i;
     int len;
@@ -940,6 +941,50 @@ static msgpack_object *parse_labels(struct flb_stackdriver *ctx, msgpack_object
     return NULL;
 }
 
+static void pack_labels(struct flb_stackdriver *ctx,
+                        msgpack_packer *mp_pck,
+                        msgpack_object *payload_labels_ptr)
+{
+    int i;
+    int ret;
+    int labels_size = 0;
+    char *val;
+    struct mk_list *head;
+    struct flb_kv *list_kv;
+    msgpack_object_kv *obj_kv = NULL;
+
+    /* Determine size of labels map */
+    labels_size = mk_list_size(&ctx->config_labels);
+    if (payload_labels_ptr != NULL &&
+        payload_labels_ptr->type == MSGPACK_OBJECT_MAP) {
+        labels_size += payload_labels_ptr->via.map.size;
+    }
+
+    msgpack_pack_map(mp_pck, labels_size);
+
+    /* pack labels from the payload */
+    if (payload_labels_ptr != NULL &&
+        payload_labels_ptr->type == MSGPACK_OBJECT_MAP) {
+
+        for (i = 0; i < payload_labels_ptr->via.map.size; i++) {
+            obj_kv = &payload_labels_ptr->via.map.ptr[i];
+            msgpack_pack_object(mp_pck, obj_kv->key);
+            msgpack_pack_object(mp_pck, obj_kv->val);
+        }
+    }
+
+    /* pack labels set in configuration */
+    /* in msgpack duplicate keys are overriden by the last set */
+    /* static label keys override payload labels */
+    mk_list_foreach(head, &ctx->config_labels){
+        list_kv = mk_list_entry(head, struct flb_kv, _head);
+        msgpack_pack_str(mp_pck, flb_sds_len(list_kv->key));
+        msgpack_pack_str_body(mp_pck, list_kv->key, flb_sds_len(list_kv->key));
+        msgpack_pack_str(mp_pck, flb_sds_len(list_kv->val));
+        msgpack_pack_str_body(mp_pck, list_kv->val, flb_sds_len(list_kv->val));
+    }
+}
+
 static void cb_results(const char *name, const char *value,
                        size_t vlen, void *data)
 {
@@ -1496,6 +1541,10 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
     /* Count number of records */
     array_size = total_records;
 
+    /* Parameters for labels */
+    msgpack_object *payload_labels_ptr;
+    int labels_size = 0;
+
     /*
      * Search each entry and validate insertId.
      * Reject the entry if insertId is invalid.
@@ -1975,17 +2024,26 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
             entry_size += 1;
         }
 
-        /* Extract labels */
-        labels_ptr = parse_labels(ctx, obj);
-        if (labels_ptr != NULL) {
-            if (labels_ptr->type != MSGPACK_OBJECT_MAP) {
-                flb_plg_error(ctx->ins, "the type of labels should be map");
-                flb_sds_destroy(operation_id);
-                flb_sds_destroy(operation_producer);
-                msgpack_unpacked_destroy(&result);
-                msgpack_sbuffer_destroy(&mp_sbuf);
-                return NULL;
-            }
+        /* Extract payload labels */
+        payload_labels_ptr = get_payload_labels(ctx, obj);
+        if (payload_labels_ptr != NULL &&
+            payload_labels_ptr->type != MSGPACK_OBJECT_MAP) {
+            flb_plg_error(ctx->ins, "the type of payload labels should be map");
+            flb_sds_destroy(operation_id);
+            flb_sds_destroy(operation_producer);
+            msgpack_unpacked_destroy(&result);
+            msgpack_sbuffer_destroy(&mp_sbuf);
+            return -1;
+        }
+
+        /* Number of parsed labels */
+        labels_size = mk_list_size(&ctx->config_labels);
+        if (payload_labels_ptr != NULL &&
+            payload_labels_ptr->type == MSGPACK_OBJECT_MAP) {
+            labels_size += payload_labels_ptr->via.map.size;
+        }
+
+        if (labels_size > 0) {
             entry_size += 1;
         }
 
@@ -2043,10 +2101,10 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
         }
 
         /* labels */
-        if (labels_ptr != NULL) {
+        if (labels_size > 0) {
             msgpack_pack_str(&mp_pck, 6);
             msgpack_pack_str_body(&mp_pck, "labels", 6);
-            msgpack_pack_object(&mp_pck, *labels_ptr);
+            pack_labels(ctx, &mp_pck, payload_labels_ptr);
         }
 
         /* Clean up id and producer if operation extracted */
@@ -2453,6 +2511,11 @@ static struct flb_config_map config_map[] = {
       0, FLB_TRUE, offsetof(struct flb_stackdriver, task_id),
       "Set the resource task id"
     },
+    {
+      FLB_CONFIG_MAP_CLIST, "labels", NULL,
+      0, FLB_TRUE, offsetof(struct flb_stackdriver, labels),
+      "Set the labels"
+    },
     {
       FLB_CONFIG_MAP_STR, "labels_key", DEFAULT_LABELS_KEY,
       0, FLB_TRUE, offsetof(struct flb_stackdriver, labels_key),

plugins/out_stackdriver/stackdriver.h

@@ -131,12 +131,16 @@ struct flb_stackdriver {
     flb_sds_t node_name;
     bool is_k8s_resource_type;
 
-    flb_sds_t labels_key;
     flb_sds_t local_resource_id;
     flb_sds_t tag_prefix;
     /* shadow tag_prefix for safe deallocation */
     flb_sds_t tag_prefix_k8s;
 
+    /* labels */
+    flb_sds_t labels_key;
+    struct mk_list *labels;
+    struct mk_list config_labels; 
+
     /* generic resources */
     flb_sds_t location;
     flb_sds_t namespace_id;

plugins/out_stackdriver/stackdriver_conf.c

@@ -24,6 +24,7 @@
 #include <fluent-bit/flb_utils.h>
 #include <fluent-bit/flb_jsmn.h>
 #include <fluent-bit/flb_sds.h>
+#include <fluent-bit/flb_kv.h>
 
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -167,6 +168,63 @@ static int read_credentials_file(const char *cred_file, struct flb_stackdriver *
     return 0;
 }
 
+/*
+ * parse_configuration_labels
+ * - Parse labels set in configuration
+ * - Returns the number of configuration labels
+ */
+
+static int parse_configuration_labels(struct flb_stackdriver *ctx)
+{
+    int ret;
+    char *p;
+    flb_sds_t key;
+    flb_sds_t val;
+    struct mk_list *head;
+    struct flb_slist_entry *entry;
+    msgpack_object_kv *kv = NULL;
+
+    if (ctx->labels) {
+        mk_list_foreach(head, ctx->labels) {
+            entry = mk_list_entry(head, struct flb_slist_entry, _head);
+
+            p = strchr(entry->str, '=');
+            if (!p) {
+                flb_plg_error(ctx->ins, "invalid key value pair on '%s'",
+                              entry->str);
+                return -1;
+            }
+
+            key = flb_sds_create_size((p - entry->str) + 1);
+            flb_sds_cat(key, entry->str, p - entry->str);
+            val = flb_sds_create(p + 1);
+            if (!key) {
+                flb_plg_error(ctx->ins,
+                              "invalid key value pair on '%s'",
+                              entry->str);
+                return -1;
+            }
+            if (!val || flb_sds_len(val) == 0) {
+                flb_plg_error(ctx->ins,
+                              "invalid key value pair on '%s'",
+                              entry->str);
+                flb_sds_destroy(key);
+                return -1;
+            }
+
+            ret = flb_kv_item_create(&ctx->config_labels, key, val);
+            flb_sds_destroy(key);
+            flb_sds_destroy(val);
+
+            if (ret == -1) {
+                return -1;
+            }
+        }
+    }
+
+    return mk_list_size(&ctx->config_labels);
+}
+
 struct flb_stackdriver *flb_stackdriver_conf_create(struct flb_output_instance *ins,
                                                     struct flb_config *config)
 {
@@ -192,6 +250,16 @@ struct flb_stackdriver *flb_stackdriver_conf_create(struct flb_output_instance *
         return NULL;
     }
 
+    /* labels */
+    flb_kv_init(&ctx->config_labels);
+    ret = parse_configuration_labels((void *)ctx);
+    if (ret == -1) {
+        flb_plg_error(ins, "unable to parse configuration labels");
+        flb_kv_release(&ctx->config_labels);
+        flb_free(ctx);
+        return NULL;
+    }
+
     /* Lookup metadata server URL */
     if (ctx->metadata_server == NULL) {
         tmp = getenv("METADATA_SERVER");
@@ -533,6 +601,7 @@ int flb_stackdriver_conf_destroy(struct flb_stackdriver *ctx)
         flb_sds_destroy(ctx->tag_prefix_k8s);
     }
 
+    flb_kv_release(&ctx->config_labels);
     flb_free(ctx);
 
     return 0;