out_stackdriver: support gzip compression (#7101)

Implement gzip compression

Signed-off-by: Catherine Fang <[email protected]>
Co-authored-by: igorpeshansky <[email protected]>

Author: CatherineF-dev

plugins/out_stackdriver/stackdriver.c

@@ -32,6 +32,7 @@
 #include <fluent-bit/flb_ra_key.h>
 #include <fluent-bit/flb_record_accessor.h>
 #include <fluent-bit/flb_log_event_decoder.h>
+#include <fluent-bit/flb_gzip.h>
 
 #include <msgpack.h>
 
@@ -2522,10 +2523,12 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
     size_t b_sent;
     flb_sds_t token;
     flb_sds_t payload_buf;
-    size_t payload_size;
+    void *compressed_payload_buffer = NULL;
+    size_t compressed_payload_size;
     struct flb_stackdriver *ctx = out_context;
     struct flb_connection *u_conn;
     struct flb_http_client *c;
+    int compressed = FLB_FALSE;
 #ifdef FLB_HAVE_METRICS
     char *name = (char *) flb_output_name(ctx->ins);
     uint64_t ts = cfl_time_now();
@@ -2563,7 +2566,6 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
         flb_upstream_conn_release(u_conn);
         FLB_OUTPUT_RETURN(FLB_RETRY);
     }
-    payload_size = flb_sds_len(payload_buf);
 
     /* Get or renew Token */
     token = get_google_token(ctx);
@@ -2581,9 +2583,22 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
         FLB_OUTPUT_RETURN(FLB_RETRY);
     }
 
+    compressed_payload_buffer = payload_buf;
+    compressed_payload_size = flb_sds_len(payload_buf);
+    if (ctx->compress_gzip == FLB_TRUE) {
+        ret = flb_gzip_compress((void *) payload_buf, flb_sds_len(payload_buf),
+                                &compressed_payload_buffer, &compressed_payload_size);
+        if (ret == -1) {
+            flb_plg_error(ctx->ins, "cannot gzip payload, disabling compression");
+        } else {
+            compressed = FLB_TRUE;
+            flb_sds_destroy(payload_buf);
+        }
+    }
+
     /* Compose HTTP Client request */
     c = flb_http_client(u_conn, FLB_HTTP_POST, FLB_STD_WRITE_URI,
-                        payload_buf, payload_size, NULL, 0, NULL, 0);
+                        compressed_payload_buffer, compressed_payload_size, NULL, 0, NULL, 0);
 
     flb_http_buffer_size(c, 4192);
 
@@ -2598,6 +2613,10 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
 
     flb_http_add_header(c, "Content-Type", 12, "application/json", 16);
     flb_http_add_header(c, "Authorization", 13, token, flb_sds_len(token));
+    /* Content Encoding: gzip */
+    if (compressed == FLB_TRUE) {
+        flb_http_set_content_encoding_gzip(c);
+    }
 
     /* Send HTTP request */
     ret = flb_http_do(c, &b_sent);
@@ -2661,8 +2680,14 @@ static void cb_stackdriver_flush(struct flb_event_chunk *event_chunk,
     update_retry_metric(ctx, event_chunk, ts, c->resp.status, ret_code);
 #endif
 
+
     /* Cleanup */
-    flb_sds_destroy(payload_buf);
+    if (compressed == FLB_TRUE) {
+        flb_free(compressed_payload_buffer);
+    }
+    else {
+        flb_sds_destroy(payload_buf);
+    }
     flb_sds_destroy(token);
     flb_http_client_destroy(c);
     flb_upstream_conn_release(u_conn);
@@ -2785,6 +2810,11 @@ static struct flb_config_map config_map[] = {
       0, FLB_TRUE, offsetof(struct flb_stackdriver, task_id),
       "Set the resource task id"
     },
+    {
+      FLB_CONFIG_MAP_STR, "compress", NULL,
+      0, FLB_FALSE, 0,
+      "Set log payload compression method. Option available is 'gzip'"
+    },
     {
       FLB_CONFIG_MAP_CLIST, "labels", NULL,
       0, FLB_TRUE, offsetof(struct flb_stackdriver, labels),

plugins/out_stackdriver/stackdriver.h

@@ -160,6 +160,9 @@ struct flb_stackdriver {
     flb_sds_t job;
     flb_sds_t task_id;
 
+    /* Internal variable to reduce string comparisons */
+    int compress_gzip;
+
     /* other */
     flb_sds_t export_to_project_id;
     flb_sds_t resource;

plugins/out_stackdriver/stackdriver_conf.c

@@ -279,6 +279,13 @@ struct flb_stackdriver *flb_stackdriver_conf_create(struct flb_output_instance *
         return NULL;
     }
 
+    /* Compress (gzip) */
+    tmp = flb_output_get_property("compress", ins);
+    ctx->compress_gzip = FLB_FALSE;
+    if (tmp && strcasecmp(tmp, "gzip") == 0) {
+        ctx->compress_gzip = FLB_TRUE;
+    }
+
     /* labels */
     flb_kv_init(&ctx->config_labels);
     ret = parse_configuration_labels((void *)ctx);