feat: add new lua-resty-events as events implementation

.github/workflows/build.yml

@@ -28,6 +28,9 @@ jobs:
           - ubuntu-20.04
         os_name:
           - linux_openresty
+        events_module:
+          - lua-resty-worker-events
+          - lua-resty-events
         test_dir:
           - t/plugin/[a-k]*
           - t/plugin/[l-z]*
@@ -169,6 +172,7 @@ jobs:
       - name: Linux Script
         env:
           TEST_FILE_SUB_DIR: ${{ matrix.test_dir }}
+          TEST_EVENTS_MODULE: ${{ matrix.events_module }}
         run: sudo -E ./ci/${{ matrix.os_name }}_runner.sh script
 
       - if: ${{ steps.cache-images.outputs.cache-hit != 'true' }}

apisix/cli/ngx_tpl.lua

@@ -199,6 +199,17 @@ stream {
         apisix.stream_init_worker()
     }
 
+    {% if (events.module or "") == "lua-resty-events" then %}
+    # the server block for lua-resty-events
+    server {
+        listen unix:{*apisix_lua_home*}/logs/stream_worker_events.sock;
+        access_log off;
+        content_by_lua_block {
+            require("resty.events.compat").run()
+        }
+    }
+    {% end %}
+
     server {
         {% for _, item in ipairs(stream_proxy.tcp or {}) do %}
         listen {*item.addr*} {% if item.tls then %} ssl {% end %} {% if enable_reuseport then %} reuseport {% end %} {% if proxy_protocol and proxy_protocol.enable_tcp_pp then %} proxy_protocol {% end %};
@@ -483,6 +494,19 @@ http {
         apisix.http_exit_worker()
     }
 
+    {% if (events.module or "") == "lua-resty-events" then %}
+    # the server block for lua-resty-events
+    server {
+        listen unix:{*apisix_lua_home*}/logs/worker_events.sock;
+        access_log off;
+        location / {
+            content_by_lua_block {
+                require("resty.events.compat").run()
+            }
+        }
+    }
+    {% end %}
+
     {% if enable_control then %}
     server {
         listen {* control_server_addr *};

apisix/events.lua

@@ -15,18 +15,27 @@
 -- limitations under the License.
 --
 
-local require = require
-local error   = error
-local ngx     = ngx
-
-local _M = {}
+local require      = require
+local error        = error
+local assert       = assert
+local tostring     = tostring
+local pairs        = pairs
+local setmetatable = setmetatable
+local ngx          = ngx
+local core         = require("apisix.core")
+
+local _M = {
+    events_module = nil,
+}
 
 _M.EVENTS_MODULE_LUA_RESTY_WORKER_EVENTS = 'lua-resty-worker-events'
 _M.EVENTS_MODULE_LUA_RESTY_EVENTS = 'lua-resty-events'
 
 
 -- use lua-resty-worker-events
-local function init_worker_events()
+local function init_resty_worker_events()
+    _M.events_module = _M.EVENTS_MODULE_LUA_RESTY_WORKER_EVENTS
+
     local we = require("resty.worker.events")
     local shm = ngx.config.subsystem == "http" and "worker-events" or "worker-events-stream"
     local ok, err = we.configure({shm = shm, interval = 0.1})
@@ -38,17 +47,53 @@ local function init_worker_events()
 end
 
 
+-- use lua-resty-events
+local function init_resty_events()
+    _M.events_module = _M.EVENTS_MODULE_LUA_RESTY_EVENTS
+
+    local listening = "unix:" .. ngx.config.prefix() .. "logs/"
+    if ngx.config.subsystem == "http" then
+        listening = listening .. "worker_events.sock"
+    else
+        listening = listening .. "stream_worker_events.sock"
+    end
+    core.log.info("subsystem: " .. ngx.config.subsystem .. " listening sock: " .. listening)
+
+    local opts = {
+        unique_timeout = 5,     -- life time of unique event data in lrucache
+        broker_id = 0,          -- broker server runs in nginx worker #0
+        listening = listening,  -- unix socket for broker listening
+    }
+
+    local we = require("resty.events.compat")
+    assert(we.configure(opts))
+    assert(we.configured())
+
+    return we
+end
+
+
 function _M.init_worker()
     if _M.inited then
-        -- Prevent duplicate initializations in the same worker to
-        -- avoid potentially unanticipated behavior
+        -- prevent duplicate initializations in the same worker to
+        -- avoid potentially unexpected behavior
         return
     end
 
     _M.inited = true
 
-    -- use lua-resty-worker-events default now
-    _M.worker_events = init_worker_events()
+    local conf = core.config.local_conf()
+    local module_name = core.table.try_read_attr(conf, "apisix", "events", "module")
+                            or _M.EVENTS_MODULE_LUA_RESTY_WORKER_EVENTS
+
+    if module_name == _M.EVENTS_MODULE_LUA_RESTY_EVENTS then
+        -- use lua-resty-events as an event module via the apisix.events.module
+        -- key in the configuration file
+        _M.worker_events = init_resty_events()
+    else
+        -- use lua-resty-worker-events default now
+        _M.worker_events = init_resty_worker_events()
+    end
 end
 
 
@@ -57,8 +102,23 @@ function _M.register(self, ...)
 end
 
 
-function _M.event_list(self, ...)
-    return self.worker_events.event_list(...)
+function _M.event_list(self, source, ...)
+    -- a patch for the lua-resty-events to support event_list
+    -- this snippet is copied from the lua-resty-worker-events lib
+    if self.events_module == _M.EVENTS_MODULE_LUA_RESTY_EVENTS then
+        local events = { _source = source }
+        for _, event in pairs({...}) do
+            events[event] = event
+        end
+        return setmetatable(events, {
+            __index = function(_, key)
+            error("event '"..tostring(key).."' is an unknown event", 2)
+            end
+        })
+    end
+
+    -- the lua-resty-worker-events has a built-in event_list implementation
+    return self.worker_events.event_list(source, ...)
 end
 
 
@@ -67,4 +127,13 @@ function _M.post(self, ...)
 end
 
 
+function _M.get_healthcheck_events_modele(self)
+    if self.events_module == _M.EVENTS_MODULE_LUA_RESTY_EVENTS then
+        return "resty.events"
+    else
+        return "resty.worker.events"
+    end
+end
+
+
 return _M

apisix/upstream.lua

@@ -19,6 +19,7 @@ local core = require("apisix.core")
 local discovery = require("apisix.discovery.init").discovery
 local upstream_util = require("apisix.utils.upstream")
 local apisix_ssl = require("apisix.ssl")
+local events = require("apisix.events")
 local error = error
 local tostring = tostring
 local ipairs = ipairs
@@ -110,10 +111,18 @@ local function create_checker(upstream)
     end
     upstream.is_creating_checker = true
 
+    core.log.debug("events module used by the healthcheck: ", events.events_module,
+                    ", module name: ",events:get_healthcheck_events_modele())
+
     local checker, err = healthcheck.new({
         name = get_healthchecker_name(healthcheck_parent),
         shm_name = "upstream-healthcheck",
         checks = upstream.checks,
+        -- the events.init_worker will be executed in the init_worker phase,
+        -- events.healthcheck_events_module is set
+        -- while the healthcheck object is executed in the http access phase,
+        -- so it can be used here
+        events_module = events:get_healthcheck_events_modele(),
     })
 
     if not checker then

ci/centos7-ci.sh

@@ -40,12 +40,19 @@ install_dependencies() {
 
     # install openresty to make apisix's rpm test work
     yum install -y yum-utils && yum-config-manager --add-repo https://openresty.org/package/centos/openresty.repo
-    wget "https://raw.githubusercontent.com/api7/apisix-build-tools/apisix-runtime/${APISIX_RUNTIME}/build-apisix-runtime-debug-centos7.sh"
-    wget "https://raw.githubusercontent.com/api7/apisix-build-tools/apisix-runtime/${APISIX_RUNTIME}/build-apisix-runtime.sh"
+
+    # TODO: disabled temporarily, waiting for APISIX 3.8 to be released to synchronize the apisix-runtime version
+    #wget "https://raw.githubusercontent.com/api7/apisix-build-tools/apisix-runtime/${APISIX_RUNTIME}/build-apisix-runtime-debug-centos7.sh"
+    #wget "https://raw.githubusercontent.com/api7/apisix-build-tools/apisix-runtime/${APISIX_RUNTIME}/build-apisix-runtime.sh"
+    wget "https://raw.githubusercontent.com/api7/apisix-build-tools/master/build-apisix-runtime-debug-centos7.sh"
+    wget "https://raw.githubusercontent.com/api7/apisix-build-tools/master/build-apisix-runtime.sh"
     chmod +x build-apisix-runtime-debug-centos7.sh
     chmod +x build-apisix-runtime.sh
     ./build-apisix-runtime-debug-centos7.sh
 
+    # patch lua-resty-events
+    sed -i 's/log(ERR, "event worker failed: ", perr)/log(ngx.WARN, "event worker failed: ", perr)/' /usr/local/openresty/lualib/resty/events/worker.lua
+
     # install luarocks
     ./utils/linux-install-luarocks.sh
 
@@ -95,7 +102,7 @@ run_case() {
     make init
     set_coredns
     # run test cases
-    FLUSH_ETCD=1 prove --timer -Itest-nginx/lib -I./ -r ${TEST_FILE_SUB_DIR} | tee /tmp/test.result
+    FLUSH_ETCD=1 TEST_EVENTS_MODULE=lua-resty-worker-events prove --timer -Itest-nginx/lib -I./ -r ${TEST_FILE_SUB_DIR} | tee /tmp/test.result
     rerun_flaky_tests /tmp/test.result
 }
 

ci/linux-install-openresty.sh

@@ -80,6 +80,11 @@ if [ "$OPENRESTY_VERSION" == "source" ]; then
     fi
 fi
 
-wget "https://raw.githubusercontent.com/api7/apisix-build-tools/apisix-runtime/${APISIX_RUNTIME}/build-apisix-runtime.sh"
+# TODO: disabled temporarily, waiting for APISIX 3.8 to be released to synchronize the apisix-runtime version
+#wget "https://raw.githubusercontent.com/api7/apisix-build-tools/apisix-runtime/${APISIX_RUNTIME}/build-apisix-runtime.sh"
+wget "https://raw.githubusercontent.com/api7/apisix-build-tools/master/build-apisix-runtime.sh"
 chmod +x build-apisix-runtime.sh
 ./build-apisix-runtime.sh latest
+
+# patch lua-resty-events
+sudo sed -i 's/log(ERR, "event worker failed: ", perr)/log(ngx.WARN, "event worker failed: ", perr)/' /usr/local/openresty/lualib/resty/events/worker.lua

ci/linux_openresty_common_runner.sh

@@ -75,7 +75,7 @@ script() {
     start_grpc_server_example
 
     # APISIX_ENABLE_LUACOV=1 PERL5LIB=.:$PERL5LIB prove -Itest-nginx/lib -r t
-    FLUSH_ETCD=1 prove --timer -Itest-nginx/lib -I./ -r $TEST_FILE_SUB_DIR | tee /tmp/test.result
+    FLUSH_ETCD=1 TEST_EVENTS_MODULE=$TEST_EVENTS_MODULE prove --timer -Itest-nginx/lib -I./ -r $TEST_FILE_SUB_DIR | tee /tmp/test.result
     rerun_flaky_tests /tmp/test.result
 }
 

ci/redhat-ci.sh

@@ -37,12 +37,18 @@ install_dependencies() {
     yum install -y openresty-openssl111 openresty-openssl111-devel pcre pcre pcre-devel xz
     yum -y install https://repos.apiseven.com/packages/centos/apache-apisix-repo-1.0-1.noarch.rpm
 
-    wget "https://raw.githubusercontent.com/api7/apisix-build-tools/apisix-runtime/${APISIX_RUNTIME}/build-apisix-runtime-debug-centos7.sh"
-    wget "https://raw.githubusercontent.com/api7/apisix-build-tools/apisix-runtime/${APISIX_RUNTIME}/build-apisix-runtime.sh"
+    # TODO: disabled temporarily, waiting for APISIX 3.8 to be released to synchronize the apisix-runtime version
+    #wget "https://raw.githubusercontent.com/api7/apisix-build-tools/apisix-runtime/${APISIX_RUNTIME}/build-apisix-runtime-debug-centos7.sh"
+    #wget "https://raw.githubusercontent.com/api7/apisix-build-tools/apisix-runtime/${APISIX_RUNTIME}/build-apisix-runtime.sh"
+    wget "https://raw.githubusercontent.com/api7/apisix-build-tools/master/build-apisix-runtime-debug-centos7.sh"
+    wget "https://raw.githubusercontent.com/api7/apisix-build-tools/master/build-apisix-runtime.sh"
     chmod +x build-apisix-runtime.sh
     chmod +x build-apisix-runtime-debug-centos7.sh
     ./build-apisix-runtime-debug-centos7.sh
 
+    # patch lua-resty-events
+    sed -i 's/log(ERR, "event worker failed: ", perr)/log(ngx.WARN, "event worker failed: ", perr)/' /usr/local/openresty/lualib/resty/events/worker.lua
+
     # install luarocks
     ./utils/linux-install-luarocks.sh
 
@@ -94,7 +100,7 @@ run_case() {
     make init
     set_coredns
     # run test cases
-    FLUSH_ETCD=1 prove --timer -Itest-nginx/lib -I./ -r ${TEST_FILE_SUB_DIR} | tee /tmp/test.result
+    FLUSH_ETCD=1 TEST_EVENTS_MODULE=lua-resty-worker-events prove --timer -Itest-nginx/lib -I./ -r ${TEST_FILE_SUB_DIR} | tee /tmp/test.result
     rerun_flaky_tests /tmp/test.result
 }
 

conf/config-default.yaml

@@ -139,6 +139,10 @@ apisix:
                                   # with the new key. Removing the old keys directly can render the data
                                   # unrecoverable.
 
+  events:                             # Event distribution module configuration
+    module: lua-resty-events          # Sets the name of the events module used.
+                                      # Supported module: lua-resty-worker-events and lua-resty-events
+
 nginx_config:                     # Config for render the template to generate nginx.conf
   # user: root                    # Set the execution user of the worker process. This is only
                                   # effective if the master process runs with super-user privileges.

t/APISIX.pm

@@ -413,6 +413,11 @@ _EOC_
 
         require "resty.core"
 
+        local events_sock_path = "$apisix_home/t/servroot/logs/stream_worker_events.sock"
+        if os.getenv("TEST_NGINX_USE_HUP") ~= "1" then
+            pcall(os.remove, events_sock_path)
+        end
+
         $stream_extra_init_by_lua_start
 
         apisix = require("apisix")
@@ -437,6 +442,14 @@ _EOC_
 
     $extra_stream_config
 
+    server {
+        listen unix:$apisix_home/t/servroot/logs/stream_worker_events.sock;
+        access_log off;
+        content_by_lua_block {
+            require("resty.events.compat").run()
+        }
+    }
+
     # fake server, only for test
     server {
         listen 1995;
@@ -516,6 +529,11 @@ _EOC_
 
     require "resty.core"
 
+    local events_sock_path = "$apisix_home/t/servroot/logs/worker_events.sock"
+    if os.getenv("TEST_NGINX_USE_HUP") ~= "1" then
+        pcall(os.remove, events_sock_path)
+    end
+
     $extra_init_by_lua_start
 
     apisix = require("apisix")
@@ -687,6 +705,18 @@ _EOC_
         }
     }
 
+_EOC_
+
+    $http_config .= <<_EOC_;
+    server {
+        listen unix:$apisix_home/t/servroot/logs/worker_events.sock;
+        access_log off;
+        location / {
+            content_by_lua_block {
+                require("resty.events.compat").run()
+            }
+        }
+    }
 _EOC_
 
     $block->set_value("http_config", $http_config);

t/cli/test_cmd.sh

@@ -181,7 +181,7 @@ fi
 
 bin/apisix quit
 
-sleep 0.5
+sleep 2
 
 if ps -ef | grep "worker process is shutting down" | grep -v "grep"; then
     echo "all workers should exited"
@@ -200,7 +200,7 @@ fi
 
 bin/apisix reload
 
-sleep 0.5
+sleep 3
 
 if ps -ef | grep "worker process is shutting down" | grep -v "grep"; then
     echo "old workers should exited"

t/discovery/consul_dump.t

@@ -471,6 +471,7 @@ location /v1/agent {
 --- response_body eval
 --- error_code eval
 [200, 200, 200, 200, 200, 200]
+--- wait: 2