Adding upstream version 1.44.3.upstream/1.44.3upstream

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

1 files changed, 1025 insertions, 0 deletions

diff --git a/fluent-bit/plugins/out_calyptia/calyptia.c b/fluent-bit/plugins/out_calyptia/calyptia.c
new file mode 100644
index 00000000..19811dcc
--- /dev/null
+++ b/fluent-bit/plugins/out_calyptia/calyptia.c
@@ -0,0 +1,1025 @@
+/* -*- 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_log.h>
+#include <fluent-bit/flb_kv.h>
+#include <fluent-bit/flb_upstream.h>
+#include <fluent-bit/flb_utils.h>
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_version.h>
+#include <fluent-bit/flb_metrics.h>
+#include <fluent-bit/flb_fstore.h>
+
+#include "calyptia.h"
+
+#include <cmetrics/cmetrics.h>
+#include <cmetrics/cmt_encode_influx.h>
+
+flb_sds_t custom_calyptia_pipeline_config_get(struct flb_config *ctx);
+
+static int get_io_flags(struct flb_output_instance *ins)
+{
+    int flags = 0;
+
+    if (ins->use_tls) {
+        flags = FLB_IO_TLS;
+    }
+    else {
+        flags = FLB_IO_TCP;
+    }
+
+    return flags;
+}
+
+static int config_add_labels(struct flb_output_instance *ins,
+                             struct flb_calyptia *ctx)
+{
+    struct mk_list *head;
+    struct flb_config_map_val *mv;
+    struct flb_slist_entry *k = NULL;
+    struct flb_slist_entry *v = NULL;
+    struct flb_kv *kv;
+
+    if (!ctx->add_labels || mk_list_size(ctx->add_labels) == 0) {
+        return 0;
+    }
+
+    /* iterate all 'add_label' definitions */
+    flb_config_map_foreach(head, mv, ctx->add_labels) {
+        if (mk_list_size(mv->val.list) != 2) {
+            flb_plg_error(ins, "'add_label' expects a key and a value, "
+                          "e.g: 'add_label version 1.8.x'");
+            return -1;
+        }
+
+        k = mk_list_entry_first(mv->val.list, struct flb_slist_entry, _head);
+        v = mk_list_entry_last(mv->val.list, struct flb_slist_entry, _head);
+
+        kv = flb_kv_item_create(&ctx->kv_labels, k->str, v->str);
+        if (!kv) {
+            flb_plg_error(ins, "could not append label %s=%s\n", k->str, v->str);
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
+static void append_labels(struct flb_calyptia *ctx, struct cmt *cmt)
+{
+    struct flb_kv *kv;
+    struct mk_list *head;
+
+    mk_list_foreach(head, &ctx->kv_labels) {
+        kv = mk_list_entry(head, struct flb_kv, _head);
+        cmt_label_add(cmt, kv->key, kv->val);
+    }
+}
+
+static void pack_str(msgpack_packer *mp_pck, char *str)
+{
+    int len;
+
+    len = strlen(str);
+    msgpack_pack_str(mp_pck, len);
+    msgpack_pack_str_body(mp_pck, str, len);
+}
+
+static void pack_env(struct flb_env *env, char *prefix,  char *key,
+                     struct flb_mp_map_header *h,
+                     msgpack_packer *mp_pck)
+{
+    int len = 0;
+    char *val;
+
+    /* prefix set in the key, if set, adjust the key name */
+    if (prefix) {
+        len = strlen(prefix);
+    }
+
+    val = (char *) flb_env_get(env, key);
+    if (val) {
+        flb_mp_map_header_append(h);
+        pack_str(mp_pck, key + len);
+        pack_str(mp_pck, val);
+    }
+}
+
+static void pack_env_metadata(struct flb_env *env,
+                              struct flb_mp_map_header *mh, msgpack_packer *mp_pck)
+{
+    char *tmp;
+    struct flb_mp_map_header h;
+    struct flb_mp_map_header meta;
+
+    /* Metadata */
+    flb_mp_map_header_append(mh);
+    pack_str(mp_pck, "metadata");
+
+    flb_mp_map_header_init(&meta, mp_pck);
+
+    /* Kubernetes */
+    tmp = (char *) flb_env_get(env, "k8s");
+    if (tmp && strcasecmp(tmp, "enabled") == 0) {
+        flb_mp_map_header_append(&meta);
+        pack_str(mp_pck, "k8s");
+
+        /* adding k8s map */
+        flb_mp_map_header_init(&h, mp_pck);
+
+        pack_env(env, "k8s.", "k8s.namespace", &h, mp_pck);
+        pack_env(env, "k8s.", "k8s.pod_name", &h, mp_pck);
+        pack_env(env, "k8s.", "k8s.node_name", &h, mp_pck);
+
+        flb_mp_map_header_end(&h);
+    }
+
+    /* AWS */
+    tmp = (char *) flb_env_get(env, "aws");
+    if (tmp && strcasecmp(tmp, "enabled") == 0) {
+        flb_mp_map_header_append(&meta);
+        pack_str(mp_pck, "aws");
+
+        /* adding aws map */
+        flb_mp_map_header_init(&h, mp_pck);
+
+        pack_env(env, "aws.", "aws.az", &h, mp_pck);
+        pack_env(env, "aws.", "aws.ec2_instance_id", &h, mp_pck);
+        pack_env(env, "aws.", "aws.ec2_instance_type", &h, mp_pck);
+        pack_env(env, "aws.", "aws.private_ip", &h, mp_pck);
+        pack_env(env, "aws.", "aws.vpc_id", &h, mp_pck);
+        pack_env(env, "aws.", "aws.ami_id", &h, mp_pck);
+        pack_env(env, "aws.", "aws.account_id", &h, mp_pck);
+        pack_env(env, "aws.", "aws.hostname", &h, mp_pck);
+
+        flb_mp_map_header_end(&h);
+    }
+    flb_mp_map_header_end(&meta);
+}
+
+static flb_sds_t get_agent_metadata(struct flb_calyptia *ctx)
+{
+    int len;
+    char *host;
+    flb_sds_t conf;
+    flb_sds_t meta;
+    struct flb_mp_map_header mh;
+    msgpack_sbuffer mp_sbuf;
+    msgpack_packer mp_pck;
+    struct flb_config *config = ctx->config;
+
+    /* init msgpack */
+    msgpack_sbuffer_init(&mp_sbuf);
+    msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
+
+    /* pack map */
+    flb_mp_map_header_init(&mh, &mp_pck);
+
+    host = (char *) flb_env_get(ctx->env, "HOSTNAME");
+    if (!host) {
+        host = "unknown";
+    }
+    len = strlen(host);
+
+    /* name */
+    flb_mp_map_header_append(&mh);
+    msgpack_pack_str(&mp_pck, 4);
+    msgpack_pack_str_body(&mp_pck, "name", 4);
+    msgpack_pack_str(&mp_pck, len);
+    msgpack_pack_str_body(&mp_pck, host, len);
+
+    /* type */
+    flb_mp_map_header_append(&mh);
+    msgpack_pack_str(&mp_pck, 4);
+    msgpack_pack_str_body(&mp_pck, "type", 4);
+    msgpack_pack_str(&mp_pck, 9);
+    msgpack_pack_str_body(&mp_pck, "fluentbit", 9);
+
+    /* rawConfig */
+    conf = custom_calyptia_pipeline_config_get(ctx->config);
+    if (conf) {
+        flb_mp_map_header_append(&mh);
+        len = flb_sds_len(conf);
+        msgpack_pack_str(&mp_pck, 9);
+        msgpack_pack_str_body(&mp_pck, "rawConfig", 9);
+        msgpack_pack_str(&mp_pck, len);
+        msgpack_pack_str_body(&mp_pck, conf, len);
+    }
+    flb_sds_destroy(conf);
+
+    /* version */
+    flb_mp_map_header_append(&mh);
+    msgpack_pack_str(&mp_pck, 7);
+    msgpack_pack_str_body(&mp_pck, "version", 7);
+    len = strlen(FLB_VERSION_STR);
+    msgpack_pack_str(&mp_pck, len);
+    msgpack_pack_str_body(&mp_pck, FLB_VERSION_STR, len);
+
+    /* edition */
+    flb_mp_map_header_append(&mh);
+    msgpack_pack_str(&mp_pck, 7);
+    msgpack_pack_str_body(&mp_pck, "edition", 7);
+    msgpack_pack_str(&mp_pck, 9);
+    msgpack_pack_str_body(&mp_pck, "community", 9);
+
+    /* machineID */
+    flb_mp_map_header_append(&mh);
+    msgpack_pack_str(&mp_pck, 9);
+    msgpack_pack_str_body(&mp_pck, "machineID", 9);
+    len = flb_sds_len(ctx->machine_id);
+    msgpack_pack_str(&mp_pck, len);
+    msgpack_pack_str_body(&mp_pck, ctx->machine_id, len);
+
+    /* fleetID */
+    if (ctx->fleet_id) {
+        flb_mp_map_header_append(&mh);
+        msgpack_pack_str(&mp_pck, 7);
+        msgpack_pack_str_body(&mp_pck, "fleetID", 7);
+        len = flb_sds_len(ctx->fleet_id);
+        msgpack_pack_str(&mp_pck, len);
+        msgpack_pack_str_body(&mp_pck, ctx->fleet_id, len);
+    }
+
+    /* pack environment metadata */
+    pack_env_metadata(config->env, &mh, &mp_pck);
+
+    /* finalize */
+    flb_mp_map_header_end(&mh);
+
+    /* convert to json */
+    meta = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size);
+    msgpack_sbuffer_destroy(&mp_sbuf);
+
+    return meta;
+}
+
+static int calyptia_http_do(struct flb_calyptia *ctx, struct flb_http_client *c,
+                            int type)
+{
+    int ret;
+    size_t b_sent;
+
+    /* append headers */
+    if (type == CALYPTIA_ACTION_REGISTER) {
+        flb_http_add_header(c,
+                            CALYPTIA_H_CTYPE, sizeof(CALYPTIA_H_CTYPE) - 1,
+                            CALYPTIA_H_CTYPE_JSON, sizeof(CALYPTIA_H_CTYPE_JSON) - 1);
+
+        flb_http_add_header(c,
+                            CALYPTIA_H_PROJECT, sizeof(CALYPTIA_H_PROJECT) - 1,
+                            ctx->api_key, flb_sds_len(ctx->api_key));
+    }
+    else if (type == CALYPTIA_ACTION_PATCH) {
+        flb_http_add_header(c,
+                            CALYPTIA_H_CTYPE, sizeof(CALYPTIA_H_CTYPE) - 1,
+                            CALYPTIA_H_CTYPE_JSON, sizeof(CALYPTIA_H_CTYPE_JSON) - 1);
+
+        flb_http_add_header(c,
+                            CALYPTIA_H_AGENT_TOKEN,
+                            sizeof(CALYPTIA_H_AGENT_TOKEN) - 1,
+                            ctx->agent_token, flb_sds_len(ctx->agent_token));
+    }
+    else if (type == CALYPTIA_ACTION_METRICS) {
+        flb_http_add_header(c,
+                            CALYPTIA_H_CTYPE, sizeof(CALYPTIA_H_CTYPE) - 1,
+                            CALYPTIA_H_CTYPE_MSGPACK,
+                            sizeof(CALYPTIA_H_CTYPE_MSGPACK) - 1);
+
+        flb_http_add_header(c,
+                            CALYPTIA_H_AGENT_TOKEN,
+                            sizeof(CALYPTIA_H_AGENT_TOKEN) - 1,
+                            ctx->agent_token, flb_sds_len(ctx->agent_token));
+    }
+#ifdef FLB_HAVE_CHUNK_TRACE
+    else if (type == CALYPTIA_ACTION_TRACE)  {
+        flb_http_add_header(c,
+                            CALYPTIA_H_CTYPE, sizeof(CALYPTIA_H_CTYPE) - 1,
+                            CALYPTIA_H_CTYPE_JSON, sizeof(CALYPTIA_H_CTYPE_JSON) - 1);
+
+        flb_http_add_header(c,
+                            CALYPTIA_H_AGENT_TOKEN,
+                            sizeof(CALYPTIA_H_AGENT_TOKEN) - 1,
+                            ctx->agent_token, flb_sds_len(ctx->agent_token));
+    }
+#endif
+
+    /* Map debug callbacks */
+    flb_http_client_debug(c, ctx->ins->callback);
+
+    /* Perform HTTP request */
+    ret = flb_http_do(c, &b_sent);
+    if (ret != 0) {
+        flb_plg_warn(ctx->ins, "http_do=%i", ret);
+        return FLB_RETRY;
+    }
+
+    if (c->resp.status != 200 && c->resp.status != 201 && c->resp.status != 204) {
+        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);
+        }
+
+        /* invalid metrics */
+        if (c->resp.status == 422) {
+            return FLB_ERROR;
+        }
+        return FLB_RETRY;;
+    }
+
+    return FLB_OK;
+}
+
+static flb_sds_t get_agent_info(char *buf, size_t size, char *k)
+{
+    int i;
+    int ret;
+    int type;
+    int len;
+    char *out_buf;
+    flb_sds_t v = NULL;
+    size_t off = 0;
+    size_t out_size;
+    msgpack_unpacked result;
+    msgpack_object root;
+    msgpack_object key;
+    msgpack_object val;
+
+    len = strlen(k);
+
+    ret = flb_pack_json(buf, size, &out_buf, &out_size, &type, NULL);
+    if (ret != 0) {
+        return NULL;
+    }
+
+    msgpack_unpacked_init(&result);
+    ret = msgpack_unpack_next(&result, out_buf, out_size, &off);
+    if (ret != MSGPACK_UNPACK_SUCCESS) {
+        flb_free(out_buf);
+        msgpack_unpacked_destroy(&result);
+        return NULL;
+    }
+
+    root = result.data;
+    if (root.type != MSGPACK_OBJECT_MAP) {
+        flb_free(out_buf);
+        msgpack_unpacked_destroy(&result);
+        return NULL;
+    }
+
+    for (i = 0; i < root.via.map.size; i++) {
+        key = root.via.map.ptr[i].key;
+        val = root.via.map.ptr[i].val;
+
+        if (key.type != MSGPACK_OBJECT_STR || val.type != MSGPACK_OBJECT_STR) {
+            continue;
+        }
+
+        if (key.via.str.size != len) {
+            continue;
+        }
+
+        if (strncmp(key.via.str.ptr, k, len) == 0) {
+            v = flb_sds_create_len(val.via.str.ptr, val.via.str.size);
+            break;
+        }
+    }
+
+    flb_free(out_buf);
+    msgpack_unpacked_destroy(&result);
+    return v;
+}
+
+/* Set the session content */
+static int store_session_set(struct flb_calyptia *ctx, char *buf, size_t size)
+{
+    int ret;
+    int type;
+    char *mp_buf;
+    size_t mp_size;
+
+    /* remove any previous session file */
+    if (ctx->fs_file) {
+        flb_fstore_file_delete(ctx->fs, ctx->fs_file);
+    }
+
+    /* create session file */
+    ctx->fs_file = flb_fstore_file_create(ctx->fs, ctx->fs_stream,
+                                          CALYPTIA_SESSION_FILE, 1024);
+    if (!ctx->fs_file) {
+        flb_plg_error(ctx->ins, "could not create new session file");
+        return -1;
+    }
+
+    /* store meta */
+    flb_fstore_file_meta_set(ctx->fs, ctx->fs_file,
+                             FLB_VERSION_STR "\n", sizeof(FLB_VERSION_STR) - 1);
+
+    /* encode */
+    ret = flb_pack_json(buf, size, &mp_buf, &mp_size, &type, NULL);
+    if (ret < 0) {
+        flb_plg_error(ctx->ins, "could not encode session information");
+        return -1;
+    }
+
+    /* store content */
+    ret = flb_fstore_file_append(ctx->fs_file, mp_buf, mp_size);
+    if (ret == -1) {
+        flb_plg_error(ctx->ins, "could not store session information");
+        flb_free(mp_buf);
+        return -1;
+    }
+
+    flb_free(mp_buf);
+    return 0;
+}
+
+static int store_session_get(struct flb_calyptia *ctx,
+                             void **out_buf, size_t *out_size)
+{
+    int ret;
+    void *buf;
+    size_t size;
+    flb_sds_t json;
+
+    ret = flb_fstore_file_content_copy(ctx->fs, ctx->fs_file,
+                                       &buf, &size);
+
+    if (size == 0) {
+        return -1;
+    }
+
+    /* decode */
+    json = flb_msgpack_raw_to_json_sds(buf, size);
+    flb_free(buf);
+    if (!json) {
+        return -1;
+    }
+
+    *out_buf = json;
+    *out_size = flb_sds_len(json);
+
+    return ret;
+}
+
+static int store_init(struct flb_calyptia *ctx)
+{
+    int ret;
+    struct flb_fstore *fs;
+    struct flb_fstore_file *fsf;
+    void *buf;
+    size_t size;
+
+    /* store context */
+    fs = flb_fstore_create(ctx->store_path, FLB_FSTORE_FS);
+    if (!fs) {
+        flb_plg_error(ctx->ins,
+                      "could not initialize 'store_path': %s",
+                      ctx->store_path);
+        return -1;
+    }
+    ctx->fs = fs;
+
+    /* stream */
+    ctx->fs_stream = flb_fstore_stream_create(ctx->fs, "calyptia");
+    if (!ctx->fs_stream) {
+        flb_plg_error(ctx->ins, "could not create storage stream");
+        return -1;
+    }
+
+    /* lookup any previous file */
+    fsf = flb_fstore_file_get(ctx->fs, ctx->fs_stream, CALYPTIA_SESSION_FILE,
+                              sizeof(CALYPTIA_SESSION_FILE) - 1);
+    if (!fsf) {
+        flb_plg_debug(ctx->ins, "no session file was found");
+        return 0;
+    }
+    ctx->fs_file = fsf;
+
+    /* retrieve session info */
+    ret = store_session_get(ctx, &buf, &size);
+    if (ret == 0) {
+        /* agent id */
+        ctx->agent_id = get_agent_info(buf, size, "id");
+
+        /* agent token */
+        ctx->agent_token = get_agent_info(buf, size, "token");
+
+        if (ctx->agent_id && ctx->agent_token) {
+            flb_plg_info(ctx->ins, "session setup OK");
+        }
+        else {
+            if (ctx->agent_id) {
+                flb_sds_destroy(ctx->agent_id);
+            }
+            if (ctx->agent_token) {
+                flb_sds_destroy(ctx->agent_token);
+            }
+        }
+        flb_sds_destroy(buf);
+    }
+
+    return 0;
+}
+
+/* Agent creation is perform on initialization using a sync upstream connection */
+static int api_agent_create(struct flb_config *config, struct flb_calyptia *ctx)
+{
+    int ret;
+    int flb_ret;
+    int flags;
+    int action = CALYPTIA_ACTION_REGISTER;
+    char uri[1024];
+    flb_sds_t meta;
+    struct flb_upstream *u;
+    struct flb_connection *u_conn;
+    struct flb_http_client *c;
+
+    /* Meta */
+    meta = get_agent_metadata(ctx);
+    if (!meta) {
+        flb_plg_error(ctx->ins, "could not retrieve metadata");
+        return -1;
+    }
+
+    /* Upstream */
+    flags = get_io_flags(ctx->ins);
+    u = flb_upstream_create(ctx->config,
+                            ctx->cloud_host, ctx->cloud_port,
+                            flags, ctx->ins->tls);
+    if (!u) {
+        flb_plg_error(ctx->ins,
+                      "could not create upstream connection on 'agent create'");
+        flb_sds_destroy(meta);
+        return -1;
+    }
+
+    /* Make it synchronous */
+    flb_stream_disable_async_mode(&u->base);
+
+    /* Get upstream connection */
+    u_conn = flb_upstream_conn_get(u);
+    if (!u_conn) {
+        flb_upstream_destroy(u);
+        flb_sds_destroy(meta);
+        return -1;
+    }
+
+    if (ctx->agent_id && ctx->agent_token) {
+        /* Patch */
+        action = CALYPTIA_ACTION_PATCH;
+        snprintf(uri, sizeof(uri) - 1, CALYPTIA_ENDPOINT_PATCH, ctx->agent_id);
+        c = flb_http_client(u_conn, FLB_HTTP_PATCH, uri,
+                            meta, flb_sds_len(meta), NULL, 0, NULL, 0);
+    }
+    else {
+        /* Create */
+        action = CALYPTIA_ACTION_REGISTER;
+        c = flb_http_client(u_conn, FLB_HTTP_POST, CALYPTIA_ENDPOINT_CREATE,
+                            meta, flb_sds_len(meta), NULL, 0, NULL, 0);
+    }
+
+    if (!c) {
+        flb_upstream_conn_release(u_conn);
+        flb_upstream_destroy(u);
+        return -1;
+    }
+
+    /* perform request */
+    flb_ret = calyptia_http_do(ctx, c, action);
+    if (flb_ret == FLB_OK &&
+        (c->resp.status == 200 || c->resp.status == 201 || c->resp.status == 204)) {
+        if (c->resp.payload_size > 0) {
+            if (action == CALYPTIA_ACTION_REGISTER) {
+                /* agent id */
+                ctx->agent_id = get_agent_info(c->resp.payload,
+                                               c->resp.payload_size,
+                                               "id");
+
+                /* agent token */
+                ctx->agent_token = get_agent_info(c->resp.payload,
+                                                  c->resp.payload_size,
+                                                  "token");
+
+                if (ctx->agent_id && ctx->agent_token) {
+                    flb_plg_info(ctx->ins, "connected to Calyptia, agent_id='%s'",
+                                 ctx->agent_id);
+
+                    if (ctx->store_path && ctx->fs) {
+                        ret = store_session_set(ctx,
+                                                c->resp.payload,
+                                                c->resp.payload_size);
+                        if (ret == -1) {
+                            flb_plg_warn(ctx->ins,
+                                         "could not store Calyptia session");
+                        }
+                    }
+                }
+            }
+        }
+
+        if (action == CALYPTIA_ACTION_PATCH) {
+            flb_plg_info(ctx->ins, "known agent registration successful");
+        }
+    }
+
+    /* release resources */
+    flb_sds_destroy(meta);
+    flb_http_client_destroy(c);
+    flb_upstream_conn_release(u_conn);
+    flb_upstream_destroy(u);
+
+    return flb_ret;
+}
+
+static struct flb_calyptia *config_init(struct flb_output_instance *ins,
+                                        struct flb_config *config)
+{
+    int ret;
+    int flags;
+    struct flb_calyptia *ctx;
+
+    /* Calyptia plugin context */
+    ctx = flb_calloc(1, sizeof(struct flb_calyptia));
+    if (!ctx) {
+        flb_errno();
+        return NULL;
+    }
+    ctx->ins = ins;
+    ctx->config = config;
+    flb_kv_init(&ctx->kv_labels);
+
+    /* Load the config map */
+    ret = flb_output_config_map_set(ins, (void *) ctx);
+    if (ret == -1) {
+        flb_free(ctx);
+        return NULL;
+    }
+
+    /* api_key */
+    if (!ctx->api_key) {
+        flb_plg_error(ctx->ins, "configuration 'api_key' is missing");
+        flb_free(ctx);
+        return NULL;
+    }
+
+    /* parse 'add_label' */
+    ret = config_add_labels(ins, ctx);
+    if (ret == -1) {
+        return NULL;
+    }
+
+    /* env reader */
+    ctx->env = flb_env_create();
+
+    /* Set context */
+    flb_output_set_context(ins, ctx);
+
+    /* Initialize optional storage */
+    if (ctx->store_path) {
+        ret = store_init(ctx);
+        if (ret == -1) {
+            return NULL;
+        }
+    }
+
+    /* the machine-id is provided by custom calyptia, which invokes this plugin. */
+    if (!ctx->machine_id) {
+        flb_plg_error(ctx->ins, "machine_id has not been set");
+        return NULL;
+    }
+
+    flb_plg_debug(ctx->ins, "machine_id=%s", ctx->machine_id);
+
+    /* Upstream */
+    flags = get_io_flags(ctx->ins);
+    ctx->u = flb_upstream_create(ctx->config,
+                                 ctx->cloud_host, ctx->cloud_port,
+                                 flags, ctx->ins->tls);
+    if (!ctx->u) {
+        return NULL;
+    }
+
+    /* Set instance flags into upstream */
+    flb_output_upstream_set(ctx->u, ins);
+
+    return ctx;
+}
+
+static int cb_calyptia_init(struct flb_output_instance *ins,
+                            struct flb_config *config, void *data)
+{
+    int ret;
+    struct flb_calyptia *ctx;
+    (void) data;
+
+    /* create config context */
+    ctx = config_init(ins, config);
+    if (!ctx) {
+        flb_plg_error(ins, "could not initialize configuration");
+        return -1;
+    }
+
+    /*
+     * This plugin instance uses the HTTP client interface, let's register
+     * it debugging callbacks.
+     */
+    flb_output_set_http_debug_callbacks(ins);
+
+    /* register/update agent */
+    ret = api_agent_create(config, ctx);
+    if (ret != FLB_OK) {
+        flb_plg_error(ctx->ins, "agent registration failed");
+        return -1;
+    }
+
+    /* metrics endpoint */
+    ctx->metrics_endpoint = flb_sds_create_size(256);
+    flb_sds_printf(&ctx->metrics_endpoint, CALYPTIA_ENDPOINT_METRICS,
+                   ctx->agent_id);
+
+#ifdef FLB_HAVE_CHUNK_TRACE
+    ctx->trace_endpoint = flb_sds_create_size(256);
+    flb_sds_printf(&ctx->trace_endpoint, CALYPTIA_ENDPOINT_TRACE, 
+                   ctx->pipeline_id);
+#endif /* FLB_HAVE_CHUNK_TRACE */
+    return 0;
+}
+
+static void debug_payload(struct flb_calyptia *ctx, void *data, size_t bytes)
+{
+    int ret;
+    size_t off = 0;
+    struct cmt *cmt;
+    cfl_sds_t out;
+
+    ret = cmt_decode_msgpack_create(&cmt, (char *) data, bytes, &off);
+    if (ret != CMT_DECODE_MSGPACK_SUCCESS) {
+        flb_plg_warn(ctx->ins, "could not unpack debug payload");
+        return;
+    }
+
+    out = cmt_encode_text_create(cmt);
+    flb_plg_info(ctx->ins, "debug payload:\n%s", out);
+    cmt_encode_text_destroy(out);
+    cmt_destroy(cmt);
+}
+
+static void cb_calyptia_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;
+    size_t off = 0;
+    size_t out_size = 0;
+    char *out_buf = NULL;
+
+/* used to create records for reporting traces to the cloud. */
+#ifdef FLB_HAVE_CHUNK_TRACE
+    flb_sds_t json;
+#endif /* FLB_HAVE_CHUNK_TRACE */
+
+    struct flb_connection *u_conn;
+    struct flb_http_client *c;
+    struct flb_calyptia *ctx = out_context;
+    struct cmt *cmt;
+    (void) i_ins;
+    (void) config;
+
+    /* Get upstream connection */
+    u_conn = flb_upstream_conn_get(ctx->u);
+    if (!u_conn) {
+        FLB_OUTPUT_RETURN(FLB_RETRY);
+    }
+
+    if (event_chunk->type == FLB_EVENT_TYPE_METRICS) {
+        /* if we have labels append them */
+        if (ctx->add_labels && mk_list_size(ctx->add_labels) > 0) {
+            ret = cmt_decode_msgpack_create(&cmt,
+                                            (char *) event_chunk->data,
+                                            event_chunk->size,
+                                            &off);
+            if (ret != CMT_DECODE_MSGPACK_SUCCESS) {
+                flb_upstream_conn_release(u_conn);
+                FLB_OUTPUT_RETURN(FLB_ERROR);
+            }
+
+            /* append labels set by config */
+            append_labels(ctx, cmt);
+
+            /* encode back to msgpack */
+            ret = cmt_encode_msgpack_create(cmt, &out_buf, &out_size);
+            if (ret != 0) {
+                cmt_destroy(cmt);
+                flb_upstream_conn_release(u_conn);
+                FLB_OUTPUT_RETURN(FLB_ERROR);
+            }
+            cmt_destroy(cmt);
+        }
+        else {
+            out_buf = (char *) event_chunk->data;
+            out_size = event_chunk->size;
+        }
+
+        /* Compose HTTP Client request */
+        c = flb_http_client(u_conn, FLB_HTTP_POST, ctx->metrics_endpoint,
+                            out_buf, out_size, NULL, 0, NULL, 0);
+        if (!c) {
+            if (out_buf != event_chunk->data) {
+                cmt_encode_msgpack_destroy(out_buf);
+            }
+            flb_upstream_conn_release(u_conn);
+            FLB_OUTPUT_RETURN(FLB_RETRY);
+        }
+
+        /* perform request: 'ret' might be FLB_OK, FLB_ERROR or FLB_RETRY */
+        ret = calyptia_http_do(ctx, c, CALYPTIA_ACTION_METRICS);
+        if (ret == FLB_OK) {
+            flb_plg_debug(ctx->ins, "metrics delivered OK");
+        }
+        else if (ret == FLB_ERROR) {
+            flb_plg_error(ctx->ins, "could not deliver metrics");
+            debug_payload(ctx, out_buf, out_size);
+        }
+
+        if (out_buf != event_chunk->data) {
+            cmt_encode_msgpack_destroy(out_buf);
+        }
+    }
+    
+#ifdef FLB_HAVE_CHUNK_TRACE
+    if (event_chunk->type == (FLB_EVENT_TYPE_LOGS | FLB_EVENT_TYPE_HAS_TRACE)) {
+        json = flb_pack_msgpack_to_json_format(event_chunk->data,
+                                               event_chunk->size,
+                                               FLB_PACK_JSON_FORMAT_STREAM,
+                                               FLB_PACK_JSON_DATE_DOUBLE,
+                                               NULL);
+        if (json == NULL) {
+            flb_upstream_conn_release(u_conn);
+            FLB_OUTPUT_RETURN(FLB_RETRY);
+        }
+        out_buf = (char *)json;
+        out_size = flb_sds_len(json);
+
+        if (flb_sds_printf(&ctx->metrics_endpoint, CALYPTIA_ENDPOINT_METRICS,
+                       ctx->agent_id) == NULL) {
+            flb_upstream_conn_release(u_conn);
+            flb_sds_destroy(json);
+            FLB_OUTPUT_RETURN(FLB_RETRY);
+        }
+        c = flb_http_client(u_conn, FLB_HTTP_POST, ctx->trace_endpoint,
+                            out_buf, out_size, NULL, 0, NULL, 0);
+        if (!c) {
+            flb_upstream_conn_release(u_conn);
+            flb_sds_destroy(json);
+            flb_sds_destroy(ctx->metrics_endpoint);
+            FLB_OUTPUT_RETURN(FLB_RETRY);
+        }
+        
+        /* perform request: 'ret' might be FLB_OK, FLB_ERROR or FLB_RETRY */
+        ret = calyptia_http_do(ctx, c, CALYPTIA_ACTION_TRACE);
+        if (ret == FLB_OK) {
+            flb_plg_debug(ctx->ins, "trace delivered OK");
+        }
+        else if (ret == FLB_ERROR) {
+            flb_plg_error(ctx->ins, "could not deliver trace");
+            debug_payload(ctx, out_buf, out_size);
+        }
+        flb_sds_destroy(json);
+    }
+#endif /* FLB_HAVE_CHUNK_TRACE */
+
+    flb_upstream_conn_release(u_conn);
+    flb_http_client_destroy(c);
+    FLB_OUTPUT_RETURN(ret);
+}
+
+static int cb_calyptia_exit(void *data, struct flb_config *config)
+{
+    struct flb_calyptia *ctx = data;
+
+    if (!ctx) {
+        return 0;
+    }
+
+    if (ctx->u) {
+        flb_upstream_destroy(ctx->u);
+    }
+
+    if (ctx->agent_id) {
+        flb_sds_destroy(ctx->agent_id);
+    }
+
+    if (ctx->agent_token) {
+        flb_sds_destroy(ctx->agent_token);
+    }
+
+    if (ctx->env) {
+        flb_env_destroy(ctx->env);
+    }
+
+    if (ctx->metrics_endpoint) {
+        flb_sds_destroy(ctx->metrics_endpoint);
+    }
+
+#ifdef FLB_HAVE_CHUNK_TRACE
+    if (ctx->trace_endpoint) {
+        flb_sds_destroy(ctx->trace_endpoint);
+    }
+#endif /* FLB_HAVE_CHUNK_TRACE */
+
+    if (ctx->fs) {
+        flb_fstore_destroy(ctx->fs);
+    }
+
+    flb_kv_release(&ctx->kv_labels);
+    flb_free(ctx);
+
+    return 0;
+}
+
+/* Configuration properties map */
+static struct flb_config_map config_map[] = {
+    {
+     FLB_CONFIG_MAP_STR, "cloud_host", CALYPTIA_HOST,
+     0, FLB_TRUE, offsetof(struct flb_calyptia, cloud_host),
+     "",
+    },
+
+    {
+     FLB_CONFIG_MAP_INT, "cloud_port", CALYPTIA_PORT,
+     0, FLB_TRUE, offsetof(struct flb_calyptia, cloud_port),
+     "",
+    },
+
+    {
+     FLB_CONFIG_MAP_STR, "api_key", NULL,
+     0, FLB_TRUE, offsetof(struct flb_calyptia, api_key),
+     "Calyptia Cloud API Key."
+    },
+    {
+     FLB_CONFIG_MAP_STR, "machine_id", NULL,
+     0, FLB_TRUE, offsetof(struct flb_calyptia, machine_id),
+     "Custom machine_id to be used when registering agent"
+    },
+    {
+     FLB_CONFIG_MAP_STR, "fleet_id", NULL,
+     0, FLB_TRUE, offsetof(struct flb_calyptia, fleet_id),
+     "Fleet ID for identifying as part of a managed fleet"
+    },
+
+    {
+     FLB_CONFIG_MAP_STR, "store_path", NULL,
+     0, FLB_TRUE, offsetof(struct flb_calyptia, store_path),
+     ""
+    },
+
+    {
+     FLB_CONFIG_MAP_SLIST_1, "add_label", NULL,
+     FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct flb_calyptia, add_labels),
+     "Label to append to the generated metric."
+    },
+
+#ifdef FLB_HAVE_CHUNK_TRACE
+    {
+     FLB_CONFIG_MAP_STR, "pipeline_id", NULL,
+     0, FLB_TRUE, offsetof(struct flb_calyptia, pipeline_id),
+     "Pipeline ID for calyptia core traces."
+    },
+#endif
+
+    /* EOF */
+    {0}
+};
+
+struct flb_output_plugin out_calyptia_plugin = {
+    .name         = "calyptia",
+    .description  = "Calyptia Cloud",
+    .cb_init      = cb_calyptia_init,
+    .cb_flush     = cb_calyptia_flush,
+    .cb_exit      = cb_calyptia_exit,
+    .config_map   = config_map,
+    .flags        = FLB_OUTPUT_NET | FLB_OUTPUT_PRIVATE | FLB_IO_OPT_TLS,
+    .event_type   = FLB_OUTPUT_METRICS
+};