diff --git a/src/aws/flb_aws_error_reporter.c b/src/aws/flb_aws_error_reporter.c index 581fd4dc72e..a9bf66c1ab9 100644 --- a/src/aws/flb_aws_error_reporter.c +++ b/src/aws/flb_aws_error_reporter.c @@ -77,8 +77,11 @@ struct flb_aws_error_reporter *flb_aws_error_reporter_create() /* clean up existing file*/ if ((f = fopen(error_reporter->file_path, "r")) != NULL) { + fclose(f); + /* file exist, try delete it*/ if (remove(error_reporter->file_path)) { + flb_sds_destroy(error_reporter->file_path); flb_free(error_reporter); flb_errno(); return NULL; @@ -115,6 +118,7 @@ int flb_aws_error_reporter_write(struct flb_aws_error_reporter *error_reporter, flb_sds_t buf; flb_sds_t buf_tmp; int deleted_message_count = 0; + int projected_size; FILE *f; if (error_reporter == NULL) { @@ -166,7 +170,34 @@ int flb_aws_error_reporter_write(struct flb_aws_error_reporter *error_reporter, message->len = flb_sds_len(buf); - /* clean up old message to provide enough space for new message*/ + projected_size = error_reporter->file_size + message->len; + mk_list_foreach_safe(head, tmp, &error_reporter->messages) { + if (projected_size <= error_reporter->max_size) { + break; + } + + tmp_message = mk_list_entry(head, struct flb_error_message, _head); + projected_size -= tmp_message->len; + deleted_message_count++; + } + + if (deleted_message_count == 0) { + f = fopen(error_reporter->file_path, "a"); + } + else { + f = fopen(error_reporter->file_path, "w"); + } + + if (f == NULL) { + flb_sds_destroy(message->data); + flb_free(message); + flb_sds_destroy(buf); + return -1; + } + + deleted_message_count = 0; + + /* clean up old messages to provide enough space for the new message */ mk_list_foreach_safe(head, tmp, &error_reporter->messages) { tmp_message = mk_list_entry(head, struct flb_error_message, _head); if (error_reporter->file_size + flb_sds_len(buf) <= error_reporter->max_size) { @@ -186,14 +217,12 @@ int flb_aws_error_reporter_write(struct flb_aws_error_reporter *error_reporter, error_reporter->file_size += message->len; if (deleted_message_count == 0) { - f = fopen(error_reporter->file_path, "a"); - fprintf(f, message->data); + fprintf(f, "%s", message->data); } else { - f = fopen(error_reporter->file_path, "w"); mk_list_foreach_safe(head, tmp, &error_reporter->messages) { tmp_message = mk_list_entry(head, struct flb_error_message, _head); - fprintf(f, tmp_message->data); + fprintf(f, "%s", tmp_message->data); } } fclose(f); @@ -233,9 +262,12 @@ void flb_aws_error_reporter_clean(struct flb_aws_error_reporter *error_reporter) /* rewrite error report file if any message is cleaned up*/ if (expired_message_count > 0) { f = fopen(error_reporter->file_path, "w"); + if (f == NULL) { + return; + } mk_list_foreach_safe(head, tmp, &error_reporter->messages) { message = mk_list_entry(head, struct flb_error_message, _head); - fprintf(f, message->data); + fprintf(f, "%s", message->data); } fclose(f); } @@ -273,4 +305,4 @@ void flb_aws_error_reporter_destroy(struct flb_aws_error_reporter *error_reporte int is_error_reporting_enabled() { return getenv(STATUS_MESSAGE_FILE_PATH_ENV) != NULL; -} \ No newline at end of file +} diff --git a/tests/internal/error_reporter.c b/tests/internal/error_reporter.c index 9ab84475fb6..9f006e1f487 100644 --- a/tests/internal/error_reporter.c +++ b/tests/internal/error_reporter.c @@ -1,23 +1,41 @@ /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ #include #include -#include #include #include +#include #include #include #include "flb_tests_internal.h" -const char* file_path = "/tmp/error.log"; +#ifdef FLB_SYSTEM_WINDOWS +#define TEST_ERROR_FILE_PATH "error.log" +#define TEST_INVALID_ERROR_FILE_PATH "NUL\\error.log" + +static int test_setenv(const char *name, const char *value, int overwrite) +{ + if (overwrite == 0 && getenv(name) != NULL) { + return 0; + } + + return _putenv_s(name, value); +} +#else +#define TEST_ERROR_FILE_PATH "/tmp/error.log" +#define TEST_INVALID_ERROR_FILE_PATH "/dev/null/error.log" +#define test_setenv(name, value, overwrite) setenv(name, value, overwrite) +#endif + +const char* file_path = TEST_ERROR_FILE_PATH; const char* error_message_1 = "[engine] scheduler could not start"; const char* error_message_2 = "[engine] scheduler could not stop"; void test_flb_aws_error_reporter_create() { - setenv(STATUS_MESSAGE_FILE_PATH_ENV, file_path, 1); + test_setenv(STATUS_MESSAGE_FILE_PATH_ENV, file_path, 1); struct flb_aws_error_reporter *error_reporter = flb_aws_error_reporter_create(); TEST_CHECK((void*) error_reporter != NULL); TEST_CHECK((void*)error_reporter->file_path != NULL); @@ -36,7 +54,7 @@ void test_flb_aws_error_reporter_write() { } error_message_3[1040] = '\0'; - setenv(STATUS_MESSAGE_FILE_PATH_ENV, file_path, 1); + test_setenv(STATUS_MESSAGE_FILE_PATH_ENV, file_path, 1); struct flb_aws_error_reporter *error_reporter = flb_aws_error_reporter_create(); flb_aws_error_reporter_write(error_reporter, error_message_1); @@ -62,7 +80,7 @@ void test_flb_aws_error_reporter_write() { void test_flb_aws_error_reporter_clean() { - setenv(STATUS_MESSAGE_FILE_PATH_ENV, file_path, 1); + test_setenv(STATUS_MESSAGE_FILE_PATH_ENV, file_path, 1); struct flb_aws_error_reporter *error_reporter = flb_aws_error_reporter_create(); flb_aws_error_reporter_write(error_reporter, error_message_1); time_t start = time(NULL); @@ -74,9 +92,30 @@ void test_flb_aws_error_reporter_clean() { flb_aws_error_reporter_destroy(error_reporter); } +void test_flb_aws_error_reporter_write_open_failure() +{ + int ret; + struct flb_aws_error_reporter *error_reporter; + + test_setenv(STATUS_MESSAGE_FILE_PATH_ENV, + TEST_INVALID_ERROR_FILE_PATH, 1); + + error_reporter = flb_aws_error_reporter_create(); + TEST_CHECK(error_reporter != NULL); + + ret = flb_aws_error_reporter_write(error_reporter, error_message_1); + TEST_CHECK(ret == -1); + TEST_CHECK(mk_list_size(&error_reporter->messages) == 0); + TEST_CHECK(error_reporter->file_size == 0); + + flb_aws_error_reporter_destroy(error_reporter); +} + TEST_LIST = { { "test_flb_aws_error_reporter_create", test_flb_aws_error_reporter_create}, {"test_flb_aws_error_reporter_write", test_flb_aws_error_reporter_write}, {"test_flb_aws_error_reporter_clean", test_flb_aws_error_reporter_clean}, + {"test_flb_aws_error_reporter_write_open_failure", + test_flb_aws_error_reporter_write_open_failure}, { 0 } -}; \ No newline at end of file +}; diff --git a/tests/runtime/out_cloudwatch.c b/tests/runtime/out_cloudwatch.c index 41fc4fa9336..c90a3e00776 100644 --- a/tests/runtime/out_cloudwatch.c +++ b/tests/runtime/out_cloudwatch.c @@ -9,6 +9,7 @@ #include "../../plugins/out_cloudwatch_logs/cloudwatch_api.h" #define ERROR_ALREADY_EXISTS "{\"__type\":\"ResourceAlreadyExistsException\"}" +#define CLOUDWATCH_ERROR_NOT_FOUND "{\"__type\":\"ResourceNotFoundException\"}" /* not a real error code, but tests that the code can respond to any error */ #define ERROR_UNKNOWN "{\"__type\":\"UNKNOWN\"}" @@ -284,6 +285,44 @@ void flb_test_cloudwatch_error_put_log_events(void) flb_destroy(ctx); } +void flb_test_cloudwatch_error_put_log_events_not_found(void) +{ + int ret; + flb_ctx_t *ctx; + int in_ffd; + int out_ffd; + + /* ResourceNotFoundException must follow the normal output retry path. */ + setenv("FLB_CLOUDWATCH_PLUGIN_UNDER_TEST", "true", 1); + setenv("TEST_PUT_LOG_EVENTS_ERROR", CLOUDWATCH_ERROR_NOT_FOUND, 1); + + ctx = flb_create(); + + in_ffd = flb_input(ctx, (char *) "lib", NULL); + TEST_CHECK(in_ffd >= 0); + flb_input_set(ctx, in_ffd, "tag", "test", NULL); + + out_ffd = flb_output(ctx, (char *) "cloudwatch_logs", NULL); + TEST_CHECK(out_ffd >= 0); + flb_output_set(ctx, out_ffd, "match", "test", NULL); + flb_output_set(ctx, out_ffd, "region", "us-west-2", NULL); + flb_output_set(ctx, out_ffd, "log_group_name", "fluent", NULL); + flb_output_set(ctx, out_ffd, "log_stream_prefix", "from-fluent-", NULL); + flb_output_set(ctx, out_ffd, "auto_create_group", "On", NULL); + flb_output_set(ctx, out_ffd, "net.keepalive", "Off", NULL); + flb_output_set(ctx, out_ffd, "Retry_Limit", "1", NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + flb_lib_push(ctx, in_ffd, (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); + + sleep(2); + flb_stop(ctx); + flb_destroy(ctx); + unsetenv("TEST_PUT_LOG_EVENTS_ERROR"); +} + void flb_test_cloudwatch_put_retention_policy_success(void) { int ret; @@ -548,6 +587,7 @@ TEST_LIST = { {"create_group_error", flb_test_cloudwatch_error_create_group }, {"create_stream_error", flb_test_cloudwatch_error_create_stream }, {"put_log_events_error", flb_test_cloudwatch_error_put_log_events }, + {"put_log_events_not_found", flb_test_cloudwatch_error_put_log_events_not_found }, {"put_retention_policy_success", flb_test_cloudwatch_put_retention_policy_success }, {"already_exists_create_group_put_retention_policy", flb_test_cloudwatch_already_exists_create_group_put_retention_policy }, {"error_put_retention_policy", flb_test_cloudwatch_error_put_retention_policy },