diff options
Diffstat (limited to 'src/fluent-bit/plugins/out_loki/loki.c')
| -rw-r--r-- | src/fluent-bit/plugins/out_loki/loki.c | 1868 |
1 files changed, 1868 insertions, 0 deletions
diff --git a/src/fluent-bit/plugins/out_loki/loki.c b/src/fluent-bit/plugins/out_loki/loki.c new file mode 100644 index 000000000..d93a3f9aa --- /dev/null +++ b/src/fluent-bit/plugins/out_loki/loki.c @@ -0,0 +1,1868 @@ +/* -*- 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_version.h> +#include <fluent-bit/flb_pack.h> +#include <fluent-bit/flb_utils.h> +#include <fluent-bit/flb_time.h> +#include <fluent-bit/flb_http_client.h> +#include <fluent-bit/flb_ra_key.h> +#include <fluent-bit/flb_thread_storage.h> +#include <fluent-bit/record_accessor/flb_ra_parser.h> +#include <fluent-bit/flb_mp.h> +#include <fluent-bit/flb_log_event_decoder.h> +#include <fluent-bit/flb_gzip.h> + +#include <ctype.h> +#include <sys/stat.h> + +#include "loki.h" + +struct flb_loki_dynamic_tenant_id_entry { + flb_sds_t value; + struct cfl_list _head; +}; + +pthread_once_t initialization_guard = PTHREAD_ONCE_INIT; + +FLB_TLS_DEFINE(struct flb_loki_dynamic_tenant_id_entry, + thread_local_tenant_id); + +void initialize_thread_local_storage() +{ + FLB_TLS_INIT(thread_local_tenant_id); +} + +static struct flb_loki_dynamic_tenant_id_entry *dynamic_tenant_id_create() { + struct flb_loki_dynamic_tenant_id_entry *entry; + + entry = (struct flb_loki_dynamic_tenant_id_entry *) \ + flb_calloc(1, sizeof(struct flb_loki_dynamic_tenant_id_entry)); + + if (entry != NULL) { + entry->value = NULL; + + cfl_list_entry_init(&entry->_head); + } + + return entry; +} + +static void dynamic_tenant_id_destroy(struct flb_loki_dynamic_tenant_id_entry *entry) { + if (entry != NULL) { + if (entry->value != NULL) { + flb_sds_destroy(entry->value); + + entry->value = NULL; + } + + if (!cfl_list_entry_is_orphan(&entry->_head)) { + cfl_list_del(&entry->_head); + } + + flb_free(entry); + } +} + +static void flb_loki_kv_init(struct mk_list *list) +{ + mk_list_init(list); +} + +static inline void safe_sds_cat(flb_sds_t *buf, const char *str, int len) +{ + flb_sds_t tmp; + + tmp = flb_sds_cat(*buf, str, len); + if (tmp) { + *buf = tmp; + } +} + +static inline void normalize_cat(struct flb_ra_parser *rp, flb_sds_t *name) +{ + int sub; + int len; + char tmp[64]; + struct mk_list *s_head; + struct flb_ra_key *key; + struct flb_ra_subentry *entry; + + /* Iterate record accessor keys */ + key = rp->key; + if (rp->type == FLB_RA_PARSER_STRING) { + safe_sds_cat(name, key->name, flb_sds_len(key->name)); + } + else if (rp->type == FLB_RA_PARSER_KEYMAP) { + safe_sds_cat(name, key->name, flb_sds_len(key->name)); + if (mk_list_size(key->subkeys) > 0) { + safe_sds_cat(name, "_", 1); + } + + sub = 0; + mk_list_foreach(s_head, key->subkeys) { + entry = mk_list_entry(s_head, struct flb_ra_subentry, _head); + + if (sub > 0) { + safe_sds_cat(name, "_", 1); + } + if (entry->type == FLB_RA_PARSER_STRING) { + safe_sds_cat(name, entry->str, flb_sds_len(entry->str)); + } + else if (entry->type == FLB_RA_PARSER_ARRAY_ID) { + len = snprintf(tmp, sizeof(tmp) -1, "%d", + entry->array_id); + safe_sds_cat(name, tmp, len); + } + sub++; + } + } +} + +static flb_sds_t normalize_ra_key_name(struct flb_loki *ctx, + struct flb_record_accessor *ra) +{ + int c = 0; + flb_sds_t name; + struct mk_list *head; + struct flb_ra_parser *rp; + + name = flb_sds_create_size(128); + if (!name) { + return NULL; + } + + mk_list_foreach(head, &ra->list) { + rp = mk_list_entry(head, struct flb_ra_parser, _head); + if (c > 0) { + flb_sds_cat(name, "_", 1); + } + normalize_cat(rp, &name); + c++; + } + + return name; +} + +void flb_loki_kv_destroy(struct flb_loki_kv *kv) +{ + /* destroy key and value */ + flb_sds_destroy(kv->key); + if (kv->val_type == FLB_LOKI_KV_STR) { + flb_sds_destroy(kv->str_val); + } + else if (kv->val_type == FLB_LOKI_KV_RA) { + flb_ra_destroy(kv->ra_val); + } + + if (kv->ra_key) { + flb_ra_destroy(kv->ra_key); + } + + if (kv->key_normalized) { + flb_sds_destroy(kv->key_normalized); + } + + flb_free(kv); +} + +int flb_loki_kv_append(struct flb_loki *ctx, char *key, char *val) +{ + int ra_count = 0; + int k_len; + int ret; + struct flb_loki_kv *kv; + + if (!key) { + return -1; + } + + if (!val && key[0] != '$') { + return -1; + } + + kv = flb_calloc(1, sizeof(struct flb_loki_kv)); + if (!kv) { + flb_errno(); + return -1; + } + + k_len = strlen(key); + if (key[0] == '$' && k_len >= 2 && isdigit(key[1])) { + flb_plg_error(ctx->ins, + "key name for record accessor cannot start with a number: %s", + key); + flb_free(kv); + return -1; + } + + kv->key = flb_sds_create(key); + if (!kv->key) { + flb_free(kv); + return -1; + } + + /* + * If the key starts with a '$', it means its a record accessor pattern and + * the key value pair will be formed using the key name and it proper value. + */ + if (key[0] == '$' && val == NULL) { + kv->ra_key = flb_ra_create(key, FLB_TRUE); + if (!kv->ra_key) { + flb_plg_error(ctx->ins, + "invalid key record accessor pattern for key '%s'", + key); + flb_loki_kv_destroy(kv); + return -1; + } + + /* Normalize 'key name' using record accessor pattern */ + kv->key_normalized = normalize_ra_key_name(ctx, kv->ra_key); + if (!kv->key_normalized) { + flb_plg_error(ctx->ins, + "could not normalize key pattern name '%s'\n", + kv->ra_key->pattern); + flb_loki_kv_destroy(kv); + return -1; + } + /* remove record keys placed as stream labels via 'labels' and 'label_keys' */ + ret = flb_slist_add(&ctx->remove_keys_derived, key); + if (ret < 0) { + flb_loki_kv_destroy(kv); + return -1; + } + ra_count++; + } + else if (val[0] == '$') { + /* create a record accessor context */ + kv->val_type = FLB_LOKI_KV_RA; + kv->ra_val = flb_ra_create(val, FLB_TRUE); + if (!kv->ra_val) { + flb_plg_error(ctx->ins, + "invalid record accessor pattern for key '%s': %s", + key, val); + flb_loki_kv_destroy(kv); + return -1; + } + ret = flb_slist_add(&ctx->remove_keys_derived, val); + if (ret < 0) { + flb_loki_kv_destroy(kv); + return -1; + } + ra_count++; + } + else { + kv->val_type = FLB_LOKI_KV_STR; + kv->str_val = flb_sds_create(val); + if (!kv->str_val) { + flb_loki_kv_destroy(kv); + return -1; + } + } + mk_list_add(&kv->_head, &ctx->labels_list); + + /* return the number of record accessor values */ + return ra_count; +} + +static void flb_loki_kv_exit(struct flb_loki *ctx) +{ + struct mk_list *tmp; + struct mk_list *head; + struct flb_loki_kv *kv; + + mk_list_foreach_safe(head, tmp, &ctx->labels_list) { + kv = mk_list_entry(head, struct flb_loki_kv, _head); + + /* unlink and destroy */ + mk_list_del(&kv->_head); + flb_loki_kv_destroy(kv); + } +} + +/* Pack a label key, it also perform sanitization of the characters */ +static int pack_label_key(msgpack_packer *mp_pck, char *key, int key_len) +{ + int i; + int k_len = key_len; + int is_digit = FLB_FALSE; + char *p; + size_t prev_size; + + /* Normalize key name using the packed value */ + if (isdigit(*key)) { + is_digit = FLB_TRUE; + k_len++; + } + + /* key: pack the length */ + msgpack_pack_str(mp_pck, k_len); + if (is_digit) { + msgpack_pack_str_body(mp_pck, "_", 1); + } + + /* save the current offset */ + prev_size = ((msgpack_sbuffer *) mp_pck->data)->size; + + /* Pack the key name */ + msgpack_pack_str_body(mp_pck, key, key_len); + + /* 'p' will point to where the key was written */ + p = (char *) (((msgpack_sbuffer*) mp_pck->data)->data + prev_size); + + /* and sanitize the key characters */ + for (i = 0; i < key_len; i++) { + if (!isalnum(p[i]) && p[i] != '_') { + p[i] = '_'; + } + } + + return 0; +} + +static flb_sds_t pack_labels(struct flb_loki *ctx, + msgpack_packer *mp_pck, + char *tag, int tag_len, + msgpack_object *map) +{ + int i; + flb_sds_t ra_val; + struct mk_list *head; + struct flb_ra_value *rval = NULL; + struct flb_loki_kv *kv; + msgpack_object k; + msgpack_object v; + struct flb_mp_map_header mh; + + + /* Initialize dynamic map header */ + flb_mp_map_header_init(&mh, mp_pck); + + mk_list_foreach(head, &ctx->labels_list) { + kv = mk_list_entry(head, struct flb_loki_kv, _head); + + /* record accessor key/value pair */ + if (kv->ra_key != NULL && kv->ra_val == NULL) { + ra_val = flb_ra_translate(kv->ra_key, tag, tag_len, *(map), NULL); + if (!ra_val || flb_sds_len(ra_val) == 0) { + /* if no value is retruned or if it's empty, just skip it */ + flb_plg_warn(ctx->ins, + "empty record accessor key translation for pattern: %s", + kv->ra_key->pattern); + } + else { + /* Pack the key and value */ + flb_mp_map_header_append(&mh); + + /* We skip the first '$' character since it won't be valid in Loki */ + pack_label_key(mp_pck, kv->key_normalized, + flb_sds_len(kv->key_normalized)); + + msgpack_pack_str(mp_pck, flb_sds_len(ra_val)); + msgpack_pack_str_body(mp_pck, ra_val, flb_sds_len(ra_val)); + } + + if (ra_val) { + flb_sds_destroy(ra_val); + } + continue; + } + + /* + * The code is a bit duplicated to be able to manage the exception of an + * invalid or empty value, on that case the k/v is skipped. + */ + if (kv->val_type == FLB_LOKI_KV_STR) { + flb_mp_map_header_append(&mh); + msgpack_pack_str(mp_pck, flb_sds_len(kv->key)); + msgpack_pack_str_body(mp_pck, kv->key, flb_sds_len(kv->key)); + msgpack_pack_str(mp_pck, flb_sds_len(kv->str_val)); + msgpack_pack_str_body(mp_pck, kv->str_val, flb_sds_len(kv->str_val)); + } + else if (kv->val_type == FLB_LOKI_KV_RA) { + /* record accessor type */ + ra_val = flb_ra_translate(kv->ra_val, tag, tag_len, *(map), NULL); + if (!ra_val || flb_sds_len(ra_val) == 0) { + flb_plg_debug(ctx->ins, "could not translate record accessor"); + } + else { + flb_mp_map_header_append(&mh); + msgpack_pack_str(mp_pck, flb_sds_len(kv->key)); + msgpack_pack_str_body(mp_pck, kv->key, flb_sds_len(kv->key)); + msgpack_pack_str(mp_pck, flb_sds_len(ra_val)); + msgpack_pack_str_body(mp_pck, ra_val, flb_sds_len(ra_val)); + } + + if (ra_val) { + flb_sds_destroy(ra_val); + } + } + } + + if (ctx->auto_kubernetes_labels == FLB_TRUE) { + rval = flb_ra_get_value_object(ctx->ra_k8s, *map); + if (rval && rval->o.type == MSGPACK_OBJECT_MAP) { + for (i = 0; i < rval->o.via.map.size; i++) { + k = rval->o.via.map.ptr[i].key; + v = rval->o.via.map.ptr[i].val; + + if (k.type != MSGPACK_OBJECT_STR || v.type != MSGPACK_OBJECT_STR) { + continue; + } + + /* append the key/value pair */ + flb_mp_map_header_append(&mh); + + /* Pack key */ + pack_label_key(mp_pck, (char *) k.via.str.ptr, k.via.str.size); + + /* Pack the value */ + msgpack_pack_str(mp_pck, v.via.str.size); + msgpack_pack_str_body(mp_pck, v.via.str.ptr, v.via.str.size); + } + } + + if (rval) { + flb_ra_key_value_destroy(rval); + } + } + + /* Check if we added any label, if no one has been set, set the defaul 'job' */ + if (mh.entries == 0) { + /* pack the default entry */ + flb_mp_map_header_append(&mh); + msgpack_pack_str(mp_pck, 3); + msgpack_pack_str_body(mp_pck, "job", 3); + msgpack_pack_str(mp_pck, 10); + msgpack_pack_str_body(mp_pck, "fluent-bit", 10); + } + flb_mp_map_header_end(&mh); + return 0; +} + +static int create_label_map_entry(struct flb_loki *ctx, + struct flb_sds_list *list, msgpack_object *val, int *ra_used) +{ + msgpack_object key; + flb_sds_t label_key; + flb_sds_t val_str; + int i; + int len; + int ret; + + if (ctx == NULL || list == NULL || val == NULL || ra_used == NULL) { + return -1; + } + + switch (val->type) { + case MSGPACK_OBJECT_STR: + label_key = flb_sds_create_len(val->via.str.ptr, val->via.str.size); + if (label_key == NULL) { + flb_errno(); + return -1; + } + + val_str = flb_ra_create_str_from_list(list); + if (val_str == NULL) { + flb_plg_error(ctx->ins, "[%s] flb_ra_create_from_list failed", __FUNCTION__); + flb_sds_destroy(label_key); + return -1; + } + + /* for debugging + printf("label_key=%s val_str=%s\n", label_key, val_str); + */ + + ret = flb_loki_kv_append(ctx, label_key, val_str); + flb_sds_destroy(label_key); + flb_sds_destroy(val_str); + if (ret == -1) { + return -1; + } + *ra_used = *ra_used + 1; + + break; + case MSGPACK_OBJECT_MAP: + len = val->via.map.size; + for (i=0; i<len; i++) { + key = val->via.map.ptr[i].key; + if (key.type != MSGPACK_OBJECT_STR) { + flb_plg_error(ctx->ins, "[%s] key is not string", __FUNCTION__); + return -1; + } + ret = flb_sds_list_add(list, (char*)key.via.str.ptr, key.via.str.size); + if (ret < 0) { + flb_plg_error(ctx->ins, "[%s] flb_sds_list_add failed", __FUNCTION__); + return -1; + } + + ret = create_label_map_entry(ctx, list, &val->via.map.ptr[i].val, ra_used); + if (ret < 0) { + return -1; + } + + ret = flb_sds_list_del_last_entry(list); + if (ret < 0) { + flb_plg_error(ctx->ins, "[%s] flb_sds_list_del_last_entry failed", __FUNCTION__); + return -1; + } + } + + break; + default: + flb_plg_error(ctx->ins, "[%s] value type is not str or map. type=%d", __FUNCTION__, val->type); + return -1; + } + return 0; +} + +static int create_label_map_entries(struct flb_loki *ctx, + char *msgpack_buf, size_t msgpack_size, int *ra_used) +{ + struct flb_sds_list *list = NULL; + msgpack_unpacked result; + size_t off = 0; + int i; + int len; + int ret; + msgpack_object key; + + if (ctx == NULL || msgpack_buf == NULL || ra_used == NULL) { + return -1; + } + + msgpack_unpacked_init(&result); + while(msgpack_unpack_next(&result, msgpack_buf, msgpack_size, &off) == MSGPACK_UNPACK_SUCCESS) { + if (result.data.type != MSGPACK_OBJECT_MAP) { + flb_plg_error(ctx->ins, "[%s] data type is not map", __FUNCTION__); + msgpack_unpacked_destroy(&result); + return -1; + } + + len = result.data.via.map.size; + for (i=0; i<len; i++) { + list = flb_sds_list_create(); + if (list == NULL) { + flb_plg_error(ctx->ins, "[%s] flb_sds_list_create failed", __FUNCTION__); + msgpack_unpacked_destroy(&result); + return -1; + } + key = result.data.via.map.ptr[i].key; + if (key.type != MSGPACK_OBJECT_STR) { + flb_plg_error(ctx->ins, "[%s] key is not string", __FUNCTION__); + flb_sds_list_destroy(list); + msgpack_unpacked_destroy(&result); + return -1; + } + + ret = flb_sds_list_add(list, (char*)key.via.str.ptr, key.via.str.size); + if (ret < 0) { + flb_plg_error(ctx->ins, "[%s] flb_sds_list_add failed", __FUNCTION__); + flb_sds_list_destroy(list); + msgpack_unpacked_destroy(&result); + return -1; + } + + ret = create_label_map_entry(ctx, list, &result.data.via.map.ptr[i].val, ra_used); + if (ret < 0) { + flb_plg_error(ctx->ins, "[%s] create_label_map_entry failed", __FUNCTION__); + flb_sds_list_destroy(list); + msgpack_unpacked_destroy(&result); + return -1; + } + + flb_sds_list_destroy(list); + list = NULL; + } + } + + msgpack_unpacked_destroy(&result); + + return 0; +} + +static int read_label_map_path_file(struct flb_output_instance *ins, flb_sds_t path, + char **out_buf, size_t *out_size) +{ + int ret; + int root_type; + char *buf = NULL; + char *msgp_buf = NULL; + FILE *fp = NULL; + struct stat st; + size_t file_size; + size_t ret_size; + + ret = access(path, R_OK); + if (ret < 0) { + flb_errno(); + flb_plg_error(ins, "can't access %s", path); + return -1; + } + + ret = stat(path, &st); + if (ret < 0) { + flb_errno(); + flb_plg_error(ins, "stat failed %s", path); + return -1; + } + file_size = st.st_size; + + fp = fopen(path, "r"); + if (fp == NULL) { + flb_plg_error(ins, "can't open %s", path); + return -1; + } + + buf = flb_malloc(file_size); + if (buf == NULL) { + flb_plg_error(ins, "malloc failed"); + fclose(fp); + return -1; + } + + ret_size = fread(buf, 1, file_size, fp); + if (ret_size < file_size && feof(fp) != 0) { + flb_plg_error(ins, "fread failed"); + fclose(fp); + flb_free(buf); + return -1; + } + + ret = flb_pack_json(buf, file_size, &msgp_buf, &ret_size, &root_type, NULL); + if (ret < 0) { + flb_plg_error(ins, "flb_pack_json failed"); + fclose(fp); + flb_free(buf); + return -1; + } + + *out_buf = msgp_buf; + *out_size = ret_size; + + fclose(fp); + flb_free(buf); + return 0; +} + +static int load_label_map_path(struct flb_loki *ctx, flb_sds_t path, int *ra_used) +{ + int ret; + char *msgpack_buf = NULL; + size_t msgpack_size; + + ret = read_label_map_path_file(ctx->ins, path, &msgpack_buf, &msgpack_size); + if (ret < 0) { + return -1; + } + + ret = create_label_map_entries(ctx, msgpack_buf, msgpack_size, ra_used); + if (ret < 0) { + flb_free(msgpack_buf); + return -1; + } + + if (msgpack_buf != NULL) { + flb_free(msgpack_buf); + } + + return 0; +} + +static int parse_labels(struct flb_loki *ctx) +{ + int ret; + int ra_used = 0; + char *p; + flb_sds_t key; + flb_sds_t val; + struct mk_list *head; + struct flb_slist_entry *entry; + + flb_loki_kv_init(&ctx->labels_list); + + if (ctx->labels) { + mk_list_foreach(head, ctx->labels) { + entry = mk_list_entry(head, struct flb_slist_entry, _head); + + /* record accessor label key ? */ + if (entry->str[0] == '$') { + ret = flb_loki_kv_append(ctx, entry->str, NULL); + if (ret == -1) { + return -1; + } + else if (ret > 0) { + ra_used++; + } + continue; + } + + p = strchr(entry->str, '='); + if (!p) { + flb_plg_error(ctx->ins, "invalid key value pair on '%s'", + entry->str); + return -1; + } + + key = flb_sds_create_size((p - entry->str) + 1); + flb_sds_cat(key, entry->str, p - entry->str); + val = flb_sds_create(p + 1); + if (!key) { + flb_plg_error(ctx->ins, + "invalid key value pair on '%s'", + entry->str); + return -1; + } + if (!val || flb_sds_len(val) == 0) { + flb_plg_error(ctx->ins, + "invalid key value pair on '%s'", + entry->str); + flb_sds_destroy(key); + return -1; + } + + ret = flb_loki_kv_append(ctx, key, val); + flb_sds_destroy(key); + flb_sds_destroy(val); + + if (ret == -1) { + return -1; + } + else if (ret > 0) { + ra_used++; + } + } + } + + /* Append label keys set in the configuration */ + if (ctx->label_keys) { + mk_list_foreach(head, ctx->label_keys) { + entry = mk_list_entry(head, struct flb_slist_entry, _head); + if (entry->str[0] != '$') { + flb_plg_error(ctx->ins, + "invalid label key, the name must start with '$'"); + return -1; + } + + ret = flb_loki_kv_append(ctx, entry->str, NULL); + if (ret == -1) { + return -1; + } + else if (ret > 0) { + ra_used++; + } + } + } + + /* label_map_path */ + if (ctx->label_map_path) { + ret = load_label_map_path(ctx, ctx->label_map_path, &ra_used); + if (ret != 0) { + flb_plg_error(ctx->ins, "failed to load label_map_path"); + } + } + + if (ctx->auto_kubernetes_labels == FLB_TRUE) { + ctx->ra_k8s = flb_ra_create("$kubernetes['labels']", FLB_TRUE); + if (!ctx->ra_k8s) { + flb_plg_error(ctx->ins, + "could not create record accessor for Kubernetes labels"); + return -1; + } + } + + /* + * If the variable 'ra_used' is greater than zero, means that record accessor is + * being used to compose the stream labels. + */ + ctx->ra_used = ra_used; + return 0; +} + +static int key_is_duplicated(struct mk_list *list, char *str, int len) +{ + struct mk_list *head; + struct flb_slist_entry *entry; + + mk_list_foreach(head, list) { + entry = mk_list_entry(head, struct flb_slist_entry, _head); + if (flb_sds_len(entry->str) == len && + strncmp(entry->str, str, len) == 0) { + return FLB_TRUE; + } + } + + return FLB_FALSE; +} + +static int prepare_remove_keys(struct flb_loki *ctx) +{ + int ret; + int len; + int size; + char *tmp; + struct mk_list *head; + struct flb_slist_entry *entry; + struct mk_list *patterns; + + patterns = &ctx->remove_keys_derived; + + /* Add remove keys set in the configuration */ + if (ctx->remove_keys) { + mk_list_foreach(head, ctx->remove_keys) { + entry = mk_list_entry(head, struct flb_slist_entry, _head); + + if (entry->str[0] != '$') { + tmp = flb_malloc(flb_sds_len(entry->str) + 2); + if (!tmp) { + flb_errno(); + continue; + } + else { + tmp[0] = '$'; + len = flb_sds_len(entry->str); + memcpy(tmp + 1, entry->str, len); + tmp[len + 1] = '\0'; + len++; + } + } + else { + tmp = entry->str; + len = flb_sds_len(entry->str); + } + + ret = key_is_duplicated(patterns, tmp, len); + if (ret == FLB_TRUE) { + if (entry->str != tmp) { + flb_free(tmp); + } + continue; + } + + ret = flb_slist_add_n(patterns, tmp, len); + if (entry->str != tmp) { + flb_free(tmp); + } + if (ret < 0) { + return -1; + } + } + size = mk_list_size(patterns); + flb_plg_debug(ctx->ins, "remove_mpa size: %d", size); + if (size > 0) { + ctx->remove_mpa = flb_mp_accessor_create(patterns); + if (ctx->remove_mpa == NULL) { + return -1; + } + } + } + + return 0; +} + +static void loki_config_destroy(struct flb_loki *ctx) +{ + if (ctx->u) { + flb_upstream_destroy(ctx->u); + } + + if (ctx->ra_k8s) { + flb_ra_destroy(ctx->ra_k8s); + } + if (ctx->ra_tenant_id_key) { + flb_ra_destroy(ctx->ra_tenant_id_key); + } + + if (ctx->remove_mpa) { + flb_mp_accessor_destroy(ctx->remove_mpa); + } + flb_slist_destroy(&ctx->remove_keys_derived); + + flb_loki_kv_exit(ctx); + flb_free(ctx); +} + +static struct flb_loki *loki_config_create(struct flb_output_instance *ins, + struct flb_config *config) +{ + int ret; + int io_flags = 0; + struct flb_loki *ctx; + struct flb_upstream *upstream; + char *compress; + + /* Create context */ + ctx = flb_calloc(1, sizeof(struct flb_loki)); + if (!ctx) { + flb_errno(); + return NULL; + } + ctx->ins = ins; + flb_loki_kv_init(&ctx->labels_list); + + /* Register context with plugin instance */ + flb_output_set_context(ins, ctx); + + /* Set networking defaults */ + flb_output_net_default(FLB_LOKI_HOST, FLB_LOKI_PORT, ins); + + /* Load config map */ + ret = flb_output_config_map_set(ins, (void *) ctx); + if (ret == -1) { + return NULL; + } + + /* Initialize final remove_keys list */ + flb_slist_create(&ctx->remove_keys_derived); + + /* Parse labels */ + ret = parse_labels(ctx); + if (ret == -1) { + return NULL; + } + + /* Load remove keys */ + ret = prepare_remove_keys(ctx); + if (ret == -1) { + return NULL; + } + + /* tenant_id_key */ + if (ctx->tenant_id_key_config) { + ctx->ra_tenant_id_key = flb_ra_create(ctx->tenant_id_key_config, FLB_FALSE); + if (!ctx->ra_tenant_id_key) { + flb_plg_error(ctx->ins, + "could not create record accessor for Tenant ID"); + } + } + + /* Compress (gzip) */ + compress = (char *) flb_output_get_property("compress", ins); + ctx->compress_gzip = FLB_FALSE; + if (compress) { + if (strcasecmp(compress, "gzip") == 0) { + ctx->compress_gzip = FLB_TRUE; + } + } + + /* Line Format */ + if (strcasecmp(ctx->line_format, "json") == 0) { + ctx->out_line_format = FLB_LOKI_FMT_JSON; + } + else if (strcasecmp(ctx->line_format, "key_value") == 0) { + ctx->out_line_format = FLB_LOKI_FMT_KV; + } + else { + flb_plg_error(ctx->ins, "invalid 'line_format' value: %s", + ctx->line_format); + return NULL; + } + + /* use TLS ? */ + if (ins->use_tls == FLB_TRUE) { + io_flags = FLB_IO_TLS; + } + else { + io_flags = FLB_IO_TCP; + } + + if (ins->host.ipv6 == FLB_TRUE) { + io_flags |= FLB_IO_IPV6; + } + + /* Create Upstream connection context */ + upstream = flb_upstream_create(config, + ins->host.name, + ins->host.port, + io_flags, + ins->tls); + if (!upstream) { + return NULL; + } + ctx->u = upstream; + flb_output_upstream_set(ctx->u, ins); + ctx->tcp_port = ins->host.port; + ctx->tcp_host = ins->host.name; + + return ctx; +} + +/* + * Convert struct flb_tm timestamp value to nanoseconds and then it pack it as + * a string. + */ +static void pack_timestamp(msgpack_packer *mp_pck, struct flb_time *tms) +{ + int len; + char buf[64]; + uint64_t nanosecs; + + /* convert to nanoseconds */ + nanosecs = flb_time_to_nanosec(tms); + + /* format as a string */ + len = snprintf(buf, sizeof(buf) - 1, "%" PRIu64, nanosecs); + + /* pack the value */ + msgpack_pack_str(mp_pck, len); + msgpack_pack_str_body(mp_pck, buf, len); +} + + +static void pack_format_line_value(flb_sds_t *buf, msgpack_object *val) +{ + int i; + int len; + char temp[512]; + msgpack_object k; + msgpack_object v; + + if (val->type == MSGPACK_OBJECT_STR) { + safe_sds_cat(buf, "\"", 1); + safe_sds_cat(buf, val->via.str.ptr, val->via.str.size); + safe_sds_cat(buf, "\"", 1); + } + else if (val->type == MSGPACK_OBJECT_NIL) { + safe_sds_cat(buf, "null", 4); + } + else if (val->type == MSGPACK_OBJECT_BOOLEAN) { + if (val->via.boolean) { + safe_sds_cat(buf, "true", 4); + } + else { + safe_sds_cat(buf, "false", 5); + } + } + else if (val->type == MSGPACK_OBJECT_POSITIVE_INTEGER) { + len = snprintf(temp, sizeof(temp)-1, "%"PRIu64, val->via.u64); + safe_sds_cat(buf, temp, len); + } + else if (val->type == MSGPACK_OBJECT_NEGATIVE_INTEGER) { + len = snprintf(temp, sizeof(temp)-1, "%"PRId64, val->via.i64); + safe_sds_cat(buf, temp, len); + } + else if (val->type == MSGPACK_OBJECT_FLOAT32 || + val->type == MSGPACK_OBJECT_FLOAT64) { + if (val->via.f64 == (double)(long long int) val->via.f64) { + len = snprintf(temp, sizeof(temp)-1, "%.1f", val->via.f64); + } + else { + len = snprintf(temp, sizeof(temp)-1, "%.16g", val->via.f64); + } + safe_sds_cat(buf, temp, len); + } + else if (val->type == MSGPACK_OBJECT_ARRAY) { + safe_sds_cat(buf, "\"[", 2); + for (i = 0; i < val->via.array.size; i++) { + v = val->via.array.ptr[i]; + if (i > 0) { + safe_sds_cat(buf, " ", 1); + } + pack_format_line_value(buf, &v); + } + safe_sds_cat(buf, "]\"", 2); + } + else if (val->type == MSGPACK_OBJECT_MAP) { + safe_sds_cat(buf, "\"map[", 5); + + for (i = 0; i < val->via.map.size; i++) { + k = val->via.map.ptr[i].key; + v = val->via.map.ptr[i].val; + + if (k.type != MSGPACK_OBJECT_STR) { + continue; + } + + if (i > 0) { + safe_sds_cat(buf, " ", 1); + } + + safe_sds_cat(buf, k.via.str.ptr, k.via.str.size); + safe_sds_cat(buf, ":", 1); + pack_format_line_value(buf, &v); + } + safe_sds_cat(buf, "]\"", 2); + } + else { + + return; + } +} + +// seek tenant id from map and set it to dynamic_tenant_id +static int get_tenant_id_from_record(struct flb_loki *ctx, msgpack_object *map, + flb_sds_t *dynamic_tenant_id) +{ + struct flb_ra_value *rval = NULL; + flb_sds_t tmp_str; + int cmp_len; + + rval = flb_ra_get_value_object(ctx->ra_tenant_id_key, *map); + + if (rval == NULL) { + flb_plg_warn(ctx->ins, "the value of %s is missing", + ctx->tenant_id_key_config); + return -1; + } + else if (rval->o.type != MSGPACK_OBJECT_STR) { + flb_plg_warn(ctx->ins, "the value of %s is not string", + ctx->tenant_id_key_config); + flb_ra_key_value_destroy(rval); + return -1; + } + + tmp_str = flb_sds_create_len(rval->o.via.str.ptr, + rval->o.via.str.size); + if (tmp_str == NULL) { + flb_plg_warn(ctx->ins, "cannot create tenant ID string from record"); + flb_ra_key_value_destroy(rval); + return -1; + } + + // check if already dynamic_tenant_id is set. + if (*dynamic_tenant_id != NULL) { + cmp_len = flb_sds_len(*dynamic_tenant_id); + + if ((rval->o.via.str.size == cmp_len) && + flb_sds_cmp(tmp_str, *dynamic_tenant_id, cmp_len) == 0) { + // tenant_id is same. nothing to do. + flb_ra_key_value_destroy(rval); + flb_sds_destroy(tmp_str); + + return 0; + } + + flb_plg_warn(ctx->ins, "Tenant ID is overwritten %s -> %s", + *dynamic_tenant_id, tmp_str); + + flb_sds_destroy(*dynamic_tenant_id); + } + + // this sds will be released after setting http header. + *dynamic_tenant_id = tmp_str; + flb_plg_debug(ctx->ins, "Tenant ID is %s", *dynamic_tenant_id); + + flb_ra_key_value_destroy(rval); + return 0; +} + +static int pack_record(struct flb_loki *ctx, + msgpack_packer *mp_pck, msgpack_object *rec, + flb_sds_t *dynamic_tenant_id) +{ + int i; + int skip = 0; + int len; + int ret; + int size_hint = 1024; + char *line; + flb_sds_t buf; + msgpack_object key; + msgpack_object val; + char *tmp_sbuf_data = NULL; + size_t tmp_sbuf_size; + msgpack_unpacked mp_buffer; + size_t off = 0; + + /* + * Get tenant id from record before removing keys. + * https://github.com/fluent/fluent-bit/issues/6207 + */ + if (ctx->ra_tenant_id_key && rec->type == MSGPACK_OBJECT_MAP) { + get_tenant_id_from_record(ctx, rec, dynamic_tenant_id); + } + + /* Remove keys in remove_keys */ + msgpack_unpacked_init(&mp_buffer); + if (ctx->remove_mpa) { + ret = flb_mp_accessor_keys_remove(ctx->remove_mpa, rec, + (void *) &tmp_sbuf_data, &tmp_sbuf_size); + if (ret == FLB_TRUE) { + ret = msgpack_unpack_next(&mp_buffer, tmp_sbuf_data, tmp_sbuf_size, &off); + if (ret != MSGPACK_UNPACK_SUCCESS) { + flb_free(tmp_sbuf_data); + msgpack_unpacked_destroy(&mp_buffer); + return -1; + } + rec = &mp_buffer.data; + } + } + + /* Drop single key */ + if (ctx->drop_single_key == FLB_TRUE && rec->type == MSGPACK_OBJECT_MAP && rec->via.map.size == 1) { + if (ctx->out_line_format == FLB_LOKI_FMT_JSON) { + rec = &rec->via.map.ptr[0].val; + } else if (ctx->out_line_format == FLB_LOKI_FMT_KV) { + val = rec->via.map.ptr[0].val; + + if (val.type == MSGPACK_OBJECT_STR) { + msgpack_pack_str(mp_pck, val.via.str.size); + msgpack_pack_str_body(mp_pck, val.via.str.ptr, val.via.str.size); + } else { + buf = flb_sds_create_size(size_hint); + if (!buf) { + msgpack_unpacked_destroy(&mp_buffer); + if (tmp_sbuf_data) { + flb_free(tmp_sbuf_data); + } + return -1; + } + pack_format_line_value(&buf, &val); + msgpack_pack_str(mp_pck, flb_sds_len(buf)); + msgpack_pack_str_body(mp_pck, buf, flb_sds_len(buf)); + flb_sds_destroy(buf); + } + + msgpack_unpacked_destroy(&mp_buffer); + if (tmp_sbuf_data) { + flb_free(tmp_sbuf_data); + } + + return 0; + } + } + + if (ctx->out_line_format == FLB_LOKI_FMT_JSON) { + line = flb_msgpack_to_json_str(size_hint, rec); + if (!line) { + if (tmp_sbuf_data) { + flb_free(tmp_sbuf_data); + } + msgpack_unpacked_destroy(&mp_buffer); + return -1; + } + len = strlen(line); + msgpack_pack_str(mp_pck, len); + msgpack_pack_str_body(mp_pck, line, len); + flb_free(line); + } + else if (ctx->out_line_format == FLB_LOKI_FMT_KV) { + if (rec->type != MSGPACK_OBJECT_MAP) { + msgpack_unpacked_destroy(&mp_buffer); + if (tmp_sbuf_data) { + flb_free(tmp_sbuf_data); + } + return -1; + } + + buf = flb_sds_create_size(size_hint); + if (!buf) { + msgpack_unpacked_destroy(&mp_buffer); + if (tmp_sbuf_data) { + flb_free(tmp_sbuf_data); + } + return -1; + } + + for (i = 0; i < rec->via.map.size; i++) { + key = rec->via.map.ptr[i].key; + val = rec->via.map.ptr[i].val; + + if (key.type != MSGPACK_OBJECT_STR) { + skip++; + continue; + } + + if (i > skip) { + safe_sds_cat(&buf, " ", 1); + } + + safe_sds_cat(&buf, key.via.str.ptr, key.via.str.size); + safe_sds_cat(&buf, "=", 1); + pack_format_line_value(&buf, &val); + } + + msgpack_pack_str(mp_pck, flb_sds_len(buf)); + msgpack_pack_str_body(mp_pck, buf, flb_sds_len(buf)); + flb_sds_destroy(buf); + } + + msgpack_unpacked_destroy(&mp_buffer); + if (tmp_sbuf_data) { + flb_free(tmp_sbuf_data); + } + + return 0; +} + +/* Initialization callback */ +static int cb_loki_init(struct flb_output_instance *ins, + struct flb_config *config, void *data) +{ + int result; + struct flb_loki *ctx; + + /* Create plugin context */ + ctx = loki_config_create(ins, config); + if (!ctx) { + flb_plg_error(ins, "cannot initialize configuration"); + return -1; + } + + result = pthread_mutex_init(&ctx->dynamic_tenant_list_lock, NULL); + + if (result != 0) { + flb_errno(); + + flb_plg_error(ins, "cannot initialize dynamic tenant id list lock"); + + loki_config_destroy(ctx); + + return -1; + } + + result = pthread_once(&initialization_guard, + initialize_thread_local_storage); + + if (result != 0) { + flb_errno(); + + flb_plg_error(ins, "cannot initialize thread local storage"); + + loki_config_destroy(ctx); + + return -1; + } + + cfl_list_init(&ctx->dynamic_tenant_list); + + /* + * This plugin instance uses the HTTP client interface, let's register + * it debugging callbacks. + */ + flb_output_set_http_debug_callbacks(ins); + + flb_plg_info(ins, + "configured, hostname=%s:%i", + ctx->tcp_host, ctx->tcp_port); + return 0; +} + +static flb_sds_t loki_compose_payload(struct flb_loki *ctx, + int total_records, + char *tag, int tag_len, + const void *data, size_t bytes, + flb_sds_t *dynamic_tenant_id) +{ + // int mp_ok = MSGPACK_UNPACK_SUCCESS; + // size_t off = 0; + flb_sds_t json; + // struct flb_time tms; + // msgpack_unpacked result; + msgpack_packer mp_pck; + msgpack_sbuffer mp_sbuf; + // msgpack_object *obj; + struct flb_log_event_decoder log_decoder; + struct flb_log_event log_event; + int ret; + + /* + * Fluent Bit uses Loki API v1 to push records in JSON format, this + * is the expected structure: + * + * { + * "streams": [ + * { + * "stream": { + * "label": "value" + * }, + * "values": [ + * [ "<unix epoch in nanoseconds>", "<log line>" ], + * [ "<unix epoch in nanoseconds>", "<log line>" ] + * ] + * } + * ] + * } + */ + + 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; + } + + /* Initialize msgpack buffers */ + msgpack_sbuffer_init(&mp_sbuf); + msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write); + + /* Main map */ + msgpack_pack_map(&mp_pck, 1); + + /* streams */ + msgpack_pack_str(&mp_pck, 7); + msgpack_pack_str_body(&mp_pck, "streams", 7); + + if (ctx->ra_used == 0 && ctx->auto_kubernetes_labels == FLB_FALSE) { + /* + * If labels are cached, there is no record accessor or custom + * keys, so it's safe to put one main stream and attach all the + * values. + */ + msgpack_pack_array(&mp_pck, 1); + + /* map content: streams['stream'] & streams['values'] */ + msgpack_pack_map(&mp_pck, 2); + + /* streams['stream'] */ + msgpack_pack_str(&mp_pck, 6); + msgpack_pack_str_body(&mp_pck, "stream", 6); + + /* Pack stream labels */ + pack_labels(ctx, &mp_pck, tag, tag_len, NULL); + + /* streams['values'] */ + msgpack_pack_str(&mp_pck, 6); + msgpack_pack_str_body(&mp_pck, "values", 6); + msgpack_pack_array(&mp_pck, total_records); + + while ((ret = flb_log_event_decoder_next( + &log_decoder, + &log_event)) == FLB_EVENT_DECODER_SUCCESS) { + msgpack_pack_array(&mp_pck, 2); + + /* Append the timestamp */ + pack_timestamp(&mp_pck, &log_event.timestamp); + pack_record(ctx, &mp_pck, log_event.body, dynamic_tenant_id); + } + } + else { + /* + * Here there are no cached labels and the labels are composed by + * each record content. To simplify the operation just create + * one stream per record. + */ + msgpack_pack_array(&mp_pck, total_records); + + while ((ret = flb_log_event_decoder_next( + &log_decoder, + &log_event)) == FLB_EVENT_DECODER_SUCCESS) { + /* map content: streams['stream'] & streams['values'] */ + msgpack_pack_map(&mp_pck, 2); + + /* streams['stream'] */ + msgpack_pack_str(&mp_pck, 6); + msgpack_pack_str_body(&mp_pck, "stream", 6); + + /* Pack stream labels */ + pack_labels(ctx, &mp_pck, tag, tag_len, log_event.body); + + /* streams['values'] */ + msgpack_pack_str(&mp_pck, 6); + msgpack_pack_str_body(&mp_pck, "values", 6); + msgpack_pack_array(&mp_pck, 1); + + msgpack_pack_array(&mp_pck, 2); + + /* Append the timestamp */ + pack_timestamp(&mp_pck, &log_event.timestamp); + pack_record(ctx, &mp_pck, log_event.body, dynamic_tenant_id); + } + } + + flb_log_event_decoder_destroy(&log_decoder); + + json = flb_msgpack_raw_to_json_sds(mp_sbuf.data, mp_sbuf.size); + + msgpack_sbuffer_destroy(&mp_sbuf); + + return json; +} + +static void payload_release(void *payload, int compressed) +{ + if (compressed) { + flb_free(payload); + } + else { + flb_sds_destroy(payload); + } +} + +static void cb_loki_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 out_ret = FLB_OK; + size_t b_sent; + flb_sds_t payload = NULL; + flb_sds_t out_buf = NULL; + size_t out_size; + int compressed = FLB_FALSE; + struct flb_loki *ctx = out_context; + struct flb_connection *u_conn; + struct flb_http_client *c; + struct flb_loki_dynamic_tenant_id_entry *dynamic_tenant_id; + + dynamic_tenant_id = FLB_TLS_GET(thread_local_tenant_id); + + if (dynamic_tenant_id == NULL) { + dynamic_tenant_id = dynamic_tenant_id_create(); + + if (dynamic_tenant_id == NULL) { + flb_errno(); + flb_plg_error(ctx->ins, "cannot allocate dynamic tenant id"); + + FLB_OUTPUT_RETURN(FLB_RETRY); + } + + FLB_TLS_SET(thread_local_tenant_id, dynamic_tenant_id); + + pthread_mutex_lock(&ctx->dynamic_tenant_list_lock); + + cfl_list_add(&dynamic_tenant_id->_head, &ctx->dynamic_tenant_list); + + pthread_mutex_unlock(&ctx->dynamic_tenant_list_lock); + } + + /* Format the data to the expected Newrelic Payload */ + payload = loki_compose_payload(ctx, + event_chunk->total_events, + (char *) event_chunk->tag, + flb_sds_len(event_chunk->tag), + event_chunk->data, event_chunk->size, + &dynamic_tenant_id->value); + + if (!payload) { + flb_plg_error(ctx->ins, "cannot compose request payload"); + + FLB_OUTPUT_RETURN(FLB_RETRY); + } + + /* Map buffer */ + out_buf = payload; + out_size = flb_sds_len(payload); + + if (ctx->compress_gzip == FLB_TRUE) { + ret = flb_gzip_compress((void *) payload, flb_sds_len(payload), (void **) &out_buf, &out_size); + if (ret == -1) { + flb_plg_error(ctx->ins, + "cannot gzip payload, disabling compression"); + } else { + compressed = FLB_TRUE; + /* payload is not longer needed */ + flb_sds_destroy(payload); + } + } + + /* 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"); + + payload_release(out_buf, compressed); + + FLB_OUTPUT_RETURN(FLB_RETRY); + } + + /* Create HTTP client context */ + c = flb_http_client(u_conn, FLB_HTTP_POST, FLB_LOKI_URI, + out_buf, out_size, + ctx->tcp_host, ctx->tcp_port, + NULL, 0); + if (!c) { + flb_plg_error(ctx->ins, "cannot create HTTP client context"); + + payload_release(out_buf, compressed); + flb_upstream_conn_release(u_conn); + + FLB_OUTPUT_RETURN(FLB_RETRY); + } + + /* Set callback context to the HTTP client context */ + flb_http_set_callback_context(c, ctx->ins->callback); + + /* User Agent */ + flb_http_add_header(c, "User-Agent", 10, "Fluent-Bit", 10); + + /* Auth headers */ + if (ctx->http_user && ctx->http_passwd) { /* Basic */ + flb_http_basic_auth(c, ctx->http_user, ctx->http_passwd); + } else if (ctx->bearer_token) { /* Bearer token */ + flb_http_bearer_auth(c, ctx->bearer_token); + } + + /* Add Content-Type header */ + flb_http_add_header(c, + FLB_LOKI_CT, sizeof(FLB_LOKI_CT) - 1, + FLB_LOKI_CT_JSON, sizeof(FLB_LOKI_CT_JSON) - 1); + + if (compressed == FLB_TRUE) { + flb_http_set_content_encoding_gzip(c); + } + + /* Add X-Scope-OrgID header */ + if (dynamic_tenant_id->value != NULL) { + flb_http_add_header(c, + FLB_LOKI_HEADER_SCOPE, sizeof(FLB_LOKI_HEADER_SCOPE) - 1, + dynamic_tenant_id->value, + flb_sds_len(dynamic_tenant_id->value)); + } + else if (ctx->tenant_id) { + flb_http_add_header(c, + FLB_LOKI_HEADER_SCOPE, sizeof(FLB_LOKI_HEADER_SCOPE) - 1, + ctx->tenant_id, flb_sds_len(ctx->tenant_id)); + } + + /* Send HTTP request */ + ret = flb_http_do(c, &b_sent); + payload_release(out_buf, compressed); + + /* Validate HTTP client return status */ + if (ret == 0) { + /* + * Only allow the following HTTP status: + * + * - 200: OK + * - 201: Created + * - 202: Accepted + * - 203: no authorative resp + * - 204: No Content + * - 205: Reset content + * + */ + if (c->resp.status == 400) { + /* + * Loki will return 400 if incoming data is out of order. + * We should not retry such data. + */ + flb_plg_error(ctx->ins, "%s:%i, HTTP status=%i Not retrying.\n%s", + ctx->tcp_host, ctx->tcp_port, c->resp.status, + c->resp.payload); + out_ret = FLB_ERROR; + } + else if (c->resp.status < 200 || c->resp.status > 205) { + if (c->resp.payload) { + flb_plg_error(ctx->ins, "%s:%i, HTTP status=%i\n%s", + ctx->tcp_host, ctx->tcp_port, c->resp.status, + c->resp.payload); + } + else { + flb_plg_error(ctx->ins, "%s:%i, HTTP status=%i", + ctx->tcp_host, ctx->tcp_port, c->resp.status); + } + out_ret = FLB_RETRY; + } + else { + if (c->resp.payload) { + flb_plg_debug(ctx->ins, "%s:%i, HTTP status=%i\n%s", + ctx->tcp_host, ctx->tcp_port, + c->resp.status, c->resp.payload); + } + else { + flb_plg_debug(ctx->ins, "%s:%i, HTTP status=%i", + ctx->tcp_host, ctx->tcp_port, + c->resp.status); + } + } + } + else { + flb_plg_error(ctx->ins, "could not flush records to %s:%i (http_do=%i)", + ctx->tcp_host, ctx->tcp_port, ret); + out_ret = FLB_RETRY; + } + + flb_http_client_destroy(c); + flb_upstream_conn_release(u_conn); + + FLB_OUTPUT_RETURN(out_ret); +} + +static void release_dynamic_tenant_ids(struct cfl_list *dynamic_tenant_list) +{ + struct cfl_list *iterator; + struct cfl_list *backup; + struct flb_loki_dynamic_tenant_id_entry *entry; + + cfl_list_foreach_safe(iterator, backup, dynamic_tenant_list) { + entry = cfl_list_entry(iterator, + struct flb_loki_dynamic_tenant_id_entry, + _head); + + dynamic_tenant_id_destroy(entry); + } +} + +static int cb_loki_exit(void *data, struct flb_config *config) +{ + struct flb_loki *ctx = data; + + if (!ctx) { + return 0; + } + + pthread_mutex_lock(&ctx->dynamic_tenant_list_lock); + + release_dynamic_tenant_ids(&ctx->dynamic_tenant_list); + + pthread_mutex_unlock(&ctx->dynamic_tenant_list_lock); + + loki_config_destroy(ctx); + + return 0; +} + +/* Configuration properties map */ +static struct flb_config_map config_map[] = { + { + FLB_CONFIG_MAP_STR, "tenant_id", NULL, + 0, FLB_TRUE, offsetof(struct flb_loki, tenant_id), + "Tenant ID used by default to push logs to Loki. If omitted or empty " + "it assumes Loki is running in single-tenant mode and no X-Scope-OrgID " + "header is sent." + }, + { + FLB_CONFIG_MAP_STR, "tenant_id_key", NULL, + 0, FLB_TRUE, offsetof(struct flb_loki, tenant_id_key_config), + "If set, X-Scope-OrgID will be the value of the key from incoming record. " + "It is useful to set X-Scode-OrgID dynamically." + }, + + { + FLB_CONFIG_MAP_CLIST, "labels", NULL, + 0, FLB_TRUE, offsetof(struct flb_loki, labels), + "labels for API requests. If no value is set, the default label is 'job=fluent-bit'" + }, + + { + FLB_CONFIG_MAP_BOOL, "auto_kubernetes_labels", "false", + 0, FLB_TRUE, offsetof(struct flb_loki, auto_kubernetes_labels), + "If set to true, it will add all Kubernetes labels to Loki labels.", + }, + + { + FLB_CONFIG_MAP_BOOL, "drop_single_key", "false", + 0, FLB_TRUE, offsetof(struct flb_loki, drop_single_key), + "If set to true and only a single key remains, the log line sent to Loki " + "will be the value of that key.", + }, + + { + FLB_CONFIG_MAP_CLIST, "label_keys", NULL, + 0, FLB_TRUE, offsetof(struct flb_loki, label_keys), + "Comma separated list of keys to use as stream labels." + }, + + { + FLB_CONFIG_MAP_CLIST, "remove_keys", NULL, + 0, FLB_TRUE, offsetof(struct flb_loki, remove_keys), + "Comma separated list of keys to remove." + }, + + { + FLB_CONFIG_MAP_STR, "line_format", "json", + 0, FLB_TRUE, offsetof(struct flb_loki, line_format), + "Format to use when flattening the record to a log line. Valid values are " + "'json' or 'key_value'. If set to 'json' the log line sent to Loki will be " + "the Fluent Bit record dumped as json. If set to 'key_value', the log line " + "will be each item in the record concatenated together (separated by a " + "single space) in the format '='." + }, + + { + FLB_CONFIG_MAP_STR, "label_map_path", NULL, + 0, FLB_TRUE, offsetof(struct flb_loki, label_map_path), + "A label map file path" + }, + + { + FLB_CONFIG_MAP_STR, "http_user", NULL, + 0, FLB_TRUE, offsetof(struct flb_loki, http_user), + "Set HTTP auth user" + }, + + { + FLB_CONFIG_MAP_STR, "http_passwd", "", + 0, FLB_TRUE, offsetof(struct flb_loki, http_passwd), + "Set HTTP auth password" + }, + + { + FLB_CONFIG_MAP_STR, "bearer_token", NULL, + 0, FLB_TRUE, offsetof(struct flb_loki, bearer_token), + "Set bearer token auth" + }, + + { + FLB_CONFIG_MAP_STR, "compress", NULL, + 0, FLB_FALSE, 0, + "Set payload compression in network transfer. Option available is 'gzip'" + }, + + /* EOF */ + {0} +}; + +/* for testing */ +static int cb_loki_format_test(struct flb_config *config, + struct flb_input_instance *ins, + void *plugin_context, + void *flush_ctx, + int event_type, + const char *tag, int tag_len, + const void *data, size_t bytes, + void **out_data, size_t *out_size) +{ + int total_records; + flb_sds_t payload = NULL; + flb_sds_t dynamic_tenant_id; + struct flb_loki *ctx = plugin_context; + + dynamic_tenant_id = NULL; + + /* Count number of records */ + total_records = flb_mp_count(data, bytes); + + payload = loki_compose_payload(ctx, total_records, + (char *) tag, tag_len, data, bytes, + &dynamic_tenant_id); + if (payload == NULL) { + if (dynamic_tenant_id != NULL) { + flb_sds_destroy(dynamic_tenant_id); + } + + return -1; + } + + *out_data = payload; + *out_size = flb_sds_len(payload); + + return 0; +} + +/* Plugin reference */ +struct flb_output_plugin out_loki_plugin = { + .name = "loki", + .description = "Loki", + .cb_init = cb_loki_init, + .cb_flush = cb_loki_flush, + .cb_exit = cb_loki_exit, + .config_map = config_map, + + /* for testing */ + .test_formatter.callback = cb_loki_format_test, + + .flags = FLB_OUTPUT_NET | FLB_IO_OPT_TLS, +}; |