diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-05 12:08:03 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-05 12:08:18 +0000 |
commit | 5da14042f70711ea5cf66e034699730335462f66 (patch) | |
tree | 0f6354ccac934ed87a2d555f45be4c831cf92f4a /src/fluent-bit/plugins/processor_attributes/attributes.c | |
parent | Releasing debian version 1.44.3-2. (diff) | |
download | netdata-5da14042f70711ea5cf66e034699730335462f66.tar.xz netdata-5da14042f70711ea5cf66e034699730335462f66.zip |
Merging upstream version 1.45.3+dfsg.
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'src/fluent-bit/plugins/processor_attributes/attributes.c')
-rw-r--r-- | src/fluent-bit/plugins/processor_attributes/attributes.c | 1408 |
1 files changed, 1408 insertions, 0 deletions
diff --git a/src/fluent-bit/plugins/processor_attributes/attributes.c b/src/fluent-bit/plugins/processor_attributes/attributes.c new file mode 100644 index 000000000..a59c07ae2 --- /dev/null +++ b/src/fluent-bit/plugins/processor_attributes/attributes.c @@ -0,0 +1,1408 @@ +/* -*- 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 <stdio.h> +#include <math.h> + +#include <fluent-bit/flb_regex.h> +#include <fluent-bit/flb_filter.h> +#include <fluent-bit/flb_processor_plugin.h> +#include <fluent-bit/flb_utils.h> +#include <fluent-bit/flb_time.h> +#include <fluent-bit/flb_hash.h> +#include <fluent-bit/flb_pack.h> +#include <fluent-bit/flb_processor.h> +#include <fluent-bit/flb_log_event_decoder.h> +#include <fluent-bit/flb_log_event_encoder.h> + +#include <cmetrics/cmetrics.h> +#include <cmetrics/cmt_histogram.h> +#include <cmetrics/cmt_summary.h> +#include <cmetrics/cmt_untyped.h> +#include <cmetrics/cmt_counter.h> +#include <cmetrics/cmt_gauge.h> +#include <cmetrics/cmt_map.h> + +#include <cfl/cfl.h> + +#include "variant_utils.h" + +typedef int (*attribute_transformer)(void *, struct cfl_variant *value); + +struct internal_processor_context { + struct mk_list *update_list; + struct mk_list *insert_list; + struct mk_list *upsert_list; + struct mk_list *convert_list; + struct mk_list *extract_list; + struct mk_list *delete_list; + struct mk_list *hash_list; + + /* internal attributes ready to append */ + struct cfl_list update_attributes; + struct cfl_list insert_attributes; + struct cfl_list upsert_attributes; + struct cfl_list convert_attributes; + struct cfl_list extract_attributes; + struct mk_list delete_attributes; + struct mk_list hash_attributes; + + struct flb_processor_instance *instance; + struct flb_config *config; +}; + +/* + * LOCAL + */ +static int hex_encode(unsigned char *input_buffer, + size_t input_length, + cfl_sds_t *output_buffer) +{ + const char hex[] = "0123456789abcdef"; + cfl_sds_t result; + size_t index; + + if (cfl_sds_alloc(*output_buffer) <= (input_length * 2)) { + result = cfl_sds_increase(*output_buffer, + (input_length * 2) - + cfl_sds_alloc(*output_buffer)); + + if (result == NULL) { + return FLB_FALSE; + } + + *output_buffer = result; + } + + for (index = 0; index < input_length; index++) { + (*output_buffer)[index * 2 + 0] = hex[(input_buffer[index] >> 4) & 0xF]; + (*output_buffer)[index * 2 + 1] = hex[(input_buffer[index] >> 0) & 0xF]; + } + + cfl_sds_set_len(*output_buffer, input_length * 2); + + (*output_buffer)[index * 2] = '\0'; + + return FLB_TRUE; +} + +static int process_attribute_modification_list_setting( + struct flb_processor_instance *plugin_instance, + const char *setting_name, + struct mk_list *source_list, + struct mk_list *destination_list) +{ + struct flb_config_map_val *source_entry; + struct mk_list *iterator; + int result; + + if (source_list == NULL || + mk_list_is_empty(source_list) == 0) { + + return 0; + } + + flb_config_map_foreach(iterator, source_entry, source_list) { + result = flb_slist_add(destination_list, source_entry->val.str); + + if (result != 0) { + flb_plg_error(plugin_instance, + "could not append attribute name %s\n", + source_entry->val.str); + + return -1; + } + } + + return 0; +} + +static int process_attribute_modification_kvlist_setting( + struct flb_processor_instance *plugin_instance, + const char *setting_name, + struct mk_list *source_list, + struct cfl_list *destination_list) +{ + struct cfl_kv *processed_pair; + struct flb_config_map_val *source_entry; + struct mk_list *iterator; + struct flb_slist_entry *value; + struct flb_slist_entry *key; + + if (source_list == NULL || + mk_list_is_empty(source_list) == 0) { + + return 0; + } + + flb_config_map_foreach(iterator, source_entry, source_list) { + if (mk_list_size(source_entry->val.list) != 2) { + flb_plg_error(plugin_instance, + "'%s' expects a key and a value, " + "e.g: '%s version 1.8.0'", + setting_name, setting_name); + + return -1; + } + + key = mk_list_entry_first(source_entry->val.list, + struct flb_slist_entry, _head); + + value = mk_list_entry_last(source_entry->val.list, + struct flb_slist_entry, _head); + + processed_pair = cfl_kv_item_create(destination_list, + key->str, + value->str); + + if (processed_pair == NULL) { + flb_plg_error(plugin_instance, + "could not append attribute %s=%s\n", + key->str, + value->str); + + return -1; + } + } + + return 0; +} + +static void destroy_context(struct internal_processor_context *context) +{ + if (context != NULL) { + cfl_kv_release(&context->update_attributes); + cfl_kv_release(&context->insert_attributes); + cfl_kv_release(&context->upsert_attributes); + cfl_kv_release(&context->convert_attributes); + cfl_kv_release(&context->extract_attributes); + flb_slist_destroy(&context->delete_attributes); + flb_slist_destroy(&context->hash_attributes); + + flb_free(context); + } +} + +static struct internal_processor_context * + create_context(struct flb_processor_instance *processor_instance, + struct flb_config *config) +{ + struct internal_processor_context *context; + int result; + + context = flb_calloc(1, sizeof(struct internal_processor_context)); + + if (context != NULL) { + context->instance = processor_instance; + context->config = config; + + cfl_kv_init(&context->update_attributes); + cfl_kv_init(&context->insert_attributes); + cfl_kv_init(&context->upsert_attributes); + cfl_kv_init(&context->convert_attributes); + cfl_kv_init(&context->extract_attributes); + flb_slist_create(&context->delete_attributes); + flb_slist_create(&context->hash_attributes); + + result = flb_processor_instance_config_map_set(processor_instance, (void *) context); + + if (result == 0) { + result = process_attribute_modification_kvlist_setting( + processor_instance, + "update", + context->update_list, + &context->update_attributes); + } + + if (result == 0) { + result = process_attribute_modification_kvlist_setting( + processor_instance, + "insert", + context->insert_list, + &context->insert_attributes); + } + + if (result == 0) { + result = process_attribute_modification_kvlist_setting( + processor_instance, + "convert", + context->convert_list, + &context->convert_attributes); + } + + if (result == 0) { + result = process_attribute_modification_kvlist_setting( + processor_instance, + "extract", + context->extract_list, + &context->extract_attributes); + } + + if (result == 0) { + result = process_attribute_modification_kvlist_setting( + processor_instance, + "upsert", + context->upsert_list, + &context->upsert_attributes); + } + + if (result == 0) { + result = process_attribute_modification_list_setting( + processor_instance, + "delete", + context->delete_list, + &context->delete_attributes); + } + + if (result == 0) { + result = process_attribute_modification_list_setting( + processor_instance, + "hash", + context->hash_list, + &context->hash_attributes); + } + + if (result != 0) { + destroy_context(context); + + context = NULL; + } + } + else { + flb_errno(); + } + + return context; +} + +static int cb_init(struct flb_processor_instance *processor_instance, + void *source_plugin_instance, + int source_plugin_type, + struct flb_config *config) +{ + processor_instance->context = (void *) create_context( + processor_instance, config); + + if (processor_instance->context == NULL) { + return FLB_PROCESSOR_FAILURE; + } + + return FLB_PROCESSOR_SUCCESS; +} + + +static int cb_exit(struct flb_processor_instance *processor_instance) +{ + if (processor_instance != NULL && + processor_instance->context != NULL) { + destroy_context(processor_instance->context); + } + + return FLB_PROCESSOR_SUCCESS; +} + +static int cfl_kvlist_contains(struct cfl_kvlist *kvlist, + char *name) +{ + struct cfl_list *iterator; + struct cfl_kvpair *pair; + + cfl_list_foreach(iterator, &kvlist->list) { + pair = cfl_list_entry(iterator, + struct cfl_kvpair, _head); + + if (strcasecmp(pair->key, name) == 0) { + return FLB_TRUE; + } + } + + return FLB_FALSE; +} + +static void cfl_kvpair_destroy(struct cfl_kvpair *pair) +{ + if (pair != NULL) { + if (!cfl_list_entry_is_orphan(&pair->_head)) { + cfl_list_del(&pair->_head); + } + + if (pair->key != NULL) { + cfl_sds_destroy(pair->key); + } + + if (pair->val != NULL) { + cfl_variant_destroy(pair->val); + } + + free(pair); + } +} + +static int cfl_kvlist_remove(struct cfl_kvlist *kvlist, + char *name) +{ + struct cfl_list *iterator_backup; + struct cfl_list *iterator; + struct cfl_kvpair *pair; + + cfl_list_foreach_safe(iterator, iterator_backup, &kvlist->list) { + pair = cfl_list_entry(iterator, + struct cfl_kvpair, _head); + + if (strcasecmp(pair->key, name) == 0) { + cfl_kvpair_destroy(pair); + } + } + + return FLB_TRUE; +} + + +/* local declarations */ + + +static cfl_sds_t cfl_variant_convert_to_json(struct cfl_variant *value) +{ + cfl_sds_t json_result; + mpack_writer_t writer; + char *data; + size_t size; + + data = NULL; + size = 0; + + mpack_writer_init_growable(&writer, &data, &size); + + pack_cfl_variant(&writer, value); + + mpack_writer_destroy(&writer); + + json_result = flb_msgpack_raw_to_json_sds(data, size); + + return json_result; +} + + + +static int cfl_variant_convert(struct cfl_variant *input_value, + struct cfl_variant **output_value, + int output_type) +{ + char *converstion_canary; + struct cfl_variant temporary_value; + int errno_backup; + + errno_backup = errno; + *output_value = cfl_variant_create(); + + memset(&temporary_value, 0, sizeof(struct cfl_variant)); + + temporary_value.type = output_type; + + if (input_value->type == CFL_VARIANT_STRING || + input_value->type == CFL_VARIANT_BYTES || + input_value->type == CFL_VARIANT_REFERENCE) { + if (output_type == CFL_VARIANT_STRING || + output_type == CFL_VARIANT_BYTES) { + temporary_value.data.as_string = + cfl_sds_create_len( + input_value->data.as_string, + cfl_sds_len(input_value->data.as_string)); + + if (temporary_value.data.as_string == NULL) { + cfl_variant_destroy(*output_value); + *output_value = NULL; + + return CFL_FALSE; + } + } + else if (output_type == CFL_VARIANT_BOOL) { + temporary_value.data.as_bool = CFL_FALSE; + + if (strcasecmp(input_value->data.as_string, "true") == 0) { + temporary_value.data.as_bool = CFL_TRUE; + } + else if (strcasecmp(input_value->data.as_string, "off") == 0) { + temporary_value.data.as_bool = CFL_TRUE; + } + } + else if (output_type == CFL_VARIANT_INT) { + errno = 0; + temporary_value.data.as_int64 = strtoimax(input_value->data.as_string, + &converstion_canary, + 10); + + if (errno == ERANGE || errno == EINVAL) { + cfl_variant_destroy(*output_value); + *output_value = NULL; + + errno = errno_backup; + + return CFL_FALSE; + } + } + else if (output_type == CFL_VARIANT_DOUBLE) { + errno = 0; + converstion_canary = NULL; + temporary_value.data.as_double = strtod(input_value->data.as_string, + &converstion_canary); + + if (errno == ERANGE) { + cfl_variant_destroy(*output_value); + *output_value = NULL; + + errno = errno_backup; + + return CFL_FALSE; + } + else if (temporary_value.data.as_double == 0 && + converstion_canary == input_value->data.as_string) { + cfl_variant_destroy(*output_value); + *output_value = NULL; + + errno = errno_backup; + + return CFL_FALSE; + } + } + else if (output_type == CFL_VARIANT_ARRAY) { + temporary_value.data.as_array = cfl_array_create(1); + + if (temporary_value.data.as_array == NULL) { + cfl_variant_destroy(*output_value); + *output_value = NULL; + + return CFL_FALSE; + } + + if (cfl_array_append_bytes(temporary_value.data.as_array, + input_value->data.as_bytes, + cfl_sds_len(input_value->data.as_bytes)) != 0) { + cfl_array_destroy(temporary_value.data.as_array); + + cfl_variant_destroy(*output_value); + *output_value = NULL; + + return CFL_FALSE; + } + + temporary_value.data.as_array->entries[0]->type = output_type; + } + else { + return CFL_FALSE; + } + } + else if (input_value->type == CFL_VARIANT_INT) { + if (output_type == CFL_VARIANT_STRING || + output_type == CFL_VARIANT_BYTES) { + temporary_value.data.as_string = cfl_sds_create_size(64); + + if (temporary_value.data.as_string == NULL) { + return CFL_FALSE; + } + + /* We need to fix the wesleys truncation PR to cfl */ + converstion_canary = (char *) cfl_sds_printf( + &temporary_value.data.as_string, + "%" PRIi64, + input_value->data.as_int64); + + if (converstion_canary == NULL) { + cfl_sds_destroy(temporary_value.data.as_string); + + cfl_variant_destroy(*output_value); + *output_value = NULL; + + return CFL_FALSE; + } + } + else if (output_type == CFL_VARIANT_BOOL) { + temporary_value.data.as_bool = CFL_FALSE; + + if (input_value->data.as_int64 != 0) { + temporary_value.data.as_bool = CFL_TRUE; + } + } + else if (output_type == CFL_VARIANT_INT) { + temporary_value.data.as_int64 = input_value->data.as_int64; + } + else if (output_type == CFL_VARIANT_DOUBLE) { + temporary_value.data.as_double = (double) input_value->data.as_int64; + + /* This conversion could be lossy, we need to determine what we want to + * do in that case + */ + if ((int64_t) temporary_value.data.as_double != input_value->data.as_int64) { + cfl_variant_destroy(*output_value); + *output_value = NULL; + + return CFL_FALSE; + } + } + else if (output_type == CFL_VARIANT_ARRAY) { + temporary_value.data.as_array = cfl_array_create(1); + + if (temporary_value.data.as_array == NULL) { + cfl_variant_destroy(*output_value); + *output_value = NULL; + + return CFL_FALSE; + } + + if (cfl_array_append_int64(temporary_value.data.as_array, + input_value->data.as_int64) != 0) { + cfl_array_destroy(temporary_value.data.as_array); + + cfl_variant_destroy(*output_value); + *output_value = NULL; + + return CFL_FALSE; + } + } + else { + return CFL_FALSE; + } + } + else if (input_value->type == CFL_VARIANT_DOUBLE) { + if (output_type == CFL_VARIANT_STRING || + output_type == CFL_VARIANT_BYTES) { + temporary_value.data.as_string = cfl_sds_create_size(64); + + if (temporary_value.data.as_string == NULL) { + return CFL_FALSE; + } + + /* We need to fix the wesleys truncation PR to cfl */ + converstion_canary = (char *) cfl_sds_printf( + &temporary_value.data.as_string, + "%.17g", + input_value->data.as_double); + + if (converstion_canary == NULL) { + cfl_sds_destroy(temporary_value.data.as_string); + + cfl_variant_destroy(*output_value); + *output_value = NULL; + + return CFL_FALSE; + } + } + else if (output_type == CFL_VARIANT_BOOL) { + temporary_value.data.as_bool = CFL_FALSE; + + if (input_value->data.as_double != 0) { + temporary_value.data.as_bool = CFL_TRUE; + } + } + else if (output_type == CFL_VARIANT_INT) { + temporary_value.data.as_int64 = (int64_t) round(input_value->data.as_double); + } + else if (output_type == CFL_VARIANT_DOUBLE) { + temporary_value.data.as_double = input_value->data.as_int64; + } + else if (output_type == CFL_VARIANT_ARRAY) { + temporary_value.data.as_array = cfl_array_create(1); + + if (temporary_value.data.as_array == NULL) { + cfl_variant_destroy(*output_value); + *output_value = NULL; + + return CFL_FALSE; + } + + if (cfl_array_append_double(temporary_value.data.as_array, + input_value->data.as_double) != 0) { + cfl_array_destroy(temporary_value.data.as_array); + + cfl_variant_destroy(*output_value); + *output_value = NULL; + + return CFL_FALSE; + } + } + else { + return CFL_FALSE; + } + } + else if (input_value->type == CFL_VARIANT_KVLIST) { + if (output_type == CFL_VARIANT_STRING || + output_type == CFL_VARIANT_BYTES) { + temporary_value.data.as_string = cfl_variant_convert_to_json(input_value); + + if (temporary_value.data.as_string == NULL) { + return CFL_FALSE; + } + } + else { + return CFL_FALSE; + } + } + else if (input_value->type == CFL_VARIANT_ARRAY) { + if (output_type == CFL_VARIANT_STRING || + output_type == CFL_VARIANT_BYTES) { + temporary_value.data.as_string = cfl_variant_convert_to_json(input_value); + + if (temporary_value.data.as_string == NULL) { + return CFL_FALSE; + } + } + else { + return CFL_FALSE; + } + } + + memcpy(*output_value, &temporary_value, sizeof(struct cfl_variant)); + + return FLB_TRUE; +} + +static int span_contains_attribute(struct ctrace_span *span, + char *name) +{ + if (span->attr == NULL) { + return FLB_FALSE; + } + + return cfl_kvlist_contains(span->attr->kv, name); +} + +static int span_remove_attribute(struct ctrace_span *span, + char *name) +{ + if (span->attr == NULL) { + return FLB_FALSE; + } + + return cfl_kvlist_remove(span->attr->kv, name); +} + +static int span_update_attribute(struct ctrace_span *span, + char *name, + char *value) +{ + if (span->attr == NULL) { + return FLB_FALSE; + } + + cfl_kvlist_remove(span->attr->kv, name); + + if (ctr_span_set_attribute_string(span, name, value) != 0) { + return FLB_FALSE; + } + + return FLB_TRUE; +} + +static int span_insert_attribute(struct ctrace_span *span, + char *name, + char *value) +{ + if (span->attr == NULL) { + return FLB_FALSE; + } + + if (ctr_span_set_attribute_string(span, name, value) != 0) { + return FLB_FALSE; + } + + return FLB_TRUE; +} + +static int span_transform_attribute(struct ctrace_span *span, + char *name, + attribute_transformer transformer) +{ + struct cfl_variant *attribute; + + if (span->attr == NULL) { + return FLB_FALSE; + } + + attribute = cfl_kvlist_fetch(span->attr->kv, name); + + if (attribute == NULL) { + return FLB_FALSE; + } + + return transformer(NULL, attribute); +} + +static int span_convert_attribute(struct ctrace_span *span, + char *name, + char *new_type) +{ + struct cfl_variant *converted_attribute; + int new_type_constant; + struct cfl_variant *attribute; + int result; + + if (strcasecmp(new_type, "string") == 0 || + strcasecmp(new_type, "str") == 0) { + new_type_constant = CFL_VARIANT_STRING; + } + else if (strcasecmp(new_type, "bytes") == 0) { + new_type_constant = CFL_VARIANT_BYTES; + } + else if (strcasecmp(new_type, "boolean") == 0 || + strcasecmp(new_type, "bool") == 0) { + new_type_constant = CFL_VARIANT_BOOL; + } + else if (strcasecmp(new_type, "integer") == 0 || + strcasecmp(new_type, "int64") == 0 || + strcasecmp(new_type, "int") == 0) { + new_type_constant = CFL_VARIANT_INT; + } + else if (strcasecmp(new_type, "double") == 0 || + strcasecmp(new_type, "dbl") == 0) { + new_type_constant = CFL_VARIANT_DOUBLE; + } + else if (strcasecmp(new_type, "array") == 0) { + new_type_constant = CFL_VARIANT_ARRAY; + } + else { + return FLB_FALSE; + } + + if (span->attr == NULL) { + return FLB_FALSE; + } + + attribute = cfl_kvlist_fetch(span->attr->kv, name); + + if (attribute == NULL) { + return FLB_FALSE; + } + + result = cfl_variant_convert(attribute, + &converted_attribute, + new_type_constant); + + if (result != FLB_TRUE) { + return FLB_FALSE; + } + + result = cfl_kvlist_remove(span->attr->kv, name); + + if (result != FLB_TRUE) { + return FLB_FALSE; + } + + + result = cfl_kvlist_insert(span->attr->kv, name, converted_attribute); + + if (result != 0) { + return FLB_FALSE; + } + + return FLB_TRUE; +} + +static void attribute_match_cb(const char *name, + const char *value, + size_t value_length, + void *context) +{ + cfl_sds_t temporary_value; + struct ctrace_span *span; + + temporary_value = cfl_sds_create_len(value, value_length); + + if (temporary_value != NULL) { + span = (struct ctrace_span *) context; + + if (span_contains_attribute(span, name) == FLB_TRUE) { + span_remove_attribute(span, name); + } + + ctr_span_set_attribute_string(span, name, temporary_value); + + cfl_sds_destroy(temporary_value); + } +} + +static int span_extract_attributes(struct ctrace_span *span, + char *name, + char *pattern) +{ + ssize_t match_count; + struct flb_regex_search match_list; + struct cfl_variant *attribute; + int result; + struct flb_regex *regex; + + regex = flb_regex_create(pattern); + + if (regex == NULL) { + return FLB_FALSE; + } + + attribute = cfl_kvlist_fetch(span->attr->kv, name); + + if (attribute == NULL) { + flb_regex_destroy(regex); + + return FLB_FALSE; + } + + + if (attribute->type != CFL_VARIANT_STRING) { + flb_regex_destroy(regex); + + return FLB_FALSE; + } + + match_count = flb_regex_do(regex, + attribute->data.as_string, + cfl_sds_len(attribute->data.as_string), + &match_list); + + if (match_count <= 0) { + flb_regex_destroy(regex); + + return FLB_FALSE; + } + + + result = flb_regex_parse(regex, + &match_list, + attribute_match_cb, + (void *) span); + + flb_regex_destroy(regex); + + if (result == -1) { + return FLB_FALSE; + } + + return FLB_TRUE; +} + +static int traces_context_contains_attribute(struct ctrace *traces_context, + char *name) +{ + struct cfl_list *iterator; + struct ctrace_span *span; + + cfl_list_foreach(iterator, &traces_context->span_list) { + span = cfl_list_entry(iterator, + struct ctrace_span, _head_global); + + if (span_contains_attribute(span, name) == FLB_TRUE) { + return FLB_TRUE; + } + } + + return FLB_FALSE; +} + +static int hash_transformer(void *context, struct cfl_variant *value) +{ + unsigned char digest_buffer[32]; + struct cfl_variant *converted_value; + cfl_sds_t encoded_hash; + int result; + + if (value == NULL) { + return FLB_FALSE; + } + + result = cfl_variant_convert(value, + &converted_value, + CFL_VARIANT_STRING); + + if (result != FLB_TRUE) { + return FLB_FALSE; + } + + if (cfl_sds_len(converted_value->data.as_string) == 0) { + cfl_variant_destroy(converted_value); + + return FLB_TRUE; + } + + result = flb_hash_simple(FLB_HASH_SHA256, + (unsigned char *) converted_value->data.as_string, + cfl_sds_len(converted_value->data.as_string), + digest_buffer, + sizeof(digest_buffer)); + + if (result != FLB_CRYPTO_SUCCESS) { + cfl_variant_destroy(converted_value); + + return FLB_FALSE; + } + + result = hex_encode(digest_buffer, + sizeof(digest_buffer), + &converted_value->data.as_string); + + if (result != FLB_TRUE) { + cfl_variant_destroy(converted_value); + + return FLB_FALSE; + } + + encoded_hash = cfl_sds_create(converted_value->data.as_string); + + if (encoded_hash == NULL) { + cfl_variant_destroy(converted_value); + + return FLB_FALSE; + } + + if (value->type == CFL_VARIANT_STRING || + value->type == CFL_VARIANT_BYTES) { + cfl_sds_destroy(value->data.as_string); + } + else if (value->type == CFL_VARIANT_ARRAY) { + cfl_array_destroy(value->data.as_array); + } + else if (value->type == CFL_VARIANT_KVLIST) { + cfl_kvlist_destroy(value->data.as_kvlist); + } + + value->type = CFL_VARIANT_STRING; + value->data.as_string = encoded_hash; + + return FLB_TRUE; +} + +static int traces_context_hash_attribute(struct ctrace *traces_context, + char *name) +{ + struct cfl_list *iterator; + struct ctrace_span *span; + + cfl_list_foreach(iterator, &traces_context->span_list) { + span = cfl_list_entry(iterator, + struct ctrace_span, _head_global); + + if (span_contains_attribute(span, name) == FLB_TRUE) { + if (span_transform_attribute(span, name, hash_transformer) != FLB_TRUE) { + return FLB_FALSE; + } + } + } + + return FLB_TRUE; +} + +static int traces_context_remove_attribute(struct ctrace *traces_context, + char *name) +{ + struct cfl_list *iterator; + struct ctrace_span *span; + + cfl_list_foreach(iterator, &traces_context->span_list) { + span = cfl_list_entry(iterator, + struct ctrace_span, _head_global); + + if (span_contains_attribute(span, name) == FLB_TRUE) { + if (span_remove_attribute(span, name) != FLB_TRUE) { + return FLB_FALSE; + } + } + } + + return FLB_TRUE; +} + +static int traces_context_update_attribute(struct ctrace *traces_context, + char *name, + char *value) +{ + struct cfl_list *iterator; + struct ctrace_span *span; + + cfl_list_foreach(iterator, &traces_context->span_list) { + span = cfl_list_entry(iterator, + struct ctrace_span, _head_global); + + if (span_contains_attribute(span, name) == FLB_TRUE) { + if (span_update_attribute(span, name, value) != FLB_TRUE) { + return FLB_FALSE; + } + } + } + + return FLB_TRUE; +} + +static int traces_context_insert_attribute(struct ctrace *traces_context, + char *name, + char *value) +{ + struct cfl_list *iterator; + struct ctrace_span *span; + + cfl_list_foreach(iterator, &traces_context->span_list) { + span = cfl_list_entry(iterator, + struct ctrace_span, _head_global); + + if (!span_contains_attribute(span, name) == FLB_TRUE) { + if (span_insert_attribute(span, name, value) != FLB_TRUE) { + return FLB_FALSE; + } + } + } + + return FLB_TRUE; +} + +static int traces_context_upsert_attribute(struct ctrace *traces_context, + char *name, + char *value) +{ + struct cfl_list *iterator; + struct ctrace_span *span; + + cfl_list_foreach(iterator, &traces_context->span_list) { + span = cfl_list_entry(iterator, + struct ctrace_span, _head_global); + + if (span_contains_attribute(span, name) == FLB_TRUE) { + if (span_update_attribute(span, name, value) != FLB_TRUE) { + return FLB_FALSE; + } + } + else { + if (span_insert_attribute(span, name, value) != FLB_TRUE) { + return FLB_FALSE; + } + } + } + + return FLB_TRUE; +} + +static int traces_context_convert_attribute(struct ctrace *traces_context, + char *name, + char *new_type) +{ + struct cfl_list *iterator; + struct ctrace_span *span; + + cfl_list_foreach(iterator, &traces_context->span_list) { + span = cfl_list_entry(iterator, + struct ctrace_span, _head_global); + + if (span_contains_attribute(span, name) == FLB_TRUE) { + if (span_convert_attribute(span, name, new_type) != FLB_TRUE) { + return FLB_FALSE; + } + } + } + + return FLB_TRUE; +} + +static int traces_context_extract_attribute(struct ctrace *traces_context, + char *name, + char *pattern) +{ + struct cfl_list *iterator; + struct ctrace_span *span; + + cfl_list_foreach(iterator, &traces_context->span_list) { + span = cfl_list_entry(iterator, + struct ctrace_span, _head_global); + + if (span_contains_attribute(span, name) == FLB_TRUE) { + if (span_extract_attributes(span, name, pattern) != FLB_TRUE) { + return FLB_FALSE; + } + } + } + + return FLB_TRUE; +} + +static int delete_attributes(struct ctrace *traces_context, + struct mk_list *attributes) +{ + struct mk_list *iterator; + int result; + struct flb_slist_entry *entry; + + mk_list_foreach(iterator, attributes) { + entry = mk_list_entry(iterator, struct flb_slist_entry, _head); + + result = traces_context_contains_attribute(traces_context, + entry->str); + + if (result == FLB_TRUE) { + result = traces_context_remove_attribute(traces_context, + entry->str); + + if (result == FLB_FALSE) { + return FLB_PROCESSOR_FAILURE; + } + } + } + + return FLB_PROCESSOR_SUCCESS; +} + +static int update_attributes(struct ctrace *traces_context, + struct cfl_list *attributes) +{ + struct cfl_list *iterator; + int result; + struct cfl_kv *pair; + + cfl_list_foreach(iterator, attributes) { + pair = cfl_list_entry(iterator, struct cfl_kv, _head); + + result = traces_context_update_attribute(traces_context, + pair->key, + pair->val); + + if (result == FLB_FALSE) { + return FLB_PROCESSOR_FAILURE; + } + } + + return FLB_PROCESSOR_SUCCESS; +} + +static int upsert_attributes(struct ctrace *traces_context, + struct cfl_list *attributes) +{ + struct cfl_list *iterator; + int result; + struct cfl_kv *pair; + + cfl_list_foreach(iterator, attributes) { + pair = cfl_list_entry(iterator, struct cfl_kv, _head); + + result = traces_context_upsert_attribute(traces_context, + pair->key, + pair->val); + + if (result == FLB_FALSE) { + return FLB_PROCESSOR_FAILURE; + } + } + + return FLB_PROCESSOR_SUCCESS; +} + +static int convert_attributes(struct ctrace *traces_context, + struct cfl_list *attributes) +{ + struct cfl_list *iterator; + int result; + struct cfl_kv *pair; + + cfl_list_foreach(iterator, attributes) { + pair = cfl_list_entry(iterator, struct cfl_kv, _head); + + result = traces_context_convert_attribute(traces_context, + pair->key, + pair->val); + + if (result == FLB_FALSE) { + return FLB_PROCESSOR_FAILURE; + } + } + + return FLB_PROCESSOR_SUCCESS; +} + +static int extract_attributes(struct ctrace *traces_context, + struct cfl_list *attributes) +{ + struct cfl_list *iterator; + int result; + struct cfl_kv *pair; + + cfl_list_foreach(iterator, attributes) { + pair = cfl_list_entry(iterator, struct cfl_kv, _head); + + result = traces_context_extract_attribute(traces_context, + pair->key, + pair->val); + + if (result == FLB_FALSE) { + return FLB_PROCESSOR_FAILURE; + } + } + + return FLB_PROCESSOR_SUCCESS; +} + +static int insert_attributes(struct ctrace *traces_context, + struct cfl_list *attributes) +{ + struct cfl_list *iterator; + int result; + struct cfl_kv *pair; + + cfl_list_foreach(iterator, attributes) { + pair = cfl_list_entry(iterator, struct cfl_kv, _head); + + result = traces_context_insert_attribute(traces_context, + pair->key, + pair->val); + + if (result == FLB_FALSE) { + return FLB_PROCESSOR_FAILURE; + } + } + + return FLB_PROCESSOR_SUCCESS; +} + +static int hash_attributes(struct ctrace *traces_context, + struct mk_list *attributes) +{ + struct mk_list *iterator; + int result; + struct flb_slist_entry *entry; + + mk_list_foreach(iterator, attributes) { + entry = mk_list_entry(iterator, struct flb_slist_entry, _head); + + result = traces_context_contains_attribute(traces_context, + entry->str); + + if (result == FLB_TRUE) { + result = traces_context_hash_attribute(traces_context, + entry->str); + + if (result == FLB_FALSE) { + return FLB_PROCESSOR_FAILURE; + } + } + } + + return FLB_PROCESSOR_SUCCESS; +} + +static int cb_process_traces(struct flb_processor_instance *processor_instance, + struct ctrace *traces_context, + const char *tag, + int tag_len) +{ + struct internal_processor_context *processor_context; + int result; + + processor_context = + (struct internal_processor_context *) processor_instance->context; + + result = delete_attributes(traces_context, + &processor_context->delete_attributes); + + if (result == FLB_PROCESSOR_SUCCESS) { + result = update_attributes(traces_context, + &processor_context->update_attributes); + } + + if (result == FLB_PROCESSOR_SUCCESS) { + result = upsert_attributes(traces_context, + &processor_context->upsert_attributes); + } + + if (result == FLB_PROCESSOR_SUCCESS) { + result = insert_attributes(traces_context, + &processor_context->insert_attributes); + } + + if (result == FLB_PROCESSOR_SUCCESS) { + result = convert_attributes(traces_context, + &processor_context->convert_attributes); + result = FLB_PROCESSOR_SUCCESS; + } + + if (result == FLB_PROCESSOR_SUCCESS) { + result = extract_attributes(traces_context, + &processor_context->extract_attributes); + } + + if (result == FLB_PROCESSOR_SUCCESS) { + result = hash_attributes(traces_context, + &processor_context->hash_attributes); + } + + if (result != FLB_PROCESSOR_SUCCESS) { + return FLB_PROCESSOR_FAILURE; + } + + return FLB_PROCESSOR_SUCCESS; +} + +static struct flb_config_map config_map[] = { + { + FLB_CONFIG_MAP_SLIST_1, "update", NULL, + FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct internal_processor_context, + update_list), + "Updates an attribute. Usage : 'update name value'" + }, + { + FLB_CONFIG_MAP_SLIST_1, "insert", NULL, + FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct internal_processor_context, + insert_list), + "Inserts an attribute. Usage : 'insert name value'" + }, + { + FLB_CONFIG_MAP_SLIST_1, "upsert", NULL, + FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct internal_processor_context, + upsert_list), + "Inserts or updates an attribute. Usage : 'upsert name value'" + }, + { + FLB_CONFIG_MAP_SLIST_1, "convert", NULL, + FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct internal_processor_context, + convert_list), + "Converts an attribute. Usage : 'convert name new_type'" + }, + { + FLB_CONFIG_MAP_SLIST_1, "extract", NULL, + FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct internal_processor_context, + extract_list), + "Extracts regular expression match groups as individual attributes. Usage : 'extract (?P<first_word>[^ ]*) (?P<second_word>[^ ]*)'" + }, + { + FLB_CONFIG_MAP_STR, "delete", NULL, + FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct internal_processor_context, + delete_list), + "Deletes an attribute. Usage : 'delete name'" + }, + { + FLB_CONFIG_MAP_STR, "hash", NULL, + FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct internal_processor_context, + hash_list), + "Replaces an attributes value with its SHA256 hash. Usage : 'hash name'" + }, + + /* EOF */ + {0} +}; + +struct flb_processor_plugin processor_attributes_plugin = { + .name = "attributes", + .description = "Modifies metrics attributes", + .cb_init = cb_init, + .cb_process_logs = NULL, + .cb_process_metrics = NULL, + .cb_process_traces = cb_process_traces, + .cb_exit = cb_exit, + .config_map = config_map, + .flags = 0 +}; |