`opentelemetry` sink batching results in malformed data (status code 400) when `max_events` is more than `1`

[navodveduth]

A note for the community

• Please vote on this issue by adding a 👍 reaction to the original issue to help the community and maintainers prioritize this request
• If you are interested in working on this issue or have submitted a pull request, please leave a comment

Problem

When using the opentelemetry sink in vector to send metrics derived from logs to an opentelemetry collector, vector repeatedly fails with 400 bad request. These errors appear in the vector agent logs, but the otel collector does not show any related error logs or indications of receiving malformed payloads. As a result, metrics are not processed by the otel collector as expected

Configuration

apiVersion: observability.kaasops.io/v1alpha1
kind: ClusterVectorPipeline
metadata:
  name: log-level-metrics-pipeline
spec:
  sources:
    kubernetes_logs:
      type: kubernetes_logs
      pod_annotation_fields:
        container_image: container_image
        container_name: container_name
        pod_name: pod_name
        pod_namespace: pod_namespace
      fingerprint_lines: 1
      ignore_older_secs: 600

  transforms:
    log_level_tagger:
      type: remap
      inputs:
        - kubernetes_logs
      source: |
        if exists(.message) {
          log_message = string!(.message)
          log_level = "INFO"

          if contains(upcase(log_message), "ERROR") {
            log_level = "ERROR"
          } else if contains(upcase(log_message), "WARN") {
            log_level = "WARN"
          } else if contains(upcase(log_message), "DEBUG") {
            log_level = "DEBUG"
          }

          .log_level = log_level

          .attributes = {
            "log_level": log_level
          }

          if exists(.pod_name) {
            .attributes.pod_name = string!(.pod_name)
          } else {
            .attributes.pod_name = "unknown_pod"
          }

          if exists(.pod_namespace) {
            .attributes.pod_namespace = string!(.pod_namespace)
          } else {
            .attributes.pod_namespace = "unknown_namespace"
          }

          .timestamp = now()
        } else {
          .log_level = "UNKNOWN"
          .attributes = {
            "log_level": "UNKNOWN",
            "pod_name": "unknown_pod",
            "pod_namespace": "unknown_namespace"
          }
        }

    log_to_metric:
      type: log_to_metric
      inputs:
        - log_level_tagger
      metrics:
        - type: counter
          name: log_level_count
          field: log_level
          tags:
            log_level: "{{attributes.log_level}}"
            pod_name: "{{attributes.pod_name}}"
            pod_namespace: "{{attributes.pod_namespace}}"

  sinks:
    otel_collector_sink:
      type: opentelemetry
      inputs:
        - log_to_metric
      protocol:
        type: http
        uri: "http://otel-collector.otel:4318/v1/logs"
        method: post
        encoding:
          codec: json
          framing:
            method: newline_delimited
      batch:
        max_events: 100
        max_bytes: 1048576
        timeout_secs: 10
      retry:
        initial_interval_secs: 1
        max_interval_secs: 30
        max_retries: 5
      healthcheck:
        enabled: true
        interval_secs: 60

Version

0.43.0

Debug Output

2024-12-18T14:40:25.520217Z ERROR sink{component_kind="sink" component_id=log-level-metrics-pipeline-otel_collector_sink component_type=opentelemetry}:request{request_id=456}: vector::sinks::util::retries: Not retriable; dropping the request. reason="Http status: 400 Bad Request" internal_log_rate_limit=true
2024-12-18T14:40:25.520229Z ERROR sink{component_kind="sink" component_id=log-level-metrics-pipeline-otel_collector_sink component_type=opentelemetry}:request{request_id=456}: vector_common::internal_event::service: Internal log [Service call failed. No retries or retries exhausted.] has been suppressed 4 times.
2024-12-18T14:40:25.520231Z ERROR sink{component_kind="sink" component_id=log-level-metrics-pipeline-otel_collector_sink component_type=opentelemetry}:request{request_id=456}: vector_common::internal_event::service: Service call failed. No retries or retries exhausted. error=None request_id=456 error_type="request_failed" stage="sending" internal_log_rate_limit=true
2024-12-18T14:40:25.520266Z ERROR sink{component_kind="sink" component_id=log-level-metrics-pipeline-otel_collector_sink component_type=opentelemetry}:request{request_id=456}: vector_common::internal_event::component_events_dropped: Internal log [Events dropped] has been suppressed 4 times.
2024-12-18T14:40:25.520268Z ERROR sink{component_kind="sink" component_id=log-level-metrics-pipeline-otel_collector_sink component_type=opentelemetry}:request{request_id=456}: vector_common::internal_event::component_events_dropped: Events dropped intentional=false count=2 reason="Service call failed. No retries or retries exhausted." internal_log_rate_limit=true
2024-12-18T14:40:26.554810Z ERROR sink{component_kind="sink" component_id=log-level-metrics-pipeline-otel_collector_sink component_type=opentelemetry}:request{request_id=457}: vector::sinks::util::retries: Internal log [Not retriable; dropping the request.] is being suppressed to avoid flooding.
2024-12-18T14:40:26.554830Z ERROR sink{component_kind="sink" component_id=log-level-metrics-pipeline-otel_collector_sink component_type=opentelemetry}:request{request_id=457}: vector_common::internal_event::service: Internal log [Service call failed. No retries or retries exhausted.] is being suppressed to avoid flooding.
2024-12-18T14:40:26.554840Z ERROR sink{component_kind="sink" component_id=log-level-metrics-pipeline-otel_collector_sink component_type=opentelemetry}:request{request_id=457}: vector_common::internal_event::component_events_dropped: Internal log [Events dropped] is being suppressed to avoid flooding.
2024-12-18T14:40:43.994791Z ERROR sink{component_kind="sink" component_id=log-level-metrics-pipeline-otel_collector_sink component_type=opentelemetry}:request{request_id=459}: vector::sinks::util::retries: Internal log [Not retriable; dropping the request.] has been suppressed 2 times.
2024-12-18T14:40:43.994821Z ERROR sink{component_kind="sink" component_id=log-level-metrics-pipeline-otel_collector_sink component_type=opentelemetry}:request{request_id=459}: vector::sinks::util::retries: Not retriable; dropping the request. reason="Http status: 400 Bad Request" internal_log_rate_limit=true
2024-12-18T14:40:43.994858Z ERROR sink{component_kind="sink" component_id=log-level-metrics-pipeline-otel_collector_sink component_type=opentelemetry}:request{request_id=459}: vector_common::internal_event::service: Internal log [Service call failed. No retries or retries exhausted.] has been suppressed 2 times.

Example Data

No response

Additional Context

Both vector and the otel collector are running in a cluster. Even with debug logging enabled on the otel collector, there are no logs showing that it received the payload or encountered any issues. However, when the same payload is sent to the otel collector using a curl request, it is logged and processed correctly

OpenTelemetry collector config:

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318
processors:
  batch: {}
  memory_limiter:
    limit_mib: 1000
    spike_limit_mib: 512
    check_interval: 5s
extensions:
  zpages: {}
exporters:
  logging:
    loglevel: debug
    sampling_initial: 5
    sampling_thereafter: 200
  prometheus:
    endpoint: 0.0.0.0:8889
    metric_expiration: 1m
service:
  extensions: [zpages]
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [logging]
    metrics:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [logging, prometheus]
    logs:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [logging, file]

References

No response

[jszwedko]

Hi @navodveduth ,

The opentelemetry sink currently only supports logs as input and not metrics. Metrics support is being tracked by #17310. I'll close this out, but let me know if I'm misunderstanding this issue. Feel free to follow the other issue for metrics support.

[pront]

Hmm, there is a bug here. This should delegate to underlying protocol, essentially this.

In this case, with this encoding:

encoding:
  codec: json
  framing:
    method: newline_delimited

the underlying HTTP sink should be able to handle it. So I suspect this has to do with the event format produced by the log_to_metric transform.

I would recommend vector tap to inspect the event format and then manually sending this event to your OTEL endpoint and observe if it's ingested or not. You could use curl or a Python script.

[seilerre]

I got the same problem, but only sending logs:

Error

2025-04-09T15:12:25.367398Z ERROR sink{component_kind="sink" component_id=otel component_type=opentelemetry}:request{request_id=25487}: vector::sinks::util::retries: Not retriable; dropping the request. reason="Http status: 400 Bad Request" internal_log_rate_limit=true
2025-04-09T15:12:25.367414Z ERROR sink{component_kind="sink" component_id=otel component_type=opentelemetry}:request{request_id=25487}: vector_common::internal_event::service: Internal log [Service call failed. No retries or retries exhausted.] has been suppressed 499 times.
2025-04-09T15:12:25.367419Z ERROR sink{component_kind="sink" component_id=otel component_type=opentelemetry}:request{request_id=25487}: vector_common::internal_event::service: Service call failed. No retries or retries exhausted. error=None request_id=25487 error_type="request_failed" stage="sending" internal_log_rate_limit=true
2025-04-09T15:12:25.367432Z ERROR sink{component_kind="sink" component_id=otel component_type=opentelemetry}:request{request_id=25487}: vector_common::internal_event::component_events_dropped: Internal log [Events dropped] has been suppressed 499 times.
2025-04-09T15:12:25.367435Z ERROR sink{component_kind="sink" component_id=otel component_type=opentelemetry}:request{request_id=25487}: vector_common::internal_event::component_events_dropped: Events dropped intentional=false count=1 reason="Service call failed. No retries or retries exhausted." internal_log_rate_limit=true
2025-04-09T15:12:25.367654Z ERROR sink{component_kind="sink" component_id=otel component_type=opentelemetry}:request{request_id=25489}: vector::sinks::util::retries: Internal log [Not retriable; dropping the request.] is being suppressed to avoid flooding.
2025-04-09T15:12:25.367667Z ERROR sink{component_kind="sink" component_id=otel component_type=opentelemetry}:request{request_id=25489}: vector_common::internal_event::service: Internal log [Service call failed. No retries or retries exhausted.] is being suppressed to avoid flooding.
2025-04-09T15:12:25.367675Z ERROR sink{component_kind="sink" component_id=otel component_type=opentelemetry}:request{request_id=25489}: vector_common::internal_event::component_events_dropped: Internal log [Events dropped] is being suppressed to avoid flooding.

Open Telemetry is happily receiving "some" logs and not logging ANY errors:

{"level":"info","ts":"2025-04-09T15:09:40.702Z","msg":"Logs","kind":"exporter","data_type":"logs","name":"debug","resource logs":51,"log records":51}
{"level":"info","ts":"2025-04-09T15:09:50.702Z","msg":"Logs","kind":"exporter","data_type":"logs","name":"debug","resource logs":100,"log records":100}
{"level":"info","ts":"2025-04-09T15:10:10.704Z","msg":"Logs","kind":"exporter","data_type":"logs","name":"debug","resource logs":51,"log records":51}
{"level":"info","ts":"2025-04-09T15:10:20.704Z","msg":"Logs","kind":"exporter","data_type":"logs","name":"debug","resource logs":100,"log records":100}
{"level":"info","ts":"2025-04-09T15:10:30.705Z","msg":"Logs","kind":"exporter","data_type":"logs","name":"debug","resource logs":1,"log records":1}

My vector config:

sources:
  docker:
    # Reads logs from Docker containers via the Docker socket
    type: docker_logs
    # endpoint: "unix:///var/run/docker.sock"
    # include_containers:
    #   - "my_container_name"
    exclude_containers:
      - otel-otel-collector
transforms:
  enrich:
      type: remap
      inputs: ["docker"]
      source: |
        # Log ID
        .logId = uuid_v4()

        # Severity
        .severity = "UNKNOWN"
        if contains(string!(.message), "info", case_sensitive: false) {
          .severity = "INFO"
          .severityNumber = 9
        } else if contains(string!(.message), "trace", case_sensitive: false) {
          .severity = "TRACE"
          .severityNumber = 1
        } else if contains(string!(.message), "debug", case_sensitive: false) {
          .severity = "DEBUG"
          .severityNumber = 5
        } else if contains(string!(.message), "warn", case_sensitive: false) {
          .severity = "WARN"
          .severityNumber = 13
        } else if contains(string!(.message), "error", case_sensitive: false) {
          .severity = "ERROR"
          .severityNumber = 17
        } else if contains(string!(.message), "fatal", case_sensitive: false) {
          .severity = "FATAL"
          .severityNumber = 21
        }
  otel_remap:
    # https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/logs/data-model.md
    # https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/protocol/file-exporter.md#examples
    inputs: ["enrich"]
    type: "remap"
    source: |
      .resourceLogs = [
        {
          "resource": {
            "attributes": [
              { "key": "host.hostname", "value": { "stringValue": .host.hostname } },
              { "key": "service.name", "value": { "stringValue": .container_name} },
              { "key": "source_type", "value": { "stringValue": .source_type } }
            ]
          },
          "scopeLogs": [
            {
              "scope": {},
              "logRecords": [
                {
                  "timeUnixNano": to_unix_timestamp!(.timestamp, unit: "nanoseconds"),
                  "body": { "stringValue": .message },
                  "severityText": .severity,
                  "severityNumber": .severityNumber,
                  "attributes": [
                    { "key": "container_created_at", "value": { "stringValue": to_string!(.container_created_at) } },
                    { "key": "log_id", "value": { "stringValue": .logId } },
                    { "key": "container_id", "value": { "stringValue": .container_id } },
                    { "key": "container_name", "value": { "stringValue": .container_name } },
                    { "key": "host", "value": { "stringValue": "${HOST_HOSTNAME}" } },
                    { "key": "image", "value": { "stringValue": .image } },
                    # Optionally include other fields like .stream, .message, etc.
                    { "key": "stream", "value": { "stringValue": .stream } }
                  ]
                }
              ]
            }
          ]
        }
      ]
      .
sinks:
  # console:
  #   inputs:
  #     - otel_remap
  #   target: stdout
  #   type: console
  #   encoding:
  #     codec: json
  otel:
    type: opentelemetry
    inputs:
      - otel_remap
    protocol:
      type: http
      uri: "http://otel-collector:4318/v1/logs"
      method: post
      batch:
        # TODO: Change this to more after https://github.com/vectordotdev/vector/issues/22188
        max_events: 1
      encoding:
        codec: json
      framing:
        method: newline_delimited
      request:
        headers:
          content-type: application/json

My Compose:

services:
  otel-collector:
    image: otel/opentelemetry-collector-contrib:0.123.0
    restart: always
    user: "0"
    env_file:
      - .env
    environment:
      - OTEL_LOG_LEVEL=debug
    volumes:
      - ./otel-collector-config.yaml:/etc/otelcol-contrib/config.yaml
      - /var/lib/docker/containers:/var/lib/docker/containers:ro
    ports:
      - 1888:1888 # pprof extension
      - 8888:8888 # Prometheus metrics exposed by the Collector
      - 8889:8889 # Prometheus exporter metrics
      - 13133:13133 # health_check extension
      - 4317:4317 # OTLP gRPC receiver
      - 4318:4318 # OTLP http receiver
      - 55679:55679 # zpages extension
    deploy:
      resources:
        limits:
          cpus: '1.00'      # maximum 50% of a single CPU
          memory: 500M      # maximum 500 MB memory usage
        reservations:
          cpus: '0.30'      # reserve 25% of a single CPU
          memory: 300M      # reserve 250 MB memory
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"

  vector:
    depends_on:
      - otel-collector
    image: timberio/vector:0.46.0-alpine
    restart: always
    user: "0"
    environment:
      - VECTOR_LOG=info
      - HOST_HOSTNAME=CHANGE
    volumes:
      # Mount Docker socket read-only so Vector can read logs directly
      - /var/run/docker.sock:/var/run/docker.sock:ro
      # Mount your Vector config file
      - ./vector.yaml:/etc/vector/vector.yaml:ro
    command: ["--config", "/etc/vector/vector.yaml"]
    # Optionally expose or publish ports for Vector's own internal metrics
    # ports:
    #   - 8686:8686
    deploy:
      resources:
        limits:
          cpus: '1.00'      # maximum 50% of a single CPU
          memory: 300M      # maximum 500 MB memory usage
        reservations:
          cpus: '0.30'      # reserve 25% of a single CPU
          memory: 300M      # reserve 250 MB memory
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"

My OTEL Config:

receivers:
  otlp:
    protocols:
      http:
        endpoint: 0.0.0.0:4318
processors:
  batch:
    timeout: 5s
    send_batch_size: 100
exporters:
  debug:
    verbosity: detailed
  kafka:
    brokers: ["redacted"] # Replace with your Kafka broker(s)
    topic: redacted
    encoding: otlp_proto
    protocol_version: "3.8.0" # Adjust according to your Kafka setup
    retry_on_failure:
      enabled: true
    sending_queue:
      queue_size: 200
    producer:
      compression: lz4
    auth:
      sasl:
        mechanism: "SCRAM-SHA-512"  # Choose "PLAIN", "SCRAM-SHA-256", or "SCRAM-SHA-512" depending on your Kafka setup
        username: "${OTEL_KAFKA_USERNAME}"
        password: "${OTEL_KAFKA_PASSWORD}"
      tls:
        insecure: false
        insecure_skip_verify: true

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [batch]
      exporters: [kafka]

[pront]

This has been discussed in other issues but currently the opentelemetry sink doesn't support batching multiple events. The following is a temporary workaround:

      batch:
        max_events: 1

Note that a single Vector event can contain multiple OTEL logs per:

vector/lib/opentelemetry-proto/src/proto/opentelemetry-proto/opentelemetry/proto/logs/v1/logs.proto

Lines 47 to 61 in be2dde4

	// A collection of ScopeLogs from a Resource.
	message ResourceLogs {
	reserved 1000;
	// The resource for the logs in this message.
	// If this field is not set then resource info is unknown.
	opentelemetry.proto.resource.v1.Resource resource = 1;
	// A list of ScopeLogs that originate from a resource.
	repeated ScopeLogs scope_logs = 2;
	// This schema_url applies to the data in the "resource" field. It does not apply
	// to the data in the "scope_logs" field which have their own schema_url field.
	string schema_url = 3;
	}