--

-- Licensed to the Apache Software Foundation (ASF) under one or more

-- contributor license agreements. See the NOTICE file distributed with

-- this work for additional information regarding copyright ownership.

-- The ASF licenses this file to You under the Apache License, Version 2.0

-- (the "License"); you may not use this file except in compliance with

-- the License. You may obtain a copy of the License at

--

-- http://www.apache.org/licenses/LICENSE-2.0

--

-- Unless required by applicable law or agreed to in writing, software

-- distributed under the License is distributed on an "AS IS" BASIS,

-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

-- See the License for the specific language governing permissions and

-- limitations under the License.

--

local plugin_name = "opentelemetry"

local core = require("apisix.core")

local plugin = require("apisix.plugin")

local process = require("ngx.process")

local always_off_sampler_new = require("opentelemetry.trace.sampling.always_off_sampler").new

local always_on_sampler_new = require("opentelemetry.trace.sampling.always_on_sampler").new

local parent_base_sampler_new = require("opentelemetry.trace.sampling.parent_base_sampler").new

local trace_id_ratio_sampler_new =

require("opentelemetry.trace.sampling.trace_id_ratio_sampler").new

local exporter_client_new = require("opentelemetry.trace.exporter.http_client").new

local otlp_exporter_new = require("opentelemetry.trace.exporter.otlp").new

local batch_span_processor_new = require("opentelemetry.trace.batch_span_processor").new

local id_generator = require("opentelemetry.trace.id_generator")

local tracer_provider_new = require("opentelemetry.trace.tracer_provider").new

local span_kind = require("opentelemetry.trace.span_kind")

local span_status = require("opentelemetry.trace.span_status")

local resource_new = require("opentelemetry.resource").new

local attr = require("opentelemetry.attribute")

local context = require("opentelemetry.context").new()

local trace_context_propagator =

require("opentelemetry.trace.propagation.text_map.trace_context_propagator").new()

local ngx = ngx

local ngx_var = ngx.var

local table = table

local type = type

local pairs = pairs

local ipairs = ipairs

local unpack = unpack

local string_format = string.format

local string_lower = string.lower

local update_time = ngx.update_time

local tostring = tostring

local lrucache = core.lrucache.new({

type = 'plugin', count = 128, ttl = 24 * 60 * 60,

})

local asterisk = string.byte("*", 1)

-- capture the library's default (random) generator once, so wrapping it below

-- stays idempotent even when create_tracer_obj re-runs after cache expiry

local original_new_ids = id_generator.new_ids

-- expects an already-lowercased string

local function is_valid_trace_id(trace_id)

if not trace_id or #trace_id ~= 32 then

return false

end

if not trace_id:match("^[0-9a-f]+$") then

return false

end

-- W3C Trace Context: all-zero trace_id is invalid

if trace_id == "00000000000000000000000000000000" then

return false

end

return true

end

local metadata_schema = {

type = "object",

properties = {

trace_id_source = {

type = "string",

enum = {"x-request-id", "random"},

description = "the source of trace id",

default = "random",

},

resource = {

type = "object",

description = "additional resource",

additionalProperties = {{type = "boolean"}, {type = "number"}, {type = "string"}},

},

collector = {

type = "object",

description = "opentelemetry collector",

properties = {

address = {type = "string", description = "host:port", default = "127.0.0.1:4318"},

request_timeout = {type = "integer", description = "second uint", default = 3},

request_headers = {

type = "object",

description = "http headers",

additionalProperties = {

one_of = {{type = "boolean"},{type = "number"}, {type = "string"}},

},

}

},

default = {address = "127.0.0.1:4318", request_timeout = 3}

},

batch_span_processor = {

type = "object",

description = "batch span processor",

properties = {

drop_on_queue_full = {

type = "boolean",

description = "if true, drop span when queue is full,"

.. " otherwise force process batches",

},

max_queue_size = {

type = "integer",

description = "maximum queue size to buffer spans for delayed processing",

},

batch_timeout = {

type = "number",

description = "maximum duration for constructing a batch",

},

inactive_timeout = {

type = "number",

description = "maximum duration for processing batches",

},

max_export_batch_size = {

type = "integer",

description = "maximum number of spans to process in a single batch",

}

},

default = {},

},

set_ngx_var = {

type = "boolean",

description = "set nginx variables",

default = false,

},

},

}

local schema = {

type = "object",

properties = {

sampler = {

type = "object",

properties = {

name = {

type = "string",

enum = {"always_on", "always_off", "trace_id_ratio", "parent_base"},

description = "sampling strategy",

default = "always_off"

},

options = {

type = "object",

properties = {

fraction = {

type = "number", description = "trace_id_ratio fraction", default = 0

},

root = {

type = "object",

description = "parent_base root sampler",

properties = {

name = {

type = "string",

enum = {"always_on", "always_off", "trace_id_ratio"},

description = "sampling strategy",

default = "always_off"

},

options = {

type = "object",

properties = {

fraction = {

type = "number",

description = "trace_id_ratio fraction parameter",

default = 0,

},

},

default = {fraction = 0}

}

},

default = {name = "always_off", options = {fraction = 0}}

},

},

default = {fraction = 0, root = {name = "always_off"}}

}

},

default = {name = "always_off", options = {fraction = 0, root = {name = "always_off"}}}

},

additional_attributes = {

type = "array",

items = {

type = "string",

minLength = 1,

}

},

additional_header_prefix_attributes = {

type = "array",

items = {

type = "string",

minLength = 1,

}

}

}

}

local _M = {

version = 0.1,

priority = 12009,

name = plugin_name,

schema = schema,

metadata_schema = metadata_schema,

}

function _M.check_schema(conf, schema_type)

if schema_type == core.schema.TYPE_METADATA then

local ok, err = core.schema.check(metadata_schema, conf)

if not ok then

return ok, err

end

local check = {"collector.address"}

core.utils.check_https(check, conf, plugin_name)

return true

end

return core.schema.check(schema, conf)

end

local hostname

local sampler_factory

function _M.init()

if process.type() ~= "worker" then

return

end

sampler_factory = {

always_off = always_off_sampler_new,

always_on = always_on_sampler_new,

parent_base = parent_base_sampler_new,

trace_id_ratio = trace_id_ratio_sampler_new,

}

hostname = core.utils.gethostname()

end

local function create_tracer_obj(conf, plugin_info)

-- id_generator is a shared module, so restore the default before applying

-- the override: without this, switching trace_id_source back to "random"

-- would keep honoring X-Request-Id until the worker restarts

id_generator.new_ids = original_new_ids

if plugin_info.trace_id_source == "x-request-id" then

id_generator.new_ids = function()

local trace_id = core.request.headers()["x-request-id"]

or ngx_var.request_id

-- a duplicated X-Request-Id makes get_headers() return a table;

-- only a plain, valid 32-hex string can be used as a trace_id,

-- anything else (UUID, table, empty, ...) falls back to the

-- default generator instead of crashing the otlp encoder

if type(trace_id) == "string" then

trace_id = string_lower(trace_id)

if is_valid_trace_id(trace_id) then

return trace_id, id_generator.new_span_id()

end

end

return original_new_ids()

end

end

-- create exporter

local exporter = otlp_exporter_new(exporter_client_new(plugin_info.collector.address,

plugin_info.collector.request_timeout,

plugin_info.collector.request_headers))

-- create span processor

local batch_span_processor = batch_span_processor_new(exporter,

plugin_info.batch_span_processor)

-- create sampler

local sampler

local sampler_name = conf.sampler.name

local sampler_options = conf.sampler.options

if sampler_name == "parent_base" then

local root_sampler

if sampler_options.root then

local name, fraction = sampler_options.root.name, sampler_options.root.options.fraction

root_sampler = sampler_factory[name](fraction)

else

root_sampler = always_off_sampler_new()

end

sampler = sampler_factory[sampler_name](root_sampler)

else

sampler = sampler_factory[sampler_name](sampler_options.fraction)

end

local resource_attrs = {attr.string("hostname", hostname)}

if plugin_info.resource then

if not plugin_info.resource["service.name"] then

table.insert(resource_attrs, attr.string("service.name", "APISIX"))

end

for k, v in pairs(plugin_info.resource) do

if type(v) == "string" then

table.insert(resource_attrs, attr.string(k, v))

end

if type(v) == "number" then

table.insert(resource_attrs, attr.double(k, v))

end

if type(v) == "boolean" then

table.insert(resource_attrs, attr.bool(k, v))

end

end

end

-- create tracer provider

local tp = tracer_provider_new(batch_span_processor, {

resource = resource_new(unpack(resource_attrs)),

sampler = sampler,

})

-- create tracer

return tp:tracer("opentelemetry-lua")

end

-- Coerce a header/var value to a string suitable for an OTel string attribute.

-- ngx.req.get_headers() returns a Lua table for multi-value headers — running

-- `tostring()` over a table emits "table: 0x..." which is useless in a span,

-- so join multi-value entries instead. Returns nil when the value cannot be

-- represented (caller should skip the attribute in that case).

local function coerce_attr_value(val)

if val == nil then

return nil

end

if type(val) == "table" then

return table.concat(val, ", ")

end

return tostring(val)

end

local function inject_attributes(attributes, wanted_attributes, source, with_prefix)

for _, key in ipairs(wanted_attributes) do

local is_key_a_match = #key >= 2 and key:byte(-1) == asterisk and with_prefix

if is_key_a_match then

local prefix = key:sub(0, -2)

for possible_key, value in pairs(source) do

if core.string.has_prefix(possible_key, prefix) then

local coerced = coerce_attr_value(value)

if coerced ~= nil then

core.table.insert(attributes, attr.string(possible_key, coerced))

end

end

end

else

-- ~= nil so boolean `false` survives instead of being silently dropped.

local val = source[key]

if val ~= nil then

local coerced = coerce_attr_value(val)

if coerced ~= nil then

core.table.insert(attributes, attr.string(key, coerced))

end

end

end

end

end

function _M.rewrite(conf, api_ctx)

local metadata = plugin.plugin_metadata(plugin_name)

if metadata == nil then

core.log.warn("plugin_metadata is required for opentelemetry plugin to working properly")

return

end

core.log.info("metadata: ", core.json.delay_encode(metadata))

local plugin_info = metadata.value

local vars = api_ctx.var

-- key the cache on modifiedIndex so the tracer is rebuilt when metadata changes

local tracer, err = core.lrucache.plugin_ctx(lrucache, api_ctx, metadata.modifiedIndex,

create_tracer_obj, conf, plugin_info)

if not tracer then

core.log.error("failed to fetch tracer object: ", err)

return

end

local span_name = vars.method

local attributes = {

attr.string("net.host.name", vars.host),

-- deprecated attributes

attr.string("http.method", vars.method),

attr.string("http.scheme", vars.scheme),

attr.string("http.target", vars.request_uri),

attr.string("http.user_agent", vars.http_user_agent),

-- new attributes

attr.string("http.request.method", vars.method),

attr.string("url.scheme", vars.scheme),

attr.string("url.path", vars.uri),

attr.string("user_agent.original", vars.http_user_agent),

}

if api_ctx.curr_req_matched then

table.insert(attributes, attr.string("apisix.route_id", api_ctx.route_id))

table.insert(attributes, attr.string("apisix.route_name", api_ctx.route_name))

table.insert(attributes, attr.string("http.route", api_ctx.curr_req_matched._path))

span_name = span_name .. " " .. api_ctx.curr_req_matched._path

end

if api_ctx.service_id then

table.insert(attributes, attr.string("apisix.service_id", api_ctx.service_id))

table.insert(attributes, attr.string("apisix.service_name", api_ctx.service_name))

end

-- extract trace context from the headers of downstream HTTP request

local upstream_context = trace_context_propagator:extract(context, ngx.req)

local ctx = tracer:start(upstream_context, span_name, {

kind = span_kind.server,

attributes = attributes,

})

if plugin_info.set_ngx_var then

local span_context = ctx:span():context()

ngx_var.opentelemetry_context_traceparent = string_format("00-%s-%s-%02x",

span_context.trace_id,

span_context.span_id,

span_context.trace_flags)

ngx_var.opentelemetry_trace_id = span_context.trace_id

ngx_var.opentelemetry_span_id = span_context.span_id

end

if not ctx:span():is_recording() and ngx.ctx.tracing then

ngx.ctx.tracing.skip = true

end

api_ctx.otel_context_token = ctx:attach()

-- inject trace context into the headers of upstream HTTP request

trace_context_propagator:inject(ctx, ngx.req)

end

local function create_child_span(tracer, parent_span_ctx, spans, span)

if not span or span.finished then

return

end

span.finished = true

local new_span_ctx, new_span = tracer:start(parent_span_ctx, span.name,

{

kind = span.kind,

attributes = span.attributes,

})

new_span.start_time = span.start_time

for _, idx in ipairs(span.child_ids or {}) do

create_child_span(tracer, new_span_ctx, spans, spans[idx])

end

if span.status then

new_span:set_status(span.status.code, span.status.message)

end

new_span:finish(span.end_time)

end

local function inject_core_spans(root_span_ctx, api_ctx, conf)

local tracing = api_ctx.ngx_ctx.tracing

if not tracing then

return

end

local span = root_span_ctx:span()

local metadata = plugin.plugin_metadata(plugin_name)

local plugin_info = metadata.value

if span and not span:is_recording() then

return

end

local inject_conf = {

sampler = {

name = "always_on",

options = conf.sampler.options

},

additional_attributes = conf.additional_attributes,

additional_header_prefix_attributes = conf.additional_header_prefix_attributes

}

-- separate key from the rewrite tracer; modifiedIndex rebuilds it on metadata change

local cache_key = "inject_core_spans#" .. tostring(metadata.modifiedIndex)

local tracer, err = core.lrucache.plugin_ctx(lrucache, api_ctx, cache_key,

create_tracer_obj, inject_conf, plugin_info)

if not tracer then

core.log.error("failed to fetch tracer object: ", err)

return

end

if #tracing.spans == 0 then

return

end

span.start_time = tracing.spans[1].start_time

local root_span = tracing.root_span

local spans = tracing.spans

for _, idx in ipairs(root_span.child_ids or {}) do

create_child_span(tracer, root_span_ctx, spans, spans[idx])

end

end

function _M.log(conf, api_ctx)

if api_ctx.otel_context_token then

-- ctx:detach() is not necessary, because of ctx is stored in ngx.ctx

local resp_source = core.response.get_response_source(api_ctx)

local status_code = ngx.status

-- get span from current context

local ctx = context:current()

local span = ctx:span()

span:set_attributes(attr.string("apisix.response_source", resp_source))

if status_code and status_code >= 500 then

span:set_status(span_status.ERROR,

resp_source .. " error: " .. status_code)

end

inject_core_spans(ctx, api_ctx, conf)

span:set_attributes(attr.int("http.status_code", status_code),

attr.int("http.response.status_code", status_code))

local attributes = {}

if conf.additional_attributes then

inject_attributes(attributes, conf.additional_attributes, api_ctx.var, false)

end

if conf.additional_header_prefix_attributes then

inject_attributes(

attributes,

conf.additional_header_prefix_attributes,

core.request.headers(api_ctx),

true

)

end

for i = 1, #attributes do

span:set_attributes(attributes[i])

end

update_time()

span:finish()

end

end

return _M