【网关开发】5.Openresty 自定义负载均衡与流量转发

背景

静态的nginx配置需要将负载均衡的服务节点信息都配置在配置文件中。现在微服务或云服务都会接入一些服务发现或者云控平台场景，经常需要更换节点，如果每次都要更新配置并且重启服务是无法接受的，所以需要网关提供动态扩展，实时更新自己负载均衡节点的能力，使用openresty网关需要使用lua扩展来实现自定义负载均衡的能力。

应用架构

实现

本文展示openresty如何开发，有一些结构体与细节可以直接去git上查看
https://github.com/zhaoshoucheng/openresty/tree/main/pkg/lua_script
使用插件地址：
https://github.com/openresty/lua-resty-balancer'
整体实现摸快
配置文件摸快：nginx的conf配置文件
lua-resty-balancer 插件：负载均衡器
upstream_context : 从etcd中获取相应的服务节点，通过负载均衡类型，获取负载均衡器

插件

编译lua-resty-balancer插件简单，进入文件夹make就可以，然后会生成librestychash.so文件，将其放入到我们可以引入到的文件夹中
例如

lua_package_cpath '/data/openresty/pkg/lua_script/?.so;;';

配置文件

upstream server_test1 {
server 0.0.0.0:1234;
balancer_by_lua_block {
    if etcd_source_module then require 'upstream.balance'.do_balance('server_test1') else ngx.exit(502) end
    }
}

etcd_source_module etcd数据模块，负责从etcd中获取服务server_test1的IP，这里知识检查作用。
do_balance 实现流量转发的模块函数

流量转发

local transform_data_simple = utils.transform_data_simple
local module_name = (...):match("(.-)[^%.]+$")
local upstream_context = require(module_name.."upstream_context")
local conf = require(module_name.."config")
local utils = require(module_name.."utils")
local ngx_balancer = require "ngx.balancer"
local cjson = require "cjson.safe"

-- 从etcd中获取数据，通过transform_data_simple 转化数据
local function get_upstream_context(name)
    local ctx = upstream_contexts[name]
    if not ctx then
        local ups = etcd_source_module:get_value(name)
        if not ups then
            return nil, "upstream configure not found: "..name
        end
        ups = transform_data_simple(ups)
        if ups then
            ctx = upstream_context.new(name, ups)
            upstream_contexts[name] = ctx
        end
    end
    return ctx
end

function _M.do_balance(ups_name)
    local ctx = ngx.ctx
    local uctx, err = get_upstream_context(ups_name)
    if not uctx then
        ngx.log(ngx.ERR, ups_name..", "..tostring(err))
        ngx.exit(502)
        return
    end
    -- 获取负载均衡器
    local b, err = uctx:get_prefered_balancer()
    if not b then
        ngx.log(ngx.ERR, "failed to get balancer: "..tostring(err))
        ngx.exit(502)
        return
    end
    -- 上次失败的节点要在下次节点中避免选择
    local key, idx
    local sn, sc = ngx_balancer.get_last_failure()
    if not sn then
        -- first call
        local ok, err = ngx_balancer.set_more_tries(3)
        if err and #err > 0 then
            ngx.log(ngx.WARN, err)
        end
        key, idx = b:find(ctx.balance_key)
        if not key then
            ngx.log(ngx.ERR, "failed to get upstream endpoint")
            ngx.exit(502)
            return
        end
    else
        -- 根据上次的负载情况选择下次的节点
        key, idx = b:next(ctx.latest_idx)
        ngx.log(ngx.WARN, "rebalancing: "..sn..", "..tostring(sc))
    end
    ngx.log(ngx.ERR,"do_balance b"..cjson.encode(b))  -- 打印负载均衡器
    -- 从所有节点中进行选择
    local peer = uctx._all_nodes[key]
    ngx.log(ngx.ERR,"do_balance uctx._all_nodes"..cjson.encode(uctx._all_nodes))    -- 打印_all_node 信息
    if not peer then
        ngx.log(ngx.ERR, "failed to get upstream endpoint: "..tostring(key))
        ngx.exit(502)
        return
    end
    ctx.latest_peer = peer
    ctx.latest_key = key
    ctx.latest_idx = idx
    -- 流量转发函数
    local ok, err = ngx_balancer.set_current_peer(peer.ip, peer.port)
    ngx.log(ngx.INFO, string.format(" do balance ip :%s point: %s", peer.ip, peer.port))  -- 打印目标节点
    if not ok then
        ngx.log(ngx.ERR, string.format("error while setting current upstream peer %s: %s", peer.ip, err))
        ngx.exit(500)
        return
    end
end

看下_all_nodes的结构，map结构key值是传入负载均衡器的值，后面会有用到

2023/01/19 14:41:32 [error] 17588#0: *234587 [lua] balance.lua:66: do_balance(): do_balance uctx._all_nodes{"10.218.22.246\u00008090":{"state":"up","fail_timeout":3000,"weight":1,"ip":"10.218.22.246","max_fail":3,"port":8090},"10.218.22.239\u00008090":{"state":"up","fail_timeout":3000,"weight":1,"ip":"10.218.22.239","max_fail":3,"port":8090}} 
while connecting to upstream, client: 10.99.4.169, server: server_test.com, request: "GET /ping HTTP/1.1", host: "server_test.com:9090"

打印每一次获得负载均衡器

2023/01/19 14:45:32 [error] 17656#0: *234610 [lua] balance.lua:67: do_balance(): do_balance b{"gcd":1,"cw":1,"nodes":{"10.218.22.239\u00008090":1,"10.218.22.246\u00008090":1},"last_id":"10.218.22.239\u00008090","max_weight":1} 
while connecting to upstream, client: 10.99.4.169, server: server_test.com, request: "GET /ping HTTP/1.1", host: "server_test.com:9090"

负载均衡器

-- 获取负载均衡器
local function get_prefered_balancer(self)
    local b = self._prefered_balancer
    local err = nil
    if not b then
        local nodes, _ = process_upstream_nodes(self._ups.nodes)
        local lb = self._ups.load_balance
        local err
        ngx.log(ngx.ERR,"get_prefered_balancer nodes"..cjson.encode(nodes))  -- 打印nodes数据
        b, err = balancers.create(lb.type, nodes)
        if not b then
            return nil, err
        end
        self._prefered_balancer = b
    end
    return b, err
end

不需要过度关注结构体的设计，用法场景不同，结构体自然不同，比较灵活，process_upstream_nodes目的就是选出目标所有可用的节点，可以结合健康检查结果，或者流量规则进行使用
最终生成的nodes就是map的key,权重的结构。将其作为负载均衡插件的参数。

2023/01/19 14:29:14 [error] 17335#0: *234481 [lua] upstream_context.lua:78: get_prefered_balancer(): get_prefered_balancer nodes{"10.218.22.239\u00008090":1,"10.218.22.246\u00008090":1} 
while connecting to upstream, client: 10.99.4.169, server: server_test.com, request: "GET /ping HTTP/1.1", host: "server_test.com:9090"

选择负载均衡器,将提供的负载均衡器封装成函数

local roundrobin = require("resty.balancer.roundrobin")
local chash = require("resty.balancer.chash")
local iphash = require("resty.balancer.iphash")
local urlhash = require("resty.balancer.urlhash")

local _M = { }
local function create(typ, ...)
    if typ == "roundrobin" or typ == "round_robin" then
        return roundrobin:new(...)
    elseif typ == "chash" then
        return chash:new(...)
    elseif typ == "iphash" or typ == "ip_hash" then
        return iphash:new(...)
    elseif typ == "urlhash" or typ == "url_hash" then
        return urlhash:new(...)
    else
        return nil, "unsupported balancer type: "..tostring(typ)
    end
end
_M.create = create
return _M

测试

round_robin 测试

ip_hash 测试

总结

当然企业级工程不可能这么简单，这只是核心的细节。我试图将复杂的整体工程拆分成一个一个的小知识点来进行分析，学习。完善自己的知识体系。所以单单是看单个知识点都比较简单。尽量避免了一些复杂的结构体和复杂的业务场景来讲解架构与知识点，这样更加的容易理解。
能把一件事情讲明白其实是一件很不容易的事情，如果有不解或者有误的地方欢迎沟通交流,有些细节代码也上传了github,欢迎查看。
https://github.com/zhaoshoucheng/openresty/tree/main/pkg/lua_script

扩展

因为github插件上提供的都是基础功能round_robin和chash，其实其他负载均衡算法也可以在这上面进行扩展。
或者实现自己的负载均衡算法。
现在提供ip_hash 和 url_hash的写法

ip_hash

local module_name = (...):match("(.-)[^%.]+$")
local chash = require(module_name.."chash")
local setmetatable = setmetatable

local _M = { _VERSION = "0.1" }

local _MT = { __index = _M }

function _M.new(_, nodes)
    return setmetatable({
        _chb = chash:new(nodes)
    }, _MT)
end

function _M:reinit(nodes)
    self._chb:reinit(nodes)
end

function _M:set(...)
    self._chb:set(...)
end

function _M:find()
    return self._chb:find(ngx.var.remote_addr) -- TODO: use xff ?
end

function _M:next(...)
    self._chb:next(...)
end

return _M

url_hash

local module_name = (...):match("(.-)[^%.]+$")
local chash = require(module_name.."chash")
local setmetatable = setmetatable

local _M = { _VERSION = "0.1" }

local _MT = { __index = _M }

function _M.new(_, nodes)
    return setmetatable({
        _chb = chash:new(nodes)
    }, _MT)
end

function _M:reinit(nodes)
    self._chb:reinit(nodes)
end

function _M:set(...)
    self._chb:set(...)
end

function _M:find()
    return self._chb:find(ngx.var.uri)
end

function _M:next(...)
    self._chb:next(...)
end

return _M