feat: support compressed responses in loggers (#10884)

apisix/plugins/response-rewrite.lua

@@ -27,9 +27,7 @@ local pairs       = pairs
 local ipairs      = ipairs
 local type        = type
 local pcall       = pcall
-local zlib        = require("ffi-zlib")
-local str_buffer  = require("string.buffer")
-local is_br_libs_loaded, brotli = pcall(require, "brotli")
+local content_decode = require("apisix.utils.content-decode")
 
 
 local lrucache = core.lrucache.new({
@@ -203,83 +201,6 @@ local function check_set_headers(headers)
 end
 
 
-local function inflate_gzip(data)
-    local inputs = str_buffer.new():set(data)
-    local outputs = str_buffer.new()
-
-    local read_inputs = function(size)
-        local data = inputs:get(size)
-        if data == "" then
-            return nil
-        end
-        return data
-    end
-
-    local write_outputs = function(data)
-        return outputs:put(data)
-    end
-
-    local ok, err = zlib.inflateGzip(read_inputs, write_outputs)
-    if not ok then
-        return nil, err
-    end
-
-    return outputs:get()
-end
-
-
-local function brotli_stream_decode(read_inputs, write_outputs)
-    -- read 64k data per times
-    local read_size = 64 * 1024
-    local decompressor = brotli.decompressor:new()
-
-    local chunk, ok, res
-    repeat
-        chunk = read_inputs(read_size)
-        if chunk then
-            ok, res = pcall(function()
-                return decompressor:decompress(chunk)
-            end)
-        else
-            ok, res = pcall(function()
-                return decompressor:finish()
-            end)
-        end
-        if not ok then
-            return false, res
-        end
-        write_outputs(res)
-    until not chunk
-
-    return true, nil
-end
-
-
-local function brotli_decode(data)
-    local inputs = str_buffer.new():set(data)
-    local outputs = str_buffer.new()
-
-    local read_inputs = function(size)
-        local data = inputs:get(size)
-        if data == "" then
-            return nil
-        end
-        return data
-    end
-
-    local write_outputs = function(data)
-        return outputs:put(data)
-    end
-
-    local ok, err = brotli_stream_decode(read_inputs, write_outputs)
-    if not ok then
-        return nil, err
-    end
-
-    return outputs:get()
-end
-
-
 function _M.check_schema(conf)
     local ok, err = core.schema.check(schema, conf)
     if not ok then
@@ -341,23 +262,19 @@ function _M.body_filter(conf, ctx)
         end
 
         local err
-        if ctx.response_encoding == "gzip" then
-            body, err = inflate_gzip(body)
-            if err ~= nil then
-                core.log.error("filters may not work as expected, inflate gzip err: ", err)
+        if ctx.response_encoding ~= nil then
+            local decoder = content_decode.dispatch_decoder(ctx.response_encoding)
+            if not decoder then
+                core.log.error("filters may not work as expected ",
+                               "due to unsupported compression encoding type: ",
+                               ctx.response_encoding)
                 return
             end
-        elseif ctx.response_encoding == "br" and is_br_libs_loaded then
-            body, err = brotli_decode(body)
+            body, err = decoder(body)
             if err ~= nil then
-                core.log.error("filters may not work as expected, brotli decode err: ", err)
+                core.log.error("filters may not work as expected: ", err)
                 return
             end
-        elseif ctx.response_encoding ~= nil then
-            core.log.error("filters may not work as expected ",
-                           "due to unsupported compression encoding type: ",
-                           ctx.response_encoding)
-            return
         end
 
         for _, filter in ipairs(conf.filters) do

apisix/utils/content-decode.lua

@@ -0,0 +1,112 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements.  See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You 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.
+--
+local pcall = pcall
+local zlib = require("ffi-zlib")
+local str_buffer = require("string.buffer")
+local is_br_libs_loaded, brotli = pcall(require, "brotli")
+local content_decode_funcs = {}
+local _M = {}
+
+
+local function inflate_gzip(data)
+    local inputs = str_buffer.new():set(data)
+    local outputs = str_buffer.new()
+
+    local read_inputs = function(size)
+        local data = inputs:get(size)
+        if data == "" then
+            return nil
+        end
+        return data
+    end
+
+    local write_outputs = function(data)
+        return outputs:put(data)
+    end
+
+    local ok, err = zlib.inflateGzip(read_inputs, write_outputs)
+    if not ok then
+        return nil, "inflate gzip err: " .. err
+    end
+
+    return outputs:get()
+end
+content_decode_funcs.gzip = inflate_gzip
+
+
+local function brotli_stream_decode(read_inputs, write_outputs)
+    -- read 64k data per times
+    local read_size = 64 * 1024
+    local decompressor = brotli.decompressor:new()
+
+    local chunk, ok, res
+    repeat
+        chunk = read_inputs(read_size)
+        if chunk then
+            ok, res = pcall(function()
+                return decompressor:decompress(chunk)
+            end)
+        else
+            ok, res = pcall(function()
+                return decompressor:finish()
+            end)
+        end
+        if not ok then
+            return false, res
+        end
+        write_outputs(res)
+    until not chunk
+
+    return true, nil
+end
+
+
+local function brotli_decode(data)
+    local inputs = str_buffer.new():set(data)
+    local outputs = str_buffer.new()
+
+    local read_inputs = function(size)
+        local data = inputs:get(size)
+        if data == "" then
+            return nil
+        end
+        return data
+    end
+
+    local write_outputs = function(data)
+        return outputs:put(data)
+    end
+
+    local ok, err = brotli_stream_decode(read_inputs, write_outputs)
+    if not ok then
+        return nil, "brotli decode err: " .. err
+    end
+
+    return outputs:get()
+end
+
+if is_br_libs_loaded then
+    content_decode_funcs.br = brotli_decode
+end
+
+
+function _M.dispatch_decoder(response_encoding)
+    return content_decode_funcs[response_encoding]
+end
+
+
+return _M

apisix/utils/log-util.lua

@@ -17,9 +17,11 @@
 local core = require("apisix.core")
 local plugin = require("apisix.plugin")
 local expr = require("resty.expr.v1")
-local ngx  = ngx
+local content_decode = require("apisix.utils.content-decode")
+local ngx = ngx
 local pairs = pairs
 local ngx_now = ngx.now
+local ngx_header = ngx.header
 local os_date = os.date
 local str_byte = string.byte
 local math_floor = math.floor
@@ -47,6 +49,7 @@ local function gen_log_format(format)
     return log_format
 end
 
+
 local function get_custom_format_log(ctx, format)
     local log_format = lru_log_format(format or "", nil, gen_log_format, format)
     local entry = core.table.new(0, core.table.nkeys(log_format))
@@ -311,7 +314,29 @@ function _M.collect_body(conf, ctx)
             if not final_body then
                 return
             end
-            ctx.resp_body = final_body
+
+            local response_encoding = ngx_header["Content-Encoding"]
+            if not response_encoding then
+                ctx.resp_body = final_body
+                return
+            end
+
+            local decoder = content_decode.dispatch_decoder(response_encoding)
+            if not decoder then
+                core.log.warn("unsupported compression encoding type: ",
+                              response_encoding)
+                ctx.resp_body = final_body
+                return
+            end
+
+            local decoded_body, err = decoder(final_body)
+            if err ~= nil then
+                core.log.warn("try decode compressed data err: ", err)
+                ctx.resp_body = final_body
+                return
+            end
+
+            ctx.resp_body = decoded_body
         end
     end
 end