in_mqtt: add test code and support payload_key

plugins/in_mqtt/mqtt.c

@@ -139,7 +139,11 @@ static int in_mqtt_exit(void *data, struct flb_config *config)
 
 /* Configuration properties map */	
 static struct flb_config_map config_map[] = {	
-
+    {
+     FLB_CONFIG_MAP_STR, "payload_key", NULL,
+     0, FLB_TRUE, offsetof(struct flb_in_mqtt_config, payload_key),
+     "Key where the payload will be preserved"
+    },
     /* EOF */	
     {0}	
 };

plugins/in_mqtt/mqtt.h

@@ -20,6 +20,7 @@
 #ifndef FLB_IN_MQTT_H
 #define FLB_IN_MQTT_H
 
+#include <fluent-bit/flb_sds.h>
 #include <fluent-bit/flb_log_event_encoder.h>
 
 #define MQTT_MSGP_BUF_SIZE 8192
@@ -28,6 +29,8 @@ struct flb_in_mqtt_config {
     char *listen;                      /* Listen interface            */
     char *tcp_port;                    /* TCP Port                    */
 
+    flb_sds_t payload_key;             /* payload key */
+
     int msgp_len;                      /* msgpack data length         */
     char msgp[MQTT_MSGP_BUF_SIZE];     /* msgpack static buffer       */
     struct flb_input_instance *ins;    /* plugin input instance       */

plugins/in_mqtt/mqtt_config.c

@@ -30,13 +30,21 @@ struct flb_in_mqtt_config *mqtt_config_init(struct flb_input_instance *ins)
 {
     char tmp[16];
     struct flb_in_mqtt_config *config;
+    int ret;
 
     config = flb_calloc(1, sizeof(struct flb_in_mqtt_config));
     if (!config) {
         flb_errno();
         return NULL;
     }
 
+    ret = flb_input_config_map_set(ins, (void*) config);
+    if (ret == -1) {
+        flb_plg_error(ins, "could not initialize config map");
+        flb_free(config);
+        return NULL;
+    }
+
     config->log_encoder = flb_log_event_encoder_create(
                             FLB_LOG_EVENT_FORMAT_DEFAULT);
 

plugins/in_mqtt/mqtt_prot.c

@@ -171,6 +171,13 @@ static int mqtt_data_append(char *topic, size_t topic_len,
                 FLB_LOG_EVENT_STRING_VALUE(topic, topic_len));
     }
 
+    if (ctx->payload_key) {
+        flb_log_event_encoder_append_body_string_length(ctx->log_encoder, flb_sds_len(ctx->payload_key));
+        flb_log_event_encoder_append_body_string_body(ctx->log_encoder, ctx->payload_key,
+                                                      flb_sds_len(ctx->payload_key));
+        flb_log_event_encoder_body_begin_map(ctx->log_encoder);
+    }
+
     /* Re-pack original KVs */
     for (i = 0;
          i < root.via.map.size &&
@@ -182,6 +189,10 @@ static int mqtt_data_append(char *topic, size_t topic_len,
                 FLB_LOG_EVENT_MSGPACK_OBJECT_VALUE(&root.via.map.ptr[i].val));
     }
 
+    if (ctx->payload_key) {
+        flb_log_event_encoder_body_commit_map(ctx->log_encoder);
+    }
+
     if (ret == FLB_EVENT_ENCODER_SUCCESS) {
         ret = flb_log_event_encoder_commit_record(ctx->log_encoder);
     }

tests/runtime/CMakeLists.txt

@@ -45,6 +45,7 @@ if(FLB_OUT_LIB)
   FLB_RT_TEST(FLB_IN_DUMMY         "in_dummy.c")
   FLB_RT_TEST(FLB_IN_HTTP          "in_http.c")
   FLB_RT_TEST(FLB_IN_ELASTICSEARCH "in_elasticsearch.c")
+  FLB_RT_TEST(FLB_IN_MQTT          "in_mqtt.c")
   FLB_RT_TEST(FLB_IN_OPENTELEMETRY "in_opentelemetry.c")
   FLB_RT_TEST(FLB_IN_RANDOM        "in_random.c")
   FLB_RT_TEST(FLB_IN_STATSD        "in_statsd.c")