summaryrefslogtreecommitdiffstats
path: root/src/fluent-bit/plugins/out_kinesis_streams/kinesis_api.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/fluent-bit/plugins/out_kinesis_streams/kinesis_api.c')
-rw-r--r--src/fluent-bit/plugins/out_kinesis_streams/kinesis_api.c987
1 files changed, 987 insertions, 0 deletions
diff --git a/src/fluent-bit/plugins/out_kinesis_streams/kinesis_api.c b/src/fluent-bit/plugins/out_kinesis_streams/kinesis_api.c
new file mode 100644
index 000000000..9124657bc
--- /dev/null
+++ b/src/fluent-bit/plugins/out_kinesis_streams/kinesis_api.c
@@ -0,0 +1,987 @@
+/* -*- 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_http_client.h>
+#include <fluent-bit/flb_utils.h>
+#include <fluent-bit/flb_base64.h>
+
+#include <monkey/mk_core.h>
+#include <msgpack.h>
+#include <string.h>
+#include <stdio.h>
+#include <stdlib.h>
+
+#ifndef FLB_SYSTEM_WINDOWS
+#include <unistd.h>
+#endif
+
+#include "kinesis_api.h"
+
+#define ERR_CODE_EXCEEDED_THROUGHPUT "ProvisionedThroughputExceededException"
+
+static struct flb_aws_header put_records_target_header = {
+ .key = "X-Amz-Target",
+ .key_len = 12,
+ .val = "Kinesis_20131202.PutRecords",
+ .val_len = 27,
+};
+
+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_records payload
+ */
+static int init_put_payload(struct flb_kinesis *ctx, struct flush *buf,
+ int *offset)
+{
+ if (!try_to_write(buf->out_buf, offset, buf->out_buf_size,
+ "{\"StreamName\":\"", 15)) {
+ goto error;
+ }
+
+ if (!try_to_write(buf->out_buf, offset, buf->out_buf_size,
+ ctx->stream_name, 0)) {
+ goto error;
+ }
+
+ if (!try_to_write(buf->out_buf, offset, buf->out_buf_size,
+ "\",\"Records\":[", 13)) {
+ goto error;
+ }
+ return 0;
+
+error:
+ return -1;
+}
+
+/*
+ * Simple and fast hashing algorithm to create random partition keys
+ */
+static flb_sds_t random_partition_key(const char *tag)
+{
+ int c;
+ unsigned long hash = 5381;
+ unsigned long hash2 = 5381;
+ flb_sds_t hash_str;
+ flb_sds_t tmp;
+ struct flb_time tm;
+
+ /* get current time */
+ flb_time_get(&tm);
+
+ /* compose hash */
+ while ((c = *tag++)) {
+ hash = ((hash << 5) + hash) + c; /* hash * 33 + c */
+ }
+ hash2 = (unsigned long) hash2 * tm.tm.tv_sec * tm.tm.tv_nsec;
+
+ /* flb_sds_printf allocs if the incoming sds is not at least 64 bytes */
+ hash_str = flb_sds_create_size(64);
+ if (!hash_str) {
+ flb_errno();
+ return NULL;
+ }
+ tmp = flb_sds_printf(&hash_str, "%lu%lu", hash % 7919, hash2 % 7919);
+ if (!tmp) {
+ flb_errno();
+ flb_sds_destroy(hash_str);
+ return NULL;
+ }
+ hash_str = tmp;
+
+ return hash_str;
+}
+
+/*
+ * Writes a log event to the output buffer
+ */
+static int write_event(struct flb_kinesis *ctx, struct flush *buf,
+ struct kinesis_event *event, int *offset)
+{
+ flb_sds_t tag_timestamp = NULL;
+
+ if (!try_to_write(buf->out_buf, offset, buf->out_buf_size,
+ "{\"Data\":\"", 9)) {
+ 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,
+ "\",\"PartitionKey\":\"", 18)) {
+ goto error;
+ }
+
+ if (!try_to_write(buf->out_buf, offset, buf->out_buf_size,
+ ctx->uuid, 10)) {
+ goto error;
+ }
+
+ tag_timestamp = random_partition_key(buf->tag);
+ if (!tag_timestamp) {
+ flb_plg_error(ctx->ins, "failed to generate partition key for %s", buf->tag);
+ goto error;
+ }
+
+ if (!try_to_write(buf->out_buf, offset, buf->out_buf_size,
+ tag_timestamp, 0)) {
+ flb_sds_destroy(tag_timestamp);
+ goto error;
+ }
+ flb_sds_destroy(tag_timestamp);
+
+ if (!try_to_write(buf->out_buf, offset, buf->out_buf_size,
+ "\"}", 2)) {
+ goto error;
+ }
+
+ return 0;
+
+error:
+ return -1;
+}
+
+/* Terminates a PutRecords payload */
+static int end_put_payload(struct flb_kinesis *ctx, struct 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;
+}
+
+
+/*
+ * 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
+ */
+static int process_event(struct flb_kinesis *ctx, struct flush *buf,
+ const msgpack_object *obj, struct flb_time *tms)
+{
+ size_t written = 0;
+ int ret;
+ size_t size;
+ size_t b64_len;
+ struct kinesis_event *event;
+ char *tmp_buf_ptr;
+ char *time_key_ptr;
+ struct tm time_stamp;
+ struct tm *tmp;
+ size_t len;
+ size_t tmp_size;
+ char *out_buf;
+
+ 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) {
+ /*
+ * negative value means failure to write to buffer,
+ * which means we ran out of space, and must send the logs
+ *
+ * TODO: This could also incorrectly be triggered if the record
+ * is larger than MAX_EVENT_SIZE
+ */
+ return 1;
+ }
+ written = (size_t) ret;
+
+ /* Discard empty messages (written == 2 means '""') */
+ if (written <= 2) {
+ flb_plg_debug(ctx->ins, "Found empty log message, %s", ctx->stream_name);
+ return 2;
+ }
+
+ if (ctx->log_key) {
+ /*
+ * 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++;
+ }
+
+ /* is (written + 1) because we still have to append newline */
+ if ((written + 1) >= MAX_EVENT_SIZE) {
+ flb_plg_warn(ctx->ins, "[size=%zu] Discarding record which is larger than "
+ "max size allowed by Kinesis, %s", written + 1,
+ ctx->stream_name);
+ return 2;
+ }
+
+ if (ctx->time_key) {
+ /* append time_key to end of json string */
+ tmp = gmtime_r(&tms->tm.tv_sec, &time_stamp);
+ if (!tmp) {
+ flb_plg_error(ctx->ins, "Could not create time stamp for %lu unix "
+ "seconds, discarding record, %s", tms->tm.tv_sec,
+ ctx->stream_name);
+ return 2;
+ }
+
+ /* format time output and return the length */
+ len = flb_aws_strftime_precision(&out_buf, ctx->time_key_format, tms);
+
+ /* how much space do we have left */
+ tmp_size = (buf->tmp_buf_size - buf->tmp_buf_offset) - written;
+ if (len > tmp_size) {
+ /* not enough space - tell caller to retry */
+ flb_free(out_buf);
+ return 1;
+ }
+
+ if (len == 0) {
+ /*
+ * when the length of out_buf is not enough for time_key_format,
+ * time_key will not be added to record.
+ */
+ flb_plg_error(ctx->ins, "Failed to add time_key %s to record, %s",
+ ctx->time_key, ctx->stream_name);
+ flb_free(out_buf);
+ }
+ else {
+ time_key_ptr = tmp_buf_ptr + written - 1;
+ memcpy(time_key_ptr, ",", 1);
+ time_key_ptr++;
+ memcpy(time_key_ptr, "\"", 1);
+ time_key_ptr++;
+ memcpy(time_key_ptr, ctx->time_key, strlen(ctx->time_key));
+ time_key_ptr += strlen(ctx->time_key);
+ memcpy(time_key_ptr, "\":\"", 3);
+ time_key_ptr += 3;
+
+ /* merge out_buf to time_key_ptr */
+ memcpy(time_key_ptr, out_buf, len);
+ flb_free(out_buf);
+ time_key_ptr += len;
+ memcpy(time_key_ptr, "\"}", 2);
+ time_key_ptr += 2;
+ written = (time_key_ptr - tmp_buf_ptr);
+ }
+ }
+
+ /* is (written + 1) because we still have to append newline */
+ if ((written + 1) >= MAX_EVENT_SIZE) {
+ flb_plg_warn(ctx->ins, "[size=%zu] Discarding record which is larger than "
+ "max size allowed by Kinesis, %s", written + 1,
+ ctx->stream_name);
+ return 2;
+ }
+
+ /* append newline to record */
+
+ tmp_size = (buf->tmp_buf_size - buf->tmp_buf_offset) - written;
+ if (tmp_size <= 1) {
+ /* no space left- tell caller to retry */
+ return 1;
+ }
+
+ memcpy(tmp_buf_ptr + written, "\n", 1);
+ written++;
+
+ /*
+ * check if event_buf is initialized and big enough
+ * Base64 encoding will increase size by ~4/3
+ */
+ size = (written * 1.5) + 4;
+ 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;
+ }
+ }
+
+ tmp_buf_ptr = buf->tmp_buf + buf->tmp_buf_offset;
+ ret = flb_base64_encode((unsigned char *) buf->event_buf, size, &b64_len,
+ (unsigned char *) tmp_buf_ptr, written);
+ if (ret != 0) {
+ flb_errno();
+ return -1;
+ }
+ written = b64_len;
+
+ 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;
+ }
+
+ /* copy serialized json to tmp_buf */
+ memcpy(tmp_buf_ptr, buf->event_buf, written);
+
+ buf->tmp_buf_offset += written;
+ event = &buf->events[buf->event_index];
+ event->json = tmp_buf_ptr;
+ event->len = written;
+ event->timestamp.tv_sec = tms->tm.tv_sec;
+ event->timestamp.tv_nsec = tms->tm.tv_nsec;
+
+ return 0;
+}
+
+/* Resets or inits a flush struct */
+static void reset_flush_buf(struct flb_kinesis *ctx, struct flush *buf) {
+ buf->event_index = 0;
+ buf->tmp_buf_offset = 0;
+ buf->data_size = PUT_RECORDS_HEADER_LEN + PUT_RECORDS_FOOTER_LEN;
+ buf->data_size += strlen(ctx->stream_name);
+}
+
+/* constructs a put payload, and then sends */
+static int send_log_events(struct flb_kinesis *ctx, struct flush *buf) {
+ int ret;
+ int offset;
+ int i;
+ struct kinesis_event *event;
+
+ if (buf->event_index <= 0) {
+ /*
+ * event_index should always be 1 more than the actual last event index
+ * when this function is called.
+ * Except in the case where send_log_events() is called at the end of
+ * process_and_send_to_kinesis. If all records were already sent, event_index
+ * will be 0. Hence this check.
+ */
+ return 0;
+ }
+
+ /* alloc out_buf if needed */
+ if (buf->out_buf == NULL || buf->out_buf_size < buf->data_size) {
+ if (buf->out_buf != NULL) {
+ flb_free(buf->out_buf);
+ }
+ buf->out_buf = flb_malloc(buf->data_size + 1);
+ if (!buf->out_buf) {
+ flb_errno();
+ return -1;
+ }
+ buf->out_buf_size = buf->data_size;
+ }
+
+ offset = 0;
+ ret = init_put_payload(ctx, buf, &offset);
+ if (ret < 0) {
+ flb_plg_error(ctx->ins, "Failed to initialize PutRecords payload, %s",
+ ctx->stream_name);
+ 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 record %d to "
+ "payload buffer, %s", i, ctx->stream_name);
+ 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 record with ','");
+ return -1;
+ }
+ }
+ }
+
+ ret = end_put_payload(ctx, buf, &offset);
+ if (ret < 0) {
+ flb_plg_error(ctx->ins, "Could not complete PutRecords payload");
+ return -1;
+ }
+ flb_plg_debug(ctx->ins, "kinesis:PutRecords: events=%d, payload=%d bytes", i, offset);
+ ret = put_records(ctx, buf, (size_t) offset, i);
+ if (ret < 0) {
+ flb_plg_error(ctx->ins, "Failed to send log records");
+ return -1;
+ }
+ buf->records_sent += i;
+
+ return 0;
+}
+
+/*
+ * Processes the msgpack object, sends the current batch if needed
+ */
+static int add_event(struct flb_kinesis *ctx, struct flush *buf,
+ const msgpack_object *obj, struct flb_time *tms)
+{
+ int ret;
+ struct kinesis_event *event;
+ int retry_add = FLB_FALSE;
+ size_t event_bytes = 0;
+
+ if (buf->event_index == 0) {
+ /* init */
+ reset_flush_buf(ctx, buf);
+ }
+
+retry_add_event:
+ retry_add = FLB_FALSE;
+ 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, %s",
+ ctx->stream_name);
+ return 0; /* 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 */
+ flb_plg_warn(ctx->ins, "Discarding large or unprocessable record, %s",
+ ctx->stream_name);
+ return 0;
+ }
+
+ event = &buf->events[buf->event_index];
+ event_bytes = event->len + PUT_RECORDS_PER_RECORD_LEN;
+
+ if ((buf->data_size + event_bytes) > PUT_RECORDS_PAYLOAD_SIZE) {
+ if (buf->event_index <= 0) {
+ /* somehow the record was larger than our entire request buffer */
+ flb_plg_warn(ctx->ins, "[size=%zu] Discarding massive log record, %s",
+ event_bytes, ctx->stream_name);
+ return 0; /* discard this record and return to caller */
+ }
+ /* do not send this event */
+ retry_add = FLB_TRUE;
+ goto send;
+ }
+
+ /* send is not needed yet, return to caller */
+ buf->data_size += event_bytes;
+ buf->event_index++;
+
+ if (buf->event_index == MAX_EVENTS_PER_PUT) {
+ goto send;
+ }
+
+ 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;
+}
+
+/*
+ * Main routine- processes msgpack and sends in batches
+ * return value is the number of events processed (number sent is stored in buf)
+ */
+int process_and_send_to_kinesis(struct flb_kinesis *ctx, struct flush *buf,
+ 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;
+ char *key_str = NULL;
+ size_t key_str_size = 0;
+ int j;
+ int ret;
+ int check = FLB_FALSE;
+ int found = FLB_FALSE;
+ 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;
+ }
+
+ while ((ret = flb_log_event_decoder_next(
+ &log_decoder,
+ &log_event)) == FLB_EVENT_DECODER_SUCCESS) {
+ map = *log_event.body;
+ map_size = map.via.map.size;
+
+ 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, &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, %s",
+ ctx->log_key, ctx->stream_name);
+ }
+ else {
+ i++;
+ }
+ continue;
+ }
+
+ ret = add_event(ctx, buf, &map, &log_event.timestamp);
+ if (ret < 0 ) {
+ goto error;
+ }
+ 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 processed */
+ buf->records_processed = i;
+ return i;
+
+error:
+ flb_log_event_decoder_destroy(&log_decoder);
+
+ return -1;
+}
+
+/*
+ * Returns number of failed records on success, -1 on failure
+ */
+static int process_api_response(struct flb_kinesis *ctx,
+ struct flb_http_client *c)
+{
+ int i;
+ int k;
+ int w;
+ int ret;
+ int failed_records = -1;
+ int root_type;
+ char *out_buf;
+ int throughput_exceeded = FLB_FALSE;
+ size_t off = 0;
+ size_t out_size;
+ msgpack_unpacked result;
+ msgpack_object root;
+ msgpack_object key;
+ msgpack_object val;
+ msgpack_object response;
+ msgpack_object response_key;
+ msgpack_object response_val;
+
+ if (strstr(c->resp.payload, "\"FailedRecordCount\":0")) {
+ return 0;
+ }
+
+ /* Convert JSON payload to msgpack */
+ ret = flb_pack_json(c->resp.payload, c->resp.payload_size,
+ &out_buf, &out_size, &root_type, NULL);
+ if (ret == -1) {
+ flb_plg_error(ctx->ins, "could not pack/validate JSON API response\n%s",
+ c->resp.payload);
+ return -1;
+ }
+
+ /* Lookup error field */
+ msgpack_unpacked_init(&result);
+ ret = msgpack_unpack_next(&result, out_buf, out_size, &off);
+ if (ret != MSGPACK_UNPACK_SUCCESS) {
+ flb_plg_error(ctx->ins, "Cannot unpack response to find error\n%s",
+ c->resp.payload);
+ failed_records = -1;
+ goto done;
+ }
+
+ root = result.data;
+ if (root.type != MSGPACK_OBJECT_MAP) {
+ flb_plg_error(ctx->ins, "unexpected payload type=%i",
+ root.type);
+ failed_records = -1;
+ goto done;
+ }
+
+ for (i = 0; i < root.via.map.size; i++) {
+ key = root.via.map.ptr[i].key;
+ if (key.type != MSGPACK_OBJECT_STR) {
+ flb_plg_error(ctx->ins, "unexpected key type=%i",
+ key.type);
+ failed_records = -1;
+ goto done;
+ }
+
+ if (key.via.str.size >= 14 &&
+ strncmp(key.via.str.ptr, "FailedRecordCount", 14) == 0) {
+ val = root.via.map.ptr[i].val;
+ if (val.type != MSGPACK_OBJECT_POSITIVE_INTEGER) {
+ flb_plg_error(ctx->ins, "unexpected 'FailedRecordCount' value type=%i",
+ val.type);
+ failed_records = -1;
+ goto done;
+ }
+
+ failed_records = val.via.u64;
+ if (failed_records == 0) {
+ /* no need to check RequestResponses field */
+ goto done;
+ }
+ }
+
+ if (key.via.str.size >= 14 &&
+ strncmp(key.via.str.ptr, "Records", 7) == 0) {
+ val = root.via.map.ptr[i].val;
+ if (val.type != MSGPACK_OBJECT_ARRAY) {
+ flb_plg_error(ctx->ins, "unexpected 'Records' value type=%i",
+ val.type);
+ failed_records = -1;
+ goto done;
+ }
+
+ if (val.via.array.size == 0) {
+ flb_plg_error(ctx->ins, "'Records' field in response is empty");
+ failed_records = -1;
+ goto done;
+ }
+
+ for (k = 0; k < val.via.array.size; k++) {
+ /* iterate through the responses */
+ response = val.via.array.ptr[k];
+ if (response.type != MSGPACK_OBJECT_MAP) {
+ flb_plg_error(ctx->ins, "unexpected 'Records[%d]' value type=%i",
+ k, response.type);
+ failed_records = -1;
+ goto done;
+ }
+ for (w = 0; w < response.via.map.size; w++) {
+ /* iterate through the response's keys */
+ response_key = response.via.map.ptr[w].key;
+ if (response_key.type != MSGPACK_OBJECT_STR) {
+ flb_plg_error(ctx->ins, "unexpected key type=%i",
+ response_key.type);
+ failed_records = -1;
+ goto done;
+ }
+ if (response_key.via.str.size >= 9 &&
+ strncmp(response_key.via.str.ptr, "ErrorCode", 9) == 0) {
+ response_val = response.via.map.ptr[w].val;
+ if (!throughput_exceeded &&
+ response_val.via.str.size >= 38 &&
+ (strncmp(response_val.via.str.ptr,
+ ERR_CODE_EXCEEDED_THROUGHPUT, 38) == 0)) {
+ throughput_exceeded = FLB_TRUE;
+ flb_plg_error(ctx->ins, "Throughput limits may have been exceeded, %s",
+ ctx->stream_name);
+ }
+ flb_plg_debug(ctx->ins, "Record %i failed with err_code=%.*s",
+ k, response_val.via.str.size,
+ response_val.via.str.ptr);
+ }
+ if (response_key.via.str.size >= 12 &&
+ strncmp(response_key.via.str.ptr, "ErrorMessage", 12) == 0) {
+ response_val = response.via.map.ptr[w].val;
+ flb_plg_debug(ctx->ins, "Record %i failed with err_msg=%.*s",
+ k, response_val.via.str.size,
+ response_val.via.str.ptr);
+ }
+ }
+ }
+ }
+ }
+
+ done:
+ flb_free(out_buf);
+ msgpack_unpacked_destroy(&result);
+ return failed_records;
+}
+
+static int plugin_under_test()
+{
+ if (getenv("FLB_KINESIS_PLUGIN_UNDER_TEST") != NULL) {
+ return FLB_TRUE;
+ }
+
+ return FLB_FALSE;
+}
+
+static 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;
+}
+
+static int partial_success()
+{
+ char *err_val = NULL;
+
+ err_val = getenv("PARTIAL_SUCCESS_CASE");
+ if (err_val != NULL && strlen(err_val) > 0) {
+ return FLB_TRUE;
+ }
+
+ return FLB_FALSE;
+}
+
+static struct flb_http_client *mock_http_call(char *error_env_var)
+{
+ /* 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 (partial_success() == FLB_TRUE) {
+ /* mocked partial failure response */
+ c->resp.payload = "{\"FailedRecordCount\":2,\"Records\":[{\"SequenceNumber\":\"49543463076548007577105092703039560359975228518395012686\",\"ShardId\":\"shardId-000000000000\"},{\"ErrorCode\":\"ProvisionedThroughputExceededException\",\"ErrorMessage\":\"Rate exceeded for shard shardId-000000000001 in stream exampleStreamName under account 111111111111.\"},{\"ErrorCode\":\"InternalFailure\",\"ErrorMessage\":\"Internal service failure.\"}]}";
+ c->resp.payload_size = strlen(c->resp.payload);
+ }
+ else {
+ /* mocked success response */
+ c->resp.payload = "{\"FailedRecordCount\":0,\"Records\":[{\"SequenceNumber\":\"49543463076548007577105092703039560359975228518395019266\",\"ShardId\":\"shardId-000000000000\"},{\"SequenceNumber\":\"49543463076570308322303623326179887152428262250726293522\",\"ShardId\":\"shardId-000000000001\"},{\"SequenceNumber\":\"49543463076570308322303623326179887152428262250726293588\",\"ShardId\":\"shardId-000000000003\"}]}";
+ c->resp.payload_size = strlen(c->resp.payload);
+ }
+ }
+
+ return c;
+}
+
+
+/*
+ * Returns -1 on failure, 0 on success
+ */
+int put_records(struct flb_kinesis *ctx, struct flush *buf,
+ size_t payload_size, int num_records)
+{
+
+ struct flb_http_client *c = NULL;
+ struct flb_aws_client *kinesis_client;
+ flb_sds_t error;
+ int failed_records = 0;
+
+ flb_plg_debug(ctx->ins, "Sending log records to stream %s",
+ ctx->stream_name);
+
+ if (plugin_under_test() == FLB_TRUE) {
+ c = mock_http_call("TEST_PUT_RECORDS_ERROR");
+ }
+ else {
+ kinesis_client = ctx->kinesis_client;
+ c = kinesis_client->client_vtable->request(kinesis_client, FLB_HTTP_POST,
+ "/", buf->out_buf, payload_size,
+ &put_records_target_header, 1);
+ }
+
+ if (c) {
+ flb_plg_debug(ctx->ins, "PutRecords http status=%d", c->resp.status);
+
+ if (c->resp.status == 200) {
+ /* Kinesis API can return partial success- check response */
+ if (c->resp.payload_size > 0) {
+ failed_records = process_api_response(ctx, c);
+ if (failed_records < 0) {
+ flb_plg_error(ctx->ins, "PutRecords response "
+ "could not be parsed, %s",
+ c->resp.payload);
+ flb_http_client_destroy(c);
+ return -1;
+ }
+ if (failed_records == num_records) {
+ flb_plg_error(ctx->ins, "PutRecords request returned "
+ "with no records successfully recieved, %s",
+ ctx->stream_name);
+ flb_http_client_destroy(c);
+ return -1;
+ }
+ if (failed_records > 0) {
+ flb_plg_error(ctx->ins, "%d out of %d records failed to be "
+ "delivered, will retry this batch, %s",
+ failed_records, num_records,
+ ctx->stream_name);
+ flb_http_client_destroy(c);
+ return -1;
+ }
+ }
+ flb_plg_debug(ctx->ins, "Sent events to %s", ctx->stream_name);
+ 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_EXCEEDED_THROUGHPUT) == 0) {
+ flb_plg_error(ctx->ins, "Throughput limits for %s "
+ "may have been exceeded.",
+ ctx->stream_name);
+ }
+ if (strncmp(error, "SerializationException", 22) == 0) {
+ /*
+ * If this happens, we habe a bug in the code
+ * User should send us the output to debug
+ */
+ flb_plg_error(ctx->ins, "<<------Bug in Code------>>");
+ printf("Malformed request: %s", buf->out_buf);
+ }
+ flb_aws_print_error(c->resp.payload, c->resp.payload_size,
+ "PutRecords", 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 records to %s", ctx->stream_name);
+ if (c) {
+ flb_http_client_destroy(c);
+ }
+ return -1;
+}
+
+
+void kinesis_flush_destroy(struct 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);
+ }
+}
generated by cgit v1.2.3 (git 2.39.1) at 2024-08-08 17:35:01 +0000