> ## Documentation Index
> Fetch the complete documentation index at: https://docs.startree.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# OpenTelemetry Message Decoder

> Decode OpenTelemetry logs, metrics, or traces from realtime streams into Pinot rows.

The OpenTelemetry message decoder converts OpenTelemetry JSON payloads into Pinot rows during realtime ingestion. It supports OpenTelemetry logs, metrics, and traces through a single decoder class selected by the `otel.data.type` decoder property.

```text theme={null}
ai.startree.pinot.plugin.inputformat.otel.OtelMessageDecoder
```

<Info>
  The decoder expects OpenTelemetry JSON payloads shaped like OTLP JSON export messages. It does not run an OTLP collector endpoint by itself; use a stream such as Kafka or another supported realtime source to deliver the JSON payloads to Pinot.
</Info>

## When to Use

Use the OpenTelemetry decoder when:

* Logs, metrics, or traces are already exported as JSON messages into a stream.
* You want Pinot tables to support observability analytics without writing a custom transform pipeline.
* You need OpenTelemetry attributes available as either JSON blobs or promoted Pinot columns.

Use the [Prometheus decoder](/corecapabilities/ingestdata/adv-concepts/realtime/decoders/prometheus) or [Prometheus remote-write decoder](/corecapabilities/ingestdata/adv-concepts/realtime/decoders/prometheus-remote-write) for Prometheus-specific payloads.

## Decoder Property

Set the decoder class and choose one data type:

```json theme={null}
{
  "stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.otel.OtelMessageDecoder",
  "stream.kafka.decoder.prop.otel.data.type": "logs"
}
```

Supported `otel.data.type` values:

| Value     | Input payload              | Extractor                    |
| --------- | -------------------------- | ---------------------------- |
| `logs`    | `resourceLogs` payloads    | OpenTelemetry log records.   |
| `metrics` | `resourceMetrics` payloads | Gauge and sum metric points. |
| `traces`  | `resourceSpans` payloads   | Span records.                |

The value is case-insensitive in practice because it is normalized before matching.

## Log Ingestion

OpenTelemetry log payloads are decoded from `resourceLogs[].scopeLogs[].logRecords[]`.

Recommended schema:

```json theme={null}
{
  "schemaName": "otel_logs",
  "dimensionFieldSpecs": [
    {"name": "severityText", "dataType": "STRING"},
    {"name": "body", "dataType": "STRING"},
    {"name": "traceId", "dataType": "STRING"},
    {"name": "spanId", "dataType": "STRING"},
    {"name": "attributes", "dataType": "JSON"},
    {"name": "resourceAttributes", "dataType": "JSON"},
    {"name": "service.name", "dataType": "STRING"}
  ],
  "metricFieldSpecs": [
    {"name": "severityNumber", "dataType": "INT"}
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "ts",
      "dataType": "TIMESTAMP",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}
```

Realtime table stream config:

```json theme={null}
{
  "streamType": "kafka",
  "stream.kafka.topic.name": "otel-logs",
  "stream.kafka.consumer.type": "lowlevel",
  "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
  "stream.kafka.broker.list": "broker-1:9092,broker-2:9092",
  "stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.otel.OtelMessageDecoder",
  "stream.kafka.decoder.prop.otel.data.type": "logs",
  "stream.kafka.consumer.prop.auto.offset.reset": "smallest"
}
```

Log output fields:

| Field                | Description                                                                                           |
| -------------------- | ----------------------------------------------------------------------------------------------------- |
| `ts`                 | Log timestamp converted from `timeUnixNano`. If missing or unparseable, the decoder uses ingest time. |
| `severityNumber`     | OpenTelemetry severity number.                                                                        |
| `severityText`       | OpenTelemetry severity text.                                                                          |
| `body`               | Log body value converted to string-compatible form.                                                   |
| `traceId`            | Trace ID associated with the log record.                                                              |
| `spanId`             | Span ID associated with the log record.                                                               |
| `attributes`         | Remaining log attributes as JSON.                                                                     |
| `resourceAttributes` | Resource attributes as JSON.                                                                          |
| Additional columns   | If a schema column matches an attribute key, the decoder can promote that attribute into the column.  |

Example query:

```sql theme={null}
SELECT ts, severityText, body, "service.name"
FROM otel_logs
WHERE severityText = 'ERROR'
  AND "service.name" = 'checkout'
ORDER BY ts DESC
LIMIT 100
```

## Metrics Ingestion

OpenTelemetry metric payloads are decoded from `resourceMetrics[].scopeMetrics[].metrics[]`.

Recommended schema:

```json theme={null}
{
  "schemaName": "otel_metrics",
  "dimensionFieldSpecs": [
    {"name": "metric", "dataType": "STRING"},
    {"name": "labels", "dataType": "JSON"},
    {"name": "unit", "dataType": "STRING"},
    {"name": "description", "dataType": "STRING"},
    {"name": "service.name", "dataType": "STRING"}
  ],
  "metricFieldSpecs": [
    {"name": "value", "dataType": "DOUBLE"}
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "ts",
      "dataType": "TIMESTAMP",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}
```

Stream config:

```json theme={null}
{
  "streamType": "kafka",
  "stream.kafka.topic.name": "otel-metrics",
  "stream.kafka.consumer.type": "lowlevel",
  "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
  "stream.kafka.broker.list": "broker-1:9092,broker-2:9092",
  "stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.otel.OtelMessageDecoder",
  "stream.kafka.decoder.prop.otel.data.type": "metrics"
}
```

Metric output fields:

| Field         | Description                                         |
| ------------- | --------------------------------------------------- |
| `metric`      | Metric name.                                        |
| `labels`      | Data point attributes as JSON.                      |
| `value`       | Numeric gauge or sum data point value.              |
| `unit`        | Metric unit, when present.                          |
| `description` | Metric description, when present.                   |
| `ts`          | Data point timestamp converted from `timeUnixNano`. |

Example query:

```sql theme={null}
SELECT DATETRUNC('minute', ts) AS minute, AVG(value)
FROM otel_metrics
WHERE metric = 'cpu.util'
  AND JSON_EXTRACT_SCALAR(labels, '$.host', 'STRING') = 'server-01'
GROUP BY minute
ORDER BY minute
```

## Trace Ingestion

OpenTelemetry trace payloads are decoded from `resourceSpans[].scopeSpans[].spans[]`.

Recommended schema:

```json theme={null}
{
  "schemaName": "otel_traces",
  "dimensionFieldSpecs": [
    {"name": "Name", "dataType": "STRING"},
    {"name": "Traceid", "dataType": "STRING"},
    {"name": "spanid", "dataType": "STRING"},
    {"name": "parent_tid", "dataType": "STRING"},
    {"name": "context", "dataType": "JSON"},
    {"name": "events", "dataType": "JSON"},
    {"name": "service.name", "dataType": "STRING"},
    {"name": "http.method", "dataType": "STRING"},
    {"name": "http.status_code", "dataType": "STRING"}
  ],
  "dateTimeFieldSpecs": [
    {
      "name": "start_time",
      "dataType": "TIMESTAMP",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    },
    {
      "name": "end_time",
      "dataType": "TIMESTAMP",
      "format": "1:MILLISECONDS:EPOCH",
      "granularity": "1:MILLISECONDS"
    }
  ]
}
```

Stream config:

```json theme={null}
{
  "streamType": "kafka",
  "stream.kafka.topic.name": "otel-traces",
  "stream.kafka.consumer.type": "lowlevel",
  "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
  "stream.kafka.broker.list": "broker-1:9092,broker-2:9092",
  "stream.kafka.decoder.class.name": "ai.startree.pinot.plugin.inputformat.otel.OtelMessageDecoder",
  "stream.kafka.decoder.prop.otel.data.type": "traces"
}
```

Trace output fields:

| Field              | Description                                                                                               |
| ------------------ | --------------------------------------------------------------------------------------------------------- |
| `Name`             | Span name.                                                                                                |
| `Traceid`          | Trace ID.                                                                                                 |
| `spanid`           | Span ID.                                                                                                  |
| `parent_tid`       | Parent span ID.                                                                                           |
| `start_time`       | Span start time converted from `startTimeUnixNano`.                                                       |
| `end_time`         | Span end time converted from `endTimeUnixNano`.                                                           |
| `context`          | Span attributes/context as JSON.                                                                          |
| `events`           | Span events as JSON.                                                                                      |
| Additional columns | If a schema column matches a span or resource attribute key, the decoder can promote it into that column. |

Example latency query:

```sql theme={null}
SELECT
  "service.name",
  PERCENTILETDIGEST(
    CAST((ToEpochMillis(end_time) - ToEpochMillis(start_time)) AS DOUBLE),
    95
  ) AS p95_ms
FROM otel_traces
WHERE start_time >= ago('PT1H')
GROUP BY "service.name"
ORDER BY p95_ms DESC
```

## Attribute Modeling

For logs and traces, attribute handling follows two patterns:

* Keep the full attributes object in `attributes`, `resourceAttributes`, or `context` JSON columns for flexible exploration.
* Promote frequently filtered attributes to top-level Pinot columns by adding matching column names to the schema.

Example promoted fields:

```json theme={null}
{
  "dimensionFieldSpecs": [
    {"name": "service.name", "dataType": "STRING"},
    {"name": "deployment.environment", "dataType": "STRING"},
    {"name": "host.name", "dataType": "STRING"}
  ]
}
```

Add JSON indexes when you frequently filter inside JSON attribute columns. Add inverted indexes to promoted low-cardinality attributes.

## Troubleshooting

| Symptom                            | Cause                                                                  | Fix                                                                          |
| ---------------------------------- | ---------------------------------------------------------------------- | ---------------------------------------------------------------------------- |
| Decoder fails during init          | `otel.data.type` is missing or not one of `logs`, `metrics`, `traces`. | Set `stream.<type>.decoder.prop.otel.data.type`.                             |
| Rows have missing timestamps       | OpenTelemetry timestamp field is missing or unparseable.               | Confirm the payload uses OTLP JSON timestamp fields such as `timeUnixNano`.  |
| Attribute filters are slow         | Query filters inside a JSON blob without an index.                     | Promote hot attributes to top-level columns or add a JSON index.             |
| Expected attribute column is null  | The schema column name does not match the OpenTelemetry attribute key. | Match names exactly, including punctuation such as `service.name`.           |
| Only one telemetry type is decoded | One table/decoder instance handles one `otel.data.type`.               | Use separate topics/tables for logs, metrics, and traces, or split upstream. |