Support gzip for otlp export

* another take on yangxikun#87

Dockerfile

@@ -1,9 +1,10 @@
 FROM openresty/openresty:1.21.4.1-0-centos
 
-RUN yum install -y gcc
+RUN yum install -y gcc git zlib-devel
 RUN yum -y --enablerepo=powertools install libyaml-devel libffi-devel
 RUN luarocks install lua-resty-http 0.16.1-0
 RUN luarocks install lua-protobuf 0.3.3
+RUN luarocks install lua-zlib 1.2
 RUN luarocks install busted 2.0.0-1
 RUN luarocks --server=http://rocks.moonscript.org install lyaml
 

lib/opentelemetry/trace/exporter/http_client.lua

@@ -32,13 +32,18 @@ function _M.new(address, timeout, headers)
     return setmetatable(self, mt)
 end
 
-function _M.do_request(self, body)
+function _M.do_request(self, body, additional_headers)
     local httpc = http.new()
     httpc:set_timeout(self.timeout * 1000)
+    local headers = self.headers
+    if additional_headers then
+        headers = additional_headers
+        for k,v in pairs(self.headers) do headers[k] = v end
+    end
 
     local res, err = httpc:request_uri(self.uri, {
         method = "POST",
-        headers = self.headers,
+        headers = headers,
         body = body,
     })
 

lib/opentelemetry/trace/exporter/otlp.lua

@@ -1,3 +1,4 @@
+local zlib = require("zlib")
 local encoder = require("opentelemetry.trace.exporter.encoder")
 local pb = require("opentelemetry.trace.exporter.pb")
 local otel_global = require("opentelemetry.global")
@@ -14,10 +15,11 @@ local mt = {
     __index = _M
 }
 
-function _M.new(http_client, timeout_ms, circuit_reset_timeout_ms, circuit_open_threshold)
+function _M.new(http_client, timeout_ms, gzip, circuit_reset_timeout_ms, circuit_open_threshold)
     local self = {
         client = http_client,
         timeout_ms = timeout_ms or DEFAULT_TIMEOUT_MS,
+        gzip = gzip or false,
         circuit = circuit.new({
             reset_timeout_ms = circuit_reset_timeout_ms,
             failure_threshold = circuit_open_threshold
@@ -40,6 +42,16 @@ local function call_collector(exporter, pb_encoded_body)
     local failures = 0
     local res
     local res_error
+    local headers
+    if exporter.gzip then
+        -- Compress (deflate) request body
+        -- the compression should be set to Best Compression and window size
+        -- should be set to 15+16, see reference below:
+        -- https://github.com/brimworks/lua-zlib/issues/4#issuecomment-26383801
+        local deflate_stream = zlib.deflate(zlib.BEST_COMPRESSION, 15+16)
+        pb_encoded_body, _, _, bytes_out = deflate_stream(pb_encoded_body, "finish")
+        headers = {["Content-Encoding"]= "gzip"}
+    end
 
     while failures < BACKOFF_RETRY_LIMIT do
         local current_time = util.gettimeofday_ms()
@@ -55,7 +67,7 @@ local function call_collector(exporter, pb_encoded_body)
         end
 
         -- Make request
-        res, res_error = exporter.client:do_request(pb_encoded_body)
+        res, res_error = exporter.client:do_request(pb_encoded_body, headers)
         local after_time = util.gettimeofday_ms()
         otel_global.metrics_reporter:record_value(
             exporter_request_duration_metric, after_time - current_time)

rockspec/opentelemetry-lua-master-0.rockspec

@@ -14,6 +14,7 @@ description = {
 dependencies = {
     "lua-protobuf = 0.3.3",
     "lua-resty-http = 0.16.1-0",
+    "lua-zlib = 1.2",
 }
 
 build = {

spec/trace/exporter/otlp_spec.lua

@@ -4,6 +4,23 @@ local context = require "opentelemetry.context"
 local tp = Global.get_tracer_provider()
 local tracer = tp:tracer("test")
 
+local function is_gzip(_, _)
+    return function(value)
+        -- check that the value starts with the two magic bytes 0x1f, 0x8b and
+        -- the compression method byte set to 0x08
+        -- reference: https://www.ietf.org/rfc/rfc1952.txt
+        return string.sub(value, 1, 3) == string.from_hex("1F8B08")
+    end
+end
+
+assert:register("matcher", "gzip", is_gzip)
+
+function string.from_hex(str)
+    return (str:gsub('..', function (cc)
+        return string.char(tonumber(cc, 16))
+    end))
+end
+
 describe("export_spans", function()
     it("invokes do_request when there are no failures", function()
         local span
@@ -17,7 +34,26 @@ describe("export_spans", function()
         stub(ngx, "log")
         cb:export_spans({ span })
         ngx.log:revert()
-        assert.spy(c.do_request).was_called_with(c, match.is_string())
+        assert.spy(c.do_request).was_called_with(c, match.is_string(), match.is_nil())
+    end)
+
+	it("invokes do_request with gzip compression when configured", function()
+        local span
+        local ctx = context.new()
+        ctx, span = tracer:start(ctx, "test span")
+        span:finish()
+
+        local c = client.new("http://localhost:8080", 10)
+        spy.on(c, "do_request")
+
+        local cb = exporter.new(c, 10000, true)
+
+        -- Supress log message, since we expect it
+        stub(ngx, "log")
+
+        cb:export_spans({ span })
+        ngx.log:revert()
+        assert.spy(c.do_request).was_called_with(c, match.is_all_of(match.is_string(), match.is_gzip()), match.is_not_nil())
     end)
 
     it("doesn't invoke protected_call when failures is equal to retry limit", function()