fix(http_proxy): flush response buffers to avoid memory bloat

Author: tkan145

gateway/src/apicast/http_proxy.lua

@@ -11,6 +11,7 @@ local file_reader = require("resty.file").file_reader
 local file_size = require("resty.file").file_size
 local client_body_reader = require("resty.http.request_reader").get_client_body_reader
 local send_response = require("resty.http.response_writer").send_response
+local proxy_response = require("resty.http.response_writer").proxy_response
 local concat = table.concat
 
 local _M = { }
@@ -165,7 +166,11 @@ local function forward_https_request(proxy_uri, uri, proxy_opts)
                 return sock:send("HTTP/1.1 502 Bad Gateway")
             end
         else
-            httpc:proxy_response(res)
+            local ok, proxy_err = proxy_response(res, DEFAULT_CHUNKSIZE)
+            if not ok then
+              ngx.log(ngx.ERR, 'failed to proxy request to: ', proxy_uri, ' err : ', proxy_err)
+              return ngx.exit(ngx.HTTP_BAD_GATEWAY)
+            end
             httpc:set_keepalive()
         end
     else

gateway/src/resty/http/proxy.lua

@@ -4,7 +4,7 @@ local http = require 'resty.resolver.http'
 local resty_url = require 'resty.url'
 local resty_env = require 'resty.env'
 local url_helper = require('resty.url_helper')
-local format = string.format
+local format    = string.format
 
 local _M = {
 

gateway/src/resty/http/response_writer.lua

@@ -3,6 +3,11 @@ local str_lower = string.lower
 local insert = table.insert
 local concat = table.concat
 
+local ngx       = ngx
+local ngx_print = ngx.print
+local ngx_flush = ngx.flush
+local ngx_header= ngx.header
+
 local _M = {
 }
 
@@ -22,13 +27,31 @@ local HOP_BY_HOP_HEADERS = {
                                     -- with this (may send chunked for example).
 }
 
-local function send(socket, data)
-    if not data or data == '' then
-        ngx.log(ngx.DEBUG, 'skipping sending nil')
-        return
+-- forward_body reads chunks from a body_reader and passes them to the callback
+-- function cb.
+-- cb(chunk) should return a true on success, or nil/false, err on failure.
+local function forward_body(reader, cb, chunksize)
+  if not reader then
+    return nil, "no body reader"
+  end
+
+  repeat
+    local chunk, read_err = reader(chunksize)
+
+    if read_err then
+      return nil, "failed to read response body: " .. read_err
     end
 
-    return socket:send(data)
+    if chunk then
+      local ok, send_err = cb(chunk)
+
+      if not ok then
+          return nil, "failed to send response body: " .. (send_err or "unknown")
+      end
+    end
+  until not chunk
+
+  return true
 end
 
 -- write_response writes response body reader to sock in the HTTP/1.x server response format,
@@ -65,25 +88,40 @@ function _M.send_response(sock, response, chunksize)
         return nil, "failed to send headers, err: " .. (err or "unknown")
     end
 
-    -- Write body
-    local reader = response.body_reader
-    repeat
-        local chunk, read_err
+    return forward_body(response.body_reader, function(chunk)
+      return sock:send(chunk)
+    end, chunksize)
+end
 
-        chunk, read_err = reader(chunksize)
-        if read_err then
-            return nil, "failed to read response body, err: " .. (err or "unknown")
-        end
+local function ngx_send_and_flush(chunk)
+    local ok, err = ngx_print(chunk)
+    if not ok then
+        return nil, "output response failed: " .. (err or "")
+    end
 
-        if chunk then
-            bytes, err = send(sock, chunk)
-            if not bytes then
-                return nil, "failed to send response body, err: " .. (err or "unknown")
-            end
+    ok, err = ngx_flush(true)
+    if not ok then
+        ngx.log(ngx.WARN, "flush response failed: ", err)
+    end
+
+    return true
+end
+
+function _M.proxy_response(res, chunksize)
+    if not res then
+        ngx.log(ngx.ERR, "no response provided")
+        return
+    end
+
+    ngx.status = res.status
+
+    for k, v in pairs(res.headers) do
+        if not HOP_BY_HOP_HEADERS[str_lower(k)] then
+            ngx_header[k] = v
         end
-    until not chunk
+    end
 
-    return true, nil
+    return forward_body(res.body_reader, ngx_send_and_flush, chunksize)
 end
 
 return _M