【网关开发】2.openresty 使用lua扩展 链接etcd数据库

标签：网关 end self lua etcd local resp

背景

在使用openresty时有一些定制化的负载均衡功能，有些元数据是保存在etcd中的，所以需要openresty与etcd进行交互，可以获取全量数据，并且使用etcd的watch功能。

上一篇

1.编译Openresty进行服务器初始化搭建

工程代码库

https://github.com/zhaoshoucheng/openresty

目录

工程目录结构

init_main.lua 在主进程启动时调用
init_worker.lua 在工作进程启动时调用
upstream 文件夹是用来定制化负载均衡功能，目前包括etcd的初始化
----> init.lua etcd初始化代码 config.lua etcd连接配置文件
resty.source etcd使用逻辑代码
----> etcdsource.lua 逻辑代码
resty.etcd etcd的sdk代码
etcd.lua、http_connect.lua、http_headers.lua、http.lua etcd依赖sdk代码

环境准备

etcd 驱动地址

https://github.com/api7/lua-resty-etcd
我们可以直接将源码 /lib/resty/etcd 文件夹和etcd.lua 获取
需要依赖 api7-lua-resty-http 和 lua-typeof

api7-lua-resty-http

https://github.com/ledgetech/lua-resty-http
需要文件：/lib/resty/ 下的http.lua、http_connect.lua、http_headers.lua
注意目录位置，不然会有依赖问题

lua-typeof

https://github.com/iresty/lua-typeof
需要文件：/lib/typeof.lua

可能遇到的问题

如果使用1.9.0 ，可能会遇到找不到pl.path的问题，这里我降低了版本，使用1.7.0
可能 api7-lua-resty-http和lua-typeof 引入的文件找不到，这个很好解决，调试时会把查找的文件列表都列出来，针对引用目录进行调整。

接口使用

lua文件引用

nginx文件中，引入lua文件
apis.conf

# 启动worker进行时执行的lua文件
init_worker_by_lua_file /data/pkg/lua_script/init_worker.lua;

init_worker.lua

require "upstream.init".on_init_worker()

config.lua 配置文件

return {
    etcd_options = {
        http_host = {
            "http://127.0.0.1:2379",
        },
        protocol = "v3",
        ssl_verify = false,
        user = "user1",
        password = "123456",
    },
    watch_path = "/openresty/demo/",
}

init.lua

-- 初始化etcd客户端
local module_name = (...):match("(.-)[^%.]+$")
local upstream_conf = require(module_name .. "config")

local function on_init_worker()
    local etcd_source =     
    require "resty.source.etcdsource".new(
        {
            etcd_conf = upstream_conf.etcd_options,
            cache_path = upstream_conf.cache_path,
            prefix = upstream_conf.watch_path,
            map_size = upstream_conf.map_size,
        }
    )

    local ok, err = etcd_source:init()
    if not ok then
        ngx.log(ngx.ERR, "failed to init obconf: "..tostring(err))
    end
end
return {
    on_init_worker = on_init_worker
}

接口使用
etcdsource.lua 启动函数部分

local function init(self)
    self._need_full_sync = true
    -- 通过定时器的方式执行
    local __on_master = function()
        return ngx.timer.at(0, init_sync_or_watch, self)
    end

    return __on_master()
end

_M.init      = init

return _M

etcdsource.lua 逻辑函数部分

local function init_sync_or_watch(p, self)
    -- 定时器自带参数处理
    if p then
        return
    end
    local ok, err
    -- 全量获取数据
    if self._need_full_sync then
        self._need_full_sync = false
        ok, err = _full_sync(self)
        if not ok then
            if err == "not master" then
                return
            end
            self._need_full_sync = true
            ngx.log(ngx.ERR, "failed to _full_sync: "..tostring(err))
            ngx.timer.at(5, init_sync_or_watch, self) -- retry
            return
        end
        if self.exit then
            return
        end
    end
    -- 增量获取数据
    ok, err = _watch_sync(self) -- blocked
    if err and err ~= "timeout" then
        ngx.log(ngx.ERR, "failed to _watch_sync: "..err)
        self._need_full_sync = not self.exit
    end
    if not self.exit then
        ngx.timer.at(err == "timeout" and 0 or 5, init_sync_or_watch, self) -- retry
    end
end

etcdsource.lua 全量获取函数

local function _full_sync(self)
    -- 获取客户端
    local _etcd, err = _get_etcd_cli(self)
    if not _etcd then
        return nil, err
    end
    -- 获取指定前缀的kv,
    local resp, err = _etcd:readdir(self.prefix)
    if not resp then
        return nil, err
    end
    if resp.status ~= 200 then
        return nil, "server response with: "..tostring(resp.status)
    end
    local kvs = resp.body.kvs or { }
    if not resp.body.kvs or #kvs == 0 then
        return nil, "server response empty kvs"
    end

    for i = 1, #kvs do
       local kv = kvs[i]
       local obj, err = json_decode(kv.value)
       if obj then
        -- 使用KV，这里只是打印一条
        ngx.log(ngx.INFO, "etcd decode data key: ", kv.key," value :",obj)
       end
    end

    local _new_revision = resp.body.header.revision
    -- 更新版本，用于增量同步
    self._etcd_revision = _new_revision
    ngx.log(ngx.INFO, "etcd decode data revision : ", _new_revision)
    return true
end

etcdsource.lua 增量获取函数

local function _watch_sync(self)
    local opts = {
        start_revision = big_int_incr(self._etcd_revision),
        timeout = 60,
        need_cancel  = true,
    }
    local _etcd, err = _get_etcd_cli(self)
    if not _etcd then
        return nil, err
    end    
    local reader, err, http_cli = _etcd:watchdir(self.prefix, opts)
    if not reader then
        return nil, err
    end
    local resp
    repeat
        resp, err = reader()
        if resp then
            if resp.result.canceled then
                ngx.log(ngx.WARN, "cancel_reason: "..tostring(resp.result.cancel_reason))
                break
            end
            if resp.compact_revision and big_int_cmp(resp.compact_revision, self._etcd_revision) > 0 then
                -- revision has been compacted since we last sync
                -- need to restart full sync
                err = "revision has been compacted since we last sync, compact_revision: "..tostring(resp.compact_revision)..", _etcd_revision: "..tostring(self._etcd_revision)
                break
            end
            -- ngx.log(ngx.INFO, "watch: "..tostring(json_encode(resp)))
            local resp_events = resp.result.events
            if resp_events and #resp_events > 0 then
                for i = 1, #resp_events do
                    local evt = resp_events[i]
                    local kv = evt.kv
                    if kv then
                        -- do something
                        ngx.log(ngx.INFO, "ectd value`"..kv.key.."value"..kv.value.."`")
                    end
                end
            end
            if resp.result.header then
                -- maintaining local revision
                local revision = resp.result.header.revision
                if revision then
                    ngx.log(ngx.INFO, "ectd __etcd_revision__`"..revision.."`")
                end
            end
        end
    until err
    return true
end

有些函数放在 source/utils.lua 中，请自行去git查看

结果验证

可能遇到的问题

在etcd.v3 会报错
ngx/re.lua:47: missing declaration for symbol 'ngx_http_lua_ffi_exec_regex'
应该是pcre引用的问题。因为我原来用的是1.15.8.1 会有这个问题，换成1.15.8.3并不能解决。
讨论链接：https://github.com/openresty/lua-resty-core/issues/258

我是用来最简单的办法，使用了1.19.3.1，编译之后可以使用

标签：网关,end,self,lua,etcd,local,resp
From： https://www.cnblogs.com/zhaosc-haha/p/16789238.html