Adding upstream version 1.44.3.upstream/1.44.3upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

7 files changed, 558 insertions, 0 deletions

diff --git a/fluent-bit/plugins/out_td/CMakeLists.txt b/fluent-bit/plugins/out_td/CMakeLists.txt
new file mode 100644
index 00000000..3641be38
--- /dev/null
+++ b/fluent-bit/plugins/out_td/CMakeLists.txt
@@ -0,0 +1,7 @@
+set(src
+  td_http.c
+  td_config.c
+  td.c)
+
+FLB_PLUGIN(out_td "${src}" "mk_core")
+target_link_libraries(flb-plugin-out_td)
diff --git a/fluent-bit/plugins/out_td/td.c b/fluent-bit/plugins/out_td/td.c
new file mode 100644
index 00000000..7836c384
--- /dev/null
+++ b/fluent-bit/plugins/out_td/td.c
@@ -0,0 +1,271 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  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_output_plugin.h>
+#include <fluent-bit/flb_network.h>
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_http_client.h>
+#include <fluent-bit/flb_time.h>
+#include <fluent-bit/flb_log_event_decoder.h>
+#include <msgpack.h>
+
+#include "td.h"
+#include "td_http.h"
+#include "td_config.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <errno.h>
+
+/*
+ * Convert the internal Fluent Bit data representation to the required
+ * one by Treasure Data cloud service.
+ *
+ * This function returns a new msgpack buffer and store the bytes length
+ * in the out_size variable.
+ */
+static char *td_format(struct flb_td *ctx, const void *data, size_t bytes, int *out_size)
+{
+    int i;
+    int ret;
+    int n_size;
+    time_t atime;
+    char *buf;
+    struct msgpack_sbuffer mp_sbuf;
+    struct msgpack_packer mp_pck;
+    msgpack_object map;
+    msgpack_sbuffer *sbuf;
+    struct flb_log_event_decoder log_decoder;
+    struct flb_log_event log_event;
+
+    /* Initialize contexts for new output */
+    msgpack_sbuffer_init(&mp_sbuf);
+    msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
+
+    ret = flb_log_event_decoder_init(&log_decoder, (char *) data, bytes);
+
+    if (ret != FLB_EVENT_DECODER_SUCCESS) {
+        flb_plg_error(ctx->ins,
+                      "Log event decoder initialization error : %d", ret);
+
+        return NULL;
+    }
+
+    while ((ret = flb_log_event_decoder_next(
+                    &log_decoder,
+                    &log_event)) == FLB_EVENT_DECODER_SUCCESS) {
+        atime = log_event.timestamp.tm.tv_sec;
+        map   = *log_event.body;
+
+        n_size = map.via.map.size + 1;
+        msgpack_pack_map(&mp_pck, n_size);
+        msgpack_pack_str(&mp_pck, 4);
+        msgpack_pack_str_body(&mp_pck, "time", 4);
+        msgpack_pack_int32(&mp_pck, atime);
+
+        for (i = 0; i < n_size - 1; i++) {
+            msgpack_pack_object(&mp_pck, map.via.map.ptr[i].key);
+            msgpack_pack_object(&mp_pck, map.via.map.ptr[i].val);
+        }
+    }
+
+    flb_log_event_decoder_destroy(&log_decoder);
+
+    /* Create new buffer */
+    sbuf = &mp_sbuf;
+    *out_size = sbuf->size;
+    buf = flb_malloc(sbuf->size);
+    if (!buf) {
+        flb_errno();
+        return NULL;
+    }
+
+    /* set a new buffer and re-initialize our MessagePack context */
+    memcpy(buf, sbuf->data, sbuf->size);
+    msgpack_sbuffer_destroy(&mp_sbuf);
+
+    return buf;
+}
+
+static int cb_td_init(struct flb_output_instance *ins, struct flb_config *config,
+                      void *data)
+{
+    struct flb_td *ctx;
+    struct flb_upstream *upstream;
+    (void) data;
+
+    ctx = td_config_init(ins);
+    if (!ctx) {
+        flb_plg_warn(ins, "Error reading configuration");
+        return -1;
+    }
+
+    if (ctx->region == FLB_TD_REGION_US) {
+        flb_output_net_default("api.treasuredata.com", 443, ins);
+    }
+    else if (ctx->region == FLB_TD_REGION_JP) {
+        flb_output_net_default("api.treasuredata.co.jp", 443, ins);
+    }
+
+    upstream = flb_upstream_create(config,
+                                   ins->host.name,
+                                   ins->host.port,
+                                   FLB_IO_TLS, ins->tls);
+    if (!upstream) {
+        flb_free(ctx);
+        return -1;
+    }
+    ctx->u = upstream;
+    flb_output_upstream_set(ctx->u, ins);
+
+    flb_output_set_context(ins, ctx);
+    return 0;
+}
+
+static void cb_td_flush(struct flb_event_chunk *event_chunk,
+                        struct flb_output_flush *out_flush,
+                        struct flb_input_instance *i_ins,
+                        void *out_context,
+                        struct flb_config *config)
+{
+    int ret;
+    int bytes_out;
+    char *pack;
+    size_t b_sent;
+    char *body = NULL;
+    struct flb_td *ctx = out_context;
+    struct flb_connection *u_conn;
+    struct flb_http_client *c;
+    (void) i_ins;
+
+    /* Convert format */
+    pack = td_format(ctx, event_chunk->data, event_chunk->size, &bytes_out);
+    if (!pack) {
+        FLB_OUTPUT_RETURN(FLB_ERROR);
+    }
+
+    /* Lookup an available connection context */
+    u_conn = flb_upstream_conn_get(ctx->u);
+    if (!u_conn) {
+        flb_plg_error(ctx->ins, "no upstream connections available");
+        flb_free(pack);
+        FLB_OUTPUT_RETURN(FLB_RETRY);
+    }
+
+    /* Compose request */
+    c = td_http_client(u_conn, pack, bytes_out, &body, ctx, config);
+    if (!c) {
+        flb_free(pack);
+        flb_upstream_conn_release(u_conn);
+        FLB_OUTPUT_RETURN(FLB_RETRY);
+    }
+
+    /* Issue HTTP request */
+    ret = flb_http_do(c, &b_sent);
+
+    /* Release Resources */
+    flb_free(pack);
+    flb_free(body);
+
+    /* Validate HTTP status */
+    if (ret == 0) {
+        /* We expect a HTTP 200 OK */
+        if (c->resp.status != 200) {
+            if (c->resp.payload_size > 0) {
+                flb_plg_warn(ctx->ins, "HTTP status %i\n%s",
+                             c->resp.status, c->resp.payload);
+            }
+            else {
+                flb_plg_warn(ctx->ins, "HTTP status %i", c->resp.status);
+            }
+            goto retry;
+        }
+        else {
+            flb_plg_info(ctx->ins, "HTTP status 200 OK");
+        }
+    }
+    else {
+        flb_plg_error(ctx->ins, "http_do=%i", ret);
+        goto retry;
+    }
+
+    /* release */
+    flb_upstream_conn_release(u_conn);
+    flb_http_client_destroy(c);
+
+    FLB_OUTPUT_RETURN(FLB_OK);
+
+ retry:
+    flb_upstream_conn_release(u_conn);
+    flb_http_client_destroy(c);
+
+    FLB_OUTPUT_RETURN(FLB_RETRY);
+}
+
+static int cb_td_exit(void *data, struct flb_config *config)
+{
+    struct flb_td *ctx = data;
+
+    if (!ctx) {
+        return 0;
+    }
+
+    flb_upstream_destroy(ctx->u);
+    flb_free(ctx);
+
+    return 0;
+}
+
+static struct flb_config_map config_map[] = {
+    {
+      FLB_CONFIG_MAP_STR, "API", (char *)NULL,
+      0, FLB_TRUE, offsetof(struct flb_td, api),
+      "Set the API key"
+    },
+    {
+      FLB_CONFIG_MAP_STR, "Database", (char *)NULL,
+      0, FLB_TRUE, offsetof(struct flb_td, db_name),
+      "Set the Database file"
+    },
+    {
+      FLB_CONFIG_MAP_STR, "Table", (char *)NULL,
+      0, FLB_TRUE, offsetof(struct flb_td, db_table),
+      "Set the Database Table"
+    },
+    {
+      FLB_CONFIG_MAP_STR, "Region", (char *)NULL,
+      0, FLB_TRUE, offsetof(struct flb_td, region_str),
+      "Set the Region: us or jp"
+    },
+    /* EOF */
+    {0}
+};
+
+/* Plugin reference */
+struct flb_output_plugin out_td_plugin = {
+    .name           = "td",
+    .description    = "Treasure Data",
+    .cb_init        = cb_td_init,
+    .cb_pre_run     = NULL,
+    .cb_flush       = cb_td_flush,
+    .cb_exit        = cb_td_exit,
+    .config_map     = config_map,
+    .flags          = FLB_IO_TLS,
+};
diff --git a/fluent-bit/plugins/out_td/td.h b/fluent-bit/plugins/out_td/td.h
new file mode 100644
index 00000000..1050289c
--- /dev/null
+++ b/fluent-bit/plugins/out_td/td.h
@@ -0,0 +1,24 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  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_OUT_TD_H
+#define FLB_OUT_TD_H
+
+
+#endif
diff --git a/fluent-bit/plugins/out_td/td_config.c b/fluent-bit/plugins/out_td/td_config.c
new file mode 100644
index 00000000..ac5be669
--- /dev/null
+++ b/fluent-bit/plugins/out_td/td_config.c
@@ -0,0 +1,86 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  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_output_plugin.h>
+#include "td_config.h"
+#include <stdlib.h>
+
+struct flb_td *td_config_init(struct flb_output_instance *ins)
+{
+    int ret;
+    struct flb_td *ctx;
+
+    
+    /* Allocate context */
+    ctx = flb_calloc(1, sizeof(struct flb_td));
+    if (!ctx) {
+        flb_errno();
+        return NULL;
+    }
+    ctx->ins      = ins;
+    ctx->fd       = -1;
+    
+    ret = flb_output_config_map_set(ins, (void *)ctx);
+    if (ret == -1) {
+        flb_plg_error(ins, "unable to load configuration");
+        flb_free(ctx);
+        return NULL;
+    }
+    
+    if (ctx->api == NULL) {
+        flb_plg_error(ins, "error reading API key value");
+        flb_free(ctx);
+        return NULL;
+    }
+
+    if (ctx->db_name == NULL) {
+        flb_plg_error(ins, "error reading Database name");
+        flb_free(ctx);
+        return NULL;
+    }
+
+    if (ctx->db_table == NULL) {
+        flb_plg_error(ins, "error reading Table name");
+        flb_free(ctx);
+        return NULL;
+    }
+
+    /* Lookup desired region */
+    if (ctx->region_str) {
+        if (strcasecmp(ctx->region_str, "us") == 0) {
+            ctx->region = FLB_TD_REGION_US;
+        }
+        else if (strcasecmp(ctx->region_str, "jp") == 0) {
+            ctx->region = FLB_TD_REGION_JP;
+        }
+        else {
+            flb_plg_error(ctx->ins, "invalid region in configuration");
+            flb_free(ctx);
+            return NULL;
+        }
+    }
+    else {
+        ctx->region = FLB_TD_REGION_US;
+    }
+
+    flb_plg_info(ctx->ins, "Treasure Data / database='%s' table='%s'",
+                 ctx->db_name, ctx->db_table);
+
+    return ctx;
+}
diff --git a/fluent-bit/plugins/out_td/td_config.h b/fluent-bit/plugins/out_td/td_config.h
new file mode 100644
index 00000000..f2412fdd
--- /dev/null
+++ b/fluent-bit/plugins/out_td/td_config.h
@@ -0,0 +1,41 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  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_TD_CONFIG_H
+#define FLB_TD_CONFIG_H
+
+#include <fluent-bit/flb_output_plugin.h>
+
+#define FLB_TD_REGION_US    0
+#define FLB_TD_REGION_JP    1
+
+struct flb_td {
+    int fd;           /* Socket to destination/backend */
+    int region;       /* TD Region end-point */
+    flb_sds_t region_str;
+    const char *api;
+    const char *db_name;
+    const char *db_table;
+    struct flb_upstream *u;
+    struct flb_output_instance *ins;
+};
+
+struct flb_td *td_config_init(struct flb_output_instance *ins);
+
+#endif
diff --git a/fluent-bit/plugins/out_td/td_http.c b/fluent-bit/plugins/out_td/td_http.c
new file mode 100644
index 00000000..53940804
--- /dev/null
+++ b/fluent-bit/plugins/out_td/td_http.c
@@ -0,0 +1,94 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  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_output_plugin.h>
+#include <fluent-bit/flb_gzip.h>
+#include <fluent-bit/flb_config.h>
+#include <fluent-bit/flb_http_client.h>
+
+#include "td_config.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+
+#define TD_HTTP_HEADER_SIZE  512
+
+struct flb_http_client *td_http_client(struct flb_connection *u_conn,
+                                       void *data, size_t len,
+                                       char **body,
+                                       struct flb_td *ctx,
+                                       struct flb_config *config)
+{
+    int ret;
+    int pos = 0;
+    int api_len;
+    size_t gz_size;
+    void *gz_data;
+    char *tmp;
+    struct flb_http_client *c;
+
+    /* Compress data */
+    ret = flb_gzip_compress(data, len, &gz_data, &gz_size);
+    if (ret == -1) {
+        flb_plg_error(ctx->ins, "error compressing data");
+        return NULL;
+    }
+
+    /* Compose URI */
+    tmp = flb_malloc(512);
+    if (!tmp) {
+        flb_free(gz_data);
+        return NULL;
+    }
+    snprintf(tmp, 256,
+             "/v3/table/import/%s/%s/msgpack.gz",
+             ctx->db_name, ctx->db_table);
+
+    /* Create client */
+    c = flb_http_client(u_conn, FLB_HTTP_PUT, tmp,
+                        gz_data, gz_size, NULL, 0, NULL, 0);
+    if (!c) {
+        flb_free(tmp);
+        flb_free(gz_data);
+        return NULL;
+    }
+
+    /* Add custom headers */
+    tmp[pos++] = 'T';
+    tmp[pos++] = 'D';
+    tmp[pos++] = '1';
+    tmp[pos++] = ' ';
+
+    api_len = strlen(ctx->api);
+    memcpy(tmp + pos, ctx->api, api_len);
+    pos += api_len;
+
+    flb_http_add_header(c,
+                        "Authorization", 13,
+                        tmp, pos);
+    flb_http_add_header(c,
+                        "Content-Type", 12,
+                        "application/gzip", 16);
+    flb_free(tmp);
+    *body = gz_data;
+
+    return c;
+}
diff --git a/fluent-bit/plugins/out_td/td_http.h b/fluent-bit/plugins/out_td/td_http.h
new file mode 100644
index 00000000..30d398fe
--- /dev/null
+++ b/fluent-bit/plugins/out_td/td_http.h
@@ -0,0 +1,35 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  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_OUT_TD_HTTP_H
+#define FLB_OUT_TD_HTTP_H
+
+#include <fluent-bit/flb_config.h>
+#include "td_config.h"
+
+char *td_http_request(void *data, size_t len,
+                      size_t *out_len,
+                      struct flb_td *ctx, struct flb_config *config);
+
+struct flb_http_client *td_http_client(struct flb_connection *u_conn,
+                                       void *data, size_t len,
+                                       char **body,
+                                       struct flb_td *ctx,
+                                       struct flb_config *config);
+#endif