feat: support reading configuration form xds(mvp) (#6614)

.github/workflows/build.yml

@@ -29,7 +29,7 @@ jobs:
         test_dir:
           - t/plugin
           - t/admin t/cli t/config-center-yaml t/control t/core t/debug t/discovery t/error_page t/misc
-          - t/node t/router t/script t/stream-node t/utils t/wasm
+          - t/node t/router t/script t/stream-node t/utils t/wasm t/xds-library
 
     runs-on: ${{ matrix.platform }}
     timeout-minutes: 90
@@ -90,6 +90,11 @@ jobs:
           sudo dpkg -i tinygo_${TINYGO_VER}_amd64.deb
           cd t/wasm && find . -type f -name "*.go" | xargs -Ip tinygo build -o p.wasm -scheduler=none -target=wasi p
 
+      - name: Build xDS library
+        run: |
+          cd t/xds-library
+          go build -o libxds.so -buildmode=c-shared main.go
+
       - name: Linux Before install
         run: sudo ./ci/${{ matrix.os_name }}_runner.sh before_install
 

apisix/cli/ngx_tpl.lua

@@ -243,6 +243,10 @@ http {
     lua_shared_dict ext-plugin {* http.lua_shared_dict["ext-plugin"] *}; # cache for ext-plugin
     {% end %}
 
+    {% if config_center == "xds" then %}
+    lua_shared_dict xds-route-config  10m;
+    {% end %}
+
     # for custom shared dict
     {% if http.custom_lua_shared_dict then %}
     {% for cache_key, cache_size in pairs(http.custom_lua_shared_dict) do %}

apisix/cli/schema.lua

@@ -28,7 +28,7 @@ local config_schema = {
         apisix = {
             properties = {
                 config_center = {
-                    enum = {"etcd", "yaml"},
+                    enum = {"etcd", "yaml", "xds"},
                 },
                 lua_module_hook = {
                     pattern = "^[a-zA-Z._-]+$",

apisix/core/config_xds.lua

@@ -0,0 +1,120 @@
+--
+-- Licensed to the Apache Software Foundation (ASF) under one or more
+-- contributor license agreements.  See the NOTICE file distributed with
+-- this work for additional information regarding copyright ownership.
+-- The ASF licenses this file to You under the Apache License, Version 2.0
+-- (the "License"); you may not use this file except in compliance with
+-- the License.  You may obtain a copy of the License at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+--
+
+--- Get configuration form ngx.shared.DICT.
+--
+-- @module core.config_xds
+
+local base              = require("resty.core.base")
+local config_local      = require("apisix.core.config_local")
+local table             = table
+local error             = error
+local is_http           = ngx.config.subsystem == "http"
+local io                = io
+local io_open           = io.open
+local io_close          = io.close
+local package           = package
+local new_tab           = base.new_tab
+local ffi               = require ("ffi")
+local C                 = ffi.C
+local route_config     = ngx.shared["xds-route-config"]
+local ngx_re_match      = ngx.re.match
+local ngx_re_gmatch     = ngx.re.gmatch
+
+local xds_lib_name = "libxds.so"
+
+
+local process
+if is_http then
+    process = require("ngx.process")
+end
+
+
+ffi.cdef[[
+extern void initial(void* route_zone_ptr);
+]]
+
+
+local _M = {
+    version = 0.1,
+    local_conf = config_local.local_conf,
+}
+
+
+-- todo: refactor this function in chash.lua and radixtree.lua
+local function load_shared_lib(lib_name)
+    local cpath = package.cpath
+    local tried_paths = new_tab(32, 0)
+    local i = 1
+
+    local iter, err = ngx_re_gmatch(cpath, "[^;]+", "jo")
+    if not iter then
+        error("failed to gmatch: " .. err)
+    end
+
+    while true do
+        local it = iter()
+        local fpath
+        fpath, err = ngx_re_match(it[0], "(.*/)",  "jo")
+        if err then
+            error("failed to match: " .. err)
+        end
+        local spath = fpath[0] .. lib_name
+
+        local f = io_open(spath)
+        if f ~= nil then
+            io_close(f)
+            return ffi.load(spath)
+        end
+        tried_paths[i] = spath
+        i = i + 1
+
+        if not it then
+            break
+        end
+    end
+
+    return nil, tried_paths
+end
+
+
+local function load_libxds(lib_name)
+    local xdsagent, tried_paths = load_shared_lib(lib_name)
+
+    if not xdsagent then
+        tried_paths[#tried_paths + 1] = 'tried above paths but can not load ' .. lib_name
+        error("can not load xds library, tried paths: " ..
+              table.concat(tried_paths, '\r\n', 1, #tried_paths))
+    end
+
+    local route_zone = C.ngx_http_lua_ffi_shdict_udata_to_zone(route_config[1])
+    local route_shd_cdata = ffi.cast("void*", route_zone)
+    xdsagent.initial(route_shd_cdata)
+end
+
+
+
+function _M.init_worker()
+    if process.type() == "privileged agent" then
+        load_libxds(xds_lib_name)
+    end
+
+    return true
+end
+
+
+return _M

apisix/core/config_yaml.lua

@@ -389,6 +389,8 @@ end
 function _M.init_worker()
     -- sync data in each non-master process
     ngx.timer.every(1, read_apisix_yaml)
+
+    return true
 end
 
 

apisix/init.lua

@@ -122,8 +122,12 @@ function _M.http_init_worker()
     plugin_config.init_worker()
     require("apisix.consumer").init_worker()
 
-    if core.config == require("apisix.core.config_yaml") then
+    if core.config.init_worker then
-        core.config.init_worker()
+        local ok, err = core.config.init_worker()
+        if not ok then
+            core.log.error("failed to init worker process of ", core.config.type,
+                           " config center, err: ", err)
+        end
     end
 
     apisix_upstream.init_worker()

t/APISIX.pm

@@ -239,7 +239,7 @@ apisix:
 _EOC_
     }
 
-    my $lua_deps_path = <<_EOC_;
+    my $lua_deps_path = $block->lua_deps_path // <<_EOC_;
     lua_package_path "$apisix_home/?.lua;$apisix_home/?/init.lua;$apisix_home/deps/share/lua/5.1/?/init.lua;$apisix_home/deps/share/lua/5.1/?.lua;$apisix_home/apisix/?.lua;$apisix_home/t/?.lua;;";
     lua_package_cpath "$apisix_home/?.so;$apisix_home/deps/lib/lua/5.1/?.so;$apisix_home/deps/lib64/lua/5.1/?.so;;";
 _EOC_
@@ -509,6 +509,7 @@ _EOC_
     lua_shared_dict ext-plugin 1m;
     lua_shared_dict kubernetes 1m;
     lua_shared_dict tars 1m;
+    lua_shared_dict xds-route-config 1m;
 
     proxy_ssl_name \$upstream_host;
     proxy_ssl_server_name on;

t/xds-library/config_xds.t

@@ -0,0 +1,105 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+use t::APISIX 'no_plan';
+
+use Cwd qw(cwd);
+my $apisix_home = $ENV{APISIX_HOME} || cwd();
+
+repeat_each(1);
+no_long_string();
+no_root_location();
+log_level("info");
+
+add_block_preprocessor(sub {
+    my ($block) = @_;
+
+    if (!$block->request) {
+        $block->set_value("request", "GET /t");
+    }
+
+    if (!$block->no_error_log) {
+        $block->set_value("no_error_log", "[error]\n[alert]");
+    }
+
+    my $lua_deps_path = $block->lua_deps_path // <<_EOC_;
+        lua_package_path "$apisix_home/?.lua;$apisix_home/?/init.lua;$apisix_home/deps/share/lua/5.1/?/init.lua;$apisix_home/deps/share/lua/5.1/?.lua;$apisix_home/apisix/?.lua;$apisix_home/t/?.lua;;";
+        lua_package_cpath "$apisix_home/?.so;$apisix_home/t/xds-library/?.so;$apisix_home/deps/lib/lua/5.1/?.so;$apisix_home/deps/lib64/lua/5.1/?.so;;";
+_EOC_
+
+    $block->set_value("lua_deps_path", $lua_deps_path);
+
+    my $extra_init_by_lua = <<_EOC_;
+    --
+    local config_xds = require("apisix.core.config_xds")
+
+    local inject = function(mod, name)
+        local old_f = mod[name]
+        mod[name] = function (...)
+            ngx.log(ngx.WARN, "config_xds run ", name)
+            return { true }
+        end
+    end
+
+    inject(config_xds, "new")
+
+_EOC_
+
+    $block->set_value("extra_init_by_lua", $extra_init_by_lua);
+
+    if (!$block->yaml_config) {
+        my $yaml_config = <<_EOC_;
+apisix:
+    node_listen: 1984
+    config_center: xds
+    enable_admin: false
+_EOC_
+
+        $block->set_value("yaml_config", $yaml_config);
+    }
+});
+
+run_tests;
+
+__DATA__
+
+=== TEST 1: load xDS library successfully
+--- config
+    location /t {
+        content_by_lua_block {
+            ngx.say("ok")
+        }
+    }
+--- no_error_log eval
+qr/can not load xDS library/
+
+
+
+=== TEST 2: read data form shdict that wirted by xDS library
+--- config
+    location /t {
+        content_by_lua_block {
+            -- wait for xds library sync data
+            ngx.sleep(1.5)
+            local core = require("apisix.core")
+            local value = ngx.shared["xds-route-config"]:get("/apisix/routes/1")
+            local route_conf, err = core.json.decode(value)
+            local json_encode = require("toolkit.json").encode
+            ngx.say(json_encode(route_conf))
+        }
+    }
+--- response_body
+{"create_time":1646972532,"id":"1","priority":0,"status":1,"update_time":1647250524,"upstream":{"hash_on":"vars","nodes":[{"host":"127.0.0.1","port":80,"priority":0,"weight":1}],"pass_host":"pass","scheme":"http","type":"roundrobin"},"uri":"/hello"}

t/xds-library/main.go

@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+/*
+#cgo LDFLAGS: -shared
+#include <stdlib.h>
+
+extern void ngx_http_lua_ffi_shdict_store(void *zone, int op,
+    const unsigned char *key, size_t key_len,
+	int value_type,
+    const unsigned char *str_value_buf, size_t str_value_len,
+    double num_value, long exptime, int user_flags, char **errmsg,
+    int *forcible);
+*/
+import "C"
+
+import (
+	"fmt"
+	"time"
+	"unsafe"
+)
+
+func main() {
+}
+
+
+//export initial
+func initial(zone unsafe.Pointer) {
+	time.Sleep(time.Second)
+	value := fmt.Sprintf(`{
+"status": 1,
+"update_time": 1647250524,
+"create_time": 1646972532,
+"uri": "/hello",
+"priority": 0,
+"id": "1",
+"upstream": {
+	"nodes": [
+		{
+			"port": 80,
+			"priority": 0,
+			"host": "127.0.0.1",
+			"weight": 1
+		}
+	],
+	"type": "roundrobin",
+	"hash_on": "vars",
+	"pass_host": "pass",
+	"scheme": "http"
+}
+}`)
+
+	write_route(zone, "/apisix/routes/1", value)
+}
+
+func write_route(zone unsafe.Pointer, key, value string) {
+	var keyCStr = C.CString(key)
+	defer C.free(unsafe.Pointer(keyCStr))
+	var keyLen = C.size_t(len(key))
+
+	var valueCStr = C.CString(value)
+	defer C.free(unsafe.Pointer(valueCStr))
+	var valueLen = C.size_t(len(value))
+
+	errMsgBuf := make([]*C.char, 1)
+	var forcible = 0
+
+	C.ngx_http_lua_ffi_shdict_store(zone, 0x0004,
+		(*C.uchar)(unsafe.Pointer(keyCStr)), keyLen,
+		4,
+		(*C.uchar)(unsafe.Pointer(valueCStr)), valueLen,
+		0, 0, 0,
+		(**C.char)(unsafe.Pointer(&errMsgBuf[0])),
+		(*C.int)(unsafe.Pointer(&forcible)),
+	)
+}