feat: doris be connection pool

Signed-off-by: composer <[email protected]>

Author: joker-star-l

plugins/out_doris/doris.c

@@ -113,17 +113,36 @@ static int http_put(struct flb_out_doris *ctx,
     msgpack_object msg_key;
     msgpack_object msg_val;
 
+    char address[1024] = {0};
+    int len = 0;
+
     /* Get upstream context and connection */
-    if (strcmp(host, ctx->host) == 0 && port == ctx->port) {
+    if (strcmp(host, ctx->host) == 0 && port == ctx->port) { // address in config
         u = ctx->u;
     }
-    else {
-        // TODO cache
-        u = flb_upstream_create(ctx->u->base.config,
-                                host,
-                                port,
-                                ctx->u->base.flags,
-                                ctx->u->base.tls_context);
+    else { // redirected address
+        len = snprintf(address, sizeof(address), "%s:%i", host, port);
+        u = flb_hash_table_get_ptr(ctx->u_pool, address, len);
+        if (!u) { // first check
+            pthread_mutex_lock(&ctx->mutex); // lock
+            u = flb_hash_table_get_ptr(ctx->u_pool, address, len);
+            if (!u) { // second check
+                u = flb_upstream_create(ctx->u->base.config,
+                                        host,
+                                        port,
+                                        ctx->u->base.flags,
+                                        ctx->u->base.tls_context);
+                if (u) {
+                    flb_hash_table_add(ctx->u_pool, address, len, u, 0);
+                }
+            }
+            pthread_mutex_unlock(&ctx->mutex); // unlock
+            if (!u) {
+                flb_plg_error(ctx->ins, "no doris be connections available to %s:%i",
+                              host, port);
+                return FLB_RETRY;
+            }
+        }
     }
     u_conn = flb_upstream_conn_get(u);
     if (!u_conn) {
@@ -276,11 +295,6 @@ static int http_put(struct flb_out_doris *ctx,
     /* Release the TCP connection */
     flb_upstream_conn_release(u_conn);
 
-    /* Release flb_upstream */
-    if (u != ctx->u) {
-        flb_upstream_destroy(u);
-    }
-
     return out_ret;
 }
 
@@ -341,7 +355,7 @@ static void cb_doris_flush(struct flb_event_chunk *event_chunk,
 
     if (ctx->add_label) {
         /* {label_prefix}_{db}_{table}_{timestamp}_{uuid} */
-        len = snprintf(label, sizeof(label) - 1, "%s_%s_%s_%lu_", ctx->label_prefix, ctx->database, ctx->table, cfl_time_now() / 1000000000L);
+        len = snprintf(label, sizeof(label), "%s_%s_%s_%lu_", ctx->label_prefix, ctx->database, ctx->table, cfl_time_now() / 1000000000L);
         flb_utils_uuid_v4_gen(label + len);
         len += 36;
     }

plugins/out_doris/doris.h

@@ -62,6 +62,11 @@ struct flb_out_doris {
     /* Upstream connection to the backend server */
     struct flb_upstream *u;
 
+    /* doris be connection pool, key: string* be_address, value: flb_upstream* u */
+    struct flb_hash_table *u_pool;
+    pthread_mutex_t mutex;
+    int mutex_initialized;
+
     /* Plugin instance */
     struct flb_output_instance *ins;
 };

plugins/out_doris/doris_conf.c

@@ -75,6 +75,7 @@ struct flb_out_doris *flb_doris_conf_create(struct flb_output_instance *ins,
     int io_flags = 0;
     const char *tmp;
     struct flb_upstream *upstream;
+    struct flb_hash_table *u_pool;
     struct flb_out_doris *ctx = NULL;
     struct flb_doris_progress_reporter *reporter = NULL;
 
@@ -177,6 +178,20 @@ struct flb_out_doris *flb_doris_conf_create(struct flb_output_instance *ins,
     /* Set instance flags into upstream */
     flb_output_upstream_set(ctx->u, ins);
 
+    /* doris be connection pool */
+    u_pool = flb_hash_table_create(FLB_HASH_TABLE_EVICT_NONE, 16, -1);
+    if (!u_pool) {
+        flb_doris_conf_destroy(ctx);
+        return NULL;
+    }
+    ctx->u_pool = u_pool;
+    if (pthread_mutex_init(&ctx->mutex, NULL) == 0) {
+        ctx->mutex_initialized = 1;
+    } else {
+        flb_doris_conf_destroy(ctx);
+        return NULL;
+    }
+
     /* create and start the progress reporter */
     if (ctx->log_progress_interval > 0) {
         reporter = flb_calloc(1, sizeof(struct flb_doris_progress_reporter));
@@ -207,15 +222,38 @@ void flb_doris_conf_destroy(struct flb_out_doris *ctx)
         return;
     }
 
-    if (ctx->u) {
-        flb_upstream_destroy(ctx->u);
-    }
-
     if (ctx->reporter) {
         ctx->reporter->running = 0;
         pthread_join(ctx->reporter_thread, NULL);
         flb_free(ctx->reporter);
     }
 
+    if (ctx->u) {
+        flb_upstream_destroy(ctx->u);
+    }
+
+    if (ctx->u_pool) {
+        int i;
+        struct mk_list *tmp;
+        struct mk_list *head;
+        struct flb_hash_table_entry *entry;
+        struct flb_hash_table_chain *table;
+
+        for (i = 0; i < ctx->u_pool->size; i++) {
+            table = &ctx->u_pool->table[i];
+            mk_list_foreach_safe(head, tmp, &table->chains) {
+                entry = mk_list_entry(head, struct flb_hash_table_entry, _head);
+                flb_upstream_destroy((struct flb_upstream*) entry->val);
+                entry->val = NULL;
+            }
+        }
+
+        flb_hash_table_destroy(ctx->u_pool);
+    }
+
+    if (ctx->mutex_initialized) {
+        pthread_mutex_destroy(&ctx->mutex);
+    }
+
     flb_free(ctx);
 }