out_http: gzip encoding support

 * Use existing miniz lib from out_td plugin
 * Move miniz to lib and write util for the same
 * Add new flag gzip_encode for out_http plugin

Signed-off-by: Kushal Bhandari <[email protected]>

CMakeLists.txt

@@ -191,6 +191,7 @@ include_directories(
   lib/sha1
   lib/msgpack-2.1.3/include
   lib/luajit-2.0.5/src/
+  lib/miniz
   ${MONKEY_INCLUDE_DIR}
   )
 
@@ -270,6 +271,9 @@ if(FLB_SQLDB)
   include_directories(lib/sqlite-amalgamation-3230000)
 endif()
 
+# Gzip by default
+add_subdirectory(lib/miniz)
+
 if(FLB_BUFFERING)
   FLB_DEFINITION(FLB_HAVE_BUFFERING)
 endif()

include/fluent-bit/flb_gzip.h

@@ -0,0 +1,26 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2018 Treasure Data Inc.
+ *
+ *  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.
+ */
+
+#ifndef FLB_GZIP_H
+#define FLB_GZIP_H
+#include "miniz.h"
+
+void *flb_gzip_compress(void *data, size_t len,
+                     size_t *out_len);
+#endif

include/fluent-bit/flb_utils.h

@@ -59,4 +59,6 @@ int flb_utils_write_str_buf(char *str, size_t str_len, char **out, size_t *out_s
 int flb_utils_url_split(char *in_url, char **out_protocol,
                         char **out_host, char **out_port, char **out_uri);
 
+void *flb_utils_gzip_compress(void *data, size_t len, size_t *out_len);
+
 #endif

lib/miniz/CMakeLists.txt

@@ -0,0 +1,7 @@
+set(src
+  miniz.c)
+
+# Tweak Miniz library
+add_definitions("-DMINIZ_NO_ARCHIVE_APIS -DMINIZ_NO_STDIO -DMINIZ_NO_TIME")
+
+add_library(miniz STATIC ${src})

plugins/out_http/http.c

@@ -24,6 +24,7 @@
 #include <fluent-bit/flb_str.h>
 #include <fluent-bit/flb_time.h>
 #include <fluent-bit/flb_utils.h>
+#include <fluent-bit/flb_gzip.h>
 #include <msgpack.h>
 
 #include <stdio.h>
@@ -353,6 +354,13 @@ int cb_http_init(struct flb_output_instance *ins, struct flb_config *config,
         }
     }
 
+    /* Gzip encode body */
+    ctx->gzip_encode = FLB_FALSE;
+    tmp = flb_output_get_property("gzip_encode", ins);
+    if (tmp) {
+        ctx->gzip_encode = flb_utils_bool(tmp);
+    }
+
     /* Date format for JSON output */
     ctx->json_date_format = FLB_JSON_DATE_DOUBLE;
     tmp = flb_output_get_property("json_date_format", ins);
@@ -427,6 +435,8 @@ void cb_http_flush(void *data, size_t bytes,
     void *body = NULL;
     uint64_t body_len;
     (void)i_ins;
+    char *gz_body;
+    size_t gz_size;
 
     struct mk_list *tmp;
     struct mk_list *head;
@@ -442,6 +452,13 @@ void cb_http_flush(void *data, size_t bytes,
         body_len = bytes;
     }
 
+    if (ctx->gzip_encode) {
+        gz_body = flb_gzip_compress(body, body_len, &gz_size);
+        free(body);
+        body = gz_body;
+        body_len = gz_size;
+    }
+
     /* Get upstream context and connection */
     u = ctx->u;
     u_conn = flb_upstream_conn_get(u);
@@ -485,6 +502,14 @@ void cb_http_flush(void *data, size_t bytes,
                         tag, tag_len);
     }
 
+    if (ctx->gzip_encode) {
+        flb_http_add_header(c,
+            FLB_HTTP_CONTENT_ENCODING,
+            sizeof(FLB_HTTP_CONTENT_ENCODING) - 1,
+            FLB_HTTP_ENCODING_GZIP,
+            sizeof(FLB_HTTP_ENCODING_GZIP) - 1);
+    }
+
     if (ctx->http_user && ctx->http_passwd) {
         flb_http_basic_auth(c, ctx->http_user, ctx->http_passwd);
     }

plugins/out_http/http.h

@@ -32,6 +32,8 @@
 #define FLB_HTTP_CONTENT_TYPE   "Content-Type"
 #define FLB_HTTP_MIME_MSGPACK   "application/msgpack"
 #define FLB_HTTP_MIME_JSON      "application/json"
+#define FLB_HTTP_CONTENT_ENCODING "Content-Encoding"
+#define FLB_HTTP_ENCODING_GZIP "gzip"
 
 struct flb_out_http_config {
     /* HTTP Auth */
@@ -65,6 +67,9 @@ struct flb_out_http_config {
     /* Arbitrary HTTP headers */
     struct mk_list headers;
     int headers_cnt;
+
+    /* gzip content-encoding */
+    int gzip_encode;
 };
 
 struct out_http_header {

plugins/out_td/CMakeLists.txt

@@ -1,11 +1,7 @@
 set(src
-  miniz/miniz.c
   td_http.c
   td_config.c
   td.c)
 
-# Tweak Miniz library
-add_definitions("-DMINIZ_NO_ARCHIVE_APIS -DMINIZ_NO_STDIO -DMINIZ_NO_TIME")
-
 FLB_PLUGIN(out_td "${src}" "mk_core")
 target_link_libraries(flb-plugin-out_td)

plugins/out_td/td_http.c

@@ -23,109 +23,14 @@
 
 #include <fluent-bit/flb_config.h>
 #include <fluent-bit/flb_http_client.h>
+#include <fluent-bit/flb_utils.h>
+#include <fluent-bit/flb_gzip.h>
 
 #include "td_config.h"
-#include "miniz/miniz.h"
 
 #define TD_HTTP_HEADER_SIZE  512
 
-static void *gzip_compress(void *data, size_t len, size_t *out_len)
-{
-    int flush;
-    int status;
-    int buf_len;
-    void *buf;
-    z_stream strm;
-
-    /* Allocate buffer */
-    buf_len = len + 32;
-    buf = flb_malloc(buf_len);
-    if (!buf) {
-        flb_errno();
-        return NULL;
-    }
-
-    /*
-     * Miniz don't support GZip format directly, instead we will:
-     *
-     * - append manual GZip magic bytes
-     * - inflate raw content
-     * - append manual CRC32 data
-     */
-
-    /* GZIP Magic bytes */
-    uint8_t *pb = buf;
-
-    pb[0] = 0x1F;
-    pb[1] = 0x8B;
-    pb[2] = 8;
-    pb[3] = 0;
-    pb[4] = 0;
-    pb[5] = 0;
-    pb[6] = 0;
-    pb[7] = 0;
-    pb[8] = 0;
-    pb[9] = 0xFF;
-    pb += 10;
-
-    /* Prepare streaming buffer context */
-    memset(&strm, '\0', sizeof(strm));
-    strm.zalloc = Z_NULL;
-    strm.zfree  = Z_NULL;
-    strm.opaque = Z_NULL;
-    strm.next_in   = data;
-    strm.avail_in  = len;
-    strm.total_out = 0;
-
-    flush = Z_NO_FLUSH;
-    deflateInit2(&strm, Z_DEFAULT_COMPRESSION,
-                 Z_DEFLATED, -Z_DEFAULT_WINDOW_BITS, 9, Z_DEFAULT_STRATEGY);
-
-    while (1) {
-        strm.next_out  = pb + strm.total_out;
-        strm.avail_out = buf_len - strm.total_out;
 
-        if (strm.avail_in == 0) {
-            flush = Z_FINISH;
-        }
-
-        status = deflate(&strm, flush);
-        if (status == Z_STREAM_END) {
-            break;
-        }
-        else if (status != Z_OK) {
-            deflateEnd(&strm);
-            flb_free(buf);
-            return NULL;
-        }
-    }
-
-    if (deflateEnd(&strm) != Z_OK) {
-        flb_free(buf);
-        return NULL;
-    }
-    *out_len = strm.total_out;
-
-    /* Construct the GZip CRC32 (footer) */
-    mz_ulong crc;
-    int footer_start = strm.total_out + 10;
-
-    crc = mz_crc32(MZ_CRC32_INIT, data, len);
-    pb = buf;
-    pb[footer_start] = crc & 0xFF;
-    pb[footer_start + 1] = (crc >> 8) & 0xFF;
-    pb[footer_start + 2] = (crc >> 16) & 0xFF;
-    pb[footer_start + 3] = (crc >> 24) & 0xFF;
-    pb[footer_start + 4] = len & 0xFF;
-    pb[footer_start + 5] = (len >> 8) & 0xFF;
-    pb[footer_start + 6] = (len >> 16) & 0xFF;
-    pb[footer_start + 7] = (len >> 24) & 0xFF;
-
-    /* Set the real buffer size for the caller */
-    *out_len += 10 + 8;
-
-    return buf;
-}
 
 struct flb_http_client *td_http_client(struct flb_upstream_conn *u_conn,
                                        void *data, size_t len,
@@ -141,7 +46,7 @@ struct flb_http_client *td_http_client(struct flb_upstream_conn *u_conn,
     struct flb_http_client *c;
 
     /* Compress data */
-    gz = gzip_compress(data, len, &gz_size);
+    gz = flb_gzip_compress(data, len, &gz_size);
     if (!gz) {
         flb_error("[td_http] error compressing data");
         return NULL;

src/CMakeLists.txt

@@ -14,6 +14,7 @@ set(src
   flb_sds.c
 
   flb_sha1.c
+  flb_gzip.c
   flb_pipe.c
   flb_meta.c
   flb_kernel.c
@@ -235,6 +236,9 @@ if(FLB_BUFFERING)
   target_link_libraries(fluent-bit-static sha1)
 endif()
 
+# gzip is enabled by default
+target_link_libraries(fluent-bit-static miniz)
+
 # Binary / Executable
 if(NOT FLB_WITHOUT_BIN)
   find_package (Threads)