feat(tracing): route/service scoping

CHANGELOG.md

@@ -96,6 +96,8 @@
   parameter that allows the selection of the IMDS protocol version.
   Defaults to `v1`, can be set to `v2` to enable IMDSv2.
   [#9962](https://github.com/Kong/kong/pull/9962)
+- **OpenTelemetry**: Support scoping with services, routes and consumers.
+  [#10096](https://github.com/Kong/kong/pull/10096)
 
 ### Fixes
 

kong/plugins/opentelemetry/handler.lua

@@ -113,7 +113,7 @@ local function process_span(span, queue)
   queue:add(pb_span)
 end
 
-function OpenTelemetryHandler:rewrite()
+function OpenTelemetryHandler:access()
   local headers = ngx_get_headers()
   local root_span = ngx.ctx.KONG_SPANS and ngx.ctx.KONG_SPANS[1]
 

kong/plugins/opentelemetry/schema.lua

@@ -30,10 +30,6 @@ local resource_attributes = Schema.define {
 return {
   name = "opentelemetry",
   fields = {
-    -- global plugin only
-    { consumer = typedefs.no_consumer },
-    { service = typedefs.no_service },
-    { route = typedefs.no_route },
     { protocols = typedefs.protocols_http }, -- TODO: support stream mode
     { config = {
       type = "record",

spec/03-plugins/37-opentelemetry/04-exporter_spec.lua

@@ -45,24 +45,46 @@ for _, strategy in helpers.each_strategy() do
     end)
 
     -- helpers
-    local function setup_instrumentations(types, config, fixtures)
+    local function setup_instrumentations(types, config, fixtures, router_scoped, service_scoped, another_global)
       local http_srv = assert(bp.services:insert {
         name = "mock-service",
         host = helpers.mock_upstream_host,
         port = helpers.mock_upstream_port,
       })
 
-      bp.routes:insert({ service = http_srv,
-                         protocols = { "http" },
-                         paths = { "/" }})
+      local http_srv2 = assert(bp.services:insert {
+        name = "mock-service2",
+        host = helpers.mock_upstream_host,
+        port = helpers.mock_upstream_port,
+      })
 
-      bp.plugins:insert({
+      local route = assert(bp.routes:insert({ service = http_srv,
+                                              protocols = { "http" },
+                                              paths = { "/" }}))
+
+      assert(bp.routes:insert({ service = http_srv2,
+                                protocols = { "http" },
+                                paths = { "/no_plugin" }}))
+
+      assert(bp.plugins:insert({
         name = "opentelemetry",
+        route = router_scoped and route,
+        service = service_scoped and http_srv,
         config = table_merge({
           endpoint = "http://127.0.0.1:" .. HTTP_SERVER_PORT,
           batch_flush_delay = 0, -- report immediately
         }, config)
-      })
+      }))
+
+      if another_global then
+        assert(bp.plugins:insert({
+          name = "opentelemetry",
+          config = table_merge({
+            endpoint = "http://127.0.0.1:" .. HTTP_SERVER_PORT,
+            batch_flush_delay = 0, -- report immediately
+          }, config)
+        }))
+      end
 
       assert(helpers.start_kong({
         proxy_listen = "0.0.0.0:" .. PROXY_PORT,
@@ -140,6 +162,62 @@ for _, strategy in helpers.each_strategy() do
       end)
     end)
 
+    for _, case in ipairs{
+      {true, true, true},
+      {true, true, nil},
+      {true, nil, true},
+      {true, nil, nil},
+      {nil, true, true},
+      {nil, true, nil},
+    } do
+      describe("#scoping for" .. (case[1] and " route" or "")
+                              .. (case[2] and " service" or "")
+                              .. (case[3] and " with global" or "")
+      , function ()
+        lazy_setup(function()
+          bp, _ = assert(helpers.get_db_utils(strategy, {
+            "services",
+            "routes",
+            "plugins",
+          }, { "opentelemetry" }))
+
+          setup_instrumentations("all", {
+            headers = {
+              ["X-Access-Token"] = "token",
+            },
+          }, nil, case[1], case[2], case[3])
+        end)
+
+        lazy_teardown(function()
+          helpers.stop_kong()
+          helpers.kill_http_server(HTTP_SERVER_PORT)
+        end)
+
+        it("works", function ()
+          local thread = helpers.http_server(HTTP_SERVER_PORT, { timeout = 10 })
+          local cli = helpers.proxy_client(7000, PROXY_PORT)
+          local r = assert(cli:send {
+            method  = "GET",
+            path    = "/no_plugin",
+          })
+          assert.res_status(200, r)
+
+          -- close client connection
+          cli:close()
+
+          local ok, err = thread:join()
+
+          -- we should have no telemetry reported
+          if case[3] then
+            assert(ok, err)
+
+          else
+            assert.is_falsy(ok)
+            assert.matches("timeout", err)
+          end
+        end)
+      end)
+    end
     describe("overwrite resource attributes #http", function ()
       lazy_setup(function()
         bp, _ = assert(helpers.get_db_utils(strategy, {
@@ -284,7 +362,6 @@ for _, strategy in helpers.each_strategy() do
           local body = fd:read("*a")
           pb_set = ngx_re.split(body, "\n")
 
-          print("pb set length: ", #pb_set)
           local count = 0
           for _, pb_data in ipairs(pb_set) do
             local decoded = assert(pb.decode("opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest", ngx.decode_base64(pb_data)))