out_kafka: refactor some configuration parsing into a separate module

Add the flb_kafka module, which will contain common kafka code. For now
only some configuration parsing was extracted from out_kafka.

Signed-off-by: Thiago Padilha <[email protected]>

Author: tchrono

include/fluent-bit/flb_kafka.h

@@ -0,0 +1,42 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_KAFKA_H
+#define FLB_KAFKA_H
+
+#include <monkey/mk_core.h>
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_config.h>
+
+#include <rdkafka.h>
+
+#define FLB_KAFKA_BROKERS             "127.0.0.1"
+#define FLB_KAFKA_TOPIC               "fluent-bit"
+
+struct flb_kafka {
+    rd_kafka_t *rk;
+    char *brokers;
+};
+
+rd_kafka_conf_t *flb_kafka_conf_create(struct flb_kafka *kafka,
+                                       struct mk_list *properties,
+                                       int with_group_id);
+
+#endif

plugins/out_kafka/kafka.c

@@ -73,7 +73,7 @@ static int cb_kafka_init(struct flb_output_instance *ins,
     struct flb_out_kafka *ctx;
 
     /* Configuration */
-    ctx = flb_kafka_conf_create(ins, config);
+    ctx = flb_out_kafka_create(ins, config);
     if (!ctx) {
         flb_plg_error(ins, "failed to initialize");
         return -1;
@@ -420,7 +420,7 @@ int produce_message(struct flb_time *tm, msgpack_object *map,
              * issue a full retry of the data chunk.
              */
             flb_time_sleep(1000);
-            rd_kafka_poll(ctx->producer, 0);
+            rd_kafka_poll(ctx->kafka.rk, 0);
 
             /* Issue a re-try */
             queue_full_retries++;
@@ -433,7 +433,7 @@ int produce_message(struct flb_time *tm, msgpack_object *map,
     }
     ctx->blocked = FLB_FALSE;
 
-    rd_kafka_poll(ctx->producer, 0);
+    rd_kafka_poll(ctx->kafka.rk, 0);
     if (ctx->format == FLB_KAFKA_FMT_JSON) {
         flb_sds_destroy(s);
     }
@@ -499,7 +499,7 @@ static int cb_kafka_exit(void *data, struct flb_config *config)
 {
     struct flb_out_kafka *ctx = data;
 
-    flb_kafka_conf_destroy(ctx);
+    flb_out_kafka_destroy(ctx);
     return 0;
 }
 

plugins/out_kafka/kafka_config.c

@@ -27,17 +27,15 @@
 #include "kafka_topic.h"
 #include "kafka_callbacks.h"
 
-struct flb_out_kafka *flb_kafka_conf_create(struct flb_output_instance *ins,
-                                            struct flb_config *config)
+struct flb_out_kafka *flb_out_kafka_create(struct flb_output_instance *ins,
+                                           struct flb_config *config)
 {
-    int ret;
     const char *tmp;
     char errstr[512];
     struct mk_list *head;
     struct mk_list *topics;
     struct flb_split_entry *entry;
     struct flb_out_kafka *ctx;
-    struct flb_kv *kv;
 
     /* Configuration context */
     ctx = flb_calloc(1, sizeof(struct flb_out_kafka));
@@ -49,55 +47,13 @@ struct flb_out_kafka *flb_kafka_conf_create(struct flb_output_instance *ins,
     ctx->blocked = FLB_FALSE;
 
     /* rdkafka config context */
-    ctx->conf = rd_kafka_conf_new();
+    ctx->conf = flb_kafka_conf_create(&ctx->kafka, &ins->properties, 0);
     if (!ctx->conf) {
         flb_plg_error(ctx->ins, "error creating context");
         flb_free(ctx);
         return NULL;
     }
 
-    /* rdkafka configuration parameters */
-    ret = rd_kafka_conf_set(ctx->conf, "client.id", "fluent-bit",
-                            errstr, sizeof(errstr));
-    if (ret != RD_KAFKA_CONF_OK) {
-        flb_plg_error(ctx->ins, "cannot configure client.id");
-    }
-
-    /* Config: Brokers */
-    tmp = flb_output_get_property("brokers", ins);
-    if (tmp) {
-        ret = rd_kafka_conf_set(ctx->conf,
-                                "bootstrap.servers",
-                                tmp,
-                                errstr, sizeof(errstr));
-        if (ret != RD_KAFKA_CONF_OK) {
-            flb_plg_error(ctx->ins, "config: %s", errstr);
-            flb_free(ctx);
-            return NULL;
-        }
-        ctx->brokers = flb_strdup(tmp);
-    }
-    else {
-        flb_plg_error(ctx->ins, "config: no brokers defined");
-        flb_free(ctx);
-        return NULL;
-    }
-
-    /* Iterate custom rdkafka properties */
-    mk_list_foreach(head, &ins->properties) {
-        kv = mk_list_entry(head, struct flb_kv, _head);
-        if (strncasecmp(kv->key, "rdkafka.", 8) == 0 &&
-            flb_sds_len(kv->key) > 8) {
-
-            ret = rd_kafka_conf_set(ctx->conf, kv->key + 8, kv->val,
-                                    errstr, sizeof(errstr));
-            if (ret != RD_KAFKA_CONF_OK) {
-                flb_plg_error(ctx->ins, "cannot configure '%s' property",
-                              kv->key + 8);
-            }
-        }
-    }
-
     /* Set our global opaque data (plugin context*/
     rd_kafka_conf_set_opaque(ctx->conf, ctx);
 
@@ -234,12 +190,12 @@ struct flb_out_kafka *flb_kafka_conf_create(struct flb_output_instance *ins,
     }
 
     /* Kafka Producer */
-    ctx->producer = rd_kafka_new(RD_KAFKA_PRODUCER, ctx->conf,
+    ctx->kafka.rk = rd_kafka_new(RD_KAFKA_PRODUCER, ctx->conf,
                                  errstr, sizeof(errstr));
-    if (!ctx->producer) {
+    if (!ctx->kafka.rk) {
         flb_plg_error(ctx->ins, "failed to create producer: %s",
                       errstr);
-        flb_kafka_conf_destroy(ctx);
+        flb_out_kafka_destroy(ctx);
         return NULL;
     }
 
@@ -280,28 +236,28 @@ struct flb_out_kafka *flb_kafka_conf_create(struct flb_output_instance *ins,
         }
     }
 
-    flb_plg_info(ctx->ins, "brokers='%s' topics='%s'", ctx->brokers, tmp);
+    flb_plg_info(ctx->ins, "brokers='%s' topics='%s'", ctx->kafka.brokers, tmp);
 #ifdef FLB_HAVE_AVRO_ENCODER
     flb_plg_info(ctx->ins, "schemaID='%s' schema='%s'", ctx->avro_fields.schema_id, ctx->avro_fields.schema_str);
 #endif
 
     return ctx;
 }
 
-int flb_kafka_conf_destroy(struct flb_out_kafka *ctx)
+int flb_out_kafka_destroy(struct flb_out_kafka *ctx)
 {
     if (!ctx) {
         return 0;
     }
 
-    if (ctx->brokers) {
-        flb_free(ctx->brokers);
+    if (ctx->kafka.brokers) {
+        flb_free(ctx->kafka.brokers);
     }
 
     flb_kafka_topic_destroy_all(ctx);
 
-    if (ctx->producer) {
-        rd_kafka_destroy(ctx->producer);
+    if (ctx->kafka.rk) {
+        rd_kafka_destroy(ctx->kafka.rk);
     }
 
     if (ctx->topic_key) {

plugins/out_kafka/kafka_config.h

@@ -26,16 +26,14 @@
 #include <fluent-bit/flb_avro.h>
 #endif
 
-#include "rdkafka.h"
+#include <fluent-bit/flb_kafka.h>
 
 #define FLB_KAFKA_FMT_JSON            0
 #define FLB_KAFKA_FMT_MSGP            1
 #define FLB_KAFKA_FMT_GELF            2
 #ifdef FLB_HAVE_AVRO_ENCODER
 #define FLB_KAFKA_FMT_AVRO            3
 #endif
-#define FLB_KAFKA_BROKERS             "127.0.0.1"
-#define FLB_KAFKA_TOPIC               "fluent-bit"
 #define FLB_KAFKA_TS_KEY              "@timestamp"
 #define FLB_KAFKA_QUEUE_FULL_RETRIES  10
 
@@ -61,9 +59,9 @@ struct flb_kafka_topic {
 };
 
 struct flb_out_kafka {
+    struct flb_kafka kafka;
     /* Config Parameters */
     int format;
-    char *brokers;
 
     /* Optional topic key for routing */
     int topic_key_len;
@@ -102,7 +100,6 @@ struct flb_out_kafka {
     int queue_full_retries;
 
     /* Internal */
-    rd_kafka_t *producer;
     rd_kafka_conf_t *conf;
 
     /* Plugin instance */
@@ -122,8 +119,8 @@ struct flb_out_kafka {
 
 };
 
-struct flb_out_kafka *flb_kafka_conf_create(struct flb_output_instance *ins,
-                                        struct flb_config *config);
-int flb_kafka_conf_destroy(struct flb_out_kafka *ctx);
+struct flb_out_kafka *flb_out_kafka_create(struct flb_output_instance *ins,
+                                           struct flb_config *config);
+int flb_out_kafka_destroy(struct flb_out_kafka *ctx);
 
 #endif

plugins/out_kafka/kafka_topic.c

@@ -30,7 +30,7 @@ struct flb_kafka_topic *flb_kafka_topic_create(char *name,
     rd_kafka_topic_t *tp;
     struct flb_kafka_topic *topic;
 
-    tp = rd_kafka_topic_new(ctx->producer, name, NULL);
+    tp = rd_kafka_topic_new(ctx->kafka.rk, name, NULL);
     if (!tp) {
         flb_plg_error(ctx->ins, "failed to create topic: %s",
                       rd_kafka_err2str(rd_kafka_last_error()));

src/CMakeLists.txt

@@ -190,6 +190,13 @@ if(FLB_LUAJIT)
     )
 endif()
 
+if(FLB_OUT_KAFKA)
+  set(src
+    ${src}
+    "flb_kafka.c"
+    )
+endif()
+
 # Link to libco
 set(extra_libs
   ${extra_libs}

src/flb_kafka.c

@@ -0,0 +1,98 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2019-2021 The Fluent Bit Authors
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include "fluent-bit/flb_config.h"
+#include "fluent-bit/flb_mem.h"
+#include "fluent-bit/flb_str.h"
+#include <fluent-bit/flb_kafka.h>
+#include <fluent-bit/flb_kv.h>
+
+#include <rdkafka.h>
+
+rd_kafka_conf_t *flb_kafka_conf_create(struct flb_kafka *kafka,
+                                       struct mk_list *properties,
+                                       int with_group_id)
+{
+    struct mk_list *head;
+    struct flb_kv *kv;
+    const char *conf;
+    rd_kafka_conf_t *kafka_cfg;
+    char errstr[512];
+
+    kafka_cfg = rd_kafka_conf_new();
+    if (!kafka_cfg) {
+        flb_error("[flb_kafka] Could not initialize kafka config object");
+        goto err;
+    }
+
+    conf = flb_config_prop_get("client_id", properties);
+    if (!conf) {
+        conf = "fluent-bit";
+    }
+    if (rd_kafka_conf_set(kafka_cfg, "client.id", conf,
+                errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
+        flb_error("[flb_kafka] cannot configure client id: %s", errstr);
+    }
+
+    if (with_group_id) {
+        conf = flb_config_prop_get("group_id", properties);
+        if (!conf) {
+            conf = "fluent-bit";
+        }
+        if (rd_kafka_conf_set(kafka_cfg, "group.id", conf,
+                    errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
+            flb_error("[flb_kafka] cannot configure group id: %s", errstr);
+        }
+    }
+
+    conf = flb_config_prop_get("brokers", properties);
+    if (conf) {
+        if (rd_kafka_conf_set(kafka_cfg, "bootstrap.servers", conf,
+                errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
+            flb_error("[flb_kafka] failed to configure brokers: %s", errstr);
+            goto err;
+        }
+        kafka->brokers = flb_strdup(conf);
+    }
+    else {
+        flb_error("config: no brokers defined");
+        goto err;
+    }
+
+    /* Iterate custom rdkafka properties */
+    mk_list_foreach(head, properties) {
+        kv = mk_list_entry(head, struct flb_kv, _head);
+        if (strncasecmp(kv->key, "rdkafka.", 8) == 0 &&
+            flb_sds_len(kv->key) > 8) {
+            if (rd_kafka_conf_set(kafka_cfg, kv->key + 8, kv->val,
+                        errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK) {
+                flb_error("[flb_kafka] cannot configure '%s' property", kv->key + 8);
+            }
+        }
+    }
+
+    return kafka_cfg;
+
+err:
+    if (kafka_cfg) {
+        flb_free(kafka_cfg);
+    }
+    return NULL;
+}