Kong 代码阅读

--- ctx 中保存有匹配的请求特征值
match_route = function(route_t, ctx)
    -- run cached matcher
    if type(matchers[route_t.match_rules]) == "function" then
        clear_tab(ctx.matches)
        return matchers[route_t.match_rules](route_t, ctx)
    end

    -- build and cache matcher

    local matchers_set = {}

    for _, bit_match_rule in pairs(MATCH_RULES) do
        --- 遍历特征 匹配
        if band(route_t.match_rules, bit_match_rule) ~= 0 then
            matchers_set[#matchers_set + 1] = matchers[bit_match_rule]
        end
    end

    matchers[route_t.match_rules] = function(route_t, ctx)
        -- clear matches context for this try on this route
        clear_tab(ctx.matches)

        -- 路由 route_t 必须满足匹配的所有特征
        for i = 1, #matchers_set do
            if not matchers_set[i](route_t, ctx) then
                return
            end
        end

        return true
    end

    return matchers[route_t.match_rules](route_t, ctx)
end

... -- init, omit
insert(category.all, route_t)

for _, host_t in ipairs(route_t.hosts) do
    if not category.routes_by_hosts[host_t.value] then
        category.routes_by_hosts[host_t.value] = {}
    end

    insert(category.routes_by_hosts[host_t.value], route_t)
end

--- omit code,
---  category.routes_by_headers
---  category.routes_by_uris
---  category.routes_by_methods
---  category.routes_by_sources
---  category.routes_by_destinations
---  category.routes_by_sni

-- header match
for _, header_name in ipairs(plain_indexes.headers) do
    if req_headers[header_name] then
        req_category = bor(req_category, MATCH_RULES.HEADER)
        hits.header_name = header_name
        break   // 匹配后结束 header 类别匹配
    end
end

-- host match

if plain_indexes.hosts[host_with_port] or plain_indexes.hosts[host_no_port] then
    req_category = bor(req_category, MATCH_RULES.HOST)
elseif ctx.req_host then
    for i = 1, #wildcard_hosts do
        local from, _, err = re_find(host_with_port, wildcard_hosts[i].regex, "ajo")
        ... -- omit
        if from then
            hits.host = wildcard_hosts[i].value
            req_category = bor(req_category, MATCH_RULES.HOST)
            break
        end
    end
end

-- uri match

for i = 1, #regex_uris do
    local from, _, err = re_find(req_uri, regex_uris[i].regex, "ajo")

    ... -- omit

    if from then
        hits.uri = regex_uris[i].value
        req_category = bor(req_category, MATCH_RULES.URI)
        break
    end
end

if not hits.uri then
    if plain_indexes.uris[req_uri] then
        hits.uri = req_uri
        req_category = bor(req_category, MATCH_RULES.URI)
    else
        for i = 1, #prefix_uris do
            if find(req_uri, prefix_uris[i].value, nil, true) == 1 then
                hits.uri = prefix_uris[i].value
                req_category = bor(req_category, MATCH_RULES.URI)
                break
            end
        end
    end
end

-- method match

if plain_indexes.methods[req_method] then
    req_category = bor(req_category, MATCH_RULES.METHOD)
end

-- src match

if plain_indexes.sources[ctx.src_ip] then
    req_category = bor(req_category, MATCH_RULES.SRC)
elseif plain_indexes.sources[ctx.src_port] then
    req_category = bor(req_category, MATCH_RULES.SRC)
else
    for i = 1, #src_trust_funcs do
        if src_trust_funcs[i](ctx.src_ip) then
            req_category = bor(req_category, MATCH_RULES.SRC)
            break
        end
    end
end

-- dst match

if plain_indexes.destinations[ctx.dst_ip] then
    req_category = bor(req_category, MATCH_RULES.DST)
elseif plain_indexes.destinations[ctx.dst_port] then
    req_category = bor(req_category, MATCH_RULES.DST)
else
    for i = 1, #dst_trust_funcs do
        if dst_trust_funcs[i](ctx.dst_ip) then
            req_category = bor(req_category, MATCH_RULES.DST)
            break
        end
    end
end

-- sni match

if plain_indexes.snis[ctx.sni] then
    req_category = bor(req_category, MATCH_RULES.SNI)
end

-- 生成通用 API 函数
local function generate_collection_endpoints(endpoints, schema, foreign_schema, foreign_field_name)
  local collection_path
  if foreign_schema then
    collection_path = fmt("/%s/:%s/%s", foreign_schema.admin_api_name or
                                        foreign_schema.name,
                                        foreign_schema.name,
                                        schema.admin_api_nested_name or
                                        schema.admin_api_name or
                                        schema.name)

  else
    collection_path = fmt("/%s", schema.admin_api_name or
                                 schema.name)
  end

  endpoints[collection_path] = {
    schema  = schema,
    methods = {
      --OPTIONS = method_not_allowed,
      --HEAD    = method_not_allowed,
      GET     = get_collection_endpoint(schema, foreign_schema, foreign_field_name),
      POST    = post_collection_endpoint(schema, foreign_schema, foreign_field_name),
      --PUT     = method_not_allowed,
      --PATCH   = method_not_allowed,
      --DELETE  = method_not_allowed,
    },
  }
end

local function post_collection_endpoint(schema, foreign_schema, foreign_field_name, method)
  return function(self, db, helpers, post_process)

    if foreign_schema then
      local foreign_entity, _, err_t = select_entity(self, db, foreign_schema)
      if err_t then
        return handle_error(err_t)
      end

      if not foreign_entity then
        return not_found()
      end

      self.args.post[foreign_field_name] = foreign_schema:extract_pk_values(foreign_entity)
    end

    -- 数据插入数据库
    local entity, _, err_t = insert_entity(self, db, schema, method)
    if err_t then
      return handle_error(err_t)
    end

    -- 调用注册回调方法
    if post_process then
      entity, _, err_t = post_process(entity)
      if err_t then
        return handle_error(err_t)
      end
    end

    return created(entity)
  end
end

function _M.new(opts)
	-- ... 忽略无关代码
  local self = assert(balancer_base.new(opts))
	-- ..
  -- inject additional properties
  self.wheel = nil   -- 用于存储负载地址的表
  self.pointer = nil -- 轮询时下个待选择地址
  self.wheelSize = opts.wheelSize or 1000 -- 负载地址表大小
  self.unassignedWheelIndices = nil -- wheel 中未分配的索引

  -- ring_balancer 对 balancer_base 继承
  -- inject overridden methods
  for name, method in pairs(ring_balancer) do
    self[name] = method
  end

  -- initialize the balancer
  self.wheel = new_tab(self.wheelSize, 0)
  self.unassignedWheelIndices = new_tab(self.wheelSize, 0)
  self.pointer = math.random(1, self.wheelSize)  -- ensure each worker starts somewhere else

  -- Create a list of entries, and randomize them.
  local unassignedWheelIndices = self.unassignedWheelIndices
  local duplicateCheck = new_tab(self.wheelSize, 0)
  -- 使用 randomlist 创建一个 1-wheelSize 范围的随机数数组, 数值不会重复. 用于将负载地址随机分布在 wheel 数组中
  local orderlist = opts.order or randomlist(self.wheelSize)

  for i = 1, self.wheelSize do
    local order = orderlist[i]
    unassignedWheelIndices[i] = order
  end

  -- Sort the hosts, to make order deterministic
  -- ...
  table_sort(hosts, function(a,b) return (a.name..":"..(a.port or "") < b.name..":"..(b.port or "")) end)
  -- Insert the hosts
  -- 添加地址, 并在 wheel 中添加索引
  for _, host in ipairs(hosts) do
    local ok, err = self:addHost(host.name, host.port, host.weight)
  end

  return self
end

function ring_balancer:redistributeIndices()
  local totalWeight = self.weight -- Host 变更时会在 base 中会修改 weight
  local movingIndexList = self.unassignedWheelIndices

  -- NOTE: calculations are based on the "remaining" indices and weights, to
  -- prevent issues due to rounding: eg. 10 equal systems with 19 indices.
  -- Calculated to get each 1.9 indices => 9 systems would get 1, last system would get 10
  -- by using "remaining" indices, the first would get 1 index, the other 9 would get 2.

  -- first; reclaim extraneous indices
  local weightLeft = totalWeight
  local indicesLeft = self.wheelSize
  local addList = {}      -- addresses that need additional indices
  local addListCount = {} -- how many extra indices the address needs
  local addCount = 0
  local dropped, added = 0, 0
	-- 便利所有负载
  for weight, address, _ in self:addressIter() do

    -- 根据权重, 计算负载索引个数
    local count
    if weightLeft == 0 then
      count = 0
    else
      count = math_floor(indicesLeft * (weight / weightLeft) + 0.0001) -- 0.0001 to bypass float arithmetic issues
    end
    local drop = #address.indices - count
    if drop > 0 then
      -- we need to reclaim some indices
      address:dropIndices(movingIndexList, drop)
      dropped = dropped + drop
    elseif drop < 0 then
      -- this one needs extra indices, so record the changes needed
      -- 记录需要添加索引的地址, 稍后为地址分配索引
      addCount = addCount + 1
      addList[addCount] = address
      addListCount[addCount] = -drop  -- negate because we need to add them
    end
    indicesLeft = indicesLeft - count
    weightLeft = weightLeft - weight
  end

  -- 为地址分配索引
  -- second: add freed indices to the recorded addresses that were short of them
  for i, address in ipairs(addList) do
    address:addIndices(movingIndexList, addListCount[i])
    added = added + addListCount[i]
  end

  return self
end

function ring_address:addIndices(availableIndicesList, count)
  count = count or #availableIndicesList
  if count > 0 then
    local myWheelIndices = self.indices
    local size = #myWheelIndices
		-- ...
    local wheel = self.host.balancer.wheel
    local lsize = #availableIndicesList + 1
    for i = 1, count do
      -- 获取一个未分配索引
      local availableIdx = lsize - i
      local wheelIdx = availableIndicesList[availableIdx]
      availableIndicesList[availableIdx] = nil
      -- 当前地址增加一个索引,(地址删除索引时使用)
      myWheelIndices[size + i] = wheelIdx
      wheel[wheelIdx] = self
    end

  end
  return self
end