Merge pull request #1572 from tkan145/THREESCALE-12258

THREESCALE-12258 Stream response back when using proxy

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

@@ -1,7 +1,9 @@
 local format = string.format
 local tostring = tostring
+local ngx = ngx
 local ngx_get_method = ngx.req.get_method
 local ngx_http_version = ngx.req.http_version
+local ngx_req_get_headers = ngx.req.get_headers
 
 local resty_url = require "resty.url"
 local url_helper = require('resty.url_helper')
@@ -11,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 = { }
 
@@ -49,9 +51,9 @@ local function forward_https_request(proxy_uri, uri, proxy_opts)
     local sock
     local opts = proxy_opts or {}
     local req_method = ngx_get_method()
-    local encoding = ngx.req.get_headers()["Transfer-Encoding"]
+    local encoding = ngx_req_get_headers()["Transfer-Encoding"]
     local is_chunked = encoding and encoding:lower() == "chunked"
-    local content_type = ngx.req.get_headers()["Content-Type"]
+    local content_type = ngx_req_get_headers()["Content-Type"]
     local content_type_is_urlencoded = content_type and content_type:lower() == "application/x-www-form-urlencoded"
     local raw = false
 
@@ -138,9 +140,9 @@ local function forward_https_request(proxy_uri, uri, proxy_opts)
 
     local request = {
         uri = uri,
-        method = ngx.req.get_method(),
-        headers = ngx.req.get_headers(0, true),
-        path = format('%s%s%s', ngx.var.uri, ngx.var.is_args, ngx.var.query_string or ''),
+        method = req_method,
+        headers = ngx_req_get_headers(0, true),
+        path = (ngx.var.uri or '') .. (ngx.var.is_args or '') .. (ngx.var.query_string or ''),
         body = body,
         proxy_uri = proxy_uri,
         proxy_options = opts
@@ -159,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
@@ -194,7 +202,7 @@ function _M.request(upstream, proxy_uri)
     local proxy_auth
 
     if proxy_uri.user or proxy_uri.password then
-        proxy_auth = "Basic " .. ngx.encode_base64(concat({ proxy_uri.user or '', proxy_uri.password or '' }, ':'))
+        proxy_auth = "Basic " .. ngx.encode_base64((proxy_uri.user or '') .. ":" .. (proxy_uri.password or ''))
     end
 
     if uri.scheme == 'http' then -- rewrite the request to use http_proxy

gateway/src/resty/http/response_writer.lua

@@ -1,5 +1,9 @@
 local fmt = string.format
 local str_lower = string.lower
+local insert = table.insert
+local concat = table.concat
+
+local ngx       = ngx
 
 local _M = {
 }
@@ -29,10 +33,48 @@ 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)
-    local bytes, err
     chunksize = chunksize or 65536
 
     if not response then
@@ -41,53 +83,51 @@ function _M.send_response(sock, response, chunksize)
     end
 
     if not sock then
-        return nil, "socket not initialized yet"
+        return "socket not initialized yet"
     end
 
-    -- Status line
-    -- TODO: get HTTP version from request
-    local status = fmt("HTTP/%d.%d %03d %s\r\n", 1, 1, response.status, response.reason)
-    bytes, err = send(sock, status)
-    if not bytes then
-        return nil, "failed to send status line, err: " .. (err or "unknown")
-    end
+    -- Build status line + headers into a single buffer to minimize send() calls
+    local buf = {
+        fmt("HTTP/1.1 %03d %s\r\n", response.status, response.reason)
+    }
 
      -- Filter out hop-by-hop headeres
      for k, v in pairs(response.headers) do
         if not HOP_BY_HOP_HEADERS[str_lower(k)] then
-            local header = fmt("%s: %s\r\n", k, v)
-            bytes, err = sock:send(header)
-            if not bytes then
-              return nil, "failed to send status line, err: " .. (err or "unknown")
-            end
+          insert(buf, k .. ": " .. v .. cr_lf)
         end
      end
 
     -- End-of-header
-    bytes, err = send(sock, cr_lf)
+    insert(buf, cr_lf)
+
+    local bytes, err = sock:send(concat(buf))
     if not bytes then
-        return nil, "failed to send status line, err: " .. (err or "unknown")
+        return "failed to send headers, err: " .. (err or "unknown")
     end
 
-    -- Write body
-    local reader = response.body_reader
-    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()