out_cloudwatch_logs: support log_group_class option

Signed-off-by: Wesley Pettit <[email protected]>

Author: PettitWesley

plugins/out_cloudwatch_logs/cloudwatch_api.c

@@ -1285,21 +1285,39 @@ int create_log_group(struct flb_cloudwatch *ctx, struct log_stream *stream)
 
     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;
+    if (ctx->log_group_class_type == LOG_CLASS_DEFAULT_TYPE) {
+        body = flb_sds_create_size(30 + strlen(stream->group));
+        if (!body) {
+            flb_sds_destroy(body);
+            flb_errno();
+            return -1;
+        }
+
+        tmp = flb_sds_printf(&body, "{\"logGroupName\":\"%s\"}", stream->group);
+        if (!tmp) {
+            flb_sds_destroy(body);
+            flb_errno();
+            return -1;
+        }
+        body = tmp;
+    } else {
+        body = flb_sds_create_size(37 + strlen(stream->group) + strlen(ctx->log_group_class));
+        if (!body) {
+            flb_sds_destroy(body);
+            flb_errno();
+            return -1;
+        }
+
+        tmp = flb_sds_printf(&body, "{\"logGroupName\":\"%s\", \"logGroupClass\":\"%s\"}", 
+                             stream->group, ctx->log_group_class);
+        if (!tmp) {
+            flb_sds_destroy(body);
+            flb_errno();
+            return -1;
+        }
+        body = tmp;
     }
-    body = tmp;
 
     if (plugin_under_test() == FLB_TRUE) {
         c = mock_http_call("TEST_CREATE_LOG_GROUP_ERROR", "CreateLogGroup");
@@ -1316,7 +1334,8 @@ int create_log_group(struct flb_cloudwatch *ctx, struct log_stream *stream)
 
         if (c->resp.status == 200) {
             /* success */
-            flb_plg_info(ctx->ins, "Created log group %s", stream->group);
+            flb_plg_info(ctx->ins, "Created log group %s with storage class %s", 
+                         stream->group, ctx->log_group_class);
             flb_sds_destroy(body);
             flb_http_client_destroy(c);
             ret = set_log_group_retention(ctx, stream);
@@ -1328,8 +1347,16 @@ int create_log_group(struct flb_cloudwatch *ctx, struct log_stream *stream)
             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);
+                    if (ctx->log_group_class_type == LOG_CLASS_INFREQUENT_ACCESS_TYPE) {
+                        flb_plg_warn(ctx->ins, "Log Group %s already exists; "
+                                     "Fluent Bit did not create this group in this execution. "
+                                     "Fluent Bit therefore was unable verify or set %s storage. "
+                                     "Check CloudWatch Console or API for the groups storage class status.",
+                                     stream->group, LOG_CLASS_INFREQUENT_ACCESS);
+                    } else {
+                        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);
@@ -1480,7 +1507,6 @@ int put_log_events(struct flb_cloudwatch *ctx, struct cw_flush *buf,
 
     struct flb_http_client *c = NULL;
     struct flb_aws_client *cw_client;
-    flb_sds_t tmp;
     int num_headers = 1;
     int retry = FLB_TRUE;
 
@@ -1510,7 +1536,7 @@ int put_log_events(struct flb_cloudwatch *ctx, struct cw_flush *buf,
         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) {
+            if (c->resp.data == NULL || c->resp.data_len == 0 || strcasestr(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, "Invalid response: full data: `%.*s`", c->resp.data_len, c->resp.data);

plugins/out_cloudwatch_logs/cloudwatch_logs.c

@@ -50,6 +50,33 @@ static struct flb_aws_header content_type_header = {
     .val_len = 26,
 };
 
+static int validate_log_group_class(struct flb_cloudwatch *ctx)
+{   
+    if (ctx->create_group == FLB_FALSE) {
+        flb_plg_error(ctx->ins, "Configuring log_group_class requires `auto_create_group On`.");
+        return -1;
+    }
+
+    if (ctx->log_group_class == NULL || strlen(ctx->log_group_class) == 0) {
+        ctx->log_group_class_type = LOG_CLASS_DEFAULT_TYPE;
+        ctx->log_group_class = LOG_CLASS_STANDARD;
+        return 0;
+    } else if (strncmp(ctx->log_group_class, LOG_CLASS_STANDARD, LOG_CLASS_STANDARD_LEN) == 0) {
+        flb_plg_debug(ctx->ins, "Using explicitly configured `log_group_class %s`, which is the default log class.", ctx->log_group_class);
+        ctx->log_group_class_type = LOG_CLASS_STANDARD_TYPE;
+        return 0;
+    } else if (strncmp(ctx->log_group_class, LOG_CLASS_INFREQUENT_ACCESS, LOG_CLASS_INFREQUENT_ACCESS_LEN) == 0) {
+        flb_plg_warn(ctx->ins, "Configured `log_group_class %s` will only apply to log groups created by Fluent Bit. "
+                     "Look for the `Created log group` info level message emitted when a group does not already exist and is created.", ctx->log_group_class);
+        ctx->log_group_class_type = LOG_CLASS_INFREQUENT_ACCESS_TYPE;
+        return 0;
+    }
+
+    flb_plg_error(ctx->ins, "The valid values for log_group_class are {%s, %s}. Invalid input was %s", LOG_CLASS_STANDARD, LOG_CLASS_INFREQUENT_ACCESS, ctx->log_group_class);
+
+    return -1;
+}
+
 static int cb_cloudwatch_init(struct flb_output_instance *ins,
                               struct flb_config *config, void *data)
 {
@@ -211,6 +238,11 @@ static int cb_cloudwatch_init(struct flb_output_instance *ins,
         ctx->sts_endpoint = (char *) tmp;
     }
 
+    ret = validate_log_group_class(ctx);
+    if (ret < 0) {
+        goto error;
+    }
+
     /* one tls instance for provider, one for cw client */
     ctx->cred_tls = flb_tls_create(FLB_TLS_CLIENT_MODE,
                                    FLB_TRUE,
@@ -652,6 +684,12 @@ static struct flb_config_map config_map[] = {
      "$HOME/.aws/ directory."
     },
 
+    {
+     FLB_CONFIG_MAP_STR, "log_group_class", "",
+     0, FLB_TRUE, offsetof(struct flb_cloudwatch, log_group_class),
+     "Specify the log storage class. Valid values are STANDARD (default) and INFREQUENT_ACCESS."
+    },
+
     /* EOF */
     {0}
 };

plugins/out_cloudwatch_logs/cloudwatch_logs.h

@@ -30,6 +30,16 @@
 #include <fluent-bit/flb_record_accessor.h>
 #include <fluent-bit/record_accessor/flb_ra_parser.h>
 
+#define LOG_CLASS_STANDARD                  "STANDARD"
+#define LOG_CLASS_STANDARD_LEN              8
+#define LOG_CLASS_INFREQUENT_ACCESS         "INFREQUENT_ACCESS"
+#define LOG_CLASS_INFREQUENT_ACCESS_LEN     17
+/* log_group_class not configured; do not send the logGroupClass field in request */
+#define LOG_CLASS_DEFAULT_TYPE              0
+/* send configured & validated string in request */
+#define LOG_CLASS_STANDARD_TYPE             1
+#define LOG_CLASS_INFREQUENT_ACCESS_TYPE    2
+
 /* buffers used for each flush */
 struct cw_flush {
     /* temporary buffer for storing the serialized event messages */
@@ -113,6 +123,8 @@ struct flb_cloudwatch {
     const char *extra_user_agent;
     const char *external_id;
     const char *profile;
+    const char *log_group_class;
+    int log_group_class_type;
     int custom_endpoint;
     /* Should the plugin create the log group */
     int create_group;