[THREESCALE-9542] Part 1: buffering policy

CHANGELOG.md

@@ -16,9 +16,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ### Added
 
 - Detect number of CPU shares when running on Cgroups V2 [PR #1410](https://github.com/3scale/apicast/pull/1410) [THREESCALE-10167](https://issues.redhat.com/browse/THREESCALE-10167)
-### Added
 
-* Add support to use Basic Authentication with the forward proxy. [PR #1409](https://github.com/3scale/APIcast/pull/1409)
+- Add support to use Basic Authentication with the forward proxy. [PR #1409](https://github.com/3scale/APIcast/pull/1409)
+
+- Added request unbuffered policy [PR #1408](https://github.com/3scale/APIcast/pull/1408) [THREESCALE-9542](https://issues.redhat.com/browse/THREESCALE-9542)
 
 ## [3.14.0] 2023-07-25
 

gateway/conf.d/apicast.conf

@@ -78,71 +78,25 @@ location @upstream {
     require('resty.ctx').apply()
   }
 
-  #{% capture proxy_cache_valid %}
-  #{#} proxy_cache $cache_zone;
-  #{#} proxy_cache_key $scheme$request_method$proxy_host$request_uri$service_id;
-  #{#} proxy_no_cache $cache_request;
-  #{#} proxy_cache_valid {{ env.APICAST_CACHE_STATUS_CODES | default: '200 302'}} {{ env.APICAST_CACHE_MAX_TIME | default: '1m' }};
-  #{% endcapture %}
-  #{{ proxy_cache_valid | replace: "#{#}", "" }}
-  #
-
-  #{% if opentelemetry != empty %}
-  #   {% capture opentelemetry_propagate_directive %}
-  #{#} opentelemetry_propagate;
-  #   {% endcapture %}
-  #   {{ opentelemetry_propagate_directive | replace: "#{#}", "" }}
-  #{% endif %}
+  proxy_request_buffering on;
+  #{% include "conf.d/upstream_shared.conf" %}
 
-  proxy_pass $proxy_pass;
+  # these are duplicated so when request is redirected here those phases are executed
+  post_action @out_of_band_authrep_action;
+  body_filter_by_lua_block { require('apicast.executor'):body_filter() }
+  header_filter_by_lua_block { require('apicast.executor'):header_filter() }
+}
 
-  proxy_http_version 1.1;
-  proxy_set_header X-Real-IP  $remote_addr;
-  proxy_set_header Host $http_host;
-  proxy_set_header X-3scale-proxy-secret-token $secret_token;
-  proxy_set_header X-3scale-debug "";
-  proxy_set_header Connection $upstream_connection_header;
-  proxy_set_header Upgrade $upstream_upgrade_header;
-
-  # This is a bit tricky. It uses liquid to set a SSL client certificate. In
-  # NGINX, all this is not executed as it is commented with '#'. However, in
-  # Liquid, all this will be evaluated. As a result, the following directives
-  # are set optionally: proxy_ssl_certificate, proxy_ssl_certificate_key,
-  # proxy_ssl_session_reuse, and proxy_ssl_password_file.
-
-  # {% if proxy_ssl_certificate != empty and proxy_ssl_certificate_key != empty %}
-  #   {% capture proxy_ssl %}
-  #{#}   proxy_ssl_certificate {{ proxy_ssl_certificate }};
-  #{#}   proxy_ssl_certificate_key {{ proxy_ssl_certificate_key }};
-  #   {% endcapture %}
-  #   {{ proxy_ssl | replace: "#{#}", "" }}
-  #
-  #   {% if proxy_ssl_password_file != empty %}
-  #     {% capture proxy_ssl %}
-  #{#}   proxy_ssl_password_file {{ proxy_ssl_password_file }};
-  #     {% endcapture %}
-  #   {{ proxy_ssl | replace: "#{#}", "" }}
-  #   {% endif %}
-  #
-  #   {% if proxy_ssl_session_reuse != empty %}
-  #     {% capture proxy_ssl %}
-  #{#}   proxy_ssl_session_reuse {{ proxy_ssl_session_reuse }};
-  #     {% endcapture %}
-  #   {{ proxy_ssl | replace: "#{#}", "" }}
-  #   {% endif %}
-  # {% endif %}
+location @upstream_request_unbuffered {
+  internal;
+
+  rewrite_by_lua_block {
+    require('resty.ctx').apply()
+  }
+
+  proxy_request_buffering off;
+  #{% include "conf.d/upstream_shared.conf" %}
 
-  # When 'upstream_retry_cases' is empty, apply the same default as NGINX.
-  # If the proxy_next_upstream directive is not declared, the retry policy
-  # will never retry.
-  # {% if upstream_retry_cases != empty %}
-  #   {% capture proxy_next_upstream %}
-  #{#}  proxy_next_upstream {{ upstream_retry_cases }};
-  #   {% endcapture %}
-  #   {{ proxy_next_upstream | replace: "#{#}", "" }}
-  # {% else %}
-  #   proxy_next_upstream error timeout;
-  # {% endif %}
   # these are duplicated so when request is redirected here those phases are executed
   post_action @out_of_band_authrep_action;
   body_filter_by_lua_block { require('apicast.executor'):body_filter() }

gateway/conf.d/upstream_shared.conf

@@ -0,0 +1,66 @@
+#{% capture proxy_cache_valid %}
+#{#} proxy_cache $cache_zone;
+#{#} proxy_cache_key $scheme$request_method$proxy_host$request_uri$service_id;
+#{#} proxy_no_cache $cache_request;
+#{#} proxy_cache_valid {{ env.APICAST_CACHE_STATUS_CODES | default: '200 302'}} {{ env.APICAST_CACHE_MAX_TIME | default: '1m' }};
+#{% endcapture %}
+#{{ proxy_cache_valid | replace: "#{#}", "" }}
+#
+
+#{% if opentelemetry != empty %}
+#   {% capture opentelemetry_propagate_directive %}
+#{#} opentelemetry_propagate;
+#   {% endcapture %}
+#   {{ opentelemetry_propagate_directive | replace: "#{#}", "" }}
+#{% endif %}
+
+proxy_pass $proxy_pass;
+
+proxy_http_version 1.1;
+
+proxy_set_header X-Real-IP  $remote_addr;
+proxy_set_header Host $http_host;
+proxy_set_header X-3scale-proxy-secret-token $secret_token;
+proxy_set_header X-3scale-debug "";
+proxy_set_header Connection $upstream_connection_header;
+proxy_set_header Upgrade $upstream_upgrade_header;
+
+# This is a bit tricky. It uses liquid to set a SSL client certificate. In
+# NGINX, all this is not executed as it is commented with '#'. However, in
+# Liquid, all this will be evaluated. As a result, the following directives
+# are set optionally: proxy_ssl_certificate, proxy_ssl_certificate_key,
+# proxy_ssl_session_reuse, and proxy_ssl_password_file.
+
+# {% if proxy_ssl_certificate != empty and proxy_ssl_certificate_key != empty %}
+#   {% capture proxy_ssl %}
+#{#}   proxy_ssl_certificate {{ proxy_ssl_certificate }};
+#{#}   proxy_ssl_certificate_key {{ proxy_ssl_certificate_key }};
+#   {% endcapture %}
+#   {{ proxy_ssl | replace: "#{#}", "" }}
+#
+#   {% if proxy_ssl_password_file != empty %}
+#     {% capture proxy_ssl %}
+#{#}   proxy_ssl_password_file {{ proxy_ssl_password_file }};
+#     {% endcapture %}
+#   {{ proxy_ssl | replace: "#{#}", "" }}
+#   {% endif %}
+#
+#   {% if proxy_ssl_session_reuse != empty %}
+#     {% capture proxy_ssl %}
+#{#}   proxy_ssl_session_reuse {{ proxy_ssl_session_reuse }};
+#     {% endcapture %}
+#   {{ proxy_ssl | replace: "#{#}", "" }}
+#   {% endif %}
+# {% endif %}
+
+# When 'upstream_retry_cases' is empty, apply the same default as NGINX.
+# If the proxy_next_upstream directive is not declared, the retry policy
+# will never retry.
+# {% if upstream_retry_cases != empty %}
+#   {% capture proxy_next_upstream %}
+#{#}  proxy_next_upstream {{ upstream_retry_cases }};
+#   {% endcapture %}
+#   {{ proxy_next_upstream | replace: "#{#}", "" }}
+# {% else %}
+#   proxy_next_upstream error timeout;
+# {% endif %}

gateway/src/apicast/policy/request_unbuffered/README.md

@@ -0,0 +1,14 @@
+# APICast Request Unbuffered
+
+This policy allows to disable request buffering
+
+## Example configuration
+
+```
+{
+    "name": "request_unbuffered",
+    "version": "builtin",
+    "configuration": {}
+}
+```
+

gateway/src/apicast/policy/request_unbuffered/apicast-policy.json

@@ -0,0 +1,13 @@
+{
+  "$schema": "http://apicast.io/policy-v1/schema#manifest#",
+  "name": "Request Unbuffered",
+  "summary": "Disable request buffering",
+  "description": [
+      "Disable request buffering. This is useful when proxying big payloads with HTTP/1.1 chunked encoding"
+  ],
+  "version": "builtin",
+  "configuration": {
+    "type": "object",
+    "properties": {}
+  }
+}

gateway/src/apicast/policy/request_unbuffered/init.lua

@@ -0,0 +1 @@
+return require('request_unbuffered')

gateway/src/apicast/policy/request_unbuffered/request_unbuffered.lua

@@ -0,0 +1,22 @@
+-- Request Unbuffered policy
+-- This policy will disable request buffering
+
+local policy = require('apicast.policy')
+local _M = policy.new('request_unbuffered')
+
+local new = _M.new
+
+--- Initialize a buffering
+-- @tparam[opt] table config Policy configuration.
+function _M.new(config)
+  local self = new(config)
+  return self
+end
+
+function _M:export()
+  return {
+    request_unbuffered = true,
+  }
+end
+
+return _M

gateway/src/apicast/upstream.lua

@@ -210,6 +210,15 @@ function _M:set_keepalive_key(context)
   end
 end
 
+local function get_upstream_location_name(context)
+    if context.upstream_location_name then
+        return context.upstream_location_name
+    end
+    if context.request_unbuffered then
+        return "@upstream_request_unbuffered"
+    end
+end
+
 --- Execute the upstream.
 --- @tparam table context any table (policy context, ngx.ctx) to store the upstream for later use by balancer
 function _M:call(context)
@@ -242,9 +251,9 @@ function _M:call(context)
 
     self:set_keepalive_key(context or {})
     if not self.servers then self:resolve() end
-    if context.upstream_location_name then
-        self.location_name = context.upstream_location_name
-    end
+
+    local upstream_location_name = get_upstream_location_name(context)
+    self:update_location(upstream_location_name)
     context[self.upstream_name] = self
 
     return exec(self)

spec/upstream_spec.lua

@@ -217,6 +217,22 @@ describe('Upstream', function()
             assert.spy(ngx.exec).was_called_with(upstream.location_name)
         end)
 
+        it('executes the upstream location when request_unbuffered provided in the context', function()
+            local contexts = {
+                ["buffered_request"] = {ctx={}, upstream_location="@upstream"},
+                ["unbuffered_request"] = {ctx={request_unbuffered=true}, upstream_location="@upstream_request_unbuffered"},
+                ["upstream_location and buffered_request"] = {ctx={upstream_location_name="@grpc", request_unbuffered=true}, upstream_location="@grpc"},
+                ["upstream_location and unbuffered_request"] = {ctx={upstream_location_name="@grpc"}, upstream_location="@grpc"},
+            }
+
+            for _, value in pairs(contexts) do
+                local upstream = Upstream.new('http://localhost')
+                upstream:call(value.ctx)
+
+                assert.spy(ngx.exec).was_called_with(value.upstream_location)
+            end
+        end)
+
         it('skips executing the upstream location when missing', function()
             local upstream = Upstream.new('http://localhost')
             upstream.location_name = nil