membphis commented on a change in pull request #2001:
URL: https://github.com/apache/apisix/pull/2001#discussion_r468326680
##########
File path: doc/admin-api.md
##########
@@ -493,7 +493,7 @@ In addition to the basic complex equalization algorithm
selection, APISIX's Upst
|Name |Optional|Description|
|------- |-----|------|
-|type |required|`roundrobin` supports the weight of the load,
`chash` consistency hash, pick one of them.|
+|type |required|`roundrobin` supports the weight of the load,
`chash` consistency hash,`ewma` minimum latency ,pick one of them.|
Review comment:
I think we'd better add a link about what is `ewma` in doc. That is
useful for the user.
##########
File path: apisix/balancer/ewma.lua
##########
@@ -0,0 +1,240 @@
+--
Review comment:
I think only `https://github.com/twitter/finagle` is not enough.
And we need to add the license for
`https://github.com/kubernetes/ingress-nginx`.
here is the full link:
https://github.com/kubernetes/ingress-nginx/blob/master/rootfs/etc/nginx/lua/balancer/ewma.lua
##########
File path: apisix/balancer/ewma.lua
##########
@@ -0,0 +1,198 @@
+-- Original Authors: Shiv Nagarajan & Scott Francis
+-- Accessed: March 12, 2018
+-- Inspiration drawn from:
+--
https://github.com/twitter/finagle/blob/1bc837c4feafc0096e43c0e98516a8e1c50c4421
+--
/finagle-core/src/main/scala/com/twitter/finagle/loadbalancer/PeakEwma.scala
+--
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements. See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You under the Apache License, Version 2.0
+-- (the "License"); you may not use this file except in compliance with
+-- the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+--
+
+
+local core = require("apisix.core")
+local ngx = ngx
+local ngx_shared = ngx.shared
+local ngx_now = ngx.now
+local math = math
+local pairs = pairs
+local next = next
+local tonumber = tonumber
+
+local _M = {}
+local DECAY_TIME = 10 -- this value is in seconds
+
+local shm_ewma = ngx_shared.balancer_ewma
+local shm_last_touched_at= ngx_shared.balancer_ewma_last_touched_at
+
+local lrucache_addr = core.lrucache.new({
+ ttl = 300, count = 1024
+})
+local lrucache_trans_format = core.lrucache.new({
+ ttl = 300, count = 256
+})
+
+
+local function decay_ewma(ewma, last_touched_at, rtt, now)
+ local td = now - last_touched_at
+ td = math.max(td, 0)
+ local weight = math.exp(-td / DECAY_TIME)
+
+ ewma = ewma * weight + rtt * (1.0 - weight)
+ return ewma
+end
+
+
+local function store_stats(upstream, ewma, now)
+ local success, err, forcible = shm_last_touched_at:set(upstream, now)
+ if not success then
+ core.log.error("balancer_ewma_last_touched_at:set failed ", err)
+ end
+ if forcible then
+ core.log.warn("balancer_ewma_last_touched_at:set valid items forcibly
overwritten")
+ end
+
+ success, err, forcible = shm_ewma:set(upstream, ewma)
+ if not success then
+ core.log.error("balancer_ewma:set failed ", err)
+ end
+ if forcible then
+ core.log.warn("balancer_ewma:set valid items forcibly overwritten")
+ end
+end
+
+
+local function get_or_update_ewma(upstream, rtt, update)
+ local ewma = shm_ewma:get(upstream) or 0
+ local now = ngx_now()
+ local last_touched_at = shm_last_touched_at:get(upstream) or 0
+ ewma = decay_ewma(ewma, last_touched_at, rtt, now)
+
+ if not update then
+ return ewma
+ end
+
+ store_stats(upstream, ewma, now)
+
+ return ewma
+end
+
+
+local function score(upstream)
+ -- Original implementation used names
+ -- Endpoints don't have names, so passing in host:Port as key instead
+ local upstream_name = upstream.host .. ":" .. upstream.port
+ return get_or_update_ewma(upstream_name, 0, false)
+end
+
+
+local function pick_and_score(peers)
+ local lowest_score_index = 1
+ local lowest_score = score(peers[lowest_score_index])
+ for i = 2, #peers do
+ local new_score = score(peers[i])
+ if new_score < lowest_score then
+ lowest_score_index, lowest_score = i, new_score
+ end
+ end
+
+ return peers[lowest_score_index], lowest_score
+end
+
+
+local function parse_addr(addr)
+ local host, port, err = core.utils.parse_addr(addr)
+ return {host = host, port = port}, err
+end
+
+
+local function _trans_format(up_nodes)
+ -- trans
+ --{"1.2.3.4:80":100,"5.6.7.8:8080":100}
+ -- into
+ -- [{"host":"1.2.3.4","port":"80"},{"host":"5.6.7.8","port":"8080"}]
+ local peers = {}
+ local res, err
+
+ for addr, _ in pairs(up_nodes) do
+ res, err = lrucache_addr(addr, nil, parse_addr, addr)
+ if not err then
+ core.table.insert(peers, res)
+ else
+ core.log.error('parse_addr error: ', addr, err)
+ end
+ end
+
+ return next(peers) and peers or nil
+end
+
+
+local function _ewma_find(ctx, up_nodes)
+ local peers
+ local endpoint
+
+ if not up_nodes
+ or core.table.nkeys(up_nodes) == 0 then
+ return nil, 'up_nodes empty'
+ end
+
+ peers = lrucache_trans_format(ctx.upstream_key, ctx.upstream_version,
+ _trans_format, up_nodes)
+ if not peers then
+ return nil, 'up_nodes trans error'
+ end
+
+ if #peers > 1 then
+ endpoint = pick_and_score(peers)
+ else
+ endpoint = peers[1]
+ end
+
+ return endpoint.host .. ":" .. endpoint.port
+end
+
+
+local function _ewma_after_balance(ctx)
+ local response_time = tonumber(ctx.var.upstream_response_time) or 0
+ local connect_time = tonumber(ctx.var.upstream_connect_time) or 0
+ local rtt = connect_time + response_time
+ local upstream = ctx.var.upstream_addr
+
+ if not upstream then
+ return nil, "no upstream addr found"
+ end
+
+ return get_or_update_ewma(upstream, rtt, true)
+end
+
+
+function _M.new(up_nodes, upstream)
+ if not shm_ewma
+ or not shm_last_touched_at then
+ return nil, "dictionary not find"
+ end
+
+ return {
+ upstream = upstream,
+ get = function (ctx)
+ return _ewma_find(ctx, up_nodes)
+ end,
+ after_balance = function(ctx)
Review comment:
`after_balance = _ewma_after_balance`, more simpler way
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
