feat(tracing): propagation module / typedef

The new propagation module replaces the propagation.lua file, providing
a more flexible and extensible way to handle tracing headers propagation
from plugins (at the moment OpenTelemetry and Zipkin). It allows
configuring the priority of tracing context extraction and what headers
to extract and inject the tracing context from/to. It also allows
clearing headers from the request after extraction to gain full control
on what is propagated to the upstream.

New configuration options:

propagation.extract: Header formats used to extract tracing context from
incoming requests.
propagation.inject: Header formats used to inject tracing context.
propagation.clear: Header names to clear after context extraction.

Each header format is now defined in its own extractor and injector
files so that individual logic for extracting and injecting the tracing
context is isolated. This is meant to improve
maintainability/testability and facilitate extending support to new
header formats.

apply PR suggestions (squash me)

Co-authored-by: Qi <add_sp@outlook.com>

changelog/unreleased/kong/propagation-module-rework.yml

@@ -0,0 +1,5 @@
+message: |
+  **OpenTelemetry, Zipkin**: the propagation module has been reworked, new
+  options allow better control over the configuration of tracing headers propagation.
+type: feature
+scope: Plugin

kong-3.7.0-0.rockspec

@@ -598,7 +598,26 @@ build = {
     ["kong.vaults.env.schema"] = "kong/vaults/env/schema.lua",
 
     ["kong.tracing.instrumentation"] = "kong/tracing/instrumentation.lua",
-    ["kong.tracing.propagation"] = "kong/tracing/propagation.lua",
+    ["kong.tracing.propagation"] = "kong/tracing/propagation/init.lua",
+    ["kong.tracing.propagation.schema"] = "kong/tracing/propagation/schema.lua",
+    ["kong.tracing.propagation.utils"] = "kong/tracing/propagation/utils.lua",
+    ["kong.tracing.propagation.extractors._base"] = "kong/tracing/propagation/extractors/_base.lua",
+    ["kong.tracing.propagation.extractors.w3c"] = "kong/tracing/propagation/extractors/w3c.lua",
+    ["kong.tracing.propagation.extractors.b3"] = "kong/tracing/propagation/extractors/b3.lua",
+    ["kong.tracing.propagation.extractors.jaeger"] = "kong/tracing/propagation/extractors/jaeger.lua",
+    ["kong.tracing.propagation.extractors.ot"] = "kong/tracing/propagation/extractors/ot.lua",
+    ["kong.tracing.propagation.extractors.gcp"] = "kong/tracing/propagation/extractors/gcp.lua",
+    ["kong.tracing.propagation.extractors.aws"] = "kong/tracing/propagation/extractors/aws.lua",
+    ["kong.tracing.propagation.extractors.datadog"] = "kong/tracing/propagation/extractors/datadog.lua",
+    ["kong.tracing.propagation.injectors._base"] = "kong/tracing/propagation/injectors/_base.lua",
+    ["kong.tracing.propagation.injectors.w3c"] = "kong/tracing/propagation/injectors/w3c.lua",
+    ["kong.tracing.propagation.injectors.b3"] = "kong/tracing/propagation/injectors/b3.lua",
+    ["kong.tracing.propagation.injectors.b3-single"] = "kong/tracing/propagation/injectors/b3-single.lua",
+    ["kong.tracing.propagation.injectors.jaeger"] = "kong/tracing/propagation/injectors/jaeger.lua",
+    ["kong.tracing.propagation.injectors.ot"] = "kong/tracing/propagation/injectors/ot.lua",
+    ["kong.tracing.propagation.injectors.gcp"] = "kong/tracing/propagation/injectors/gcp.lua",
+    ["kong.tracing.propagation.injectors.aws"] = "kong/tracing/propagation/injectors/aws.lua",
+    ["kong.tracing.propagation.injectors.datadog"] = "kong/tracing/propagation/injectors/datadog.lua",
     ["kong.tracing.request_id"] = "kong/tracing/request_id.lua",
     ["kong.tracing.tracing_context"] = "kong/tracing/tracing_context.lua",
 

kong/db/schema/typedefs.lua

@@ -2,6 +2,7 @@
 -- @module kong.db.schema.typedefs
 local utils = require "kong.tools.utils"
 local queue_schema = require "kong.tools.queue_schema"
+local propagation_schema = require "kong.tracing.propagation.schema"
 local openssl_pkey = require "resty.openssl.pkey"
 local openssl_x509 = require "resty.openssl.x509"
 local Schema = require "kong.db.schema"
@@ -882,6 +883,8 @@ typedefs.jwk = Schema.define {
 
 typedefs.queue = queue_schema
 
+typedefs.propagation = propagation_schema
+
 local function validate_lua_expression(expression)
   local sandbox = require "kong.tools.sandbox"
   return sandbox.validate_safe(expression)

kong/plugins/opentelemetry/handler.lua

@@ -14,10 +14,6 @@ local ngx_ERR = ngx.ERR
 local ngx_DEBUG = ngx.DEBUG
 local ngx_now = ngx.now
 local ngx_update_time = ngx.update_time
-local ngx_req = ngx.req
-local ngx_get_headers = ngx_req.get_headers
-local propagation_parse = propagation.parse
-local propagation_set = propagation.set
 local null = ngx.null
 local encode_traces = otlp.encode_traces
 local encode_span = otlp.transform_span
@@ -91,8 +87,8 @@ local function http_export(conf, spans)
   return ok, err
 end
 
-function OpenTelemetryHandler:access(conf)
-  local headers = ngx_get_headers()
+
+local function get_inject_ctx(extracted_ctx, conf)
   local root_span = ngx.ctx.KONG_SPANS and ngx.ctx.KONG_SPANS[1]
 
   -- get the global tracer when available, or instantiate a new one
@@ -109,17 +105,21 @@ function OpenTelemetryHandler:access(conf)
   end
 
   local injected_parent_span = tracing_context.get_unlinked_span("balancer") or root_span
-  local header_type, trace_id, span_id, parent_id, parent_sampled, _ = propagation_parse(headers, conf.header_type)
+  local trace_id = extracted_ctx.trace_id
+  local span_id = extracted_ctx.span_id
+  local parent_id = extracted_ctx.parent_id
+  local parent_sampled = extracted_ctx.should_sample
 
   -- Overwrite trace ids
   -- with the value extracted from incoming tracing headers
   if trace_id then
     -- to propagate the correct trace ID we have to set it here
-    -- before passing this span to propagation.set()
+    -- before passing this span to propagation
     injected_parent_span.trace_id = trace_id
     -- update the Tracing Context with the trace ID extracted from headers
     tracing_context.set_raw_trace_id(trace_id)
   end
+
   -- overwrite root span's parent_id
   if span_id then
     root_span.parent_id = span_id
@@ -147,7 +147,22 @@ function OpenTelemetryHandler:access(conf)
   -- Set the sampled flag for the outgoing header's span
   injected_parent_span.should_sample = sampled
 
-  propagation_set(conf.header_type, header_type, injected_parent_span, "w3c")
+  extracted_ctx.trace_id      = injected_parent_span.trace_id
+  extracted_ctx.span_id       = injected_parent_span.span_id
+  extracted_ctx.should_sample = injected_parent_span.should_sample
+  extracted_ctx.parent_id     = injected_parent_span.parent_id
+
+  -- return the injected ctx (data to be injected with outgoing tracing headers)
+  return extracted_ctx
+end
+
+
+function OpenTelemetryHandler:access(conf)
+  propagation.propagate(
+    propagation.get_plugin_params(conf),
+    get_inject_ctx,
+    conf
+  )
 end
 
 

kong/plugins/opentelemetry/schema.lua

@@ -71,16 +71,25 @@ return {
               default = nil }},
         { header_type = { description = "All HTTP requests going through the plugin are tagged with a tracing HTTP request. This property codifies what kind of tracing header the plugin expects on incoming requests.",
               type = "string",
+              deprecation = {
+                message = "opentelemetry: config.header_type is deprecated, please use config.propagation options instead",
+                removal_in_version = "4.0",
+                old_default = "preserve" },
               required = false,
               default = "preserve",
-              one_of = { "preserve", "ignore", "b3", "b3-single", "w3c", "jaeger", "ot", "aws", "gcp" } } },
+              one_of = { "preserve", "ignore", "b3", "b3-single", "w3c", "jaeger", "ot", "aws", "gcp", "datadog" } } },
         { sampling_rate = {
           description = "Tracing sampling rate for configuring the probability-based sampler. When set, this value supersedes the global `tracing_sampling_rate` setting from kong.conf.",
           type = "number",
           between = {0, 1},
           required = false,
           default = nil,
         } },
+        { propagation = typedefs.propagation {
+          default = {
+            default_format = "w3c",
+          },
+        } },
       },
     }, },
   },

kong/plugins/zipkin/handler.lua

@@ -88,24 +88,44 @@ end
 local initialize_request
 
 
-local function get_context(conf, ctx)
+local function get_context(conf, ctx, extracted_ctx)
   local zipkin = ctx.zipkin
   if not zipkin then
-    initialize_request(conf, ctx)
+    initialize_request(conf, ctx, extracted_ctx)
     zipkin = ctx.zipkin
   end
   return zipkin
 end
 
 
 if subsystem == "http" then
-  initialize_request = function(conf, ctx)
+  initialize_request = function(conf, ctx, extracted_ctx)
     local req = kong.request
 
     local req_headers = req.get_headers()
 
-    local header_type, trace_id, span_id, parent_id, should_sample, baggage =
-      propagation.parse(req_headers, conf.header_type)
+    extracted_ctx       = extracted_ctx
+                          or propagation.extract(propagation.get_plugin_params(conf))
+                          or {}
+    local trace_id      = extracted_ctx.trace_id
+    local should_sample = extracted_ctx.should_sample
+    local baggage       = extracted_ctx.baggage
+
+    -- Some formats (e.g. W3C) only provide one span_id, which is the id of the
+    -- span that the header represents, it is meant to be used as the parent of
+    -- the server span (span generated by the receiver) and is in fact
+    -- sometimes called parent_id.
+    -- Other formats (e.g. B3) support two span IDs, usually span_id and
+    -- parent_id. In that case the span (and its ID) is shared between client
+    -- and server and the parent_id identifies its parent.
+    local parent_id, span_id
+    if extracted_ctx.reuse_span_id then
+      span_id   = extracted_ctx.span_id
+      parent_id = extracted_ctx.parent_id
+
+    else
+      parent_id = extracted_ctx.span_id
+    end
 
     local method = req.get_method()
 
@@ -168,23 +188,38 @@ if subsystem == "http" then
 
     ctx.zipkin = {
       request_span = request_span,
-      header_type = header_type,
       proxy_span = nil,
       header_filter_finished = false,
     }
   end
 
 
-  function ZipkinLogHandler:access(conf) -- luacheck: ignore 212
-    local zipkin = get_context(conf, kong.ctx.plugin)
+  local function get_inject_ctx(extract_ctx, conf)
+    local zipkin = get_context(conf, kong.ctx.plugin, extract_ctx)
     local ngx_ctx = ngx.ctx
 
     local access_start =
       ngx_ctx.KONG_ACCESS_START and ngx_ctx.KONG_ACCESS_START * 1000
       or ngx_now_mu()
-    get_or_add_proxy_span(zipkin, access_start)
 
-    propagation.set(conf.header_type, zipkin.header_type, zipkin.proxy_span, conf.default_header_type)
+    local proxy_span = get_or_add_proxy_span(zipkin, access_start)
+
+    local inject_ctx = extract_ctx
+    inject_ctx.trace_id      = proxy_span.trace_id      or inject_ctx.trace_id      or nil
+    inject_ctx.span_id       = proxy_span.span_id       or inject_ctx.span_id       or nil
+    inject_ctx.parent_id     = proxy_span.parent_id     or inject_ctx.parent_id     or nil
+    inject_ctx.should_sample = proxy_span.should_sample or inject_ctx.should_sample or nil
+    inject_ctx.baggage       = proxy_span.baggage       or inject_ctx.baggage       or nil
+    return inject_ctx
+  end
+
+
+  function ZipkinLogHandler:access(conf) -- luacheck: ignore 212
+    propagation.propagate(
+      propagation.get_plugin_params(conf),
+      get_inject_ctx,
+      conf
+    )
   end
 
 

kong/plugins/zipkin/schema.lua

@@ -56,9 +56,13 @@ return {
           { include_credential = { description = "Specify whether the credential of the currently authenticated consumer should be included in metadata sent to the Zipkin server.", type = "boolean", required = true, default = true } },
           { traceid_byte_count = { description = "The length in bytes of each request's Trace ID.", type = "integer", required = true, default = 16, one_of = { 8, 16 } } },
           { header_type = { description = "All HTTP requests going through the plugin are tagged with a tracing HTTP request. This property codifies what kind of tracing header the plugin expects on incoming requests", type = "string", required = true, default = "preserve",
-                            one_of = { "preserve", "ignore", "b3", "b3-single", "w3c", "jaeger", "ot", "aws", "gcp" } } },
+                            one_of = { "preserve", "ignore", "b3", "b3-single", "w3c", "jaeger", "ot", "aws", "datadog", "gcp" },
+                            deprecation = { message = "zipkin: config.header_type is deprecated, please use config.propagation options instead", removal_in_version = "4.0", old_default = "preserve" }
+                          } },
           { default_header_type = { description = "Allows specifying the type of header to be added to requests with no pre-existing tracing headers and when `config.header_type` is set to `\"preserve\"`. When `header_type` is set to any other value, `default_header_type` is ignored.", type = "string", required = true, default = "b3",
-                            one_of = { "b3", "b3-single", "w3c", "jaeger", "ot", "aws", "gcp" } } },
+                            one_of = { "b3", "b3-single", "w3c", "jaeger", "ot", "aws", "datadog", "gcp" },
+                            deprecation = { message = "zipkin: config.default_header_type is deprecated, please use config.propagation.default_format instead", removal_in_version = "4.0", old_default = "b3" }
+                          } },
           { tags_header = { description = "The Zipkin plugin will add extra headers to the tags associated with any HTTP requests that come with a header named as configured by this property.", type = "string", required = true, default = "Zipkin-Tags" } },
           { static_tags = {  description = "The tags specified on this property will be added to the generated request traces.", type = "array", elements = static_tag,
                             custom_validator = validate_static_tags } },
@@ -70,6 +74,11 @@ return {
           { phase_duration_flavor = { description = "Specify whether to include the duration of each phase as an annotation or a tag.", type = "string", required = true, default = "annotations",
                                       one_of = { "annotations", "tags" } } },
           { queue = typedefs.queue },
+          { propagation = typedefs.propagation {
+            default = {
+              default_format = "b3",
+            },
+          } },
         },
     }, },
   },