allow chaning service endpoint inside the middleware

lua/concurredis.lua

@@ -74,8 +74,9 @@ concurredis.disable_bgsave = function(fun)
     local save = red:config('get', 'save')
 
     local res, err = pcall(fun, red)
-
-    assert(red:config('set', unpack(save)))
+    if save then
+      assert(red:config('set', unpack(save)))
+    end
     assert(res, err)
   end)
 

lua/middlewares/brainslug.lua

@@ -4,13 +4,11 @@ local collector       = require 'collector'
 local Service         = require 'models.service'
 local statsd          = require 'statsd_wrapper'
 
-local function get_endpoint_host()
-  return string.match(ngx.var._endpoint_url, "^.+://([^/]+)")
+local function extract_host(url)
+  return string.match(url, "^.+://([^/]+)")
 end
 
-local pass_response = function(req, res, endpoint_url)
-  local start    = ngx.now()
-
+local pass_response = function(req, res)
   -- override original request body with middelware's version
   -- TODO: if req.body= would set the request body, we would not have to do it here
   if req.body then
@@ -21,19 +19,19 @@ local pass_response = function(req, res, endpoint_url)
   -- nginx is unable to resolve private names like localhost.  We
   -- hardcode the resolution of localhost to 127.0.0.0 for now,
   -- and remove trailing slash
-  ngx.var._endpoint_url  = endpoint_url:gsub('localhost', '127.0.0.1', 1):gsub('/$', '')
-  ngx.var._endpoint_host = get_endpoint_host()
+  ngx.var._endpoint_url  = req.endpoint:gsub('localhost', '127.0.0.1', 1):gsub('/$', '')
+  ngx.var._endpoint_host = extract_host(ngx.var._endpoint_url)
 
   ngx.var._path = req.uri or '/'
 
+  local start    = ngx.now()
   local response_data = ngx.location.capture("/___pass", {
     method         = ngx["HTTP_" .. req.method],
     args           = req.args,
     ctx = { headers = req.headers },
     always_forward_body = true,
     copy_all_vars = true
   })
-
   local elapsed_time = ngx.now() - start
 
   statsd.time('proxy.real_request', elapsed_time)
@@ -50,7 +48,8 @@ end
 
 local function get_generic_path(req, service_id, status)
   -- if successful request, we must build the swagger spec
-  local generic_path = nil
+  local generic_path
+
   if status >= 200 and status < 300 then
     local service = Service:find(service_id)
     if not service then
@@ -86,7 +85,6 @@ end
 return function(req, next_middleware, config)
   local start        = ngx.now()
   local trace        = config.trace
-  local endpoint_url = config.endpoint_url
   local service_id   = tonumber(config.service_id)
 
   fill_trace_with_req(trace, req)
@@ -95,9 +93,9 @@ return function(req, next_middleware, config)
 
   local res = next_middleware()
 
-  trace.time          = pass_response(req, res, endpoint_url)
+  trace.time          = pass_response(req, res)
   trace.generic_path  = get_generic_path(req, service_id, res.status)
-  trace.endpoint      = assert(get_endpoint_host(), "Endpoint host expected")
+  trace.endpoint      = assert(extract_host(req.endpoint), "Endpoint host expected")
 
   -- Feel free to refactor this one, but we needed to show full original url in the UI
   trace.req.endpoint = ngx.var._endpoint_url

lua/middlewares/sanitizer.lua

@@ -1,6 +1,6 @@
 return function(req, next_middleware, config)
   if req.headers then
-    req.headers.Host = config.endpoint
+    req.headers.Host = string.match(req.endpoint,  "://([^/]+)")
   end
   local res = next_middleware()
   if res.headers then

lua/models/pipeline.lua

@@ -285,11 +285,11 @@ Pipeline.execute = function(pipeline, endpoint_url)
 
   local trace      = Trace:new(req)
 
+  req.endpoint = endpoint_url
+
   trace.service_id = pipeline.service_id
 
-  rack:use(sanitizer, {
-    endpoint = string.match(endpoint_url, "://([^/]+)")
-  })
+  rack:use(sanitizer)
 
   local ok, res = pcall(function()
     for _,middleware in ipairs(get_active_sorted_middlewares(pipeline)) do
@@ -298,8 +298,7 @@ Pipeline.execute = function(pipeline, endpoint_url)
 
     rack:use(brainslug, {
       trace         = trace,
-      service_id    = pipeline.service_id,
-      endpoint_url  = endpoint_url
+      service_id    = pipeline.service_id
     })
 
     return rack:run(req)

lua/rack.lua

@@ -157,7 +157,8 @@ function Rack:create_initial_request()
     args          = ngx.req.get_uri_args(),
     method        = ngx.var.request_method,
     scheme        = scheme,
-    uri           = uri,
+    uri           = uri, -- # TODO: deprecate this
+    path          = uri,
     host          = host
   }, bodybuilder_mt)
 end

spec/fixtures/pipelines/change_endpoint_pipeline.json

@@ -0,0 +1,14 @@
+{ "_id": 1,
+  "service_id": 1,
+  "middlewares": {
+    "do-nothing":{ "name": "do-nothing",
+      "uuid": "do-nothing",
+      "position": 0,
+      "code": "return function(req, next_middleware) req.endpoint = 'http://lvh-me.lvh.me:10002'; return next_middleware() end",
+      "config": {},
+      "active": true,
+      "spec_id": 0,
+      "description": "pass through of the request and returns response"
+    }
+  }
+}

spec/fixtures/pipelines/service_two_pipeline.json

@@ -0,0 +1,4 @@
+{ "_id": 2,
+  "service_id": 2,
+  "middlewares": {}
+}

spec/fixtures/services/lvh_service.json

@@ -0,0 +1,9 @@
+{ "_id": 2,
+  "name": "lvh",
+  "description": "the echo api",
+  "endpoints": [
+    { "url": "http://lvh.me:8081",
+      "code": "lvh"
+    }
+  ]
+}

spec/proxy/echo_spec.rb

@@ -13,6 +13,8 @@ def get_response_key(key)
   let(:service) { load_fixture('services',  'echo_service') }
   before(:each) { service }
 
+  let(:headers) { get_response_key('headers') }
+
   it 'returns the request as a query string when using the echo pipeline' do
     load_fixture('pipelines', 'empty_pipeline')
 
@@ -25,6 +27,13 @@ def get_response_key(key)
     get_response_key('path').should eq("/echo")
   end
 
+  it 'can change endpoint' do
+    load_fixture('services',  'lvh_service')
+    load_fixture('pipelines', 'service_two_pipeline')
+    load_fixture('pipelines', 'change_endpoint_pipeline')
+    expect(headers).to include('host' => 'lvh.me:8081')
+  end
+
   it 'uses a minimal middleware that does nothing. different url' do
     load_fixture('pipelines', 'echo_pipeline')
     get_json("#{host}/ech", host: proxy)['path'].should eq("/ech")
@@ -109,7 +118,7 @@ def get_response_key(key)
     jor.wait_for_async_locks
     last_trace_id = get_json('http://localhost:7071/api/traces/last_id')['last_id']
     last_trace = get_json("http://localhost:7071/api/traces/#{last_trace_id}")
-    expect(last_trace).to include('endpoint' => '127.0.0.1:8081')
+    expect(last_trace).to include('endpoint' => 'localhost:8081')
     expect(last_trace).to include('service_id' => service['_id'])
     expect(last_trace).to include('starred' => false)
   end