This repository has been archived by the owner on Apr 27, 2021. It is now read-only.
enable setting upstreams dynamically via HTTP endpoint #16
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| local ngx_balancer = require("ngx.balancer") | ||
| local ngx_upstream = require("ngx.upstream") | ||
| local math = require("math") | ||
| local json = require("cjson") | ||
| local resty_chash = require("resty.chash") | ||
| local resty_roundrobin = require("resty.roundrobin") | ||
| local configuration = require("configuration") | ||
| local util = require("util") | ||
|
|
||
| local DEFAULT_ALGORITHM = "round_robin" | ||
|
|
||
| local _M = {} | ||
|
|
||
| local function balance_least_conn(endpoints) | ||
| local servers, nodes = {}, {} | ||
| local str_null = string.char(0) | ||
|
|
||
| for _, endpoint in ipairs(endpoints) do | ||
| local id = endpoint.address .. str_null .. endpoint.port | ||
| servers[id] = endpoint | ||
| nodes[id] = 1 | ||
| end | ||
|
|
||
| -- TODO(elvinefendi) move this out of hot path and do it in process_backends_data function instead | ||
| local chash = resty_chash:new(nodes) | ||
|
|
||
| local id = chash:find() | ||
| local endpoint = servers[id] | ||
| return endpoint.address, endpoint.port | ||
| end | ||
|
|
||
| function _M.call() | ||
| ngx_balancer.set_more_tries(1) | ||
|
|
||
| local lb = configuration.get_lb(ngx.var.proxy_upstream_name) | ||
| local host_port_string = lb:find() | ||
| ngx.log(ngx.INFO, "selected host_port_string: " .. host_port_string) | ||
| local host, port = util.split_pair(host_port_string, ":") | ||
|
|
||
| local ok, err = ngx_balancer.set_current_peer(host, port) | ||
| if ok then | ||
| ngx.log(ngx.INFO, "current peer is set to " .. host_port_string) | ||
| else | ||
| ngx.log(ngx.ERR, "error while setting current upstream peer to: " .. tostring(err)) | ||
| end | ||
| end | ||
|
|
||
| return _M |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,126 @@ | ||
| local router = require("router") | ||
| local json = require("cjson") | ||
| local util = require("util") | ||
|
|
||
| -- key's are backend names | ||
| -- value's are respective load balancing algorithm name to use for the backend | ||
| local backend_lb_algorithms = ngx.shared.backend_lb_algorithms | ||
|
|
||
| -- key's are always going to be ngx.var.proxy_upstream_name, a uniqueue identifier of an app's Backend object | ||
| -- currently it is built our of namepsace, service name and service port | ||
| -- value is JSON encoded ingress.Backend object.Backend object, for more info refer to internal//ingress/types.go | ||
| local backends_data = ngx.shared.backends_data | ||
|
|
||
| -- TODO(elvinefendi) this is for future iteration when/if we decide for example to dynamically configure certificates | ||
| -- similar to backends_data | ||
| -- local servers_data = ngx.shared.servers_data | ||
|
|
||
| -- measured in seconds | ||
| -- for an Nginx worker to pick up the new list of upstream peers | ||
| -- it will take <the delay until controller POSTed the backend object to the Nginx endpoint> + BACKEND_PROCESSING_DELAY | ||
| local BACKEND_PROCESSING_DELAY = 1 | ||
|
|
||
| local _M = {} | ||
|
|
||
| local lbs = {} | ||
| local backends = {} | ||
|
|
||
| function _M.get_lb(backend_name) | ||
| return lbs[backend_name] | ||
| end | ||
|
|
||
| local resty_roundrobin = require("resty.roundrobin") | ||
|
|
||
| -- TODO(elvinefendi) make this consider lb_alg instead of always using round robin | ||
| local function update_backend(backend) | ||
| ngx.log(ngx.INFO, "updating backend: " .. backend.name) | ||
|
|
||
| local servers, nodes = {}, {} | ||
|
|
||
| for _, endpoint in ipairs(backend.endpoints) do | ||
| id = endpoint.address .. ":" .. endpoint.port | ||
| servers[id] = endpoint | ||
| nodes[id] = 1 | ||
| end | ||
|
|
||
| local rr = lbs[backend.name] | ||
| if rr then | ||
| rr:reinit(nodes) | ||
| else | ||
| rr = resty_roundrobin:new(nodes) | ||
| end | ||
|
|
||
| lbs[backend.name] = rr | ||
| backends[backend.name] = backend | ||
| end | ||
|
|
||
| -- this function will be periodically called in every worker to decode backends and store them in local backends variable | ||
| local function process_backends_data() | ||
| -- 0 here means get all the keys which can be slow if there are many keys | ||
| -- TODO(elvinefendi) think about storing comma separated backend names in another dictionary and using that to | ||
| -- fetch the list of them here insted of blocking the access to shared dictionary | ||
| backend_names = backends_data:get_keys(0) | ||
|
|
||
| for _, backend_name in pairs(backend_names) do | ||
| backend_data = backends_data:get(backend_name) | ||
|
|
||
| local ok, backend = pcall(json.decode, backend_data) | ||
|
|
||
| if ok then | ||
| if not util.deep_compare(backends[backend_name], backend, true) then | ||
| update_backend(backend) | ||
| end | ||
| else | ||
| ngx.log(ngx.ERROR, "could not parse backend_json: " .. tostring(backend)) | ||
| end | ||
| end | ||
| end | ||
|
|
||
| function _M.init_worker() | ||
| _, err = ngx.timer.every(BACKEND_PROCESSING_DELAY, process_backends_data) | ||
| if err then | ||
| ngx.log(ngx.ERROR, "error when setting up timer.every for process_backends_data: " .. tostring(err)) | ||
| end | ||
| end | ||
|
|
||
| function _M.call() | ||
| local r = router.new() | ||
|
|
||
| r:match({ | ||
| POST = { | ||
| ["/configuration/backends/:name"] = function(params) | ||
| ngx.req.read_body() -- explicitly read the req body | ||
|
|
||
| local success, err = backends_data:set(params.name, ngx.req.get_body_data()) | ||
| if not success then | ||
| return err | ||
| end | ||
|
|
||
| -- TODO(elvinefendi) also check if it is a supported algorith | ||
| if params.lb_alg ~=nil and params.lb_alg ~= "" then | ||
| success, err = backend_lb_algorithms:set(params.name, params.lb_alg) | ||
| if not success then | ||
| return err | ||
| end | ||
| end | ||
|
|
||
| ngx.status = 201 | ||
| ngx.log(ngx.INFO, "backend data was updated for " .. params.name .. ": " .. tostring(ngx.req.get_body_data())) | ||
| end | ||
| } | ||
| }) | ||
|
|
||
| local ok, errmsg = r:execute(ngx.var.request_method, ngx.var.request_uri, ngx.req.get_uri_args()) | ||
| if ok then | ||
| if errmsg then | ||
| ngx.status = 400 | ||
| ngx.print(tostring(errmsg)) | ||
| end | ||
| else | ||
| ngx.log(ngx.ERROR, tostring(errmsg)) | ||
| ngx.status = 404 | ||
| ngx.print("Not found!") | ||
| end | ||
| end | ||
|
|
||
| return _M |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| local _M = {} | ||
|
|
||
| function _M.split_pair(pair, seperator) | ||
| local i = pair:find(seperator) | ||
| if i == nil then | ||
| return pair, nil | ||
| else | ||
| local name = pair:sub(1, i - 1) | ||
| local value = pair:sub(i + 1, -1) | ||
| return name, value | ||
| end | ||
| end | ||
|
|
||
| local function deep_compare(t1, t2, ignore_mt) | ||
| local ty1 = type(t1) | ||
| local ty2 = type(t2) | ||
| if ty1 ~= ty2 then return false end | ||
| -- non-table types can be directly compared | ||
| if ty1 ~= 'table' and ty2 ~= 'table' then return t1 == t2 end | ||
| -- as well as tables which have the metamethod __eq | ||
| local mt = getmetatable(t1) | ||
| if not ignore_mt and mt and mt.__eq then return t1 == t2 end | ||
| for k1,v1 in pairs(t1) do | ||
| local v2 = t2[k1] | ||
| if v2 == nil or not deep_compare(v1,v2) then return false end | ||
| end | ||
| for k2,v2 in pairs(t2) do | ||
| local v1 = t1[k2] | ||
| if v1 == nil or not deep_compare(v1,v2) then return false end | ||
| end | ||
| return true | ||
| end | ||
| _M.deep_compare = deep_compare | ||
|
|
||
| return _M |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@zrdaley this is how I ended up implementing it. The idea is to detect only Endpoints changes in case of any other change we fall back to reloading.