tests(opentelemetry): OTel formatted logs

This commit adds unit and integration tests for the OTel-formatted logs
feature.

spec/01-unit/26-observability/05-logs_spec.lua

@@ -0,0 +1,87 @@
+require "kong.tools.utils"
+
+
+describe("Observability/Logs unit tests", function()
+  describe("maybe_push()", function()
+    local o11y_logs, maybe_push, get_request_logs, get_worker_logs
+    local old_ngx, old_kong
+
+    lazy_setup(function()
+      old_ngx = _G.ngx
+      old_kong = _G.kong
+
+      _G.ngx = {
+        config = { subsystem = "http" },
+        ctx = {},
+        DEBUG = ngx.DEBUG,
+        INFO = ngx.INFO,
+        WARN = ngx.WARN,
+      }
+
+      _G.kong = {
+        configuration = {
+          log_level = "info",
+        },
+      }
+
+      o11y_logs = require "kong.observability.logs"
+      maybe_push = o11y_logs.maybe_push
+      get_request_logs = o11y_logs.get_request_logs
+      get_worker_logs = o11y_logs.get_worker_logs
+    end)
+
+    before_each(function()
+      _G.ngx.ctx = {}
+    end)
+
+    lazy_teardown(function()
+      _G.ngx = old_ngx
+      _G.kong = old_kong
+    end)
+
+    it("has no effect when no log line is provided", function()
+      maybe_push(1, ngx.INFO)
+      local worker_logs = get_worker_logs()
+      assert.same({}, worker_logs)
+      local request_logs = get_request_logs()
+      assert.same({}, request_logs)
+    end)
+
+    it("has no effect when log line is empty", function()
+      maybe_push(1, ngx.INFO, "")
+      local worker_logs = get_worker_logs()
+      assert.same({}, worker_logs)
+      local request_logs = get_request_logs()
+      assert.same({}, request_logs)
+    end)
+
+    it("has no effect when log level is lower than the configured value", function()
+      maybe_push(1, ngx.DEBUG, "Don't mind me, I'm just a debug log")
+      local worker_logs = get_worker_logs()
+      assert.same({}, worker_logs)
+      local request_logs = get_request_logs()
+      assert.same({}, request_logs)
+    end)
+
+    it("generates worker-scoped log entries", function()
+      local log_level = ngx.WARN
+      local body = "Careful! I'm a warning!"
+
+      maybe_push(1, log_level, body, true, 123, ngx.null, nil, function()end, { foo = "bar" })
+      local worker_logs = get_worker_logs()
+      assert.equals(1, #worker_logs)
+
+      local logged_entry = worker_logs[1]
+      assert.same(log_level, logged_entry.log_level)
+      assert.matches(body .. "true123nilnilfunction:%s0x%x+table:%s0x%x+", logged_entry.body)
+      assert.is_table(logged_entry.attributes)
+      assert.is_number(logged_entry.attributes["introspection.current.line"])
+      assert.is_string(logged_entry.attributes["introspection.name"])
+      assert.is_string(logged_entry.attributes["introspection.namewhat"])
+      assert.is_string(logged_entry.attributes["introspection.source"])
+      assert.is_string(logged_entry.attributes["introspection.what"])
+      assert.is_number(logged_entry.observed_time_unix_nano)
+      assert.is_number(logged_entry.time_unix_nano)
+    end)
+  end)
+end)

spec/02-integration/14-observability/05-logs_spec.lua

@@ -0,0 +1,86 @@
+local helpers = require "spec.helpers"
+
+for _, strategy in helpers.each_strategy() do
+  describe("Observability Logs", function ()
+    describe("ngx.log patch", function()
+      local proxy_client
+      local post_function_access = [[
+        local threads = {}
+        local n_threads = 100
+
+        for i = 1, n_threads do
+          threads[i] = ngx.thread.spawn(function()
+            ngx.log(ngx.INFO, "thread_" .. i .. " logged")
+          end)
+        end
+
+        for i = 1, n_threads do
+          ngx.thread.wait(threads[i])
+        end
+      ]]
+
+      lazy_setup(function()
+        local bp = helpers.get_db_utils(strategy, {
+          "routes",
+          "services",
+          "plugins",
+        })
+
+        local http_srv = assert(bp.services:insert {
+          name = "mock-service",
+          host = helpers.mock_upstream_host,
+          port = helpers.mock_upstream_port,
+        })
+
+        local logs_route = assert(bp.routes:insert({ service = http_srv,
+                                                     protocols = { "http" },
+                                                     paths = { "/logs" }}))
+
+        assert(bp.plugins:insert({
+          name = "post-function",
+          route = logs_route,
+          config = {
+            access = { post_function_access },
+          },
+        }))
+
+        -- only needed to enable the log collection hook
+        assert(bp.plugins:insert({
+          name = "opentelemetry",
+          route = logs_route,
+          config = {
+            logs_endpoint = "http://" .. helpers.mock_upstream_host .. ":" .. helpers.mock_upstream_port,
+          }
+        }))
+
+        helpers.start_kong({
+          database = strategy,
+          nginx_conf = "spec/fixtures/custom_nginx.template",
+          plugins = "opentelemetry,post-function",
+        })
+        proxy_client = helpers.proxy_client()
+      end)
+
+      lazy_teardown(function()
+        if proxy_client then
+          proxy_client:close()
+        end
+        helpers.stop_kong()
+      end)
+
+      it("does not produce yielding and concurrent executions", function ()
+        local res = assert(proxy_client:send {
+          method  = "GET",
+          path    = "/logs",
+        })
+        assert.res_status(200, res)
+
+        -- plugin produced logs:
+        assert.logfile().has.line("thread_1 logged", true, 10)
+        assert.logfile().has.line("thread_100 logged", true, 10)
+        -- plugin did not produce concurrent accesses to ngx.log:
+        assert.logfile().has.no.line("[error]", true)
+      end)
+    end)
+  end)
+end

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

@@ -43,6 +43,14 @@ local pb_decode_span = function(data)
   return pb.decode("opentelemetry.proto.trace.v1.Span", data)
 end
 
+local pb_encode_log = function(data)
+  return pb.encode("opentelemetry.proto.logs.v1.LogRecord", data)
+end
+
+local pb_decode_log = function(data)
+  return pb.decode("opentelemetry.proto.logs.v1.LogRecord", data)
+end
+
 describe("Plugin: opentelemetry (otlp)", function()
   local old_ngx_get_phase
 
@@ -66,7 +74,7 @@ describe("Plugin: opentelemetry (otlp)", function()
     ngx.ctx.KONG_SPANS = nil
   end)
 
-  it("encode/decode pb", function ()
+  it("encode/decode pb (traces)", function ()
     local N = 10000
 
     local test_spans = {
@@ -125,6 +133,41 @@ describe("Plugin: opentelemetry (otlp)", function()
     end
   end)
 
+  it("encode/decode pb (logs)", function ()
+    local N = 10000
+
+    local test_logs = {}
+
+    for _ = 1, N do
+      local now_ns = time_ns()
+
+      local log = {
+        time_unix_nano = now_ns,
+        observed_time_unix_nano = now_ns,
+        log_level = ngx.INFO,
+        span_id = rand_bytes(8),
+        body = "log line",
+        attributes = {
+          foo = "bar",
+          test = true,
+          version = 0.1,
+        },
+      }
+      insert(test_logs, log)
+    end
+
+    local trace_id = rand_bytes(16)
+    local flags = tonumber(rand_bytes(1))
+    local prepared_logs = otlp.prepare_logs(test_logs, trace_id, flags)
+
+    for _, prepared_log in ipairs(prepared_logs) do
+      local decoded_log = pb_decode_log(pb_encode_log(prepared_log))
+
+      local ok, err = table_compare(prepared_log, decoded_log)
+      assert.is_true(ok, err)
+    end
+  end)
+
   it("check lengths of trace_id and span_id ", function ()
     local TRACE_ID_LEN, PARENT_SPAN_ID_LEN = 16, 8
     local default_span = {