aws: error_reporter: Plug SEGV occurrences on cloudwatch logs

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;
-}
+}

tests/internal/error_reporter.c

@@ -1,23 +1,41 @@
 /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
 #include <stdio.h>
 #include <stdlib.h>
-#include <unistd.h>
 #include <time.h>
 #include <monkey/mk_core/mk_list.h>
 
+#include <fluent-bit/flb_compat.h>
 #include <fluent-bit/flb_mem.h>
 #include <fluent-bit/aws/flb_aws_error_reporter.h>
 
 #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 }
-};
+};

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 },