apisix etcd机制

local router          = require("apisix.router")

function _M.http_init_worker()
    .....
    router.http_init_worker()
    .....
end

function _M.http_init_worker()
    local conf = core.config.local_conf()
    local router_http_name = "radixtree_uri"

    if conf and conf.apisix and conf.apisix.router then
        router_http_name = conf.apisix.router.http or router_http_name
        ......
    end

    local router_http = require("apisix.http.router." .. router_http_name)
    attach_http_router_common_methods(router_http) -- 这里 attach 两个方法给到模块
    router_http.init_worker(filter) -- 这里调用 init_worker 完成初始化
    _M.router_http = router_http

    ......
end

local function attach_http_router_common_methods(http_router)
    if http_router.routes == nil then
        http_router.routes = function ()
            if not http_router.user_routes then
                return nil, nil
            end
            -- 返回模块中的 user_routes 的值值和版本
            local user_routes = http_router.user_routes
            return user_routes.values, user_routes.conf_version
        end
    end

    if http_router.init_worker == nil then
        http_router.init_worker = function (filter)
            http_router.user_routes = http_route.init_worker(filter)
            -- 这里引用 local http_route = require("apisix.http.route") 的 init_worker
            -- 获取 routes， 赋值给模块中的 user_routes
        end
    end
end

function _M.init_worker(filter)
    local user_routes, err = core.config.new("/routes", {
            automatic = true,
            item_schema = core.schema.route,
            checker = check_route,
            filter = filter,
        })
    if not user_routes then
        error("failed to create etcd instance for fetching /routes : " .. err)
    end

    return user_routes
end

local require = require
local core = require("apisix.core")
local base_router = require("apisix.http.route")
local get_services = require("apisix.http.service").services
local cached_router_version
local cached_service_version

local _M = {}

local uri_routes = {}
local uri_router

function _M.match(api_ctx)
    local user_routes = _M.user_routes
    local _, service_version = get_services()
    if not cached_router_version or cached_router_version ~= user_routes.conf_version
        or not cached_service_version or cached_service_version ~= service_version
    then
        uri_router = base_router.create_radixtree_uri_router(user_routes.values,
                                                             uri_routes, true)
        cached_router_version = user_routes.conf_version
        cached_service_version = service_version
    end

    if not uri_router then
        core.log.error("failed to fetch valid `uri_with_parameter` router: ")
        return true
    end

    return _M.matching(api_ctx)
end


function _M.matching(api_ctx)
    core.log.info("route match mode: radixtree_uri_with_parameter")
    return base_router.match_uri(uri_router, api_ctx)
end


return _M

function _M.init_worker(filter)
    local user_routes, err = core.config.new("/routes", {
            automatic = true,  -- 注意这个标志
            item_schema = core.schema.route,
            checker = check_route,
            filter = filter,
        })
    if not user_routes then
        error("failed to create etcd instance for fetching /routes : " .. err)
    end

    return user_routes
end

local config_provider = local_conf.deployment and local_conf.deployment.config_provider
                      or "etcd"
log.info("use config_provider: ", config_provider)
local config = require("apisix.core.config_" .. config_provider)
config.type = config_provider

return {
  config      = config,
}

---
-- Create a new connection to communicate with the control plane.
-- This function should be used in the `init_worker_by_lua` phase.
--

function _M.new(key, opts)
    ......
    -- etcd 相关配置
    local etcd_conf = local_conf.etcd
    local prefix = etcd_conf.prefix
    local resync_delay = etcd_conf.resync_delay
    if not resync_delay or resync_delay < 0 then
        resync_delay = 5
    end
    local health_check_timeout = etcd_conf.health_check_timeout
    if not health_check_timeout or health_check_timeout < 0 then
        health_check_timeout = 10
    end

    local automatic = opts and opts.automatic
    -- 初始化对象
    local obj = setmetatable({
        etcd_cli = nil,
        key = key and prefix .. key,
        automatic = automatic,  -- true
        item_schema = item_schema,
        checker = checker,
        sync_times = 0,   -- 同步次数
        running = true,
        conf_version = 0,  -- 当前数据版本号，可以通过这个判断是否有更新
        values = nil,      -- 所有同步过来的数据
        need_reload = true,  -- 注意这里，首次 new 的时候，need_reload 是 true
        watching_stream = nil,
        routes_hash = nil,
        prev_index = 0,     -- 上一次执行同步etcd 返回的 modifiedIndex
        last_err = nil,
        last_err_time = nil,
        resync_delay = resync_delay,
        health_check_timeout = health_check_timeout,
        timeout = timeout,  -- 注意，这里没有设置timeout, 那么在连接的时候，http_waitdir 将使用代码中的默认值 ` local ok, err = self.watch_sema:wait(timeout or 60)`
        single_item = single_item,
        filter = filter_fun,
    }, mt)

    if automatic then
        ......
        -- 如果 之前加载过，直接拿上一次数据全量加载
        if loaded_configuration[key] then
            local res = loaded_configuration[key]
            loaded_configuration[key] = nil -- tried to load

            log.notice("use loaded configuration ", key)

            local dir_res, headers = res.body, res.headers
            load_full_data(obj, dir_res, headers)
        end
   
        -- 启动一个定时器， local ngx_timer_at = ngx.timer.at
        ngx_timer_at(0, _automatic_fetch, obj)

    else
        -- 初始化 etcd_cli
        local etcd_cli, err = get_etcd()
        if not etcd_cli then
            return nil, "failed to start an etcd instance: " .. err
        end
        obj.etcd_cli = etcd_cli
    end

    -- 将 `/routes` 放入 created_obj
    if key then
        created_obj[key] = obj
    end

    return obj
end
    

local function _automatic_fetch(premature, self)
            ......
            local ok, err = sync_data(self)
            ......
            
    if not exiting() and self.running then
        ngx_timer_at(0, _automatic_fetch, self)
    end
end

local function sync_data(self)
    if not self.key then
        return nil, "missing 'key' arguments"
    end

    init_watch_ctx(self.key)

    if self.need_reload then
        local res, err = readdir(self.etcd_cli, self.key)
        if not res then
            return false, err
        end

        local dir_res, headers = res.body.list or res.body.node or {}, res.headers
        log.debug("readdir key: ", self.key, " res: ",
                  json.delay_encode(dir_res))

        if self.values then
            for i, val in ipairs(self.values) do
                config_util.fire_all_clean_handlers(val)
            end

            self.values = nil
            self.values_hash = nil
        end

        load_full_data(self, dir_res, headers)

        return true
    end

    local dir_res, err = waitdir(self)

    if not dir_res then
        if err == "compacted" then
            self.need_reload = true
            log.error("waitdir [", self.key, "] err: ", err,
                     ", will read the configuration again via readdir")
            return false
        end

        return false, err
    end