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, 0 insertions, 1868 deletions
diff --git a/src/fluent-bit/plugins/out_loki/loki.c b/src/fluent-bit/plugins/out_loki/loki.c deleted file mode 100644 index d93a3f9aa..000000000 --- a/src/fluent-bit/plugins/out_loki/loki.c +++ /dev/null @@ -1,1868 +0,0 @@ -/* -*- 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, -}; |