Don't pick tried endpoint & count the latest in ewma balancer

fixes https://github.com/kubernetes/ingress-nginx/issues/6632

rootfs/etc/nginx/lua/balancer/ewma.lua

@@ -18,6 +18,7 @@ local string = string
 local tonumber = tonumber
 local setmetatable = setmetatable
 local string_format = string.format
+local table_insert = table.insert
 local ngx_log = ngx.log
 local INFO = ngx.INFO
 
@@ -105,10 +106,15 @@ local function get_or_update_ewma(upstream, rtt, update)
 end
 
 
+local function get_upstream_name(upstream)
+   return upstream.address .. ":" .. upstream.port
+end
+
+
 local function score(upstream)
   -- Original implementation used names
   -- Endpoints don't have names, so passing in IP:Port as key instead
-  local upstream_name = upstream.address .. ":" .. upstream.port
+  local upstream_name = get_upstream_name(upstream)
   return get_or_update_ewma(upstream_name, 0, false)
 end
 
@@ -135,6 +141,7 @@ local function pick_and_score(peers, k)
       lowest_score_index, lowest_score = i, new_score
     end
   end
+
   return peers[lowest_score_index], lowest_score
 end
 
@@ -146,7 +153,7 @@ local function calculate_slow_start_ewma(self)
   local endpoints_count = 0
 
   for _, endpoint in pairs(self.peers) do
-    local endpoint_string = endpoint.address .. ":" .. endpoint.port
+    local endpoint_string = get_upstream_name(endpoint)
     local ewma = ngx.shared.balancer_ewma:get(endpoint_string)
 
     if ewma then
@@ -170,20 +177,50 @@ function _M.balance(self)
   if #peers > 1 then
     local k = (#peers < PICK_SET_SIZE) and #peers or PICK_SET_SIZE
     local peer_copy = util.deepcopy(peers)
-    endpoint, ewma_score = pick_and_score(peer_copy, k)
+
+    local tried_endpoints
+    if not ngx.ctx.balancer_ewma_tried_endpoints then
+      tried_endpoints = {}
+      ngx.ctx.balancer_ewma_tried_endpoints = tried_endpoints
+    else
+      tried_endpoints = ngx.ctx.balancer_ewma_tried_endpoints
+    end
+
+    local filtered_peers
+    for _, peer in ipairs(peer_copy) do
+      if not tried_endpoints[get_upstream_name(peer)] then
+        if not filtered_peers then
+          filtered_peers = {}
+        end
+        table_insert(filtered_peers, peer)
+      end
+    end
+
+    if not filtered_peers then
+      ngx.log(ngx.WARN, "all endpoints have been retried")
+      filtered_peers = peer_copy
+    end
+
+    if #filtered_peers > 1 then
+      endpoint, ewma_score = pick_and_score(filtered_peers, k)
+    else
+      endpoint, ewma_score = filtered_peers[1], score(filtered_peers[1])
+    end
+
+    tried_endpoints[get_upstream_name(endpoint)] = true
   end
 
   ngx.var.balancer_ewma_score = ewma_score
 
   -- TODO(elvinefendi) move this processing to _M.sync
-  return endpoint.address .. ":" .. endpoint.port
+  return get_upstream_name(endpoint)
 end
 
 function _M.after_balance(_)
-  local response_time = tonumber(split.get_first_value(ngx.var.upstream_response_time)) or 0
-  local connect_time = tonumber(split.get_first_value(ngx.var.upstream_connect_time)) or 0
+  local response_time = tonumber(split.get_last_value(ngx.var.upstream_response_time)) or 0
+  local connect_time = tonumber(split.get_last_value(ngx.var.upstream_connect_time)) or 0
   local rtt = connect_time + response_time
-  local upstream = split.get_first_value(ngx.var.upstream_addr)
+  local upstream = split.get_last_value(ngx.var.upstream_addr)
 
   if util.is_blank(upstream) then
     return