# Load OpenTelemetry agent traces into MotherDuck
> Land agent spans in S3 with the OpenTelemetry Collector, load new files with a scheduled Flight, and roll them up into one row per agent run.
This guide builds the pipeline described in the [agent observability overview](/getting-started/agent-observability/). The OpenTelemetry Collector writes your agents' spans to S3 as OTLP JSON files. A Flight loads the new files every five minutes into a `spans` table and rebuilds the recent rows of an `agent_runs` table, with one row per agent run. You're done when the check query at the end shows spans from the last hour.

## Prerequisites

- Agents instrumented with an OpenTelemetry SDK that records the [generative AI semantic conventions](https://opentelemetry.io/docs/specs/semconv/gen-ai/) and exports OTLP to a Collector.
- The OpenTelemetry Collector Contrib distribution (`otelcol-contrib`), which includes the AWS S3 exporter, running with AWS credentials that can write to your bucket.
- An S3 bucket and a prefix for the trace files, such as `s3://<bucket>/otel/traces`.
- An [S3 secret in MotherDuck](/integrations/cloud-storage/amazon-s3/) that can read that prefix, created by the user or service account that will own the Flight.
- A plan with [scheduled Flights](/concepts/flights/#availability-and-plan-limits).
- For the optional labeling step, access to [`prompt_jev`](/sql-reference/motherduck-sql-reference/ai-functions/prompt-jev/), which is in preview.

## Export spans to S3

Add the AWS S3 exporter to the Collector's traces pipeline. The `batch` processor groups spans into one file per batch, and `marshaler: otlp_json` writes each file as OTLP JSON:

```yaml
receivers:
  otlp:
    protocols:
      grpc:
      http:
processors:
  batch:
    send_batch_size: 8192
    timeout: 10s
exporters:
  awss3:
    s3uploader:
      region: <aws_region>
      s3_bucket: <bucket>
      s3_prefix: otel/traces
    marshaler: otlp_json
service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch]
      exporters: [awss3]
```

If the Collector already exports to a tracing tool, add `awss3` to that pipeline's `exporters` list instead of replacing it.

Restart the Collector and send some agent traffic. The exporter writes files partitioned by time, such as `otel/traces/year=2026/month=11/day=03/hour=15/minute=30/traces_<id>.json`. List them from MotherDuck to confirm the files arrive and the secret can read them:

```sql
SELECT file
FROM glob('s3://<bucket>/otel/traces/**/*.json')
LIMIT 5;
```

## Create the tables

Create a database for the traces, and three tables in it: `spans` holds one row per span, `agent_runs` one row per trace, and `loaded_files` records which files the Flight has already loaded.

```sql
CREATE DATABASE IF NOT EXISTS agent_traces;
USE agent_traces;

CREATE TABLE IF NOT EXISTS spans (
    trace_id       VARCHAR,
    span_id        VARCHAR,
    parent_span_id VARCHAR,
    span_name      VARCHAR,
    service_name   VARCHAR,
    start_time     TIMESTAMPTZ,
    end_time       TIMESTAMPTZ,
    status_code    VARCHAR,
    status_message VARCHAR,
    operation      VARCHAR,
    agent_name     VARCHAR,
    model          VARCHAR,
    tool_name      VARCHAR,
    input_tokens   BIGINT,
    output_tokens  BIGINT,
    error_type     VARCHAR,
    attributes     MAP(VARCHAR, VARCHAR)
);

CREATE TABLE IF NOT EXISTS agent_runs (
    trace_id      VARCHAR,
    started_at    TIMESTAMPTZ,
    ended_at      TIMESTAMPTZ,
    agent_name    VARCHAR,
    model_calls   BIGINT,
    tool_calls    BIGINT,
    error_spans   BIGINT,
    input_tokens  BIGINT,
    output_tokens BIGINT
);

CREATE TABLE IF NOT EXISTS loaded_files (
    file      VARCHAR,
    loaded_at TIMESTAMPTZ
);
```

`spans` keeps every attribute in the `attributes` map and copies the ones you filter on most into columns of their own. To filter on another attribute often, add a column and fill it from the map.

## Create the loading Flight

Each run of this Flight does the following in one transaction:

1. Lists the files in yesterday's and today's folders that aren't in `loaded_files`. The exporter partitions by time, so older folders hold nothing new.
2. Flattens those files into one row per span. OTLP JSON stores times as nanoseconds since the epoch, span status as a number, and attributes as a list of key-value pairs.
3. Skips spans that are already loaded, and keeps one copy of a span that appears in two files, both of which a Collector retry can cause.
4. Records the files in `loaded_files`.
5. Rebuilds `agent_runs` for traces whose root span started in the last three hours, so a run is complete even when its spans arrived in different files.

A run that fails loads nothing, and the next run picks up the same files. The `DATABASE` and `LANDING_PREFIX` config values keep your names out of the Python source:

```sql
SELECT flight_id, flight_name, current_version
FROM MD_CREATE_FLIGHT(
    name := 'load_agent_traces',
    config := MAP {
        'DATABASE': 'agent_traces',
        'LANDING_PREFIX': 's3://<bucket>/otel/traces'
    },
    requirements_txt := 'duckdb==1.5.6',
    source_code := $flight$
import os
from datetime import datetime, timedelta, timezone

import duckdb

LOAD_SPANS = """
INSERT INTO spans BY NAME
WITH resource_spans AS (
    SELECT unnest(json_extract(json, '$.resourceSpans')::JSON[]) AS rs
    FROM read_ndjson_objects($files, maximum_object_size = 268435456)
),
scope_spans AS (
    SELECT rs, unnest(json_extract(rs, '$.scopeSpans')::JSON[]) AS ss
    FROM resource_spans
),
parsed AS (
    SELECT
        span ->> 'traceId' AS trace_id,
        span ->> 'spanId' AS span_id,
        nullif(span ->> 'parentSpanId', '') AS parent_span_id,
        span ->> 'name' AS span_name,
        list_filter(
            json_extract(rs, '$.resource.attributes')::JSON[],
            lambda a: (a ->> 'key') = 'service.name'
        )[1] -> 'value' ->> 'stringValue' AS service_name,
        make_timestamp(((span ->> 'startTimeUnixNano')::HUGEINT // 1000)::BIGINT) AT TIME ZONE 'UTC' AS start_time,
        make_timestamp(((span ->> 'endTimeUnixNano')::HUGEINT // 1000)::BIGINT) AT TIME ZONE 'UTC' AS end_time,
        CASE (span -> 'status' ->> 'code')::INTEGER
            WHEN 1 THEN 'OK'
            WHEN 2 THEN 'ERROR'
            ELSE 'UNSET'
        END AS status_code,
        span -> 'status' ->> 'message' AS status_message,
        map_from_entries([
            {
                k: a ->> 'key',
                v: coalesce(
                    a -> 'value' ->> 'stringValue',
                    a -> 'value' ->> 'intValue',
                    a -> 'value' ->> 'doubleValue',
                    a -> 'value' ->> 'boolValue'
                )
            }
            FOR a IN json_extract(span, '$.attributes')::JSON[]
        ]) AS attributes
    FROM (SELECT rs, unnest(json_extract(ss, '$.spans')::JSON[]) AS span FROM scope_spans)
)
SELECT
    *,
    attributes['gen_ai.operation.name'] AS operation,
    attributes['gen_ai.agent.name'] AS agent_name,
    attributes['gen_ai.request.model'] AS model,
    attributes['gen_ai.tool.name'] AS tool_name,
    attributes['gen_ai.usage.input_tokens']::BIGINT AS input_tokens,
    attributes['gen_ai.usage.output_tokens']::BIGINT AS output_tokens,
    attributes['error.type'] AS error_type
FROM parsed
ANTI JOIN (
    SELECT trace_id, span_id FROM spans WHERE start_time >= now() - INTERVAL 1 DAY
) AS loaded USING (trace_id, span_id)
QUALIFY row_number() OVER (PARTITION BY trace_id, span_id) = 1
"""

ROLLUP_RUNS = """
DELETE FROM agent_runs WHERE started_at >= now() - INTERVAL 3 HOUR;

INSERT INTO agent_runs
SELECT
    trace_id,
    min(start_time) AS started_at,
    max(end_time) AS ended_at,
    max(agent_name) FILTER (WHERE parent_span_id IS NULL) AS agent_name,
    count(*) FILTER (WHERE operation = 'chat') AS model_calls,
    count(*) FILTER (WHERE operation = 'execute_tool') AS tool_calls,
    count(*) FILTER (WHERE status_code = 'ERROR') AS error_spans,
    sum(input_tokens) AS input_tokens,
    sum(output_tokens) AS output_tokens
FROM spans
WHERE trace_id IN (
    SELECT trace_id
    FROM spans
    WHERE parent_span_id IS NULL
        AND start_time >= now() - INTERVAL 3 HOUR
)
GROUP BY trace_id;
"""

def main():
    con = duckdb.connect(f"md:{os.environ['DATABASE']}")
    prefix = os.environ["LANDING_PREFIX"].rstrip("/")

    # The exporter partitions files by time, so yesterday's and today's
    # folders hold every file that can still be new.
    today = datetime.now(timezone.utc)
    days = [today - timedelta(days=1), today]
    patterns = [f"{prefix}/year={d:%Y}/month={d:%m}/day={d:%d}/**/*.json" for d in days]
    new_files = """
        SELECT file FROM glob(?) UNION SELECT file FROM glob(?)
        EXCEPT SELECT file FROM loaded_files
        ORDER BY file
    """
    files = [row[0] for row in con.execute(new_files, patterns).fetchall()]
    if not files:
        print("no new trace files")
        return

    con.execute("BEGIN TRANSACTION")
    con.execute(LOAD_SPANS, {"files": files})
    con.execute("INSERT INTO loaded_files SELECT unnest($files), now()", {"files": files})
    con.execute(ROLLUP_RUNS)
    con.execute("COMMIT")
    print(f"loaded {len(files)} trace files")

if __name__ == "__main__":
    main()
$flight$
);
```

Pin `duckdb` to the version MotherDuck runs, as described in [packages and runtime](/key-tasks/flights/packages-and-runtime/). The Flight runs as the account that creates it, so create it as a [service account](/key-tasks/flights/flights-authentication-config-and-secrets/#service-accounts-for-production-flights) for a pipeline that shouldn't depend on one person, and create the S3 secret as that account too.

## Run it and schedule it

Store the Flight ID in a variable, then trigger one run by hand:

```sql
SET VARIABLE trace_flight_id = (
    SELECT flight_id
    FROM MD_LIST_FLIGHTS()
    WHERE flight_name = 'load_agent_traces'
    ORDER BY created_at DESC
    LIMIT 1
);

SELECT * FROM MD_RUN_FLIGHT(flight_id := getvariable('trace_flight_id'));
```

Runs are asynchronous. Check the run history until the run finishes:

```sql
SELECT run_number, status, created_at
FROM MD_LIST_FLIGHT_RUNS(flight_id := getvariable('trace_flight_id'))
ORDER BY run_number DESC
LIMIT 5;
```

When the run succeeds, its log ends with `loaded <n> trace files`. Then schedule the Flight every five minutes:

```sql
CALL MD_UPDATE_FLIGHT(
    flight_id := getvariable('trace_flight_id'),
    schedule_cron := '*/5 * * * *'
);
```

A shorter interval keeps the tables fresher, and each run loads fewer files at a time. See [scheduling and runs](/key-tasks/flights/scheduling-and-runs/) for cron syntax.

## Check the pipeline

After a few scheduled runs, check that files keep loading and that runs are assembled:

```sql
SELECT
    (SELECT max(loaded_at) FROM loaded_files) AS last_load,
    (SELECT max(end_time) FROM spans) AS latest_span,
    (SELECT count(*) FROM spans WHERE start_time >= now() - INTERVAL 1 HOUR) AS spans_last_hour,
    (SELECT count(*) FROM agent_runs WHERE started_at >= now() - INTERVAL 1 HOUR) AS runs_last_hour,
    (SELECT count(*) FROM agent_runs WHERE error_spans > 0 AND started_at >= now() - INTERVAL 1 HOUR) AS runs_with_errors;
```

`last_load` should be at most five minutes old, `latest_span` close behind it, and `spans_last_hour` and `runs_last_hour` above zero while your agents are running.

## Label failed tool calls with prompt_jev

Error messages are free text, so you can't group by them. To label each failed tool call with its cause, create a table for the labels:

```sql
CREATE TABLE IF NOT EXISTS span_failure_causes (
    trace_id   VARCHAR,
    span_id    VARCHAR,
    cause      VARCHAR,
    confidence DOUBLE
);
```

Then classify the failures that aren't labeled yet. Add this statement to `ROLLUP_RUNS` in the Flight to label new failures as they land, or run it on its own:

```sql
INSERT INTO span_failure_causes
SELECT
    trace_id,
    span_id,
    (classified.verdict).choice AS cause,
    (classified.verdict).confidence AS confidence
FROM (
    SELECT
        trace_id,
        span_id,
        prompt_jev(
            status_message,
            'What caused this tool call to fail?',
            choice := ['timeout', 'invalid_arguments', 'auth', 'rate_limited', 'upstream_error']
        ) AS verdict
    FROM spans
    ANTI JOIN span_failure_causes USING (trace_id, span_id)
    WHERE operation = 'execute_tool'
        AND status_code = 'ERROR'
) AS classified;
```

`prompt_jev` is billed in [AI Units](/about-motherduck/billing/pricing#ai-function-pricing) by input tokens, so the `ANTI JOIN` keeps each failure from being classified twice. Its requests are processed by TypeSafe in the United States, so check what your error messages contain before you send them.

## Troubleshooting

| Symptom | Fix |
|---|---|
| The `glob` query returns no files | Check `s3_prefix` and the partition layout under it. With `compression: gzip` on the exporter, the files end in `.json.gz`: change both patterns in the Flight to `*.json.gz`. |
| An access error on S3 | The S3 secret must belong to the account that owns the Flight, and its `SCOPE` must cover the prefix. |
| A JSON object size error | A batch file is larger than `maximum_object_size`. Raise it, or lower `send_batch_size` in the Collector. |
| `spans` fills, but attribute columns stay `NULL` | Your instrumentation uses other attribute names. Look at the keys in `attributes` and change the column mapping at the end of `LOAD_SPANS`. |

## Related

- [Agent observability overview](/getting-started/agent-observability/)
- [Ingest S3 Parquet files on a schedule](/key-tasks/flights/ingest-s3-parquet-files-on-a-schedule/)
- [Loading data best practices](/key-tasks/loading-data-into-motherduck/considerations-for-loading-data/)
- [`prompt_jev` reference](/sql-reference/motherduck-sql-reference/ai-functions/prompt-jev/)


---

## Docs feedback

MotherDuck accepts optional user-submitted feedback about this page at `GET https://motherduck.com/docs/api/feedback/agent`.
For agents and automated tools, feedback submission should be user-confirmed before sending.

URL-encode query parameter values and send a GET request:

```text
GET https://motherduck.com/docs/api/feedback/agent?page_path=%2Fkey-tasks%2Fai-and-motherduck%2Fload-opentelemetry-agent-traces%2F&page_title=Load%20OpenTelemetry%20agent%20traces%20into%20MotherDuck&text=<url-encoded user feedback, max 2000 characters>
```

Optionally append `&source=<url-encoded interface identifier>` such as `claude.ai` or `chatgpt`.

`page_path` and `text` are required; `page_title` and `source` are optional. Responses: `200 {"feedback_id": "<uuid>"}`, `400` for malformed query parameters, and `429` when rate-limited.