feat: add more spans to opentelemetry plugin (#12686)

Signed-off-by: Nic <[email protected]>
Co-authored-by: Ashish Tiwari <[email protected]>
Co-authored-by: AlinsRan <[email protected]>

Author: 3 people

apisix/cli/config.lua

@@ -84,7 +84,8 @@ local _M = {
         neg_ttl = 60,
         neg_count = 512
       }
-    }
+    },
+    tracing = false
   },
   nginx_config = {
     error_log = "logs/error.log",

apisix/core/response.lua

@@ -19,6 +19,7 @@
 --
 -- @module core.response
 
+local tracer = require("apisix.tracer")
 local encode_json = require("cjson.safe").encode
 local ngx = ngx
 local arg = ngx.arg
@@ -86,6 +87,9 @@ function resp_exit(code, ...)
     end
 
     if code then
+        if code >= 400 then
+            tracer.finish_all(ngx.ctx, tracer.status.ERROR, "response code " .. code)
+        end
         return ngx_exit(code)
     end
 end

apisix/init.lua

@@ -47,6 +47,8 @@ local debug           = require("apisix.debug")
 local pubsub_kafka    = require("apisix.pubsub.kafka")
 local resource        = require("apisix.resource")
 local trusted_addresses_util = require("apisix.utils.trusted-addresses")
+local tracer          = require("apisix.tracer")
+
 local discovery = require("apisix.discovery.init").discovery
 local ngx             = ngx
 local get_method      = ngx.req.get_method
@@ -202,6 +204,9 @@ function _M.ssl_client_hello_phase()
     local ngx_ctx = ngx.ctx
     local api_ctx = core.tablepool.fetch("api_ctx", 0, 32)
     ngx_ctx.api_ctx = api_ctx
+    api_ctx.ngx_ctx = ngx_ctx
+
+    local span = tracer.start(ngx_ctx, "ssl_client_hello_phase", tracer.kind.server)
 
     local ok, err = router.router_ssl.match_and_set(api_ctx, true, sni)
 
@@ -215,18 +220,23 @@ function _M.ssl_client_hello_phase()
             core.log.error("failed to fetch ssl config: ", err)
         end
         core.log.error("failed to match any SSL certificate by SNI: ", sni)
+        span:set_status(tracer.status.ERROR, "no matched SSL")
+        span:finish(ngx_ctx)
         ngx_exit(-1)
     end
 
     ok, err = apisix_ssl.set_protocols_by_clienthello(ngx_ctx.matched_ssl.value.ssl_protocols)
     if not ok then
         core.log.error("failed to set ssl protocols: ", err)
+        span:set_status(tracer.status.ERROR, "failed set protocols")
+        span:finish(ngx_ctx)
         ngx_exit(-1)
     end
 
     -- in stream subsystem, ngx.ssl.server_name() return hostname of ssl session in preread phase,
     -- so that we can't get real SNI without recording it in ngx.ctx during client_hello phase
     ngx.ctx.client_hello_sni = sni
+    span:finish(ngx_ctx)
 end
 
 
@@ -480,7 +490,6 @@ local function common_phase(phase_name)
 end
 
 
-
 function _M.handle_upstream(api_ctx, route, enable_websocket)
     -- some plugins(ai-proxy...) request upstream by http client directly
     if api_ctx.bypass_nginx_upstream then
@@ -677,9 +686,12 @@ function _M.http_access_phase()
     -- always fetch table from the table pool, we don't need a reused api_ctx
     local api_ctx = core.tablepool.fetch("api_ctx", 0, 32)
     ngx_ctx.api_ctx = api_ctx
+    api_ctx.ngx_ctx = ngx_ctx
 
     core.ctx.set_vars_meta(api_ctx)
 
+    local span = tracer.start(ngx_ctx, "apisix.phase.access", tracer.kind.server)
+
     if not verify_https_client(api_ctx) then
         return core.response.exit(400)
     end
@@ -717,10 +729,13 @@ function _M.http_access_phase()
 
     handle_x_forwarded_headers(api_ctx)
 
+    local match_span = tracer.start(ngx_ctx, "http_router_match", tracer.kind.internal)
     router.router_http.match(api_ctx)
 
     local route = api_ctx.matched_route
     if not route then
+        match_span:set_status(tracer.status.ERROR, "no matched route")
+        match_span:finish(ngx.ctx)
         -- run global rule when there is no matching route
         local global_rules, conf_version = apisix_global_rules.global_rules()
         plugin.run_global_rules(api_ctx, global_rules, conf_version, nil)
@@ -729,6 +744,7 @@ function _M.http_access_phase()
         return core.response.exit(404,
                     {error_msg = "404 Route Not Found"})
     end
+    match_span:finish(ngx_ctx)
 
     core.log.info("matched route: ",
                   core.json.delay_encode(api_ctx.matched_route, true))
@@ -785,7 +801,6 @@ function _M.http_access_phase()
     else
         local plugins = plugin.filter(api_ctx, route)
         api_ctx.plugins = plugins
-
         plugin.run_plugin("rewrite", plugins, api_ctx)
         if api_ctx.consumer then
             local changed
@@ -821,6 +836,7 @@ function _M.http_access_phase()
         end
         plugin.run_plugin("access", plugins, api_ctx)
     end
+    span:finish(ngx_ctx)
 
     _M.handle_upstream(api_ctx, route, enable_websocket)
 
@@ -879,6 +895,8 @@ end
 
 
 function _M.http_header_filter_phase()
+    local ngx_ctx = ngx.ctx
+    local span = tracer.start(ngx_ctx, "apisix.phase.header_filter", tracer.kind.server)
     core.response.set_header("Server", ver_header)
 
     local up_status = get_var("upstream_status")
@@ -901,6 +919,9 @@ function _M.http_header_filter_phase()
         end
         core.response.set_header("Apisix-Plugins", core.table.concat(deduplicate, ", "))
     end
+    span:finish(ngx_ctx)
+
+    tracer.start(ngx_ctx, "apisix.phase.body_filter", tracer.kind.server)
 end
 
 
@@ -1056,6 +1077,7 @@ function _M.http_log_phase()
     if not api_ctx then
         return
     end
+    tracer.finish_all(api_ctx.ngx_ctx)
 
     if not api_ctx.var.apisix_upstream_response_time or
     api_ctx.var.apisix_upstream_response_time == "" then
@@ -1081,6 +1103,9 @@ function _M.http_log_phase()
         core.tablepool.release("matched_route_record", api_ctx.curr_req_matched)
     end
 
+    tracer.release(api_ctx.ngx_ctx)
+    api_ctx.ngx_ctx = nil
+
     core.tablepool.release("api_ctx", api_ctx)
 end
 

apisix/plugin.lua

@@ -38,6 +38,7 @@ local tostring      = tostring
 local error         = error
 local getmetatable  = getmetatable
 local setmetatable  = setmetatable
+local tracer    = require("apisix.tracer")
 -- make linter happy to avoid error: getting the Lua global "load"
 -- luacheck: globals load, ignore lua_load
 local lua_load          = load
@@ -1228,7 +1229,10 @@ function _M.run_plugin(phase, plugins, api_ctx)
             plugin_run = true
             run_meta_pre_function(conf, api_ctx, plugins[i]["name"])
             api_ctx._plugin_name = plugins[i]["name"]
+            local span = tracer.start(api_ctx.ngx_ctx, "apisix.phase." .. phase
+                                        .. ".plugins." .. api_ctx._plugin_name)
             phase_func(conf, api_ctx)
+            span:finish(api_ctx.ngx_ctx)
             api_ctx._plugin_name = nil
         end
     end
@@ -1301,6 +1305,7 @@ end
 
 function _M.run_global_rules(api_ctx, global_rules, conf_version, phase_name)
     if global_rules and #global_rules > 0 then
+        local span = tracer.start(api_ctx.ngx_ctx, "run_global_rules", tracer.kind.internal)
         local orig_conf_type = api_ctx.conf_type
         local orig_conf_version = api_ctx.conf_version
         local orig_conf_id = api_ctx.conf_id
@@ -1335,6 +1340,7 @@ function _M.run_global_rules(api_ctx, global_rules, conf_version, phase_name)
         api_ctx.conf_type = orig_conf_type
         api_ctx.conf_version = orig_conf_version
         api_ctx.conf_id = orig_conf_id
+        span:finish(api_ctx.ngx_ctx)
     end
 end
 

apisix/plugins/opentelemetry.lua

@@ -48,6 +48,7 @@ local pairs   = pairs
 local ipairs  = ipairs
 local unpack  = unpack
 local string_format = string.format
+local update_time = ngx.update_time
 
 local lrucache = core.lrucache.new({
     type = 'plugin', count = 128, ttl = 24 * 60 * 60,
@@ -327,10 +328,17 @@ function _M.rewrite(conf, api_ctx)
 
     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("uri.path", vars.uri),
+        attr.string("user_agent.original", vars.http_user_agent),
     }
 
     if api_ctx.curr_req_matched then
@@ -376,48 +384,96 @@ function _M.rewrite(conf, api_ctx)
       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
 
 
-function _M.delayed_body_filter(conf, api_ctx)
-    if api_ctx.otel_context_token and ngx.arg[2] then
-        local ctx = context:current()
-        ctx:detach(api_ctx.otel_context_token)
-        api_ctx.otel_context_token = nil
+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
 
-        -- get span from current context
-        local span = ctx:span()
-        local upstream_status = core.response.get_upstream_status(api_ctx)
-        if upstream_status and upstream_status >= 500 then
-            span:set_status(span_status.ERROR,
-                            "upstream response status: " .. upstream_status)
-        end
 
-        span:set_attributes(attr.int("http.status_code", upstream_status))
+local function inject_core_spans(root_span_ctx, api_ctx, conf)
+    local tracing = api_ctx.ngx_ctx.tracing
+    if not tracing then
+        return
+    end
 
-        span:finish()
+    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
+    }
+    local tracer, err = core.lrucache.plugin_ctx(lrucache, api_ctx, nil,
+                                                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
 
 
--- body_filter maybe not called because of empty http body response
--- so we need to check if the span has finished in log phase
 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 upstream_status = core.response.get_upstream_status(api_ctx)
 
         -- get span from current context
-        local span = context:current():span()
+        local ctx = context:current()
+        local span = ctx:span()
         if upstream_status and upstream_status >= 500 then
             span:set_status(span_status.ERROR,
                     "upstream response status: " .. upstream_status)
         end
 
+        inject_core_spans(ctx, api_ctx, conf)
+        span:set_attributes(attr.int("http.status_code", upstream_status),
+                            attr.int("http.response.status_code", upstream_status))
+        update_time()
         span:finish()
     end
 end

apisix/secret.lua

@@ -18,6 +18,7 @@
 local require   = require
 local core      = require("apisix.core")
 local string    = require("apisix.core.string")
+local tracer    = require("apisix.tracer")
 
 local local_conf = require("apisix.core.config_local").local_conf()
 
@@ -28,6 +29,7 @@ local byte      = string.byte
 local type      = type
 local pcall     = pcall
 local pairs     = pairs
+local ngx       = ngx
 
 local _M = {}
 
@@ -148,16 +150,20 @@ local function fetch_by_uri_secret(secret_uri)
         return nil, "no secret conf, secret_uri: " .. secret_uri
     end
 
+    local span = tracer.start(ngx.ctx, "fetch_secret", tracer.kind.client)
     local ok, sm = pcall(require, "apisix.secret." .. opts.manager)
     if not ok then
         return nil, "no secret manager: " .. opts.manager
     end
 
     local value, err = sm.get(conf, opts.key)
     if err then
+        span:set_status(tracer.status.ERROR, err)
+        span:finish(ngx.ctx)
         return nil, err
     end
 
+    span:finish(ngx.ctx)
     return value
 end