Re: [I] bug: dynamic upstream, one Eureka node is unavailable, half of the requests are lost after reloading [apisix]

["liquanzhou (via GitHub)" <gi...@apache.org>]

liquanzhou commented on issue #12610:
URL: https://github.com/apache/apisix/issues/12610#issuecomment-3432001253

   通过ai生成代码修复,测试是可以解决一个eureka节点宕机后,reload还能正常加载列表,可以当做一个参考
   
   日志中会有持续的eureka节点失败的警告
   2025/10/22 19:55:26 [warn] 7225#7225: *40628 [lua] init.lua:200: failed to fetch registry from http://10.250.200.98:8761/eureka/: timeout, context: ngx.timer
   
   `--
   -- 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 local_conf         = require("apisix.core.config_local").local_conf()
   local http               = require("resty.http")
   local core               = require("apisix.core")
   local ipmatcher          = require("resty.ipmatcher")
   local ipairs             = ipairs
   local tostring           = tostring
   local type               = type
   local math_random        = math.random
   local ngx                = ngx
   local ngx_timer_at       = ngx.timer.at
   local ngx_timer_every    = ngx.timer.every
   local string_sub         = string.sub
   local str_find           = core.string.find
   local log                = core.log
   local semaphore = require("ngx.semaphore")
   
   local default_weight
   local applications
   local init_sema
   local initial_fetched = false
   
   local _M = {
       version = 0.1,
   }
   
   
   local function service_info()
       local host = local_conf.discovery and
           local_conf.discovery.eureka and local_conf.discovery.eureka.host
       if not host then
           log.error("do not set eureka.host")
           return
       end
   
       local basic_auth
       -- TODO Add health check to get healthy nodes.
       local url = host[math_random(#host)]
       local auth_idx = str_find(url, "@")
       if auth_idx then
           local protocol_idx = str_find(url, "://")
           local protocol = string_sub(url, 1, protocol_idx + 2)
           local user_and_password = string_sub(url, protocol_idx + 3, auth_idx - 1)
           local other = string_sub(url, auth_idx + 1)
           url = protocol .. other
           basic_auth = "Basic " .. ngx.encode_base64(user_and_password)
       end
       if local_conf.discovery.eureka.prefix then
           url = url .. local_conf.discovery.eureka.prefix
       end
       if string_sub(url, #url) ~= "/" then
           url = url .. "/"
       end
   
       return url, basic_auth
   end
   
   
   local function request(request_uri, basic_auth, method, path, query, body)
       log.info("eureka uri:", request_uri, ".")
       local url = request_uri .. path
       local headers = core.table.new(0, 5)
       headers['Connection'] = 'Keep-Alive'
       headers['Accept'] = 'application/json'
   
       if basic_auth then
           headers['Authorization'] = basic_auth
       end
   
       if body and 'table' == type(body) then
           local err
           body, err = core.json.encode(body)
           if not body then
               return nil, 'invalid body : ' .. err
           end
           -- log.warn(method, url, body)
           headers['Content-Type'] = 'application/json'
       end
   
       local httpc = http.new()
       local timeout = local_conf.discovery.eureka.timeout
       local connect_timeout = timeout and timeout.connect or 2000
       local send_timeout = timeout and timeout.send or 2000
       local read_timeout = timeout and timeout.read or 5000
       log.info("connect_timeout:", connect_timeout, ", send_timeout:", send_timeout,
               ", read_timeout:", read_timeout, ".")
       httpc:set_timeouts(connect_timeout, send_timeout, read_timeout)
       return httpc:request_uri(url, {
           version = 1.1,
           method = method,
           headers = headers,
           query = query,
           body = body,
           ssl_verify = false,
       })
   end
   
   
   local function parse_instance(instance)
       local status = instance.status
       local overridden_status = instance.overriddenstatus or instance.overriddenStatus
       if overridden_status and overridden_status ~= "UNKNOWN" then
           status = overridden_status
       end
   
       if status ~= "UP" then
           return
       end
       local port
       if tostring(instance.port["@enabled"]) == "true" and instance.port["$"] then
           port = instance.port["$"]
           -- secure = false
       end
       if tostring(instance.securePort["@enabled"]) == "true" and instance.securePort["$"] then
           port = instance.securePort["$"]
           -- secure = true
       end
       local ip = instance.ipAddr
       if not ipmatcher.parse_ipv4(ip) and
               not ipmatcher.parse_ipv6(ip) then
           log.error(instance.app, " service ", instance.hostName, " node IP ", ip,
                   " is invalid(must be IPv4 or IPv6).")
           return
       end
       return ip, port, instance.metadata
   end
   
   
   local function build_endpoints()
       local host_list = local_conf.discovery and local_conf.discovery.eureka and local_conf.discovery.eureka.host
       if not host_list or #host_list == 0 then
           log.error("do not set eureka.host")
           return nil
       end
   
       local endpoints = core.table.new(#host_list, 0)
       for _, h in ipairs(host_list) do
           local url = h
           local basic_auth
           local auth_idx = str_find(url, "@")
           if auth_idx then
               local protocol_idx = str_find(url, "://")
               local protocol = string_sub(url, 1, protocol_idx + 2)
               local user_and_password = string_sub(url, protocol_idx + 3, auth_idx - 1)
               local other = string_sub(url, auth_idx + 1)
               url = protocol .. other
               basic_auth = "Basic " .. ngx.encode_base64(user_and_password)
           end
           if local_conf.discovery.eureka.prefix then
               url = url .. local_conf.discovery.eureka.prefix
           end
           if string_sub(url, #url) ~= "/" then
               url = url .. "/"
           end
           core.table.insert(endpoints, { url = url, auth = basic_auth })
       end
       return endpoints
   end
   
   
   local function fetch_full_registry(premature)
       if premature then
           return
       end
   
       -- 遍历所有 eureka 端点，直到成功
       local endpoints = build_endpoints()
       if not endpoints or #endpoints == 0 then
           return
       end
   
       local res, err
       local used_endpoint
       local start = math_random(#endpoints)
       for i = 0, #endpoints - 1 do
           local ep = endpoints[((start + i) % #endpoints) + 1]
           log.info("eureka uri:", ep.url, ".")
           local r, e = request(ep.url, ep.auth, "GET", "apps")
           if r and r.body and r.status == 200 then
               res = r
               used_endpoint = ep
               break
           end
           log.warn("failed to fetch registry from ", ep.url, ": ", e or (r and ("status=" .. tostring(r.status)) or "unknown"))
       end
   
       if not res then
           log.error("failed to fetch registry from all eureka hosts")
           return
       end
   
       local json_str = res.body
       local data, derr = core.json.decode(json_str)
       if not data then
           log.error("invalid response body: ", json_str, " err: ", derr)
           return
       end
       local apps = data.applications.application
       local up_apps = core.table.new(0, #apps)
       for _, app in ipairs(apps) do
           for _, instance in ipairs(app.instance) do
               local ip, port, metadata = parse_instance(instance)
               if ip and port then
                   local nodes = up_apps[app.name]
                   if not nodes then
                       nodes = core.table.new(#app.instance, 0)
                       up_apps[app.name] = nodes
                   end
                   core.table.insert(nodes, {
                       host = ip,
                       port = port,
                       weight = metadata and metadata.weight or default_weight,
                       metadata = metadata,
                   })
                   if metadata then
                       metadata.weight = nil
                   end
               end
           end
       end
       applications = up_apps
       log.info("successfully updated service registry, services count=",
                core.table.nkeys(up_apps), "; source=", used_endpoint and used_endpoint.url or "unknown")
       if not initial_fetched then
           initial_fetched = true
           if init_sema then
               init_sema:post(1)
           end
       end
   end
   
   
   function _M.nodes(service_name)
       if not applications then
           if init_sema then
               local ok, err = init_sema:wait(3)
               if not ok then
                   log.warn("wait eureka initial fetch timeout: ", err)
               end
           end
       end
   
       if not applications then
           log.error("failed to fetch nodes for : ", service_name)
           return
       end
   
       return applications[service_name]
   end
   
   
   -- 注释掉文件缓存相关依赖与变量，避免写盘/读盘
   -- local core_io = require("apisix.core.io")
   -- local io_open = io.open
   -- local cache_file = ngx.config.prefix() .. "logs/eureka_registry.json"
   -- local save_registry_cache
   -- local load_registry_cache
   -- 将文件缓存函数整体注释掉，避免写盘与读盘
   --[[
   local function save_registry_cache(apps)
       local body, err = core.json.encode({ applications = apps, ts = ngx.now() })
       if not body then
           log.error("encode eureka registry cache failed: ", err)
           return
       end
       local f, ferr = io_open(cache_file, "w")
       if not f then
           log.error("open eureka registry cache file failed: ", ferr,
                     ", path: ", cache_file)
           return
       end
       f:write(body)
       f:close()
   end
   
   local function load_registry_cache()
       local body = core_io.get_file(cache_file)
       if not body then
           return nil, "no cache file"
       end
       local data, err = core.json.decode(body)
       if not data or not data.applications then
           return nil, "invalid cache format: " .. (err or "")
       end
       return data.applications
   end
   ]]
   -- 移除保存缓存调用
   -- save_registry_cache(up_apps)
   
   
   function _M.init_worker()
       default_weight = local_conf.discovery.eureka.weight or 100
       log.info("default_weight:", default_weight, ".")
       local fetch_interval = local_conf.discovery.eureka.fetch_interval or 30
       log.info("fetch_interval:", fetch_interval, ".")
       init_sema = semaphore.new()
   
       ngx_timer_at(0, fetch_full_registry)
       ngx_timer_every(fetch_interval, fetch_full_registry)
   end
   
   
   function _M.dump_data()
       return {config = local_conf.discovery.eureka, services = applications or {}}
   end
   
   
   return _M
   `


-- 
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.

To unsubscribe, e-mail: notifications-unsubscribe@apisix.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org