perf(http_proxy): stream the response to avoid memory bloat

Author: tkan145

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ### Fixed
 - Correct FAPI header to `x-fapi-interaction-id` [PR #1557](https://github.com/3scale/APIcast/pull/1557) [THREESCALE-11957](https://issues.redhat.com/browse/THREESCALE-11957)
 - Only validate oidc setting if authentication method is set to oidc [PR #1568](https://github.com/3scale/APIcast/pull/1568) [THREESCALE-11441](https://issues.redhat.com/browse/THREESCALE-11441)
+- Reduce memory consumption when returning large response that has been routed through a proxy server. [PR #1572](https://github.com/3scale/APIcast/pull/1572) [THREESCALE-12258](https://issues.redhat.com/browse/THREESCALE-12258)
 
 ### Added
 - Update APIcast schema manifest [PR #1550](https://github.com/3scale/APIcast/pull/1550)

gateway/src/apicast/http_proxy.lua

@@ -13,7 +13,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 concat = table.concat
+local proxy_response = require("resty.http.response_writer").proxy_response
 
 local _M = { }
 
@@ -161,13 +161,19 @@ local function forward_https_request(proxy_uri, uri, proxy_opts)
 
     if res then
         if opts.request_unbuffered and raw then
-            local bytes, err = send_response(sock, res, DEFAULT_CHUNKSIZE)
-            if not bytes then
+            err = send_response(sock, res, DEFAULT_CHUNKSIZE)
+            if err then
                 ngx.log(ngx.ERR, "failed to send response: ", err)
-                return sock:send("HTTP/1.1 502 Bad Gateway")
+                sock:close()
+                return ngx.exit(ngx.HTTP_BAD_GATEWAY)
             end
         else
-            httpc:proxy_response(res)
+            err = proxy_response(res, DEFAULT_CHUNKSIZE)
+            if err then
+                ngx.log(ngx.ERR, 'failed to proxy request to: ', proxy_uri, ' err : ', err)
+                httpc:close()
+                return
+            end
             httpc:set_keepalive()
         end
     else

gateway/src/resty/http/response_writer.lua

@@ -3,6 +3,8 @@ local str_lower = string.lower
 local insert = table.insert
 local concat = table.concat
 
+local ngx       = ngx
+
 local _M = {
 }
 
@@ -31,6 +33,45 @@ local function send(socket, data)
     return socket:send(data)
 end
 
+local function send_chunk(chunk)
+    if not chunk then
+        return nil
+    end
+
+    local ok, err = ngx.print(chunk)
+    if not ok then
+        return "output response failed: " .. (err or "")
+    end
+
+    return nil
+end
+
+-- 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 "no body reader"
+  end
+
+  local buffer_size = chunksize or 65536
+
+  repeat
+      local buffer, read_err, send_err
+      buffer, read_err = reader(buffer_size)
+      if read_err then
+          return "failed to read response body: " .. read_err
+      end
+
+      if buffer then
+          send_err = cb(buffer)
+          if send_err then
+              return "failed to send response body: " .. (send_err or "unknown")
+          end
+      end
+  until not buffer
+end
+
 -- write_response writes response body reader to sock in the HTTP/1.x server response format,
 -- The connection is closed if send() fails or when returning a non-zero
 function _M.send_response(sock, response, chunksize)
@@ -42,7 +83,7 @@ function _M.send_response(sock, response, chunksize)
     end
 
     if not sock then
-        return nil, "socket not initialized yet"
+        return "socket not initialized yet"
     end
 
     -- Build status line + headers into a single buffer to minimize send() calls
@@ -62,32 +103,31 @@ function _M.send_response(sock, response, chunksize)
 
     local bytes, err = sock:send(concat(buf))
     if not bytes then
-        return nil, "failed to send headers, err: " .. (err or "unknown")
-    end
-
-    -- Write body
-    local reader = response.body_reader
-    if not reader then
-      return nil, "no body reader"
+        return "failed to send headers, err: " .. (err or "unknown")
     end
 
-    repeat
-        local chunk, read_err
-
-        chunk, read_err = reader(chunksize)
-        if read_err then
-            return nil, "failed to read response body, err: " .. (err or "unknown")
+    return forward_body(response.body_reader, function(chunk)
+        bytes, err = send(sock, chunk)
+        if not bytes then
+            return "failed to send response body, err: " .. (err or "unknown")
         end
+    end, chunksize)
+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
-        end
-    until not chunk
+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
+  end
 
-    return true, nil
+  return forward_body(res.body_reader, send_chunk, chunksize)
 end
 
 return _M

spec/http_proxy_spec.lua

@@ -20,6 +20,8 @@ describe('http_proxy', function()
 
       local resty_http_proxy = require 'resty.http.proxy'
       stub(resty_http_proxy, 'new', function() return httpc end)
+      local http_writer = require 'resty.http.response_writer'
+      stub(http_writer, 'proxy_response')
     end
 
     before_each(function()