在nginx/openresty上实现动态upstream的一种方法

一、背景

在使用nginx作为WEB服务器时，通过反向代理把后端服务的接口映射出来供前端应用是一种常见的做法。同时，为了保障后端服务始终可用，通常一个后端服务会启动不止一个实例，并通过upstream把这些后端服务实例管理起来。而为了确保后端服务实例处于可用状态，也通常会在nginx的upstream上配置健康检查，健康检查的时间间隔通常也是按秒来设置的。

nginx的upstream服务器健康检查在80%的场景下是没有问题。但是如果后端接口的并发量较高，经常都有人访问，后端服务发布时，在服务还没有完成启动前，极有可能让一些客户得到502的响应，即后端服务超时。因此，如果能够在发布前将待发布的后端实例端口从nginx的upstream服务器列表剔除，待发布成功且后端服务可以正常提供服务时再将它放回upstream服务器列表，这个种情况将得到完美的解决。

二、本例实现路径

在nginx启用一个共享内存区域，用于存储upstream的后台服务器地址
在nginx的配置阶段初始化默认的upstream后端服务器列表
用balancer_by_lua_block来初始化一个upstream
正常配置反向代码
在nginx对外提供接口，用于变更upstream后端服务器列表

三、准备工作

为了进行后续配置，请确保已经在linux环境下安装与配置好了 openresty。

四、配置过程

1. 初始化upstream服务地址存储空间

lua_shared_dict upstream_dict 1m;

这个配置需要放到与server/upstream同一层级，如：

lua_shared_dict upstream_dict 1m;
upstream dynamic_upstream {
    ...
}

由于lua_shared_dict当中的value不支持复杂数据类型，后续upstream将通过如下key-value的形式在upstream_dict进行存储：

{//这是根节点
    "[upstream前缀][upstream后端服务地址]": "[upstream后端服务地址]"
}

注意：[]当中的内容为点位符。

实际应用中upstream_dict所存储数据示例：

{
    "nodejs_http://127.0.0.1:4000": "http://127.0.0.1:4000"
}

如果upstream后端服务较多，想节约一些内存，也可以考虑使用在key里面使用后端服务地址的hash。

2. 初始化upstream默认的后端服务地址列表

a. 为了完成这项工作，单独写了一个lua模块，用于upstream的后端服务各项管理工作。

local upsteam_servers = {}
local inited_postfix = '__upstream_inited'
function string.starts(String, Start)
    return string.sub(String, 1, string.len(Start)) == Start
end
function string.ends(String, Start)
    local whole_length = string.len(String)
    local part_length = string.len(Start)
    return string.sub(String, whole_length - part_length + 1) == Start
end
-- 加载upstream服务器列表
function upsteam_servers.get_server_list(shared_dict_name, prefix)
    local the_table = ngx.shared[shared_dict_name]
    ngx.log(ngx.DEBUG, 'Trying to get all server from ', shared_dict_name,
            ' with prefix ', prefix)
    if nil == the_table then return {} end
    local result = {}
    local the_keys = the_table:get_keys()
    for _, key in pairs(the_keys) do
        ngx.log(ngx.DEBUG, 'checking key ', key, ' if match prefix ', prefix)
        if string.starts(key, prefix) and false ==
            string.ends(key, inited_postfix) then
            table.insert(result, the_table:get(key))
        end
    end
    return result
end

-- find server from usptream list
function upsteam_servers.find(shared_dict_name, upstream_prefix, upstream_server)
    local upsteam_servers_data = upsteam_servers.get_server_list(
                                     shared_dict_name, upstream_prefix)
    return upsteam_servers.find_from_list(upsteam_servers_data, upstream_server)
end

function upsteam_servers.find_from_list(upsteam_servers_data, upstream_server)
    for k, v in pairs(upsteam_servers_data) do
        if v == server then return k end
    end
    return -1
end
-- remove server from list
function upsteam_servers.delete(shared_dict_name, upstream_prefix, server)
    ngx.log(ngx.DEBUG, "delete :: upstream = ", server, ' with upstream prefix ',
            upstream_prefix, ' from ', shared_dict_name)
    ngx.shared[shared_dict_name]:delete(upstream_prefix .. server)
    ngx.log(ngx.DEBUG, 'removed server is ', server)
    local server_list = upsteam_servers.get_server_list(shared_dict_name,
                                                        upstream_prefix)
    return #server_list
end
-- add server into list
function upsteam_servers.add(shared_dict_name, upstream_prefix, server)
    ngx.log(ngx.DEBUG, "add :: upstream = ", server, ' with upstream prefix ',
            upstream_prefix, ' from ', shared_dict_name)
    local the_dict = ngx.shared[shared_dict_name]
    if nil == the_dict then
        ngx.log(ngx.DEBUG, 'dict ', shared_dict_name, ' was not existed ')
        return -1
    end
    local succ, err, forcible = the_dict:set(upstream_prefix .. server, server)
    ngx.log(ngx.DEBUG, "success ? ", succ, " error ? ", err, " forcible ? ",
            forcible)
    ngx.log(ngx.DEBUG, 'added server ', server,
            ' into server list with upstream prefix ', upstream_prefix)
    local server_list = upsteam_servers.get_server_list(shared_dict_name,
                                                        upstream_prefix)
    return #server_list
end

-- get all server list
function upsteam_servers.all(shared_dict_name, upstream_prefix)
    return upsteam_servers.get_server_list(shared_dict_name, upstream_prefix)
end

function upsteam_servers.all(shared_dict_name, upstream_prefix)
    return upsteam_servers.get_server_list(shared_dict_name, upstream_prefix)
end
-- 检查某个upstream是否初始化完成
function upsteam_servers.is_upstream_inited(shared_dict_name, upstream_prefix)
    local the_dict = ngx.shared[shared_dict_name]
    if nil == the_dict then return false end
    return the_dict:get(upstream_prefix .. inited_postfix) == true
end
-- 设置某个upstream是否初始化完成
function upsteam_servers.set_upstream_inited(shared_dict_name, upstream_prefix,
                                             value)
    local the_dict = ngx.shared[shared_dict_name]
    if nil == the_dict then return false end
    the_dict:set(upstream_prefix .. inited_postfix, value)
end
return upsteam_servers

本例将以上脚本保存到lua_package_path目录下，并命名为dynamic_upstream.lua，这样在其它地方便可以通过local us = require 'dynamic_upstream'进行加载与引用。

b. 初始化服务器列表

以下样本需要添加到nginx的顶层配置，与server/upstream同级。

init_by_lua_block {
    local us = require "dynamic_upstream"
    local dict_name = 'upstream_dict'
    -- 注意以下前缀，后续有关upstream与upstream后端服务地址
    local upstream_prefix = 'nodejs_'
    if us.is_upstream_inited(dict_name, upstream_prefix) == false then
        us.add(dict_name, upstream_prefix,'127.0.0.1:3001')
        local count = us.add(dict_name, upstream_prefix,'127.0.0.1:3000')
        us.set_upstream_inited(dict_name, upstream_prefix, true)
        ngx.log(ngx.INFO, "total servers' count = ", count)
    end
}

#其它配置，方便看清上下文
upstream dynamic_upstream {
	server 0.0.0.1 fail_timeout=3;
    ...

3. 配置upstream

具体配置如下：

upstream dynamic_upstream {
    server 0.0.0.1 fail_timeout=3;
    balancer_by_lua_block {
        local balancer = require "ngx.balancer";
        local us = require "dynamic_upstream"
        local server_table = us.all('upstream_dict', 'nodejs_')
        if nil == server_table or #server_table == 0 then
        	return ngx.exit(500)
        end
        ngx.log(ngx.INFO, 'total servers count ', #server_table)
        local chosen_server = server_table[math.random(#server_table)]
        ok, err = balancer.set_more_tries(#server_table - 1)
        if not ok then
        	ngx.log(ngx.ERR, "set_more_tries failed: ", err)
        end
        ngx.log(ngx.ERR, 'chosen server ', chosen_server)
        ok, err = balancer.set_current_peer(chosen_server)
        if not ok then
        	ngx.log(ngx.ERR, "set_current_peer failed: ", err)
        return ngx.exit(500)
        end
    }
    keepalive 10;
}

balancer_by_lua_block的配置，还未仔细斟酌每处可能存在的坑，生产环境应用建议根据实际需要进一步调整。

4. 配置反向代理与upstream后端服务管理接口

a. 添加lua模块

为了可以在多台服务器都使用usptream管理功能，现把管理代码抽取成为一个模块。

local api = {}
local dus = require 'dynamic_upstream'
function api.handle_api(dict_name, upstream_prefix)
  local action = ngx.var.arg_action
  local upstream_item = ngx.var.arg_server
  if action ~= 'list' and (nil == action or nil == upstream_item) then
    ngx.say('invalid parameters, both action and server query params are needed.')
    return
  end
  if action == 'add' then
    local total = dus.add(dict_name, upstream_prefix, upstream_item);
    ngx.say('ok')
    return
  end
  if action == 'remove' then
    local total = dus.delete(dict_name, upstream_prefix, upstream_item)
    if false == total then
      ngx.say('failed')
      return
    end
    ngx.say('ok')
    return
  end
  if action == 'list' then
    for k,v in pairs(dus.all(dict_name, upstream_prefix)) do
      ngx.say(v)
    end
  end
end
return api

本例将以上脚本保存到lua_package_path目录下，并命名为dynamic-upstream-api.lua，这样在其它地方便可以通过local api = require 'dynamic-upstream-api'进行加载与引用。

b. 配置nginx

location /api {
    proxy_pass http://dynamic_upstream;
	# 其它设置...
}
location /_/upstreams {
    allow 127.0.0.1;
    allow ::1;
    deny all;
    content_by_lua_block {
        local dict_name = 'upstream_dict'
        local upstream_prefix = 'nodejs_'
        local api = require 'dynamic-upstream-api'
        api.handle_api(dict_name, upstream_prefix)
    }
}

c. upstream管理接口的用法

查询upstream后端服务列表

curl "http://{host}/_/upstreams?action=list"

返回的数据示例：

127.0.0.1:4000
127.0.0.1:4001

添加upstream后端服务地址

curl "http://{host}/_/upstreams?action=add&server={server_address}"

删除upstream后端服务地址

curl "http://{host}/_/upstreams?action=remove&server={server_address}"

五、upstream后端服务建议发布流程

0. 注意事项

避免并行发布同一个后端服务

1. 调用“查询后端服务列表”接口，确保有足够的后端服务实例可用

避免因为没有后端可用导致用户端异常