Merge pull request #1025 from 3scale/upstream-config-policy

[THREESCALE-2166] Add Upstream Connection policy

Author: davidor

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 - Ability to configure client certificate chain depth [PR #1006](https://github.com/3scale/APIcast/pull/1006)
 - You can filter services by endpoint name using Regexp [PR #1022](https://github.com/3scale/APIcast/pull/1022) [THREESCALE-1524](https://issues.jboss.org/browse/THREESCALE-1524)
+- "Upstream Connection" policy. It allows to configure several options for the connections to the upstream [PR #1025](https://github.com/3scale/APIcast/pull/1025), [THREESCALE-2166](https://issues.jboss.org/browse/THREESCALE-2166)
 
 ### Fixed
 

gateway/src/apicast/balancer.lua

@@ -1,6 +1,7 @@
 local assert = assert
 local round_robin = require 'resty.balancer.round_robin'
 local resty_url = require 'resty.url'
+local inspect = require 'inspect'
 
 local _M = { default_balancer = round_robin.new() }
 
@@ -52,6 +53,23 @@ local function get_peer(balancer, upstream)
   return peer
 end
 
+local function set_timeouts(balancer, timeouts)
+  if not timeouts then return end
+
+  local _, err = balancer:set_timeouts(
+      timeouts.connect_timeout,
+      timeouts.send_timeout,
+      timeouts.read_timeout
+  )
+
+  if err then
+    ngx.log(ngx.WARN,
+            'Error while setting balancer timeouts: ',
+            inspect(timeouts),
+            ' err: ', err)
+  end
+end
+
 function _M.call(_, context, bal)
   local balancer = assert(bal or _M.default_balancer, 'missing balancer')
   local upstream, peer, err, ok
@@ -71,6 +89,9 @@ function _M.call(_, context, bal)
   ok, err = balancer:set_current_peer(peer[1], peer[2] or upstream.uri.port or resty_url.default_port(upstream.uri.scheme))
 
   if ok then
+    -- context.upstream_connection_opts is set by the "upstream_connection" policy.
+    set_timeouts(balancer, context.upstream_connection_opts)
+
     -- I wish there would be a nicer way, but unfortunately ngx.exit(ngx.OK) does not
     -- terminate the current phase handler and will evaluate all remaining balancer phases.
     context[upstream] = peer

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

@@ -0,0 +1,26 @@
+{
+  "$schema": "http://apicast.io/policy-v1/schema#manifest#",
+  "name": "Upstream connection",
+  "summary": "Allows to configure several options for the connections to the upstream",
+  "description": "Allows to configure several options for the connections to the upstream",
+  "version": "builtin",
+  "configuration": {
+    "type": "object",
+    "properties": {
+      "connect_timeout": {
+        "description": "Timeout for establishing a connection (in seconds).",
+        "type": "integer"
+      },
+      "send_timeout": {
+        "description": "Timeout between two successive write operations (in seconds).",
+        "type": "number",
+        "exclusiveMinimum": 0
+      },
+      "read_timeout": {
+        "description": "Timeout between two successive read operations (in seconds).",
+        "type": "number",
+        "exclusiveMinimum": 0
+      }
+    }
+  }
+}

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

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

gateway/src/apicast/policy/upstream_connection/upstream_connection.lua

@@ -0,0 +1,31 @@
+-- Upstream Connection Policy
+--- This policy exposes some parameters related with the connection to the
+--- upstream.
+
+local tonumber = tonumber
+
+local _M = require('apicast.policy').new('Upstream connection policy')
+
+local new = _M.new
+
+function _M.new(config)
+  local self = new(config)
+
+  self.connect_timeout = tonumber(config.connect_timeout)
+  self.send_timeout = tonumber(config.send_timeout)
+  self.read_timeout = tonumber(config.read_timeout)
+
+  return self
+end
+
+function _M:export()
+  return {
+    upstream_connection_opts = {
+      connect_timeout = self.connect_timeout,
+      send_timeout = self.send_timeout,
+      read_timeout = self.read_timeout
+    }
+  }
+end
+
+return _M

gateway/src/resty/balancer.lua

@@ -137,4 +137,16 @@ function _M.set_peer(self, peers)
   end
 end
 
+function _M:set_timeouts(connect_timeout, send_timeout, read_timeout)
+  local ngx_balancer = self.balancer
+
+  if not ngx_balancer then
+    return nil, 'balancer not available'
+  end
+
+  -- If one of the values is nil, the default applies:
+  -- https://github.com/openresty/lua-resty-core/blob/master/lib/ngx/balancer.md#set_timeouts
+  return ngx_balancer.set_timeouts(connect_timeout, send_timeout, read_timeout)
+end
+
 return _M

spec/balancer_spec.lua

@@ -23,5 +23,37 @@ describe('apicast.balancer', function()
       assert(apicast_balancer:call({ upstream =  upstream }, balancer))
       assert.spy(balancer.set_current_peer).was.called_with(balancer, '127.0.0.2', 443)
     end)
+
+    it('sets the timeouts in the balancer when they are received in the context', function()
+      local timeouts = {
+        connect_timeout = 1,
+        send_timeout = 2,
+        read_timeout = 3,
+      }
+
+      local balancer = setmetatable({
+        set_current_peer = spy.new(function() return true end),
+        set_timeouts = spy.new(function() return true end),
+      }, { __index = apicast_balancer.default_balancer })
+
+      local upstream = Upstream.new('https://127.0.0.2')
+
+      upstream.servers = {
+        { address = '127.0.0.2', port = nil }
+      }
+
+      apicast_balancer:call(
+          { upstream =  upstream,
+            upstream_connection_opts = timeouts },
+          balancer
+      )
+
+      assert.spy(balancer.set_timeouts).was_called_with(
+          balancer,
+          timeouts.connect_timeout,
+          timeouts.send_timeout,
+          timeouts.read_timeout
+      )
+    end)
   end)
 end)

spec/policy/upstream_connection/upstream_connection_spec.lua

@@ -0,0 +1,27 @@
+local UpstreamConnectionPolicy = require('apicast.policy.upstream_connection')
+
+describe('Upstream connection policy', function()
+  describe('.export', function()
+    it('returns the timeouts included in the config', function()
+      local config_timeouts = {
+        connect_timeout = 1,
+        send_timeout = 2,
+        read_timeout = 3
+      }
+      local policy = UpstreamConnectionPolicy.new(config_timeouts)
+
+      local exported = policy:export()
+
+      assert.same(config_timeouts, exported.upstream_connection_opts)
+    end)
+
+    it('does not return timeout params that is not in the config', function()
+      local config_timeouts = { connect_timeout = 1 } -- Missing send and read
+      local policy = UpstreamConnectionPolicy.new(config_timeouts)
+
+      local exported = policy:export()
+
+      assert.same(config_timeouts, exported.upstream_connection_opts)
+    end)
+  end)
+end)

spec/resty/balancer_spec.lua

@@ -127,4 +127,26 @@ describe('resty.balancer', function()
     end)
 
   end)
+
+  describe('set_timeouts', function()
+    it('forwards the call to the internal balancer', function()
+      local b = resty_balancer.new(function() end)
+      b.balancer = { } -- Internal balancer
+      stub.new(b.balancer, 'set_timeouts', function() end)
+
+      b:set_timeouts(1, 2, 3)
+
+      assert.stub(b.balancer.set_timeouts).was_called_with(1, 2, 3)
+    end)
+
+    it('returns an error when the internal balancer has not been set', function()
+      local b = resty_balancer.new(function() end)
+      b.balancer = nil -- Internal balancer
+
+      local res, err = b:set_timeouts(1, 2, 3)
+
+      assert.is_nil(res)
+      assert.equals('balancer not available', err)
+    end)
+  end)
 end)

t/apicast-policy-upstream-connection.t

@@ -0,0 +1,61 @@
+use lib 't';
+use Test::APIcast::Blackbox 'no_plan';
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: Set timeouts
+In this test we set some timeouts to 1s. To force a read timeout, the upstream
+returns part of the response, then waits 3s (more than the timeout defined),
+and after that, it returns the rest of the response.
+This test uses the "ignore_response" section, because we know that the response
+is not going to be complete and that makes the Test::Nginx framework raise an
+error. With "ignore_response" that error is ignored.
+--- configuration
+{
+  "services": [
+    {
+      "id": 42,
+      "proxy": {
+        "api_backend": "http://example.com:80/",
+        "policy_chain": [
+          {
+            "name": "apicast.policy.upstream_connection",
+            "configuration": {
+              "connect_timeout": 1,
+              "send_timeout": 1,
+              "read_timeout": 1
+            }
+          },
+          {
+            "name": "apicast.policy.upstream",
+            "configuration": {
+              "rules": [
+                {
+                  "regex": "/",
+                  "url": "http://test:$TEST_NGINX_SERVER_PORT"
+                }
+              ]
+            }
+          }
+        ]
+      }
+    }
+  ]
+}
+--- upstream
+  location / {
+     content_by_lua_block {
+       ngx.say("first part")
+       ngx.flush(true)
+       ngx.sleep(3)
+       ngx.say("yay, second part")
+     }
+  }
+--- request
+GET /
+--- ignore_response
+--- error_log
+upstream timed out
+--- error_code: