feat(discovery): add kubernetes discovery data warm-up in master process

apisix/discovery/init.lua

@@ -37,6 +37,16 @@ function discovery.init_worker()
     end
 end
 
+function discovery.init()
+    if discovery_type then
+        for discovery_name, _ in pairs(discovery_type) do
+            if discovery[discovery_name].init then
+                discovery[discovery_name].init()
+            end
+        end
+    end
+end
+
 return {
     version = 0.1,
     discovery = discovery

apisix/discovery/kubernetes/informer_factory.lua

@@ -258,7 +258,7 @@ local function watch(httpc, apiserver, informer)
 end
 
 
-local function list_watch(informer, apiserver)
+local function list_watch(informer, apiserver, skip_watch)
     local ok
     local reason, message
     local httpc = http.new()
@@ -301,7 +301,9 @@ local function list_watch(informer, apiserver)
     if informer.post_list then
         informer:post_list()
     end
-
+    if skip_watch then
+        return true
+    end
     core.log.info("begin to watch ", informer.kind)
     informer.fetch_state = "watching"
     ok, reason, message = watch(httpc, apiserver, informer)

apisix/discovery/kubernetes/init.lua

@@ -506,7 +506,18 @@ local _M = {
 }
 
 
-local function start_fetch(handle)
+local function start_fetch(handle, immediate)
+    if immediate then
+        -- Execute immediately and synchronously without using timer
+        core.log.info(process.type(), " process immediately pulls the k8s endpoints list")
+        local ok, status = pcall(handle.list_watch, handle, handle.apiserver, true)
+        if not ok then
+            core.log.error("list_watch failed, kind: ", handle.kind,
+                           ", reason: RuntimeException, message: ", status)
+        end
+        return
+    end
+
     local timer_runner
     timer_runner = function(premature)
         if premature then
@@ -558,7 +569,7 @@ local function single_mode_init(conf)
                 "please check your APISIX version")
     end
 
-    if process.type() ~= "privileged agent" then
+    if process.type() == "worker" then
         ctx = endpoint_dict
         return
     end
@@ -605,7 +616,7 @@ local function single_mode_init(conf)
         default_weight = default_weight
     }, { __index = endpoints_informer })
 
-    start_fetch(ctx)
+    start_fetch(ctx, process.type() == "master")
 end
 
 
@@ -653,7 +664,7 @@ end
 local function multiple_mode_init(confs)
     ctx = core.table.new(#confs, 0)
 
-    if process.type() ~= "privileged agent" then
+    if process.type() == "worker" then
         multiple_mode_worker_init(confs)
         return
     end
@@ -716,7 +727,7 @@ local function multiple_mode_init(confs)
     end
 
     for _, item in pairs(ctx) do
-        start_fetch(item)
+        start_fetch(item, process.type() == "master")
     end
 end
 
@@ -762,6 +773,9 @@ function _M.init_worker()
 end
 
 
+_M.init = _M.init_worker
+
+
 local function dump_endpoints_from_dict(endpoint_dict)
     local keys, err = endpoint_dict:get_keys(0)
     if err then

apisix/init.lua

@@ -106,6 +106,11 @@ function _M.http_init(args)
         end
     end
 
+    local discovery = require("apisix.discovery.init").discovery
+    if discovery and discovery.init then
+        discovery.init()
+    end
+
     xrpc.init()
 end
 

t/kubernetes/discovery/kubernetes.t

@@ -39,8 +39,8 @@ _EOC_
 env MyPort=6443;
 env KUBERNETES_SERVICE_HOST=127.0.0.1;
 env KUBERNETES_SERVICE_PORT=6443;
-env KUBERNETES_CLIENT_TOKEN=$::token_value;
 env KUBERNETES_CLIENT_TOKEN_FILE=$::token_file;
+env KUBERNETES_CLIENT_TOKEN=$::token_value;
 _EOC_
 
     $block->set_value("main_config", $main_config);
@@ -200,7 +200,7 @@ deployment:
 discovery:
   kubernetes:
     service:
-        host: "sample.com"
+        host: "localhost"
     shared_size: "2m"
     client:
         token: ${KUBERNETES_CLIENT_TOKEN}
@@ -209,7 +209,7 @@ GET /compare
 {
   "service": {
     "schema": "https",
-    "host": "sample.com",
+    "host": "localhost",
     "port": "${KUBERNETES_SERVICE_PORT}"
   },
   "client": {
@@ -276,14 +276,14 @@ discovery:
   kubernetes:
   - id: "debug"
     service:
-        host: "1.cluster.com"
+        schema: "http"
+        host: "127.0.0.1"
         port: "6445"
     client:
         token: ${KUBERNETES_CLIENT_TOKEN}
   - id: "release"
     service:
-        schema: "http"
-        host: "2.cluster.com"
+        host: "127.0.0.1"
         port: "${MyPort}"
     client:
         token: ${KUBERNETES_CLIENT_TOKEN}
@@ -295,8 +295,8 @@ GET /compare
   {
     "id": "debug",
     "service": {
-      "schema": "https",
-      "host": "1.cluster.com",
+      "schema": "http",
+      "host": "127.0.0.1",
       "port": "6445"
     },
     "client": {
@@ -309,8 +309,8 @@ GET /compare
   {
     "id": "release",
     "service": {
-      "schema": "http",
-      "host": "2.cluster.com",
+      "schema": "https",
+      "host": "127.0.0.1",
       "port": "${MyPort}"
     },
     "client": {
@@ -385,6 +385,8 @@ qr/re-read the token value/
 --- grep_error_log_out
 re-read the token value
 re-read the token value
+re-read the token value
+re-read the token value
 
 
 
@@ -421,3 +423,34 @@ GET /compare
 Content-type: application/json
 --- response_body
 true
+
+
+
+=== TEST 9: test master process immediately pulls k8s endpoints list (warm-up)
+--- yaml_config
+apisix:
+  node_listen: 1984
+  config_center: yaml
+deployment:
+  role: data_plane
+  role_data_plane:
+    config_provider: yaml
+discovery:
+  kubernetes:
+    client:
+        token: ${KUBERNETES_CLIENT_TOKEN}
+--- config
+    location /t {
+        content_by_lua_block {
+            ngx.say("pass")
+        }
+    }
+--- log_level: info
+--- request
+GET /t
+--- response_body
+pass
+--- grep_error_log eval
+qr/master process immediately pulls the k8s endpoints list/
+--- grep_error_log_out
+master process immediately pulls the k8s endpoints list

t/kubernetes/discovery/kubernetes2.t

@@ -92,8 +92,8 @@ _EOC_
     my $main_config = $block->main_config // <<_EOC_;
 env KUBERNETES_SERVICE_HOST=127.0.0.1;
 env KUBERNETES_SERVICE_PORT=6443;
-env KUBERNETES_CLIENT_TOKEN=$::token_value;
 env KUBERNETES_CLIENT_TOKEN_FILE=$::token_file;
+env KUBERNETES_CLIENT_TOKEN=$::token_value;
 _EOC_
 
     $block->set_value("main_config", $main_config);

t/kubernetes/discovery/kubernetes3.t

@@ -115,8 +115,8 @@ _EOC_
     my $main_config = $block->main_config // <<_EOC_;
 env KUBERNETES_SERVICE_HOST=127.0.0.1;
 env KUBERNETES_SERVICE_PORT=6443;
-env KUBERNETES_CLIENT_TOKEN=$::token_value;
 env KUBERNETES_CLIENT_TOKEN_FILE=$::token_file;
+env KUBERNETES_CLIENT_TOKEN=$::token_value;
 _EOC_
 
     $block->set_value("main_config", $main_config);
@@ -246,15 +246,6 @@ _EOC_
             content_by_lua_block {
                 local http = require("resty.http")
                 local healthcheck_uri = "http://127.0.0.1:7085" .. "/status/ready"
-                for i = 1, 4 do
-                    local httpc = http.new()
-                    local res, _ = httpc:request_uri(healthcheck_uri, {method = "GET", keepalive = false})
-                    if res.status == 200 then
-                        ngx.status = res.status
-                        return
-                    end
-                    ngx.sleep(1)
-                end
                 local httpc = http.new()
                 local res, _ = httpc:request_uri(healthcheck_uri, {method = "GET", keepalive = false})
                 ngx.status = res.status
@@ -533,15 +524,11 @@ GET /dump
 === TEST 7: test pre_list and post_list work  for single-k8s with endpoint_slices
 --- log_level: info
 --- yaml_config eval: $::single_yaml_config
---- extra_init_by_lua
+--- extra_init_by_lua_start
     local ngx = ngx
-    local core = require("apisix.core")
 
     local dict = ngx.shared["kubernetes"]
-    local ok,err = dict:set("dirty_key", true)
-    if not ok then
-        core.log.error("set dirty_key to dict fail, err: ", err)
-    end
+    assert(dict:set("dirty_key", true))
 --- request
 GET /ready_check
 --- no_error_log
@@ -556,15 +543,11 @@ kubernetes discovery module found dirty data in shared dict, key: dirty_key
 === TEST 8: test pre_list and post_list work for multi-k8s with endpoint_slices
 --- log_level: info
 --- yaml_config eval: $::yaml_config
---- extra_init_by_lua
+--- extra_init_by_lua_start
     local ngx = ngx
-    local core = require("apisix.core")
 
     local dict = ngx.shared["kubernetes-first"]
-    local ok,err = dict:set("dirty_key", true)
-    if not ok then
-        core.log.error("set dirty_key to dict fail, err: ", err)
-    end
+    assert(dict:set("dirty_key", true))
 --- request
 GET /ready_check
 --- no_error_log
@@ -593,15 +576,11 @@ discovery:
     client:
       token_file: "/tmp/var/run/secrets/kubernetes.io/serviceaccount/token"
     watch_endpoint_slices: false
---- extra_init_by_lua
+--- extra_init_by_lua_start
     local ngx = ngx
-    local core = require("apisix.core")
 
     local dict = ngx.shared["kubernetes"]
-    local ok,err = dict:set("dirty_key", true)
-    if not ok then
-        core.log.error("set dirty_key to dict fail, err: ", err)
-    end
+    assert(dict:set("dirty_key", true))
 --- request
 GET /ready_check
 --- no_error_log
@@ -639,15 +618,11 @@ discovery:
       client:
         token_file: "/tmp/var/run/secrets/kubernetes.io/serviceaccount/token"
       watch_endpoint_slices: false
---- extra_init_by_lua
+--- extra_init_by_lua_start
     local ngx = ngx
-    local core = require("apisix.core")
 
     local dict = ngx.shared["kubernetes-first"]
-    local ok,err = dict:set("dirty_key", true)
-    if not ok then
-        core.log.error("set dirty_key to dict fail, err: ", err)
-    end
+    assert(dict:set("dirty_key", true))
 --- request
 GET /ready_check
 --- no_error_log
@@ -688,10 +663,8 @@ discovery:
 --- request
 GET /ready_check
 --- error_code: 503
---- grep_error_log eval
+--- error_log eval
 qr/connect apiserver failed/
---- grep_error_log_out
-connect apiserver failed
 
 
 

t/kubernetes/discovery/kubernetes4.t

@@ -243,8 +243,8 @@ _EOC_
     my $main_config = $block->main_config // <<_EOC_;
 env KUBERNETES_SERVICE_HOST=127.0.0.1;
 env KUBERNETES_SERVICE_PORT=6443;
-env KUBERNETES_CLIENT_TOKEN=$::token_value;
 env KUBERNETES_CLIENT_TOKEN_FILE=$::token_file;
+env KUBERNETES_CLIENT_TOKEN=$::token_value;
 _EOC_
 
     $block->set_value("main_config", $main_config);