otlp_http_json_to_event.rb

# Convert an OTLP/HTTP JSON request into a regular Logstash event.
#
# OTLP JSON represents attributes as arrays of {"key", "value"} objects and
# wraps every attribute value in a typed object such as {"stringValue", "x"}.
# This script removes that encoding noise: attributes become hashes and typed
# values become normal Ruby values. The OTLP request hierarchy is retained, so
# it works for logs, metrics, and traces.
#
# The OTLP request must be supplied in the required `source` script parameter.
# The source can contain either JSON text or an already-decoded map. The
# converted map replaces the event root:
#
# filter {
#   ruby {
#     path => "/etc/logstash/otlp_http_json_to_event.rb"
#     script_params => { "source" => "[otlp][payload]" }
#   }
# }

require "json"

OTLP_VALUE_KEYS = [
  "stringValue", "boolValue", "intValue", "doubleValue", "bytesValue",
  "arrayValue", "kvlistValue"
].freeze

def register(params)
  @source = params.fetch("source")
  raise ArgumentError, "source must be a non-empty field reference" unless @source.is_a?(String) && !@source.empty?
end

def filter(event)
  payload = event.get(@source)

  raise ArgumentError, "OTLP payload was not found in #{@source.inspect}" if payload.nil?

  payload = JSON.parse(payload) if payload.is_a?(String)
  unless payload.is_a?(Hash)
    raise ArgumentError, "OTLP payload must be a JSON object, got #{payload.class}"
  end

  converted = convert_otlp_value(payload)

  # Replace the HTTP input event rather than nesting the converted request in
  # another field. Keep Logstash's standard event metadata intact.
  standard_fields = {}
  ["@timestamp", "@version"].each do |field|
    value = event.get(field)
    standard_fields[field] = value unless value.nil?
  end
  event.to_hash.keys.each { |field| event.remove(field) }
  converted.each { |field, value| event.set(field, value) }
  standard_fields.each { |field, value| event.set(field, value) unless converted.key?(field) }

  [event]
end

def convert_otlp_value(value)
  case value
  when Array
    value.map { |item| convert_otlp_value(item) }
  when Hash
    return unwrap_otlp_value(value) if otlp_typed_value?(value)

    value.each_with_object({}) do |(key, item), converted|
      converted[key] = key == "attributes" ? convert_otlp_attributes(item) : convert_otlp_value(item)
    end
  else
    value
  end
end

def otlp_typed_value?(value)
  value.keys.length == 1 && OTLP_VALUE_KEYS.include?(value.keys.first)
end

def unwrap_otlp_value(value)
  key = value.keys.first
  raw_value = value[key]

  case key
  when "intValue"
    Integer(raw_value)
  when "doubleValue"
    Float(raw_value)
  when "arrayValue"
    convert_otlp_value(raw_value.fetch("values", []))
  when "kvlistValue"
    convert_otlp_attributes(raw_value.fetch("values", []))
  else
    raw_value
  end
end

def convert_otlp_attributes(attributes)
  return convert_otlp_value(attributes) unless attributes.is_a?(Array)

  # Do not turn arbitrary arrays into maps. Only OTLP KeyValue arrays are
  # collapsed, which also makes malformed payloads pass through predictably.
  return convert_otlp_value(attributes) unless attributes.all? do |attribute|
    attribute.is_a?(Hash) && attribute.keys.sort == ["key", "value"]
  end

  attributes.each_with_object({}) do |attribute, converted|
    converted[attribute["key"]] = convert_otlp_value(attribute["value"])
  end
end

test "converts an OTLP HTTP JSON request from the configured source" do
  parameters do
    { "source" => "otlp" }
  end

  in_event do
    {
      "otlp" => '{"resourceLogs":[{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"checkout"}},{"key":"service.instance.id","value":{"intValue":"42"}}]},"scopeLogs":[{"logRecords":[{"body":{"stringValue":"payment received"},"attributes":[{"key":"http.status_code","value":{"intValue":"200"}},{"key":"successful","value":{"boolValue":true}}]}]}]}]}'
    }
  end

  expect("expands the source to the root as normal maps and scalar values") do |events|
    record = events.first.get("resourceLogs").first.fetch("scopeLogs").first.fetch("logRecords").first
    resource_attributes = events.first.get("resourceLogs").first.fetch("resource").fetch("attributes")

    events.size == 1 &&
      events.first.get("otlp").nil? &&
      resource_attributes == { "service.name" => "checkout", "service.instance.id" => 42 } &&
      record.fetch("body") == "payment received" &&
      record.fetch("attributes") == { "http.status_code" => 200, "successful" => true }
  end
end

test "unwraps nested array and key-value-list attribute values" do
  parameters do
    { "source" => "otlp" }
  end

  in_event do
    {
      "otlp" => {
        "attributes" => [
          {
            "key" => "request",
            "value" => {
              "kvlistValue" => {
                "values" => [
                  { "key" => "attempt", "value" => { "intValue" => "3" } },
                  { "key" => "regions", "value" => { "arrayValue" => { "values" => [{ "stringValue" => "pt" }, { "stringValue" => "us" }] } } }
                ]
              }
            }
          }
        ]
      }
    }
  end

  expect("converts nested OTLP collection values") do |events|
    events.size == 1 && events.first.get("attributes") == {
      "request" => { "attempt" => 3, "regions" => ["pt", "us"] }
    }
  end
end