merge from master
This commit is contained in:
liuwei
commit
ce6e564f82
86 changed files with 1448 additions and 3530 deletions
|
|
@ -5,38 +5,17 @@
|
|||
-- /finagle-core/src/main/scala/com/twitter/finagle/loadbalancer/PeakEwma.scala
|
||||
|
||||
|
||||
local resty_lock = require("resty.lock")
|
||||
local util = require("util")
|
||||
local split = require("util.split")
|
||||
|
||||
local DECAY_TIME = 10 -- this value is in seconds
|
||||
local LOCK_KEY = ":ewma_key"
|
||||
local PICK_SET_SIZE = 2
|
||||
|
||||
local ewma_lock = resty_lock:new("locks", {timeout = 0, exptime = 0.1})
|
||||
local balancer_ewma = {}
|
||||
local balancer_ewma_last_touched_at = {}
|
||||
|
||||
local _M = { name = "ewma" }
|
||||
|
||||
local function lock(upstream)
|
||||
local _, err = ewma_lock:lock(upstream .. LOCK_KEY)
|
||||
if err then
|
||||
if err ~= "timeout" then
|
||||
ngx.log(ngx.ERR, string.format("EWMA Balancer failed to lock: %s", tostring(err)))
|
||||
end
|
||||
end
|
||||
|
||||
return err
|
||||
end
|
||||
|
||||
local function unlock()
|
||||
local ok, err = ewma_lock:unlock()
|
||||
if not ok then
|
||||
ngx.log(ngx.ERR, string.format("EWMA Balancer failed to unlock: %s", tostring(err)))
|
||||
end
|
||||
|
||||
return err
|
||||
end
|
||||
|
||||
local function decay_ewma(ewma, last_touched_at, rtt, now)
|
||||
local td = now - last_touched_at
|
||||
td = (td > 0) and td or 0
|
||||
|
|
@ -47,40 +26,18 @@ local function decay_ewma(ewma, last_touched_at, rtt, now)
|
|||
end
|
||||
|
||||
local function get_or_update_ewma(upstream, rtt, update)
|
||||
local lock_err = nil
|
||||
if update then
|
||||
lock_err = lock(upstream)
|
||||
end
|
||||
local ewma = ngx.shared.balancer_ewma:get(upstream) or 0
|
||||
if lock_err ~= nil then
|
||||
return ewma, lock_err
|
||||
end
|
||||
local ewma = balancer_ewma[upstream] or 0
|
||||
|
||||
local now = ngx.now()
|
||||
local last_touched_at = ngx.shared.balancer_ewma_last_touched_at:get(upstream) or 0
|
||||
local last_touched_at = balancer_ewma_last_touched_at[upstream] or 0
|
||||
ewma = decay_ewma(ewma, last_touched_at, rtt, now)
|
||||
|
||||
if not update then
|
||||
return ewma, nil
|
||||
end
|
||||
|
||||
local success, err, forcible = ngx.shared.balancer_ewma_last_touched_at:set(upstream, now)
|
||||
if not success then
|
||||
ngx.log(ngx.WARN, "balancer_ewma_last_touched_at:set failed " .. err)
|
||||
end
|
||||
if forcible then
|
||||
ngx.log(ngx.WARN, "balancer_ewma_last_touched_at:set valid items forcibly overwritten")
|
||||
end
|
||||
|
||||
success, err, forcible = ngx.shared.balancer_ewma:set(upstream, ewma)
|
||||
if not success then
|
||||
ngx.log(ngx.WARN, "balancer_ewma:set failed " .. err)
|
||||
end
|
||||
if forcible then
|
||||
ngx.log(ngx.WARN, "balancer_ewma:set valid items forcibly overwritten")
|
||||
end
|
||||
|
||||
unlock()
|
||||
balancer_ewma_last_touched_at[upstream] = now
|
||||
balancer_ewma[upstream] = ewma
|
||||
return ewma, nil
|
||||
end
|
||||
|
||||
|
|
@ -153,8 +110,8 @@ function _M.sync(self, backend)
|
|||
self.peers = backend.endpoints
|
||||
|
||||
-- TODO: Reset state of EWMA per backend
|
||||
ngx.shared.balancer_ewma:flush_all()
|
||||
ngx.shared.balancer_ewma_last_touched_at:flush_all()
|
||||
balancer_ewma = {}
|
||||
balancer_ewma_last_touched_at = {}
|
||||
end
|
||||
|
||||
function _M.new(self, backend)
|
||||
|
|
|
|||
|
|
@ -94,22 +94,17 @@ local function set_cookie(self, value)
|
|||
end
|
||||
end
|
||||
|
||||
local function pick_random(instance)
|
||||
local index = math.random(instance.npoints)
|
||||
return instance:next(index)
|
||||
end
|
||||
|
||||
function _M.balance(self)
|
||||
local cookie, err = ck:new()
|
||||
if not cookie then
|
||||
ngx.log(ngx.ERR, err)
|
||||
return pick_random(self.instance)
|
||||
ngx.log(ngx.ERR, "error while initializing cookie: " .. tostring(err))
|
||||
return
|
||||
end
|
||||
|
||||
local key = cookie:get(self.cookie_name)
|
||||
if not key then
|
||||
local tmp_endpoint_string = pick_random(self.instance)
|
||||
key = encrypted_endpoint_string(self, tmp_endpoint_string)
|
||||
local random_str = string.format("%s.%s", ngx.now(), ngx.worker.pid())
|
||||
key = encrypted_endpoint_string(self, random_str)
|
||||
set_cookie(self, key)
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -26,10 +26,6 @@ describe("Balancer ewma", function()
|
|||
local instance = balancer_ewma:new(backend)
|
||||
|
||||
local stats = { ["10.184.7.40:8080"] = 0.5, ["10.184.97.100:8080"] = 0.3 }
|
||||
ngx.shared.balancer_ewma.get = function(self, key) return stats[key] end
|
||||
local t = ngx.now()-10
|
||||
ngx.shared.balancer_ewma_last_touched_at.get = function(self, key) return t end
|
||||
|
||||
|
||||
local peer = instance:balance()
|
||||
assert.equal("10.184.97.100:8080", peer)
|
||||
|
|
@ -52,13 +48,7 @@ describe("Balancer ewma", function()
|
|||
endpoints = { { address = "10.184.7.40", port = "8080", maxFails = 0, failTimeout = 0 } }
|
||||
}
|
||||
|
||||
local s1 = spy.on(ngx.shared.balancer_ewma, "flush_all")
|
||||
local s2 = spy.on(ngx.shared.balancer_ewma_last_touched_at, "flush_all")
|
||||
|
||||
instance:sync(new_backend)
|
||||
|
||||
assert.spy(s1).was_not_called()
|
||||
assert.spy(s2).was_not_called()
|
||||
end)
|
||||
|
||||
it("updates endpoints", function()
|
||||
|
|
@ -77,13 +67,7 @@ describe("Balancer ewma", function()
|
|||
local new_backend = util.deepcopy(backend)
|
||||
new_backend.endpoints[1].maxFails = 3
|
||||
|
||||
local s1 = spy.on(ngx.shared.balancer_ewma, "flush_all")
|
||||
local s2 = spy.on(ngx.shared.balancer_ewma_last_touched_at, "flush_all")
|
||||
|
||||
instance:sync(new_backend)
|
||||
|
||||
assert.spy(s1).was_called()
|
||||
assert.spy(s2).was_called()
|
||||
end)
|
||||
end)
|
||||
end)
|
||||
|
|
|
|||
|
|
@ -116,7 +116,8 @@ describe("Sticky", function()
|
|||
local cookie_instance = {
|
||||
set = function(self, payload)
|
||||
assert.equal(payload.key, test_backend.sessionAffinityConfig.cookieSessionAffinity.name)
|
||||
assert.equal(payload.value, util[test_backend_hash_fn .. "_digest"](test_backend_endpoint))
|
||||
local expected_len = #util[test_backend_hash_fn .. "_digest"]("anything")
|
||||
assert.equal(#payload.value, expected_len)
|
||||
assert.equal(payload.path, ngx.var.location_path)
|
||||
assert.equal(payload.domain, nil)
|
||||
assert.equal(payload.httponly, true)
|
||||
|
|
|
|||
|
|
@ -12,6 +12,10 @@
|
|||
# setup custom paths that do not require root access
|
||||
pid /tmp/nginx.pid;
|
||||
|
||||
{{ if $cfg.UseGeoIP2 }}
|
||||
load_module /etc/nginx/modules/ngx_http_geoip2_module.so;
|
||||
{{ end }}
|
||||
|
||||
{{ if $cfg.EnableModsecurity }}
|
||||
load_module /etc/nginx/modules/ngx_http_modsecurity_module.so;
|
||||
{{ end }}
|
||||
|
|
@ -23,8 +27,8 @@ load_module /etc/nginx/modules/ngx_http_opentracing_module.so;
|
|||
daemon off;
|
||||
|
||||
worker_processes {{ $cfg.WorkerProcesses }};
|
||||
{{ if gt (len $cfg.WorkerCpuAffinity) 0 }}
|
||||
worker_cpu_affinity {{ $cfg.WorkerCpuAffinity }};
|
||||
{{ if gt (len $cfg.WorkerCPUAffinity) 0 }}
|
||||
worker_cpu_affinity {{ $cfg.WorkerCPUAffinity }};
|
||||
{{ end }}
|
||||
|
||||
{{ if ne .MaxOpenFiles 0 }}
|
||||
|
|
@ -123,6 +127,26 @@ http {
|
|||
geoip_proxy_recursive on;
|
||||
{{ end }}
|
||||
|
||||
{{ if $cfg.UseGeoIP2 }}
|
||||
# https://github.com/leev/ngx_http_geoip2_module#example-usage
|
||||
|
||||
geoip2 /etc/nginx/geoip/GeoLite2-City.mmdb {
|
||||
$geoip2_city_country_code source=$the_real_ip country iso_code;
|
||||
$geoip2_city_country_name source=$the_real_ip country names en;
|
||||
$geoip2_city source=$the_real_ip city names en;
|
||||
$geoip2_postal_code source=$the_real_ip postal code;
|
||||
$geoip2_dma_code source=$the_real_ip location metro_code;
|
||||
$geoip2_latitude source=$the_real_ip location latitude;
|
||||
$geoip2_longitude source=$the_real_ip location longitude;
|
||||
$geoip2_region_code source=$the_real_ip subdivisions 0 iso_code;
|
||||
$geoip2_region_name source=$the_real_ip subdivisions 0 names en;
|
||||
}
|
||||
|
||||
geoip2 /etc/nginx/geoip/GeoLite2-ASN.mmdb {
|
||||
$geoip2_asn source=$the_real_ip autonomous_system_number;
|
||||
}
|
||||
{{ end }}
|
||||
|
||||
aio threads;
|
||||
aio_write on;
|
||||
|
||||
|
|
@ -322,7 +346,7 @@ http {
|
|||
# If no such header is provided, it can provide a random value.
|
||||
map $http_x_request_id $req_id {
|
||||
default $http_x_request_id;
|
||||
{{ if $cfg.GenerateRequestId }}
|
||||
{{ if $cfg.GenerateRequestID }}
|
||||
"" $request_id;
|
||||
{{ end }}
|
||||
}
|
||||
|
|
@ -891,9 +915,23 @@ stream {
|
|||
|
||||
waf:set_option("mode", "{{ $location.LuaRestyWAF.Mode }}")
|
||||
waf:set_option("storage_zone", "waf_storage")
|
||||
|
||||
{{ if $location.LuaRestyWAF.AllowUnknownContentTypes }}
|
||||
waf:set_option("allow_unknown_content_types", true)
|
||||
{{ else }}
|
||||
waf:set_option("allowed_content_types", { "text/html", "text/json", "application/json" })
|
||||
{{ end }}
|
||||
|
||||
waf:set_option("event_log_level", ngx.WARN)
|
||||
|
||||
{{ if gt $location.LuaRestyWAF.ScoreThreshold 0 }}
|
||||
waf:set_option("score_threshold", {{ $location.LuaRestyWAF.ScoreThreshold }})
|
||||
{{ end }}
|
||||
|
||||
{{ if not $location.LuaRestyWAF.ProcessMultipartBody }}
|
||||
waf:set_option("process_multipart_body", false)
|
||||
{{ end }}
|
||||
|
||||
{{ if $location.LuaRestyWAF.Debug }}
|
||||
waf:set_option("debug", true)
|
||||
waf:set_option("event_log_request_arguments", true)
|
||||
|
|
@ -1077,7 +1115,7 @@ stream {
|
|||
{{ $proxySetHeader }} X-Forwarded-Host $best_http_host;
|
||||
{{ $proxySetHeader }} X-Forwarded-Port $pass_port;
|
||||
{{ $proxySetHeader }} X-Forwarded-Proto $pass_access_scheme;
|
||||
{{ if $all.Cfg.ProxyAddOriginalUriHeader }}
|
||||
{{ if $all.Cfg.ProxyAddOriginalURIHeader }}
|
||||
{{ $proxySetHeader }} X-Original-URI $request_uri;
|
||||
{{ end }}
|
||||
{{ $proxySetHeader }} X-Scheme $pass_access_scheme;
|
||||
|
|
@ -1135,7 +1173,6 @@ stream {
|
|||
proxy_set_header X-Service-Port $service_port;
|
||||
{{ end }}
|
||||
|
||||
{{ if not (empty $location.Backend) }}
|
||||
{{ buildProxyPass $server.Hostname $all.Backends $location }}
|
||||
{{ if (or (eq $location.Proxy.ProxyRedirectFrom "default") (eq $location.Proxy.ProxyRedirectFrom "off")) }}
|
||||
proxy_redirect {{ $location.Proxy.ProxyRedirectFrom }};
|
||||
|
|
@ -1143,10 +1180,6 @@ stream {
|
|||
proxy_redirect {{ $location.Proxy.ProxyRedirectFrom }} {{ $location.Proxy.ProxyRedirectTo }};
|
||||
{{ end }}
|
||||
{{ else }}
|
||||
# No endpoints available for the request
|
||||
return 503;
|
||||
{{ end }}
|
||||
{{ else }}
|
||||
# Location denied. Reason: {{ $location.Denied | printf "%q" }}
|
||||
return 503;
|
||||
{{ end }}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue