Manual rules/remaps for batching/buffering?

[target-san]

Question

Hello!

Vector's default implementation for opentelmetry/http sinks automatically batches JSON records into JSON array, which doesn't seem to be correct OTLP protocol. The best I could do is disable framing completely and limit max events per send to 1. Is it possible to somehow specify how batching should be performed? Or maybe there are some examples how to do manual batching via "reduce" transformer? The best I can guess is to remap events, then reduce them, then remap again into OTLP payload, then send via raw HTTP.

Thanks

P.S: Two VRL files mentioned in remap serve to first transform Rust's tracing JSON logs into more convenient form and then transform that form into OTLP payloads. Adding them just for completeness, though they don't affect question directly

Vector Config

config.yaml:

sources:
  backend_logs:
    type: file
    include: ["backend.log"]
    # Used for debug only
    ignore_checkpoints: true

transforms:
  backend_service_name:
    type: remap
    inputs:
      - backend_logs
    source: |
      .service_name = "backend"
  backend_remap:
    type: remap
    inputs:
      - backend_service_name
    files:
      - rust-logs.vrl
      - otlp-payload.vrl

sinks:
  backend_openobserve:
    type: http
    inputs:
      - backend_remap
    uri: "https://api.eu1.honeycomb.io/v1/logs"
    method: post
    encoding:
      codec: json
    headers:
      Content-Type: "application/json"
      X-Honeycomb-Team: <API key>
      X-Honeycomb-Dataset: "backend"
    batch:
      max_events: 1
      timeout_secs: 1
    framing:
      method: bytes
    request:
      timeout_secs: 30  # backend_openobserve:

rust-logs.vrl

# Converts Rust's tracing JSON output into format ingestible by OTLP payload former

tstamp = ""
message = ""
level = "INFO"
service_name = null
attributes = {}

# Keep service name if it was assigned before parsing
if is_string(.service_name) {
    service_name = .service_name
}

# Parse JSON message first (if needed)
if is_string(.message) {
    . = parse_json!(string!(.message))
}

# Fetch timestamp from JSON, it's mandatory
tstamp = .timestamp

# Handle message. It's either on top level or inside "fields"

if exists(.message) {
    message = to_string!(.message)
}

# Flatten "fields" into attributes - except `.fields.message` which replaces global message

if exists(.fields) && is_object(.fields) {
    if exists(.fields.message) {
        message = to_string!(.fields.message)
        del(.fields.message)
    }
    # Merge all fields from .fields into "attributes"
    attributes = merge!(attributes, .fields)
}

# Process spans - collect from "span" object and "spans" array
spans_to_process = []

# Collect span from "span" object field
if exists(.span) && is_object(.span) {
    spans_to_process = push(spans_to_process, .span)
}

# Collect spans from "spans" array field
if exists(.spans) && is_array(.spans) {
    for_each(array!(.spans)) -> |_index, span_item| {
        if is_object(span_item) {
            spans_to_process = push(spans_to_process, span_item)
        }
    }
}

# Process each collected span
for_each(spans_to_process) -> |_index, span_obj| {
    # Get span name (default to "unknown" if missing)
    span_name = get(span_obj, ["name"]) ?? "unknown"
    
    # Process each field in the span object
    for_each(object!(span_obj)) -> |field_name, field_value| {
        # Skip the "name" field since we use it as prefix
        if field_name != "name" {
            # Create field name: {span_name}_{field_name}
            new_field_name = join!([span_name, field_name], "_")
            # Assign to root object
            attributes = set!(attributes, [new_field_name], field_value)
        }
    }
}

# 3. Rename code location fields
if exists(.target) {
    attributes.code_target = .target
}

if exists(.filename) {
    attributes.code_file = .filename
}

if exists(.line_number) {
    attributes.code_line = .line_number
}

# Form and return final object
. = {
    "timestamp": tstamp,
    "message": message,
    "level": level,
    "attributes": attributes
}

if is_string(service_name) {
    .service_name = service_name
}

.

otlp-payload.vrl:

# This VRL expects following incoming JSON format:
# * `timestamp` - timestamp in `%Y-%M-%DT%h:%m:%s.%ms%Z` format
# * `message` - log message, goes into body
# * `level` - log level, `ERROR`/`WARN`/`INFO`/`DEBUG`/`TRACE`
# * `service_name` - goes into resource.attributes."service.name" if present
# * `attributes` - object with event's custom attributes; all are stringified ATM

# OTLP's severity table
otlp_severity = {
  "TRACE": 1,
  "TRACE2": 2,
  "TRACE3": 3,
  "TRACE4": 4,
  "DEBUG": 5,
  "DEBUG2": 6,
  "DEBUG3": 7,
  "DEBUG4": 8,
  "INFO": 9,
  "INFO2": 10,
  "INFO3": 11,
  "INFO4": 12,
  "WARN": 13,
  "WARN2": 14,
  "WARN3": 15,
  "WARN4": 16,
  "ERROR": 17,
  "ERROR2": 18,
  "ERROR3": 19,
  "ERROR4": 20,
  "FATAL": 21,
  "FATAL2": 22,
  "FATAL3": 23,
  "FATAL4": 24,
}

# Severity text and number from level
# Default level is info
severity_text = "INFO"
severity_number = 9

if exists(.level) {
    severity_text = string!(.level)

    level_num, err = get(otlp_severity, [severity_text])
    # Keep unknown levels at "INFO"
    if err == null {
        severity_number = level_num
    }
}

# Compose attributes into OTEL format
attributes = []

for_each(object!(.attributes)) -> |field_name, field_value| {
    attributes = push(attributes, {
        "key": field_name,
        "value": { "stringValue": to_string!(field_value) }
    })
}

tstamp = parse_timestamp!(.timestamp, "%+")

# Compose final event
event = {
    "scopeLogs": [{
        "logRecords": [{
            "timeUnixNano": to_unix_timestamp!(tstamp, unit: "nanoseconds"),
            "body": { "stringValue": string!(.message) },
            "severityText": severity_text,
            "severityNumber": severity_number,
            "attributes": attributes
        }]
    }]
}

# Add service name if set
if exists(.service_name) && is_string(.service_name) {
    event.resource = {
        "attributes": [
            { "key": "service.name", "value": { "stringValue": .service_name } },
        ]
    }
}

. = {
    "resourceLogs": [event]
}

.

Vector Logs

No response