diff options
Diffstat (limited to 'src/fluent-bit/plugins/out_cloudwatch_logs/cloudwatch_api.c')
| -rw-r--r-- | src/fluent-bit/plugins/out_cloudwatch_logs/cloudwatch_api.c | 1564 |
1 files changed, 0 insertions, 1564 deletions
diff --git a/src/fluent-bit/plugins/out_cloudwatch_logs/cloudwatch_api.c b/src/fluent-bit/plugins/out_cloudwatch_logs/cloudwatch_api.c deleted file mode 100644 index 8043968cf..000000000 --- a/src/fluent-bit/plugins/out_cloudwatch_logs/cloudwatch_api.c +++ /dev/null @@ -1,1564 +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_compat.h> -#include <fluent-bit/flb_info.h> -#include <fluent-bit/flb_output.h> -#include <fluent-bit/flb_utils.h> -#include <fluent-bit/flb_slist.h> -#include <fluent-bit/flb_time.h> -#include <fluent-bit/flb_pack.h> -#include <fluent-bit/flb_macros.h> -#include <fluent-bit/flb_config_map.h> -#include <fluent-bit/flb_output_plugin.h> -#include <fluent-bit/flb_log_event_decoder.h> - -#include <fluent-bit/flb_sds.h> -#include <fluent-bit/flb_aws_credentials.h> -#include <fluent-bit/flb_aws_util.h> -#include <fluent-bit/flb_mem.h> -#include <fluent-bit/flb_info.h> -#include <fluent-bit/flb_http_client.h> -#include <fluent-bit/flb_utils.h> -#include <fluent-bit/flb_intermediate_metric.h> - -#include <monkey/mk_core.h> -#include <msgpack.h> -#include <string.h> -#include <stdio.h> - -#ifndef FLB_SYSTEM_WINDOWS -#include <unistd.h> -#endif - -#include "cloudwatch_api.h" - -#define ERR_CODE_ALREADY_EXISTS "ResourceAlreadyExistsException" -#define ERR_CODE_INVALID_SEQUENCE_TOKEN "InvalidSequenceTokenException" -#define ERR_CODE_NOT_FOUND "ResourceNotFoundException" -#define ERR_CODE_DATA_ALREADY_ACCEPTED "DataAlreadyAcceptedException" - -#define AMZN_REQUEST_ID_HEADER "x-amzn-RequestId" - -#define ONE_DAY_IN_MILLISECONDS 86400000 -#define FOUR_HOURS_IN_SECONDS 14400 - - -static struct flb_aws_header create_group_header = { - .key = "X-Amz-Target", - .key_len = 12, - .val = "Logs_20140328.CreateLogGroup", - .val_len = 28, -}; - -static struct flb_aws_header put_retention_policy_header = { - .key = "X-Amz-Target", - .key_len = 12, - .val = "Logs_20140328.PutRetentionPolicy", - .val_len = 32, -}; - -static struct flb_aws_header create_stream_header = { - .key = "X-Amz-Target", - .key_len = 12, - .val = "Logs_20140328.CreateLogStream", - .val_len = 29, -}; - -static struct flb_aws_header put_log_events_header[] = { - { - .key = "X-Amz-Target", - .key_len = 12, - .val = "Logs_20140328.PutLogEvents", - .val_len = 26, - }, - { - .key = "x-amzn-logs-format", - .key_len = 18, - .val = "", - .val_len = 0, - }, -}; - -int plugin_under_test() -{ - if (getenv("FLB_CLOUDWATCH_PLUGIN_UNDER_TEST") != NULL) { - return FLB_TRUE; - } - - return FLB_FALSE; -} - -char *mock_error_response(char *error_env_var) -{ - char *err_val = NULL; - char *error = NULL; - int len = 0; - - err_val = getenv(error_env_var); - if (err_val != NULL && strlen(err_val) > 0) { - error = flb_malloc(strlen(err_val) + sizeof(char)); - if (error == NULL) { - flb_errno(); - return NULL; - } - - len = strlen(err_val); - memcpy(error, err_val, len); - error[len] = '\0'; - return error; - } - - return NULL; -} - -struct flb_http_client *mock_http_call(char *error_env_var, char *api) -{ - /* create an http client so that we can set the response */ - struct flb_http_client *c = NULL; - char *error = mock_error_response(error_env_var); - - c = flb_calloc(1, sizeof(struct flb_http_client)); - if (!c) { - flb_errno(); - flb_free(error); - return NULL; - } - mk_list_init(&c->headers); - - if (error != NULL) { - c->resp.status = 400; - /* resp.data is freed on destroy, payload is supposed to reference it */ - c->resp.data = error; - c->resp.payload = c->resp.data; - c->resp.payload_size = strlen(error); - } - else { - c->resp.status = 200; - c->resp.payload = ""; - c->resp.payload_size = 0; - if (strcmp(api, "PutLogEvents") == 0) { - /* mocked success response */ - c->resp.payload = "{\"nextSequenceToken\": \"" - "49536701251539826331025683274032969384950891766572122113\"}"; - c->resp.payload_size = strlen(c->resp.payload); - } - else { - c->resp.payload = ""; - c->resp.payload_size = 0; - } - } - - return c; -} - -int compare_events(const void *a_arg, const void *b_arg) -{ - struct cw_event *r_a = (struct cw_event *) a_arg; - struct cw_event *r_b = (struct cw_event *) b_arg; - - if (r_a->timestamp < r_b->timestamp) { - return -1; - } - else if (r_a->timestamp == r_b->timestamp) { - return 0; - } - else { - return 1; - } -} - -static inline int try_to_write(char *buf, int *off, size_t left, - const char *str, size_t str_len) -{ - if (str_len <= 0){ - str_len = strlen(str); - } - if (left <= *off+str_len) { - return FLB_FALSE; - } - memcpy(buf+*off, str, str_len); - *off += str_len; - return FLB_TRUE; -} - -/* - * Writes the "header" for a put log events payload - */ -static int init_put_payload(struct flb_cloudwatch *ctx, struct cw_flush *buf, - struct log_stream *stream, int *offset) -{ - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - "{\"logGroupName\":\"", 17)) { - goto error; - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - stream->group, 0)) { - goto error; - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - "\",\"logStreamName\":\"", 19)) { - goto error; - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - stream->name, 0)) { - goto error; - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - "\",", 2)) { - goto error; - } - - if (stream->sequence_token) { - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - "\"sequenceToken\":\"", 17)) { - goto error; - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - stream->sequence_token, 0)) { - goto error; - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - "\",", 2)) { - goto error; - } - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - "\"logEvents\":[", 13)) { - goto error; - } - - return 0; - -error: - return -1; -} - -/* - * Writes a log event to the output buffer - */ -static int write_event(struct flb_cloudwatch *ctx, struct cw_flush *buf, - struct cw_event *event, int *offset) -{ - char ts[50]; - - if (!snprintf(ts, 50, "%llu", event->timestamp)) { - goto error; - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - "{\"timestamp\":", 13)) { - goto error; - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - ts, 0)) { - goto error; - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - ",\"message\":\"", 12)) { - goto error; - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - event->json, event->len)) { - goto error; - } - - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - "\"}", 2)) { - goto error; - } - - return 0; - -error: - return -1; -} - -/* Terminates a PutLogEvents payload */ -static int end_put_payload(struct flb_cloudwatch *ctx, struct cw_flush *buf, - int *offset) -{ - if (!try_to_write(buf->out_buf, offset, buf->out_buf_size, - "]}", 2)) { - return -1; - } - buf->out_buf[*offset] = '\0'; - - return 0; -} - -static unsigned long long stream_time_span(struct log_stream *stream, - struct cw_event *event) -{ - if (stream->oldest_event == 0 || stream->newest_event == 0) { - return 0; - } - - if (stream->oldest_event > event->timestamp) { - return stream->newest_event - event->timestamp; - } - else if (stream->newest_event < event->timestamp) { - return event->timestamp - stream->oldest_event; - } - - return stream->newest_event - stream->oldest_event; -} - -/* returns FLB_TRUE if time span is less than 24 hours, FLB_FALSE if greater */ -static int check_stream_time_span(struct log_stream *stream, - struct cw_event *event) -{ - unsigned long long span = stream_time_span(stream, event); - - if (span < ONE_DAY_IN_MILLISECONDS) { - return FLB_TRUE; - } - - return FLB_FALSE; -} - -/* sets the oldest_event and newest_event fields */ -static void set_stream_time_span(struct log_stream *stream, struct cw_event *event) -{ - if (stream->oldest_event == 0 || stream->oldest_event > event->timestamp) { - stream->oldest_event = event->timestamp; - } - - if (stream->newest_event == 0 || stream->newest_event < event->timestamp) { - stream->newest_event = event->timestamp; - } -} - -/* - * Truncate log if needed. If truncated, only `written` is modified - * returns FLB_TRUE if truncated - */ -static int truncate_log(const struct flb_cloudwatch *ctx, const char *log_buffer, - size_t *written) { - size_t trailing_backslash_count = 0; - - if (*written > MAX_EVENT_LEN) { - flb_plg_warn(ctx->ins, "[size=%zu] Truncating event which is larger than " - "max size allowed by CloudWatch", *written); - *written = MAX_EVENT_LEN; - - /* remove trailing unescaped backslash if inadvertently synthesized */ - while (trailing_backslash_count < *written && - log_buffer[(*written - 1) - trailing_backslash_count] == '\\') { - trailing_backslash_count++; - } - if (trailing_backslash_count % 2 == 1) { - /* odd number of trailing backslashes, remove unpaired backslash */ - (*written)--; - } - return FLB_TRUE; - } - return FLB_FALSE; -} - - -/* - * Processes the msgpack object - * -1 = failure, record not added - * 0 = success, record added - * 1 = we ran out of space, send and retry - * 2 = record could not be processed, discard it - * Returns 0 on success, -1 on general errors, - * and 1 if we ran out of space to write the event - * which means a send must occur - */ -int process_event(struct flb_cloudwatch *ctx, struct cw_flush *buf, - const msgpack_object *obj, struct flb_time *tms) -{ - size_t written; - int ret; - size_t size; - int offset = 0; - struct cw_event *event; - char *tmp_buf_ptr; - - tmp_buf_ptr = buf->tmp_buf + buf->tmp_buf_offset; - ret = flb_msgpack_to_json(tmp_buf_ptr, - buf->tmp_buf_size - buf->tmp_buf_offset, - obj); - if (ret <= 0) { - /* - * failure to write to buffer, - * which means we ran out of space, and must send the logs - */ - return 1; - } - written = (size_t) ret; - /* Discard empty messages (written == 2 means '""') */ - if (written <= 2) { - flb_plg_debug(ctx->ins, "Found empty log message"); - return 2; - } - - /* the json string must be escaped, unless the log_key option is used */ - if (ctx->log_key == NULL) { - /* - * check if event_buf is initialized and big enough - * If all chars need to be hex encoded (impossible), 6x space would be - * needed - */ - size = written * 6; - if (buf->event_buf == NULL || buf->event_buf_size < size) { - flb_free(buf->event_buf); - buf->event_buf = flb_malloc(size); - buf->event_buf_size = size; - if (buf->event_buf == NULL) { - flb_errno(); - return -1; - } - } - offset = 0; - if (!flb_utils_write_str(buf->event_buf, &offset, size, - tmp_buf_ptr, written)) { - return -1; - } - written = offset; - - tmp_buf_ptr = buf->tmp_buf + buf->tmp_buf_offset; - if ((buf->tmp_buf_size - buf->tmp_buf_offset) < written) { - /* not enough space, send logs */ - return 1; - } - - /* truncate log, if needed */ - truncate_log(ctx, buf->event_buf, &written); - - /* copy serialized json to tmp_buf */ - if (!strncpy(tmp_buf_ptr, buf->event_buf, written)) { - return -1; - } - } - else { - /* - * flb_msgpack_to_json will encase the value in quotes - * We don't want that for log_key, so we ignore the first - * and last character - */ - written -= 2; - tmp_buf_ptr++; /* pass over the opening quote */ - buf->tmp_buf_offset++; /* advance tmp_buf past opening quote */ - - /* truncate log, if needed */ - truncate_log(ctx, tmp_buf_ptr, &written); - } - - /* add log to events list */ - buf->tmp_buf_offset += written; - event = &buf->events[buf->event_index]; - event->json = tmp_buf_ptr; - event->len = written; - event->timestamp = (unsigned long long) (tms->tm.tv_sec * 1000ull + - tms->tm.tv_nsec/1000000); - - return 0; -} - -/* Resets or inits a cw_flush struct */ -void reset_flush_buf(struct flb_cloudwatch *ctx, struct cw_flush *buf) { - buf->event_index = 0; - buf->tmp_buf_offset = 0; - buf->event_index = 0; - buf->data_size = PUT_LOG_EVENTS_HEADER_LEN + PUT_LOG_EVENTS_FOOTER_LEN; - if (buf->current_stream != NULL) { - buf->data_size += strlen(buf->current_stream->name); - buf->data_size += strlen(buf->current_stream->group); - if (buf->current_stream->sequence_token) { - buf->data_size += strlen(buf->current_stream->sequence_token); - } - } -} - -/* sorts events, constructs a put payload, and then sends */ -int send_log_events(struct flb_cloudwatch *ctx, struct cw_flush *buf) { - int ret; - int offset; - int i; - struct cw_event *event; - - if (buf->event_index <= 0) { - return 0; - } - - /* events must be sorted by timestamp in a put payload */ - qsort(buf->events, buf->event_index, sizeof(struct cw_event), compare_events); - -retry: - buf->current_stream->newest_event = 0; - buf->current_stream->oldest_event = 0; - - offset = 0; - ret = init_put_payload(ctx, buf, buf->current_stream, &offset); - if (ret < 0) { - flb_plg_error(ctx->ins, "Failed to initialize PutLogEvents payload"); - return -1; - } - - for (i = 0; i < buf->event_index; i++) { - event = &buf->events[i]; - ret = write_event(ctx, buf, event, &offset); - if (ret < 0) { - flb_plg_error(ctx->ins, "Failed to write log event %d to " - "payload buffer", i); - return -1; - } - if (i != (buf->event_index - 1)) { - if (!try_to_write(buf->out_buf, &offset, buf->out_buf_size, - ",", 1)) { - flb_plg_error(ctx->ins, "Could not terminate log event with ','"); - return -1; - } - } - } - - ret = end_put_payload(ctx, buf, &offset); - if (ret < 0) { - flb_plg_error(ctx->ins, "Could not complete PutLogEvents payload"); - return -1; - } - - flb_plg_debug(ctx->ins, "cloudwatch:PutLogEvents: events=%d, payload=%d bytes", i, offset); - ret = put_log_events(ctx, buf, buf->current_stream, (size_t) offset); - if (ret < 0) { - flb_plg_error(ctx->ins, "Failed to send log events"); - return -1; - } - else if (ret > 0) { - goto retry; - } - - return 0; -} - - /* - * Processes the msgpack object, sends the current batch if needed - * -1 = failure, event not added - * 0 = success, event added - * 1 = event been skipped - * Returns 0 on success, -1 on general errors, - * and 1 if we found a empty event or a large event. - */ -int add_event(struct flb_cloudwatch *ctx, struct cw_flush *buf, - struct log_stream *stream, - const msgpack_object *obj, struct flb_time *tms) -{ - int ret; - struct cw_event *event; - int retry_add = FLB_FALSE; - int event_bytes = 0; - - if (buf->event_index > 0 && buf->current_stream != stream) { - /* we already have events for a different stream, send them first */ - retry_add = FLB_TRUE; - goto send; - } - -retry_add_event: - buf->current_stream = stream; - retry_add = FLB_FALSE; - if (buf->event_index == 0) { - /* init */ - reset_flush_buf(ctx, buf); - } - - ret = process_event(ctx, buf, obj, tms); - if (ret < 0) { - return -1; - } - else if (ret == 1) { - if (buf->event_index <= 0) { - /* somehow the record was larger than our entire request buffer */ - flb_plg_warn(ctx->ins, "Discarding massive log record"); - return 1; /* discard this record and return to caller */ - } - /* send logs and then retry the add */ - retry_add = FLB_TRUE; - goto send; - } - else if (ret == 2) { - /* - * discard this record and return to caller - * only happens for empty records in this plugin - */ - return 1; - } - - event = &buf->events[buf->event_index]; - event_bytes = event->len + PUT_LOG_EVENTS_PER_EVENT_LEN; - - if (check_stream_time_span(stream, event) == FLB_FALSE) { - /* do not send this event */ - retry_add = FLB_TRUE; - goto send; - } - - if ((buf->data_size + event_bytes) > PUT_LOG_EVENTS_PAYLOAD_SIZE) { - if (buf->event_index <= 0) { - /* somehow the record was larger than our entire request buffer */ - flb_plg_warn(ctx->ins, "Discarding massive log record"); - return 0; /* discard this record and return to caller */ - } - /* do not send this event */ - retry_add = FLB_TRUE; - goto send; - } - - buf->data_size += event_bytes; - set_stream_time_span(stream, event); - buf->event_index++; - - if (buf->event_index == MAX_EVENTS_PER_PUT) { - goto send; - } - - /* send is not needed yet, return to caller */ - return 0; - -send: - ret = send_log_events(ctx, buf); - reset_flush_buf(ctx, buf); - if (ret < 0) { - return -1; - } - - if (retry_add == FLB_TRUE) { - goto retry_add_event; - } - - return 0; -} - -int should_add_to_emf(struct flb_intermediate_metric *an_item) -{ - /* Valid for cpu plugin */ - if (strncmp(an_item->key.via.str.ptr, "cpu_", 4) == 0 - || strncmp(an_item->key.via.str.ptr, "user_p", 6) == 0 - || strncmp(an_item->key.via.str.ptr, "system_p", 8) == 0) { - return 1; - } - - /* Valid for mem plugin */ - if (strncmp(an_item->key.via.str.ptr, "Mem.total", 9) == 0 - || strncmp(an_item->key.via.str.ptr, "Mem.used", 8) == 0 - || strncmp(an_item->key.via.str.ptr, "Mem.free", 8) == 0 - || strncmp(an_item->key.via.str.ptr, "Swap.total", 10) == 0 - || strncmp(an_item->key.via.str.ptr, "Swap.used", 9) == 0 - || strncmp(an_item->key.via.str.ptr, "Swap.free", 9) == 0) { - return 1; - } - - return 0; -} - -int pack_emf_payload(struct flb_cloudwatch *ctx, - struct mk_list *flb_intermediate_metrics, - const char *input_plugin, - struct flb_time tms, - msgpack_sbuffer *mp_sbuf, - msgpack_unpacked *mp_result, - msgpack_object *emf_payload) -{ - int total_items = mk_list_size(flb_intermediate_metrics) + 1; - - struct mk_list *metric_temp; - struct mk_list *metric_head; - struct flb_intermediate_metric *an_item; - msgpack_unpack_return mp_ret; - - /* Serialize values into the buffer using msgpack_sbuffer_write */ - msgpack_packer mp_pck; - msgpack_packer_init(&mp_pck, mp_sbuf, msgpack_sbuffer_write); - msgpack_pack_map(&mp_pck, total_items); - - /* Pack the _aws map */ - msgpack_pack_str(&mp_pck, 4); - msgpack_pack_str_body(&mp_pck, "_aws", 4); - - msgpack_pack_map(&mp_pck, 2); - - msgpack_pack_str(&mp_pck, 9); - msgpack_pack_str_body(&mp_pck, "Timestamp", 9); - msgpack_pack_long_long(&mp_pck, tms.tm.tv_sec * 1000L); - - msgpack_pack_str(&mp_pck, 17); - msgpack_pack_str_body(&mp_pck, "CloudWatchMetrics", 17); - msgpack_pack_array(&mp_pck, 1); - - msgpack_pack_map(&mp_pck, 3); - - msgpack_pack_str(&mp_pck, 9); - msgpack_pack_str_body(&mp_pck, "Namespace", 9); - - if (ctx->metric_namespace) { - msgpack_pack_str(&mp_pck, flb_sds_len(ctx->metric_namespace)); - msgpack_pack_str_body(&mp_pck, ctx->metric_namespace, - flb_sds_len(ctx->metric_namespace)); - } - else { - msgpack_pack_str(&mp_pck, 18); - msgpack_pack_str_body(&mp_pck, "fluent-bit-metrics", 18); - } - - msgpack_pack_str(&mp_pck, 10); - msgpack_pack_str_body(&mp_pck, "Dimensions", 10); - - struct mk_list *head, *inner_head; - struct flb_split_entry *dimension_list, *entry; - struct mk_list *csv_values; - if (ctx->metric_dimensions) { - msgpack_pack_array(&mp_pck, mk_list_size(ctx->metric_dimensions)); - - mk_list_foreach(head, ctx->metric_dimensions) { - dimension_list = mk_list_entry(head, struct flb_split_entry, _head); - csv_values = flb_utils_split(dimension_list->value, ',', 256); - msgpack_pack_array(&mp_pck, mk_list_size(csv_values)); - - mk_list_foreach(inner_head, csv_values) { - entry = mk_list_entry(inner_head, struct flb_split_entry, _head); - msgpack_pack_str(&mp_pck, entry->len); - msgpack_pack_str_body(&mp_pck, entry->value, entry->len); - } - flb_utils_split_free(csv_values); - } - } - else { - msgpack_pack_array(&mp_pck, 0); - } - - msgpack_pack_str(&mp_pck, 7); - msgpack_pack_str_body(&mp_pck, "Metrics", 7); - - if (strcmp(input_plugin, "cpu") == 0) { - msgpack_pack_array(&mp_pck, 3); - } - else if (strcmp(input_plugin, "mem") == 0) { - msgpack_pack_array(&mp_pck, 6); - } - else { - msgpack_pack_array(&mp_pck, 0); - } - - mk_list_foreach_safe(metric_head, metric_temp, flb_intermediate_metrics) { - an_item = mk_list_entry(metric_head, struct flb_intermediate_metric, _head); - if (should_add_to_emf(an_item) == 1) { - msgpack_pack_map(&mp_pck, 2); - msgpack_pack_str(&mp_pck, 4); - msgpack_pack_str_body(&mp_pck, "Name", 4); - msgpack_pack_object(&mp_pck, an_item->key); - msgpack_pack_str(&mp_pck, 4); - msgpack_pack_str_body(&mp_pck, "Unit", 4); - msgpack_pack_str(&mp_pck, strlen(an_item->metric_unit)); - msgpack_pack_str_body(&mp_pck, an_item->metric_unit, - strlen(an_item->metric_unit)); - } - } - - /* Pack the metric values for each record */ - mk_list_foreach_safe(metric_head, metric_temp, flb_intermediate_metrics) { - an_item = mk_list_entry(metric_head, struct flb_intermediate_metric, _head); - msgpack_pack_object(&mp_pck, an_item->key); - msgpack_pack_object(&mp_pck, an_item->value); - } - - /* - * Deserialize the buffer into msgpack_object instance. - */ - - mp_ret = msgpack_unpack_next(mp_result, mp_sbuf->data, mp_sbuf->size, NULL); - - if (mp_ret != MSGPACK_UNPACK_SUCCESS) { - flb_plg_error(ctx->ins, "msgpack_unpack returned non-success value %i", mp_ret); - return -1; - } - - *emf_payload = mp_result->data; - return 0; -} - -/* - * Main routine- processes msgpack and sends in batches which ignore the empty ones - * return value is the number of events processed and send. - */ -int process_and_send(struct flb_cloudwatch *ctx, const char *input_plugin, - struct cw_flush *buf, flb_sds_t tag, - const char *data, size_t bytes) -{ - int i = 0; - size_t map_size; - msgpack_object map; - msgpack_object_kv *kv; - msgpack_object key; - msgpack_object val; - msgpack_unpacked mp_emf_result; - msgpack_object emf_payload; - /* msgpack::sbuffer is a simple buffer implementation. */ - msgpack_sbuffer mp_sbuf; - - struct log_stream *stream; - - char *key_str = NULL; - size_t key_str_size = 0; - int j; - int ret; - int check = FLB_FALSE; - int found = FLB_FALSE; - - /* Added for EMF support */ - struct flb_intermediate_metric *metric; - struct mk_list *tmp; - struct mk_list *head; - struct flb_intermediate_metric *an_item; - - int intermediate_metric_type; - char *intermediate_metric_unit; - struct flb_log_event_decoder log_decoder; - struct flb_log_event log_event; - - 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 -1; - } - - if (strncmp(input_plugin, "cpu", 3) == 0) { - intermediate_metric_type = GAUGE; - intermediate_metric_unit = PERCENT; - } - else if (strncmp(input_plugin, "mem", 3) == 0) { - intermediate_metric_type = GAUGE; - intermediate_metric_unit = BYTES; - } - - /* unpack msgpack */ - while ((ret = flb_log_event_decoder_next( - &log_decoder, - &log_event)) == FLB_EVENT_DECODER_SUCCESS) { - - /* Get the record/map */ - map = *log_event.body; - map_size = map.via.map.size; - - stream = get_log_stream(ctx, tag, map); - if (!stream) { - flb_plg_debug(ctx->ins, "Couldn't determine log group & stream for record with tag %s", tag); - goto error; - } - - if (ctx->log_key) { - key_str = NULL; - key_str_size = 0; - check = FLB_FALSE; - found = FLB_FALSE; - - kv = map.via.map.ptr; - - for(j=0; j < map_size; j++) { - key = (kv+j)->key; - if (key.type == MSGPACK_OBJECT_BIN) { - key_str = (char *) key.via.bin.ptr; - key_str_size = key.via.bin.size; - check = FLB_TRUE; - } - if (key.type == MSGPACK_OBJECT_STR) { - key_str = (char *) key.via.str.ptr; - key_str_size = key.via.str.size; - check = FLB_TRUE; - } - - if (check == FLB_TRUE) { - if (strncmp(ctx->log_key, key_str, key_str_size) == 0) { - found = FLB_TRUE; - val = (kv+j)->val; - ret = add_event(ctx, buf, stream, &val, - &log_event.timestamp); - if (ret < 0 ) { - goto error; - } - } - } - - } - if (found == FLB_FALSE) { - flb_plg_error(ctx->ins, "Could not find log_key '%s' in record", - ctx->log_key); - } - - if (ret == 0) { - i++; - } - - continue; - } - - if (strncmp(input_plugin, "cpu", 3) == 0 - || strncmp(input_plugin, "mem", 3) == 0) { - /* Added for EMF support: Construct a list */ - struct mk_list flb_intermediate_metrics; - mk_list_init(&flb_intermediate_metrics); - - kv = map.via.map.ptr; - - /* - * Iterate through the record map, extract intermediate metric data, - * and add to the list. - */ - for (i = 0; i < map_size; i++) { - metric = flb_calloc(1, sizeof(struct flb_intermediate_metric)); - if (!metric) { - goto error; - } - - metric->key = (kv + i)->key; - metric->value = (kv + i)->val; - metric->metric_type = intermediate_metric_type; - metric->metric_unit = intermediate_metric_unit; - metric->timestamp = log_event.timestamp; - - mk_list_add(&metric->_head, &flb_intermediate_metrics); - - } - - /* The msgpack object is only valid during the lifetime of the - * sbuffer & the unpacked result. - */ - msgpack_sbuffer_init(&mp_sbuf); - msgpack_unpacked_init(&mp_emf_result); - - ret = pack_emf_payload(ctx, - &flb_intermediate_metrics, - input_plugin, - log_event.timestamp, - &mp_sbuf, - &mp_emf_result, - &emf_payload); - - /* free the intermediate metric list */ - - mk_list_foreach_safe(head, tmp, &flb_intermediate_metrics) { - an_item = mk_list_entry(head, struct flb_intermediate_metric, _head); - mk_list_del(&an_item->_head); - flb_free(an_item); - } - - if (ret != 0) { - flb_plg_error(ctx->ins, "Failed to convert EMF metrics to msgpack object. ret=%i", ret); - msgpack_unpacked_destroy(&mp_emf_result); - msgpack_sbuffer_destroy(&mp_sbuf); - goto error; - } - ret = add_event(ctx, buf, stream, &emf_payload, - &log_event.timestamp); - - msgpack_unpacked_destroy(&mp_emf_result); - msgpack_sbuffer_destroy(&mp_sbuf); - - } else { - ret = add_event(ctx, buf, stream, &map, - &log_event.timestamp); - } - - if (ret < 0 ) { - goto error; - } - - if (ret == 0) { - i++; - } - } - flb_log_event_decoder_destroy(&log_decoder); - - /* send any remaining events */ - ret = send_log_events(ctx, buf); - reset_flush_buf(ctx, buf); - if (ret < 0) { - return -1; - } - - /* return number of events */ - return i; - -error: - flb_log_event_decoder_destroy(&log_decoder); - - return -1; -} - -struct log_stream *get_or_create_log_stream(struct flb_cloudwatch *ctx, - flb_sds_t stream_name, - flb_sds_t group_name) -{ - int ret; - struct log_stream *new_stream; - struct log_stream *stream; - struct mk_list *tmp; - struct mk_list *head; - time_t now; - - /* check if the stream already exists */ - now = time(NULL); - mk_list_foreach_safe(head, tmp, &ctx->streams) { - stream = mk_list_entry(head, struct log_stream, _head); - if (strcmp(stream_name, stream->name) == 0 && strcmp(group_name, stream->group) == 0) { - return stream; - } - else { - /* check if stream is expired, if so, clean it up */ - if (stream->expiration < now) { - mk_list_del(&stream->_head); - log_stream_destroy(stream); - } - } - } - - /* create the new stream */ - new_stream = flb_calloc(1, sizeof(struct log_stream)); - if (!new_stream) { - flb_errno(); - return NULL; - } - new_stream->name = flb_sds_create(stream_name); - if (new_stream->name == NULL) { - flb_errno(); - return NULL; - } - new_stream->group = flb_sds_create(group_name); - if (new_stream->group == NULL) { - flb_errno(); - return NULL; - } - - ret = create_log_stream(ctx, new_stream, FLB_TRUE); - if (ret < 0) { - log_stream_destroy(new_stream); - return NULL; - } - new_stream->expiration = time(NULL) + FOUR_HOURS_IN_SECONDS; - - mk_list_add(&new_stream->_head, &ctx->streams); - return new_stream; -} - -struct log_stream *get_log_stream(struct flb_cloudwatch *ctx, flb_sds_t tag, - const msgpack_object map) -{ - flb_sds_t group_name = NULL; - flb_sds_t stream_name = NULL; - flb_sds_t tmp_s = NULL; - int free_group = FLB_FALSE; - int free_stream = FLB_FALSE; - struct log_stream *stream; - - /* templates take priority */ - if (ctx->ra_stream) { - stream_name = flb_ra_translate_check(ctx->ra_stream, tag, flb_sds_len(tag), - map, NULL, FLB_TRUE); - } - - if (ctx->ra_group) { - group_name = flb_ra_translate_check(ctx->ra_group, tag, flb_sds_len(tag), - map, NULL, FLB_TRUE); - } - - if (stream_name == NULL) { - if (ctx->stream_name) { - stream_name = ctx->stream_name; - } else { - free_stream = FLB_TRUE; - /* use log_stream_prefix */ - stream_name = flb_sds_create(ctx->log_stream_prefix); - if (!stream_name) { - flb_errno(); - if (group_name) { - flb_sds_destroy(group_name); - } - return NULL; - } - - tmp_s = flb_sds_cat(stream_name, tag, flb_sds_len(tag)); - if (!tmp_s) { - flb_errno(); - flb_sds_destroy(stream_name); - if (group_name) { - flb_sds_destroy(group_name); - } - return NULL; - } - stream_name = tmp_s; - } - } else { - free_stream = FLB_TRUE; - } - - if (group_name == NULL) { - group_name = ctx->group_name; - } else { - free_group = FLB_TRUE; - } - - flb_plg_debug(ctx->ins, "Using stream=%s, group=%s", stream_name, group_name); - - stream = get_or_create_log_stream(ctx, stream_name, group_name); - - if (free_group == FLB_TRUE) { - flb_sds_destroy(group_name); - } - if (free_stream == FLB_TRUE) { - flb_sds_destroy(stream_name); - } - return stream; -} - - -static int set_log_group_retention(struct flb_cloudwatch *ctx, struct log_stream *stream) -{ - if (ctx->log_retention_days <= 0) { - /* no need to set */ - return 0; - } - - struct flb_http_client *c = NULL; - struct flb_aws_client *cw_client; - flb_sds_t body; - flb_sds_t tmp; - flb_sds_t error; - - flb_plg_info(ctx->ins, "Setting retention policy on log group %s to %dd", stream->group, ctx->log_retention_days); - - body = flb_sds_create_size(68 + strlen(stream->group)); - if (!body) { - flb_sds_destroy(body); - flb_errno(); - return -1; - } - - /* construct CreateLogGroup request body */ - tmp = flb_sds_printf(&body, "{\"logGroupName\":\"%s\",\"retentionInDays\":%d}", stream->group, ctx->log_retention_days); - if (!tmp) { - flb_sds_destroy(body); - flb_errno(); - return -1; - } - body = tmp; - - if (plugin_under_test() == FLB_TRUE) { - c = mock_http_call("TEST_PUT_RETENTION_POLICY_ERROR", "PutRetentionPolicy"); - } - else { - cw_client = ctx->cw_client; - c = cw_client->client_vtable->request(cw_client, FLB_HTTP_POST, - "/", body, strlen(body), - &put_retention_policy_header, 1); - } - - if (c) { - flb_plg_debug(ctx->ins, "PutRetentionPolicy http status=%d", c->resp.status); - - if (c->resp.status == 200) { - /* success */ - flb_plg_info(ctx->ins, "Set retention policy to %d", ctx->log_retention_days); - flb_sds_destroy(body); - flb_http_client_destroy(c); - return 0; - } - - /* Check error */ - if (c->resp.payload_size > 0) { - error = flb_aws_error(c->resp.payload, c->resp.payload_size); - if (error != NULL) { - /* some other error occurred; notify user */ - flb_aws_print_error(c->resp.payload, c->resp.payload_size, - "PutRetentionPolicy", ctx->ins); - flb_sds_destroy(error); - } - else { - /* error can not be parsed, print raw response to debug */ - flb_plg_debug(ctx->ins, "Raw response: %s", c->resp.payload); - } - } - } - - flb_plg_error(ctx->ins, "Failed to putRetentionPolicy"); - if (c) { - flb_http_client_destroy(c); - } - flb_sds_destroy(body); - - return -1; -} - -int create_log_group(struct flb_cloudwatch *ctx, struct log_stream *stream) -{ - struct flb_http_client *c = NULL; - struct flb_aws_client *cw_client; - flb_sds_t body; - flb_sds_t tmp; - flb_sds_t error; - int ret; - - flb_plg_info(ctx->ins, "Creating log group %s", stream->group); - - body = flb_sds_create_size(25 + strlen(stream->group)); - if (!body) { - flb_sds_destroy(body); - flb_errno(); - return -1; - } - - /* construct CreateLogGroup request body */ - tmp = flb_sds_printf(&body, "{\"logGroupName\":\"%s\"}", stream->group); - if (!tmp) { - flb_sds_destroy(body); - flb_errno(); - return -1; - } - body = tmp; - - if (plugin_under_test() == FLB_TRUE) { - c = mock_http_call("TEST_CREATE_LOG_GROUP_ERROR", "CreateLogGroup"); - } - else { - cw_client = ctx->cw_client; - c = cw_client->client_vtable->request(cw_client, FLB_HTTP_POST, - "/", body, strlen(body), - &create_group_header, 1); - } - - if (c) { - flb_plg_debug(ctx->ins, "CreateLogGroup http status=%d", c->resp.status); - - if (c->resp.status == 200) { - /* success */ - flb_plg_info(ctx->ins, "Created log group %s", stream->group); - flb_sds_destroy(body); - flb_http_client_destroy(c); - ret = set_log_group_retention(ctx, stream); - return ret; - } - - /* Check error */ - if (c->resp.payload_size > 0) { - error = flb_aws_error(c->resp.payload, c->resp.payload_size); - if (error != NULL) { - if (strcmp(error, ERR_CODE_ALREADY_EXISTS) == 0) { - flb_plg_info(ctx->ins, "Log Group %s already exists", - stream->group); - flb_sds_destroy(body); - flb_sds_destroy(error); - flb_http_client_destroy(c); - ret = set_log_group_retention(ctx, stream); - return ret; - } - /* some other error occurred; notify user */ - flb_aws_print_error(c->resp.payload, c->resp.payload_size, - "CreateLogGroup", ctx->ins); - flb_sds_destroy(error); - } - else { - /* error can not be parsed, print raw response to debug */ - flb_plg_debug(ctx->ins, "Raw response: %s", c->resp.payload); - } - } - } - - flb_plg_error(ctx->ins, "Failed to create log group"); - if (c) { - flb_http_client_destroy(c); - } - flb_sds_destroy(body); - return -1; -} - -int create_log_stream(struct flb_cloudwatch *ctx, struct log_stream *stream, - int can_retry) -{ - - struct flb_http_client *c = NULL; - struct flb_aws_client *cw_client; - flb_sds_t body; - flb_sds_t tmp; - flb_sds_t error; - int ret; - - flb_plg_info(ctx->ins, "Creating log stream %s in log group %s", - stream->name, stream->group); - - body = flb_sds_create_size(50 + strlen(stream->group) + - strlen(stream->name)); - if (!body) { - flb_sds_destroy(body); - flb_errno(); - return -1; - } - - /* construct CreateLogStream request body */ - tmp = flb_sds_printf(&body, - "{\"logGroupName\":\"%s\",\"logStreamName\":\"%s\"}", - stream->group, - stream->name); - if (!tmp) { - flb_sds_destroy(body); - flb_errno(); - return -1; - } - body = tmp; - - cw_client = ctx->cw_client; - if (plugin_under_test() == FLB_TRUE) { - c = mock_http_call("TEST_CREATE_LOG_STREAM_ERROR", "CreateLogStream"); - } - else { - c = cw_client->client_vtable->request(cw_client, FLB_HTTP_POST, - "/", body, strlen(body), - &create_stream_header, 1); - } - - if (c) { - flb_plg_debug(ctx->ins,"CreateLogStream http status=%d", - c->resp.status); - - if (c->resp.status == 200) { - /* success */ - flb_plg_info(ctx->ins, "Created log stream %s", stream->name); - flb_sds_destroy(body); - flb_http_client_destroy(c); - return 0; - } - - /* Check error */ - if (c->resp.payload_size > 0) { - error = flb_aws_error(c->resp.payload, c->resp.payload_size); - if (error != NULL) { - if (strcmp(error, ERR_CODE_ALREADY_EXISTS) == 0) { - flb_plg_info(ctx->ins, "Log Stream %s already exists", - stream->name); - flb_sds_destroy(body); - flb_sds_destroy(error); - flb_http_client_destroy(c); - return 0; - } - - if (strcmp(error, ERR_CODE_NOT_FOUND) == 0) { - flb_sds_destroy(body); - flb_sds_destroy(error); - flb_http_client_destroy(c); - - if (ctx->create_group == FLB_TRUE) { - flb_plg_info(ctx->ins, "Log Group %s not found. Will attempt to create it.", - stream->group); - ret = create_log_group(ctx, stream); - if (ret < 0) { - return -1; - } else { - if (can_retry == FLB_TRUE) { - /* retry stream creation */ - return create_log_stream(ctx, stream, FLB_FALSE); - } else { - /* we failed to create the stream */ - return -1; - } - } - } else { - flb_plg_error(ctx->ins, "Log Group %s not found and `auto_create_group` disabled.", - stream->group); - } - return -1; - } - /* some other error occurred; notify user */ - flb_aws_print_error(c->resp.payload, c->resp.payload_size, - "CreateLogStream", ctx->ins); - flb_sds_destroy(error); - } - else { - /* error can not be parsed, print raw response to debug */ - flb_plg_debug(ctx->ins, "Raw response: %s", c->resp.payload); - } - } - } - - flb_plg_error(ctx->ins, "Failed to create log stream"); - if (c) { - flb_http_client_destroy(c); - } - flb_sds_destroy(body); - return -1; -} - -/* - * Returns -1 on failure, 0 on success, and 1 for a sequence token error, - * which means the caller can retry. - */ -int put_log_events(struct flb_cloudwatch *ctx, struct cw_flush *buf, - struct log_stream *stream, size_t payload_size) -{ - - struct flb_http_client *c = NULL; - struct flb_aws_client *cw_client; - flb_sds_t tmp; - flb_sds_t error; - int num_headers = 1; - int retry = FLB_TRUE; - - flb_plg_debug(ctx->ins, "Sending log events to log stream %s", stream->name); - - /* stream is being used, update expiration */ - stream->expiration = time(NULL) + FOUR_HOURS_IN_SECONDS; - - if (ctx->log_format != NULL) { - put_log_events_header[1].val = (char *) ctx->log_format; - put_log_events_header[1].val_len = strlen(ctx->log_format); - num_headers = 2; - } - -retry_request: - if (plugin_under_test() == FLB_TRUE) { - c = mock_http_call("TEST_PUT_LOG_EVENTS_ERROR", "PutLogEvents"); - } - else { - cw_client = ctx->cw_client; - c = cw_client->client_vtable->request(cw_client, FLB_HTTP_POST, - "/", buf->out_buf, payload_size, - put_log_events_header, num_headers); - } - - if (c) { - flb_plg_debug(ctx->ins, "PutLogEvents http status=%d", c->resp.status); - - if (c->resp.status == 200) { - if (c->resp.data == NULL || c->resp.data_len == 0 || strstr(c->resp.data, AMZN_REQUEST_ID_HEADER) == NULL) { - /* code was 200, but response is invalid, treat as failure */ - if (c->resp.data != NULL) { - flb_plg_debug(ctx->ins, "Could not find sequence token in " - "response: response body is empty: full data: `%.*s`", c->resp.data_len, c->resp.data); - } - flb_http_client_destroy(c); - - if (retry == FLB_TRUE) { - flb_plg_debug(ctx->ins, "issuing immediate retry for invalid response"); - retry = FLB_FALSE; - goto retry_request; - } - flb_plg_error(ctx->ins, "Recieved code 200 but response was invalid, %s header not found", - AMZN_REQUEST_ID_HEADER); - return -1; - } - - - /* success */ - if (c->resp.payload_size > 0) { - flb_plg_debug(ctx->ins, "Sent events to %s", stream->name); - tmp = flb_json_get_val(c->resp.payload, c->resp.payload_size, - "nextSequenceToken"); - if (tmp) { - if (stream->sequence_token != NULL) { - flb_sds_destroy(stream->sequence_token); - } - stream->sequence_token = tmp; - - flb_http_client_destroy(c); - return 0; - } - else { - flb_plg_error(ctx->ins, "Could not find sequence token in " - "response: %s", c->resp.payload); - } - } - - flb_http_client_destroy(c); - return 0; - } - - /* Check error */ - if (c->resp.payload_size > 0) { - error = flb_aws_error(c->resp.payload, c->resp.payload_size); - if (error != NULL) { - if (strcmp(error, ERR_CODE_INVALID_SEQUENCE_TOKEN) == 0) { - /* - * This case will happen when we do not know the correct - * sequence token; we can find it in the error response - * and retry. - */ - flb_plg_debug(ctx->ins, "Sequence token was invalid, " - "will retry"); - tmp = flb_json_get_val(c->resp.payload, c->resp.payload_size, - "expectedSequenceToken"); - if (tmp) { - if (stream->sequence_token != NULL) { - flb_sds_destroy(stream->sequence_token); - } - stream->sequence_token = tmp; - flb_sds_destroy(error); - flb_http_client_destroy(c); - /* tell the caller to retry */ - return 1; - } - } else if (strcmp(error, ERR_CODE_DATA_ALREADY_ACCEPTED) == 0) { - /* not sure what causes this but it counts as success */ - flb_plg_info(ctx->ins, "Got %s, a previous retry must have succeeded asychronously", ERR_CODE_DATA_ALREADY_ACCEPTED); - flb_sds_destroy(error); - flb_http_client_destroy(c); - /* success */ - return 0; - } - /* some other error occurred; notify user */ - flb_aws_print_error(c->resp.payload, c->resp.payload_size, - "PutLogEvents", ctx->ins); - flb_sds_destroy(error); - } - else { - /* error could not be parsed, print raw response to debug */ - flb_plg_debug(ctx->ins, "Raw response: %s", c->resp.payload); - } - } - } - - flb_plg_error(ctx->ins, "Failed to send log events"); - if (c) { - flb_http_client_destroy(c); - } - return -1; -} - - -void cw_flush_destroy(struct cw_flush *buf) -{ - if (buf) { - flb_free(buf->tmp_buf); - flb_free(buf->out_buf); - flb_free(buf->events); - flb_free(buf->event_buf); - flb_free(buf); - } -} |