fix(hybrid) allow graceful exit of cp and dp nodes (#6306)

### Summary

Starting control plane with:
```
KONG_ROLE=control_plane \
KONG_CLUSTER_CERT=cluster.crt \
KONG_CLUSTER_CERT_KEY=cluster.key \
./bin/kong start -p cp
```

Then connecting data plane to it with:
```
KONG_ROLE=data_plane \
KONG_CLUSTER_CONTROL_PLANE=cp.test:8005 \
KONG_CLUSTER_CERT=cluster.crt \
KONG_CLUSTER_CERT_KEY=cluster.key \
KONG_LUA_SSL_TRUSTED_CERTIFICATE=cluster.crt \
KONG_DATABASE=off \
./bin/kong start -p dp
```

And then gracefully stopping control plane with:

```
KONG_ROLE=control_plane \
KONG_CLUSTER_CERT=cluster.crt \
KONG_CLUSTER_CERT_KEY=cluster.key \
./bin/kong quit -p cp
```

Will result (after 10 secs):

```
Timeout, Kong stopped forcefully
```

This PR is to fix that so that what happens is this instead (after around 1 sec):
```
Kong stopped (gracefully)
```

kong/clustering.lua

@@ -35,6 +35,7 @@ local deflate_gzip = utils.deflate_gzip
 local MAX_PAYLOAD = 4 * 1024 * 1024 -- 4MB
 local PING_INTERVAL = 30 -- 30 seconds
 local WS_OPTS = {
+  timeout = 5000,
   max_payload_len = MAX_PAYLOAD,
 }
 local ngx_ERR = ngx.ERR
@@ -163,7 +164,12 @@ local function communicate(premature, conf)
         return
       end
 
-      ngx_sleep(PING_INTERVAL)
+      for _ = 1, PING_INTERVAL do
+        ngx_sleep(1)
+        if exiting() then
+          return
+        end
+      end
     end
   end)
 
@@ -326,7 +332,10 @@ function _M.handle_cp_websocket()
   end)
 
   while not exiting() do
-    local ok, err = sem:wait(10)
+    local ok, err = sem:wait(5)
+    if exiting() then
+      return
+    end
     if ok then
       local payload = table_remove(queue, 1)
       assert(payload, "config queue can not be empty after semaphore returns")
@@ -351,10 +360,8 @@ function _M.handle_cp_websocket()
         end
       end
 
-    else -- not ok
-      if err ~= "timeout" then
-        ngx_log(ngx_ERR, "semaphore wait error: ", err)
-      end
+    elseif err ~= "timeout" then
+      ngx_log(ngx_ERR, "semaphore wait error: ", err)
     end
   end
 end
@@ -394,11 +401,28 @@ local function push_config_timer(premature, semaphore, delay)
   end
 
   while not exiting() do
-    local ok, err = semaphore:wait(10)
+    local ok, err = semaphore:wait(1)
+    if exiting() then
+      return
+    end
     if ok then
       ok, err = pcall(push_config)
       if ok then
-        ngx.sleep(delay)
+        local sleep_left = delay
+        while sleep_left > 0 do
+          if sleep_left <= 1 then
+            ngx.sleep(sleep_left)
+            break
+          end
+
+          ngx.sleep(1)
+
+          if exiting() then
+            return
+          end
+
+          sleep_left = sleep_left - 1
+        end
 
       else
         ngx_log(ngx_ERR, "export and pushing config failed: ", err)

spec/02-integration/02-cmd/12-hybrid_spec.lua

@@ -101,3 +101,55 @@ describe("kong hybrid", function()
     end)
   end)
 end)
+
+
+for _, strategy in helpers.each_strategy() do
+  if strategy ~= "off" then
+    describe("kong hybrid with #" .. strategy .. " backend", function()
+      lazy_setup(function()
+        helpers.get_db_utils(strategy, {
+        }) -- runs migrations
+
+        assert(helpers.start_kong({
+          role = "control_plane",
+          cluster_cert = "spec/fixtures/kong_clustering.crt",
+          cluster_cert_key = "spec/fixtures/kong_clustering.key",
+          lua_ssl_trusted_certificate = "spec/fixtures/kong_clustering.crt",
+          database = strategy,
+          prefix = "servroot",
+          cluster_listen = "127.0.0.1:9005",
+          nginx_conf = "spec/fixtures/custom_nginx.template",
+        }))
+
+        assert(helpers.start_kong({
+          role = "data_plane",
+          database = "off",
+          prefix = "servroot2",
+          cluster_cert = "spec/fixtures/kong_clustering.crt",
+          cluster_cert_key = "spec/fixtures/kong_clustering.key",
+          lua_ssl_trusted_certificate = "spec/fixtures/kong_clustering.crt",
+          cluster_control_plane = "127.0.0.1:9005",
+          proxy_listen = "0.0.0.0:9002",
+        }))
+      end)
+
+      lazy_teardown(function()
+        helpers.kill_all()
+      end)
+
+      it("quits gracefully", function()
+        local ok, err, msg = helpers.kong_exec("quit --prefix servroot")
+        assert.equal("", err)
+        assert.equal("Kong stopped (gracefully)\n", msg)
+        assert.equal(true, ok)
+
+        ok, err, msg = helpers.kong_exec("quit --prefix servroot2", {
+          DATABASE="off"
+        })
+        assert.equal("", err)
+        assert.equal("Kong stopped (gracefully)\n", msg)
+        assert.equal(true, ok)
+      end)
+    end)
+  end
+end