test(plugin): test for fix of otel reconfigure issue #KAG-1061

CHANGELOG.md

@@ -122,6 +122,8 @@
   [#10327](https://github.com/Kong/kong/pull/10327)
 - **OAuth2**: fix an issue that OAuth2 token was being cached to nil while access to the wrong service first.
   [#10522](https://github.com/Kong/kong/pull/10522)
+- **OpenTelemetry**: fix an issue that reconfigure of OpenTelemetry does not take effect.
+  [#10172](https://github.com/Kong/kong/pull/10172)
 
 #### PDK
 

spec/03-plugins/37-opentelemetry/06-regression_spec.lua

@@ -0,0 +1,112 @@
+local helpers = require "spec.helpers"
+
+for _, strategy in helpers.each_strategy() do
+  describe("opentelemetry regression #" .. strategy, function()
+    local bp
+    setup(function()
+      bp = assert(helpers.get_db_utils(strategy, {
+        "services",
+        "routes",
+        "plugins",
+      }, { "opentelemetry" }))
+    end)
+
+    describe("#KAG-1061", function ()
+      if strategy == "off" then
+        return -- not relevant
+      end
+
+      local mock1, mock2
+      local mock_port1, mock_port2
+      setup(function()  
+        mock_port1 = helpers.get_available_port()
+        mock_port2 = helpers.get_available_port()
+
+        local http_srv = assert(bp.services:insert {
+          name = "mock-service",
+          host = helpers.mock_upstream_host,
+          port = helpers.mock_upstream_port,
+        })
+
+        local route = assert(bp.routes:insert({ service = http_srv,
+                                                protocols = { "http" },
+                                                paths = { "/" }}))
+        bp.plugins:insert({
+          name = "opentelemetry",
+          instance_name = "test1",
+          route = route,
+          service = http_srv,
+          config = {
+            endpoint = "http://127.0.0.1:" .. mock_port1,
+            batch_flush_delay = 0, -- report immediately
+          }
+        })
+
+        assert(helpers.start_kong({
+          database = strategy,
+          nginx_conf = "spec/fixtures/custom_nginx.template",
+          plugins = "opentelemetry",
+          tracing_instrumentations = "all",
+        }))
+        -- we do not wait too long for the mock to receive the request
+        mock1 = helpers.http_mock(mock_port1, {
+          timeout = 5,
+        })
+        mock2 = helpers.http_mock(mock_port2, {
+          timeout = 5,
+        })
+      end)
+
+      teardown(function()
+        helpers.stop_kong()
+      end)
+
+      it("test", function ()
+        local client = assert(helpers.proxy_client())
+        local res = assert(client:send {
+          method = "GET",
+          path = "/",
+        })
+        assert.res_status(200, res)
+
+        -- sent to mock1
+        assert(mock1())
+
+        local admin_client = assert(helpers.admin_client())
+        local res = assert(admin_client:send {
+          method = "PATCH",
+          path = "/plugins/test1",
+          body = {
+            config = {
+              endpoint = "http://127.0.0.1:" .. mock_port2,
+            }
+          },
+          headers = {
+            ["Content-Type"] = "application/json",
+          }
+        })
+        assert.res_status(200, res)
+
+        -- keep sending requests until the reconfigure takes effect and
+        -- the traces are sent to mock2
+        local done
+        local send_co = coroutine.create(function ()
+          local time = 0
+          while not done and time < 10 do
+            local res = assert(client:send {
+              method = "GET",
+              path = "/",
+            })
+            assert.res_status(200, res)
+            time = time + 1
+          end
+        end)
+
+        coroutine.resume(send_co)
+
+        assert(mock2())
+        done = true
+      end) 
+    end)
+  end)
+end

spec/helpers.lua

@@ -1286,6 +1286,8 @@ local function kill_tcp_server(port)
 end
 
 
+-- If it applies, please use `http_mock`, the coroutine variant of `http_server`, which is
+-- more determinative and less flaky.
 --- Starts a local HTTP server.
 -- Accepts a single connection and then closes. Sends a 200 ok, 'Connection:
 -- close' response.
@@ -1345,6 +1347,183 @@ local function http_server(port, opts)
 end
 
 
+local code_status = {
+  [200] = "OK",
+  [201] = "Created",
+  [202] = "Accepted",
+  [203] = "Non-Authoritative Information",
+  [204] = "No Content",
+  [205] = "Reset Content",
+  [206] = "Partial Content",
+  [207] = "Multi-Status",
+  [300] = "Multiple Choices",
+  [301] = "Moved Permanently",
+  [302] = "Found",
+  [303] = "See Other",
+  [304] = "Not Modified",
+  [305] = "Use Proxy",
+  [307] = "Temporary Redirect",
+  [308] = "Permanent Redirect",
+  [400] = "Bad Request",
+  [401] = "Unauthorized",
+  [402] = "Payment Required",
+  [403] = "Forbidden",
+  [404] = "Not Found",
+  [405] = "Method Not Allowed",
+  [406] = "Not Acceptable",
+  [407] = "Proxy Authentication Required",
+  [408] = "Request Timeout",
+  [409] = "Conflict",
+  [410] = "Gone",
+  [411] = "Length Required",
+  [412] = "Precondition Failed",
+  [413] = "Payload Too Large",
+  [414] = "URI Too Long",
+  [415] = "Unsupported Media Type",
+  [416] = "Range Not Satisfiable",
+  [417] = "Expectation Failed",
+  [418] = "I'm a teapot",
+  [422] = "Unprocessable Entity",
+  [423] = "Locked",
+  [424] = "Failed Dependency",
+  [426] = "Upgrade Required",
+  [428] = "Precondition Required",
+  [429] = "Too Many Requests",
+  [431] = "Request Header Fields Too Large",
+  [451] = "Unavailable For Legal Reasons",
+  [500] = "Internal Server Error",
+  [501] = "Not Implemented",
+  [502] = "Bad Gateway",
+  [503] = "Service Unavailable",
+  [504] = "Gateway Timeout",
+  [505] = "HTTP Version Not Supported",
+  [506] = "Variant Also Negotiates",
+  [507] = "Insufficient Storage",
+  [508] = "Loop Detected",
+  [510] = "Not Extended",
+  [511] = "Network Authentication Required",
+}
+
+
+local EMPTY = {}
+
+
+local function handle_response(code, body, headers)
+  if not code then
+    code = 500
+    body = ""
+    headers = EMPTY
+  end
+
+  local head_str = ""
+
+  for k, v in pairs(headers or EMPTY) do
+    head_str = head_str .. k .. ": " .. v .. "\r\n"
+  end
+
+  return code .. " " .. code_status[code] .. " HTTP/1.1" .. "\r\n" ..
+          "Content-Length: " .. #body .. "\r\n" ..
+          "Connection: close\r\n" ..
+          head_str ..
+          "\r\n" ..
+          body
+end
+
+
+local function handle_request(client, response)
+  local lines = {}
+  local headers = {}
+  local line, err
+
+  local content_length
+  repeat
+    line, err = client:receive("*l")
+    if err then
+      return nil, err
+    else
+      local k, v = line:match("^([^:]+):%s*(.+)$")
+      if k then
+        headers[k] = v
+        if k:lower() == "content-length" then
+          content_length = tonumber(v)
+        end
+      end
+      table.insert(lines, line)
+    end
+  until line == ""
+
+  local method = lines[1]:match("^(%S+)%s+(%S+)%s+(%S+)$")
+  local method_lower = method:lower()
+
+  local body
+  if content_length then
+    body = client:receive(content_length)
+
+  elseif method_lower == "put" or method_lower == "post" then
+    body = client:receive("*a")
+  end
+
+  local response_str
+  local meta = getmetatable(response)
+  if type(response) == "function" or (meta and meta.__call) then
+    response_str = response(lines, body, headers)
+
+  elseif type(response) == "table" and response.code then
+    response_str = handle_response(response.code, response.body, response.headers)
+
+  elseif type(response) == "table" and response[1] then
+    response_str = handle_response(response[1], response[2], response[3])
+
+  elseif type(response) == "string" then
+    response_str = response
+
+  elseif response == nil then
+    response_str = "HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n"
+  end
+
+
+  client:send(response_str)
+  return lines, body, headers
+end
+
+
+--- Start a local HTTP server with coroutine.
+-- local mock = helpers.http_mock(1234, { timeout = 0.1 })
+-- wait for a request, and respond with the custom response
+-- the request is returned as the function's return values
+-- return nil, err if error
+-- local lines, body, headers = mock(custom_response)
+-- local lines, body, headers = mock()
+-- mock("closing", true) -- close the server
+local function http_mock(port, opts)
+  local socket = require "socket"
+  local server = assert(socket.tcp())
+  server:settimeout(opts and opts.timeout or 60)
+  assert(server:setoption('reuseaddr', true))
+  assert(server:bind("*", port))
+  assert(server:listen())
+  return coroutine.wrap(function(response, exit)
+    local lines, body, headers
+    -- start listening
+    while not exit do
+      local client, err = server:accept()
+      if err then
+        lines, body = false, err
+
+      else
+        lines, body, headers = handle_request(client, response)
+        client:close()
+      end
+
+      response, exit = coroutine.yield(lines, body, headers)
+    end
+
+    server:close()
+    return true
+  end)
+end
+
+
 --- Stops a local HTTP server.
 -- A server previously created with `http_server` can be stopped prematurely by
 -- calling this function.
@@ -3657,6 +3836,7 @@ end
   udp_server = udp_server,
   kill_tcp_server = kill_tcp_server,
   http_server = http_server,
+  http_mock = http_mock,
   kill_http_server = kill_http_server,
   get_proxy_ip = get_proxy_ip,
   get_proxy_port = get_proxy_port,