kong--kong/kong/plugins/opentelemetry/traces.lua

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.observability.otlp"
local otel_utils = require "kong.plugins.opentelemetry.utils"
local clone = require "table.clone"

local to_hex = require "resty.string".to_hex
local bor = require "bit".bor

local ngx = ngx
local kong = kong
local ngx_log = ngx.log
local ngx_ERR = ngx.ERR
local ngx_DEBUG = ngx.DEBUG

local http_export_request = otel_utils.http_export_request
local get_headers = otel_utils.get_headers
local _log_prefix = otel_utils._log_prefix
local encode_traces = otlp.encode_traces
local encode_span = otlp.transform_span


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
  local tracer = kong.tracing.name == "noop" and kong.tracing.new("otel")
                 or kong.tracing

  -- make propagation work with tracing disabled
  if not root_span then
    root_span = tracer.start_span("root")
    root_span:set_attribute("kong.propagation_only", true)

    -- since tracing is disabled, turn off sampling entirely for this trace
    kong.ctx.plugin.should_sample = false
  end

  local injected_parent_span = tracing_context.get_unlinked_span("balancer") or root_span
  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
  local flags = extracted_ctx.w3c_flags or extracted_ctx.flags

  -- 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
    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

  elseif parent_id then
    root_span.parent_id = parent_id
  end

  -- Configure the sampled flags
  local sampled
  if kong.ctx.plugin.should_sample == false then
    sampled = false

  else
    -- Sampling decision for the current trace.
    local err
    -- get_sampling_decision() depends on the value of the trace id: call it
    -- after the trace_id is updated
    sampled, err = tracer:get_sampling_decision(parent_sampled, conf.sampling_rate)
    if err then
      ngx_log(ngx_ERR, _log_prefix, "sampler failure: ", err)
    end
  end
  tracer:set_should_sample(sampled)
  -- Set the sampled flag for the outgoing header's span
  injected_parent_span.should_sample = sampled

  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

  flags = flags or 0x00
  local sampled_flag = sampled and 1 or 0
  local out_flags = bor(flags,  sampled_flag)
  tracing_context.set_flags(out_flags)

  -- return the injected ctx (data to be injected with outgoing tracing headers)
  return extracted_ctx
end


local function access(conf)
  propagation.propagate(
    propagation.get_plugin_params(conf),
    get_inject_ctx,
    conf
  )
end


local function header_filter(conf)
  if conf.http_response_header_for_traceid then
    local trace_id = tracing_context.get_raw_trace_id()
    if not trace_id then
      local root_span = ngx.ctx.KONG_SPANS and ngx.ctx.KONG_SPANS[1]
      trace_id = root_span and root_span.trace_id
    end
    if trace_id then
      trace_id = to_hex(trace_id)
      kong.response.add_header(conf.http_response_header_for_traceid, trace_id)
    end
  end
end


local function http_export_traces(params, spans)
  local conf = params.conf
  local resource_attributes = params.options.compiled_resource_attributes
                              or conf.resource_attributes
  local headers = get_headers(conf.headers)

  local payload = encode_traces(spans, resource_attributes)

  local ok, err = http_export_request({
    connect_timeout = conf.connect_timeout,
    send_timeout = conf.send_timeout,
    read_timeout = conf.read_timeout,
    endpoint = conf.traces_endpoint,
  }, payload, headers)

  if ok then
    ngx_log(ngx_DEBUG, _log_prefix, "exporter sent ", #spans,
          " spans to ", conf.traces_endpoint)

  else
    ngx_log(ngx_ERR, _log_prefix, err)
  end

  return ok, err
end


local function log(conf, options)
  ngx_log(ngx_DEBUG, _log_prefix, "total spans in current request: ", ngx.ctx.KONG_SPANS and #ngx.ctx.KONG_SPANS)

  kong.tracing.process_span(function (span)
    if span.should_sample == false or kong.ctx.plugin.should_sample == false then
      -- ignore
      return
    end

    -- overwrite
    local trace_id = tracing_context.get_raw_trace_id()
    if trace_id then
      span.trace_id = trace_id
    end

    local params = {
      conf = conf,
      options = options,
    }
    local queue_conf = clone(Queue.get_plugin_params("opentelemetry", conf))
    queue_conf.name = queue_conf.name .. ":traces"

    local ok, err = Queue.enqueue(
      queue_conf,
      http_export_traces,
      params,
      encode_span(span)
    )
    if not ok then
      kong.log.err("Failed to enqueue span to log server: ", err)
    end
  end)
end


return {
  access = access,
  header_filter = header_filter,
  log = log,
}