feat(o11y/d11y): prepare instrumentation for active tracing

* support array and map type attributes
* support configurable sink for tracer
* refactor OTLP code to be accessible outside OTel

changelog/unreleased/kong/feat-tracing-pdk-attributes.yml

@@ -0,0 +1,3 @@
+message: Array and Map type span attributes are now supported by the tracing PDK
+type: feature
+scope: PDK

kong-3.9.0-0.rockspec

@@ -598,8 +598,6 @@ build = {
     ["kong.plugins.opentelemetry.migrations.001_331_to_332"] = "kong/plugins/opentelemetry/migrations/001_331_to_332.lua",
     ["kong.plugins.opentelemetry.handler"] = "kong/plugins/opentelemetry/handler.lua",
     ["kong.plugins.opentelemetry.schema"]  = "kong/plugins/opentelemetry/schema.lua",
-    ["kong.plugins.opentelemetry.proto"]  = "kong/plugins/opentelemetry/proto.lua",
-    ["kong.plugins.opentelemetry.otlp"]  = "kong/plugins/opentelemetry/otlp.lua",
     ["kong.plugins.opentelemetry.traces"] = "kong/plugins/opentelemetry/traces.lua",
     ["kong.plugins.opentelemetry.logs"] = "kong/plugins/opentelemetry/logs.lua",
     ["kong.plugins.opentelemetry.utils"] = "kong/plugins/opentelemetry/utils.lua",
@@ -674,6 +672,9 @@ build = {
 
     ["kong.observability.logs"] = "kong/observability/logs.lua",
 
+    ["kong.observability.otlp.proto"] = "kong/observability/otlp/proto.lua",
+    ["kong.observability.otlp"] = "kong/observability/otlp/init.lua",
+
     ["kong.timing"] = "kong/timing/init.lua",
     ["kong.timing.context"] = "kong/timing/context.lua",
     ["kong.timing.hooks"] = "kong/timing/hooks/init.lua",

kong/plugins/opentelemetry/otlp.lua → kong/observability/otlp/init.lua

@@ -1,15 +1,16 @@
-require "kong.plugins.opentelemetry.proto"
+require "kong.observability.otlp.proto"
 local pb = require "pb"
 local new_tab = require "table.new"
 local nkeys = require "table.nkeys"
 local tablepool = require "tablepool"
-local deep_copy = require("kong.tools.table").deep_copy
+local kong_table = require "kong.tools.table"
 
 local kong = kong
 local insert = table.insert
 local tablepool_fetch = tablepool.fetch
 local tablepool_release = tablepool.release
-local table_merge = require("kong.tools.table").table_merge
+local table_merge = kong_table.table_merge
+local deep_copy = kong_table.deep_copy
 local setmetatable = setmetatable
 
 local TRACE_ID_LEN = 16
@@ -33,19 +34,41 @@ local TYPE_TO_ATTRIBUTE_TYPES = {
   boolean = "bool_value",
 }
 
+local function transform_value(key, value)
+  if type(value) == "table" then
+    if kong_table.is_array(value) then
+      local entries = new_tab(#value, 0)
+      for _, v in ipairs(value) do
+        insert(entries, transform_value(nil, v))
+      end
+      return { array_value = { values = entries } }
+    else
+      local entries = new_tab(nkeys(value), 0)
+      for k, v in pairs(value) do
+        insert(entries, {
+          key = k,
+          value = transform_value(k, v)
+        })
+      end
+      return { kvlist_value = { values = entries } }
+    end
+  end
+
+  local attribute_type = key and KEY_TO_ATTRIBUTE_TYPES[key]
+                         or TYPE_TO_ATTRIBUTE_TYPES[type(value)]
+  return attribute_type and { [attribute_type] = value } or EMPTY_TAB
+end
+
 local function transform_attributes(attr)
   if type(attr) ~= "table" then
     error("invalid attributes", 2)
   end
 
   local pb_attributes = new_tab(nkeys(attr), 0)
   for k, v in pairs(attr) do
-
-    local attribute_type = KEY_TO_ATTRIBUTE_TYPES[k] or TYPE_TO_ATTRIBUTE_TYPES[type(v)]
-
     insert(pb_attributes, {
       key = k,
-      value = attribute_type and { [attribute_type] = v } or EMPTY_TAB
+      value = transform_value(k, v),
     })
   end
 

kong/pdk/tracing.lua

@@ -234,13 +234,7 @@ local function link_span(tracer, span, name, options)
   span.linked = true
 
   -- insert the span to ctx
-  local ctx = ngx.ctx
-  local spans = ctx.KONG_SPANS
-  if not spans then
-    spans = tablepool_fetch(POOL_SPAN_STORAGE, 10, 0)
-    spans[0] = 0 -- span counter
-    ctx.KONG_SPANS = spans
-  end
+  local spans = tracer.get_spans()
 
   local len = spans[0] + 1
   spans[len] = span
@@ -321,7 +315,8 @@ function span_mt:set_attribute(key, value)
    vtyp = type(value)
   end
 
-  if vtyp ~= "string" and vtyp ~= "number" and vtyp ~= "boolean" and vtyp ~= nil then
+  -- TODO: any invalid type left?
+  if vtyp ~= "string" and vtyp ~= "number" and vtyp ~= "boolean" and vtyp ~= "table" and vtyp ~= nil then
     -- we should not error out here, as most of the caller does not catch
     -- errors, and they are hooking to core facilities, which may cause
     -- unexpected behavior.
@@ -427,14 +422,19 @@ local noop_tracer = {}
 noop_tracer.name = "noop"
 noop_tracer.start_span = function() return noop_span end
 noop_tracer.create_span = function() return noop_span end
+noop_tracer.get_spans = NOOP
+noop_tracer.get_root_span = NOOP
+noop_tracer.init_spans = NOOP
 noop_tracer.link_span = NOOP
 noop_tracer.active_span = NOOP
 noop_tracer.set_active_span = NOOP
 noop_tracer.process_span = NOOP
 noop_tracer.set_should_sample = NOOP
 noop_tracer.get_sampling_decision = NOOP
+noop_tracer.spans_table_key = "noop"
 
 local VALID_TRACING_PHASES = {
+  ssl_cert = true,
   rewrite = true,
   access = true,
   header_filter = true,
@@ -446,9 +446,11 @@ local VALID_TRACING_PHASES = {
 --- New Tracer
 local function new_tracer(name, options)
   name = name or "default"
+  local namespace = options and options.namespace or "KONG"
+  local cache_key = namespace .. "_" .. name
 
-  if tracer_memo[name] then
-    return tracer_memo[name]
+  if tracer_memo[cache_key] then
+    return tracer_memo[cache_key]
   end
 
   local self = {
@@ -463,7 +465,8 @@ local function new_tracer(name, options)
 
   options.sampling_rate = options.sampling_rate or 1.0
   self.sampler = get_trace_id_based_sampler(options.sampling_rate)
-  self.active_span_key = name .. "_" .. "active_span"
+  self.active_span_key = namespace .. "_" .. "active_span"
+  self.spans_table_key = namespace .. "_" .. "SPANS"
 
   --- Get the active span
   -- Returns the root span by default
@@ -501,7 +504,7 @@ local function new_tracer(name, options)
   -- @function kong.tracing.start_span
   -- @phases rewrite, access, header_filter, response, body_filter, log, admin_api
   -- @tparam string name span name
-  -- @tparam table options TODO(mayo)
+  -- @tparam table options
   -- @treturn table span
   function self.start_span(...)
     if not VALID_TRACING_PHASES[ngx.get_phase()] then
@@ -519,6 +522,26 @@ local function new_tracer(name, options)
     return link_span(...)
   end
 
+  function self.init_spans()
+    local spans = tablepool_fetch(POOL_SPAN_STORAGE, 10, 0)
+    spans[0] = 0 -- span counter
+    ngx.ctx[self.spans_table_key] = spans
+    return spans
+  end
+
+  function self.get_spans()
+    return ngx.ctx[self.spans_table_key] or self.init_spans()
+  end
+
+  function self.get_root_span()
+    local spans = self.get_spans()
+    if not spans then
+      return
+    end
+
+    return spans[1]
+  end
+
   --- Batch process spans
   -- Please note that socket is not available in the log phase, use `ngx.timer.at` instead
   --
@@ -532,12 +555,12 @@ local function new_tracer(name, options)
       error("processor must be a function", 2)
     end
 
-    local ctx = ngx.ctx
-    if not ctx.KONG_SPANS then
+    local spans = self.get_spans()
+    if not spans then
       return
     end
 
-    for _, span in ipairs(ctx.KONG_SPANS) do
+    for _, span in ipairs(spans) do
       if span.tracer and span.tracer.name == self.name then
         processor(span, ...)
       end
@@ -549,12 +572,12 @@ local function new_tracer(name, options)
   -- @function kong.tracing:set_should_sample
   -- @tparam bool should_sample value for the sample parameter
   function self:set_should_sample(should_sample)
-    local ctx = ngx.ctx
-    if not ctx.KONG_SPANS then
+    local spans = self.get_spans()
+    if not spans then
       return
     end
 
-    for _, span in ipairs(ctx.KONG_SPANS) do
+    for _, span in ipairs(spans) do
       if span.is_recording ~= false then
         span.should_sample = should_sample
       end
@@ -579,7 +602,7 @@ local function new_tracer(name, options)
     local ctx = ngx.ctx
 
     local sampled
-    local root_span = ctx.KONG_SPANS and ctx.KONG_SPANS[1]
+    local root_span = self.get_root_span()
     local trace_id = tracing_context.get_raw_trace_id(ctx)
     local sampling_rate = plugin_sampling_rate or kong.configuration.tracing_sampling_rate
 
@@ -617,11 +640,11 @@ noop_tracer.new = new_tracer
 
 local global_tracer
 tracer_mt.set_global_tracer = function(tracer)
-  if type(tracer) ~= "table" or getmetatable(tracer) ~= tracer_mt then
+  if type(tracer) ~= "table" or
+      (getmetatable(tracer) ~= tracer_mt and tracer.name ~= "noop") then
     error("invalid tracer", 2)
   end
 
-  tracer.active_span_key = "active_span"
   global_tracer = tracer
   -- replace kong.pdk.tracer
   if kong then

kong/plugins/opentelemetry/logs.lua

@@ -1,6 +1,6 @@
 local Queue = require "kong.tools.queue"
 local o11y_logs = require "kong.observability.logs"
-local otlp = require "kong.plugins.opentelemetry.otlp"
+local otlp = require "kong.observability.otlp"
 local tracing_context = require "kong.observability.tracing.tracing_context"
 local otel_utils = require "kong.plugins.opentelemetry.utils"
 local clone = require "table.clone"

kong/plugins/opentelemetry/traces.lua

@@ -1,7 +1,7 @@
 local Queue = require "kong.tools.queue"
 local propagation = require "kong.observability.tracing.propagation"
 local tracing_context = require "kong.observability.tracing.tracing_context"
-local otlp = require "kong.plugins.opentelemetry.otlp"
+local otlp = require "kong.observability.otlp"
 local otel_utils = require "kong.plugins.opentelemetry.utils"
 local clone = require "table.clone"
 

spec/01-unit/26-observability/01-tracer_pdk_spec.lua

@@ -177,6 +177,14 @@ describe("Tracer PDK", function()
       assert.spy(log_spy).was_called_with(ngx.ERR, match.is_string())
 
       assert.error(function() span:set_attribute(123, 123) end)
+
+      -- array attribute value is allowed
+      span:set_attribute("key1", { "value1", "value2" })
+      assert.same({ "value1", "value2" }, span.attributes["key1"])
+
+      -- map attribute value is allowed
+      span:set_attribute("key1", { key1 = "value1", key2 = "value2" })
+      assert.same({ key1 = "value1", key2 = "value2" }, span.attributes["key1"])
     end)
 
     it("fails add_event", function ()

spec/03-plugins/37-opentelemetry/01-otlp_spec.lua

@@ -1,6 +1,6 @@
 require "spec.helpers"
-require "kong.plugins.opentelemetry.proto"
-local otlp = require "kong.plugins.opentelemetry.otlp"
+require "kong.observability.otlp.proto"
+local otlp = require "kong.observability.otlp"
 local pb = require "pb"
 
 local fmt = string.format
@@ -74,6 +74,18 @@ describe("Plugin: opentelemetry (otlp)", function()
     ngx.ctx.KONG_SPANS = nil
   end)
 
+  local function assert_contains_attribute(span, attr_name, attr_type)
+    assert.is_table(span.attributes)
+    for _, attr in ipairs(span.attributes) do
+      if attr.key == attr_name then
+        assert.is_table(attr.value)
+        assert.not_nil(attr.value[attr_type])
+        return
+      end
+    end
+    assert.fail(fmt("attribute %s not found", attr_name))
+  end
+
   it("encode/decode pb (traces)", function ()
     local N = 10000
 
@@ -112,6 +124,11 @@ describe("Plugin: opentelemetry (otlp)", function()
           int = i,
           bool = (i % 2 == 0 and true) or false,
           double = i / (N * 1000),
+          array = { "one", "two", "three" },
+          map = {
+            key1 = "value1",
+            key2 = "value2",
+          }
         },
       })
 
@@ -120,14 +137,27 @@ describe("Plugin: opentelemetry (otlp)", function()
       span:finish()
 
       insert(test_spans, table.clone(span))
-      span:release()
     end
 
     for _, test_span in ipairs(test_spans) do
       local pb_span = otlp.transform_span(test_span)
       local pb_data = pb_encode_span(pb_span)
       local decoded_span = pb_decode_span(pb_data)
 
+      if decoded_span.name == "full-span" then
+        assert_contains_attribute(decoded_span, "foo", "string_value")
+        assert_contains_attribute(decoded_span, "test", "bool_value")
+        assert_contains_attribute(decoded_span, "version", "double_value")
+
+      else
+        assert_contains_attribute(decoded_span, "str", "string_value")
+        assert_contains_attribute(decoded_span, "int", "double_value")
+        assert_contains_attribute(decoded_span, "bool", "bool_value")
+        assert_contains_attribute(decoded_span, "double", "double_value")
+        assert_contains_attribute(decoded_span, "array", "array_value")
+        assert_contains_attribute(decoded_span, "map", "kvlist_value")
+      end
+
       local ok, err = table_compare(pb_span, decoded_span)
       assert.is_true(ok, err)
     end

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

@@ -1,4 +1,4 @@
-require "kong.plugins.opentelemetry.proto"
+require "kong.observability.otlp.proto"
 local helpers = require "spec.helpers"
 local pb = require "pb"
 local pl_file = require "pl.file"

spec/03-plugins/37-opentelemetry/05-otelcol_spec.lua

@@ -1,4 +1,4 @@
-require "kong.plugins.opentelemetry.proto"
+require "kong.observability.otlp.proto"
 local helpers = require "spec.helpers"
 local kong_table = require "kong.tools.table"
 local ngx_re = require "ngx.re"