Spans of a trace arrive over a few seconds: a session window per trace partition merges them into one record when the trace goes quiet.

Tracing: an SDK sends one event for each step of a request (the HTTP call, the database query, the cache lookup), and retries when it gets no answer. A trace is finished when no more events arrive, and what you want to keep is one record per trace. The same shape counts visits, sessions and anything else that ends when it goes quiet.

Each trace is a partition, and each event’s transactionId is the trace and the step, so a retried event is stored once. A stream with a session window closes a trace after two seconds of quiet, adds it up, and pushes one message per finished trace.

The SDK sends the events of trace-7f3a to the partition trace-7f3a: http 120 ms, db 45 ms, cache 3 ms, and a retried db event, which is a duplicate and stored once. A stream with a session window of 2 seconds closes the trace after 2 seconds of quiet, adds it up and pushes one message to the queue traces: 3 spans, 168 ms. — Events merged into one record per trace by a session window.

// A trace is complete when it has been quiet for 2 seconds.
const merger = await Stream
  .from(queen.queue('trace-events'))
  .windowSession({ gap: 2, idleFlushMs: 500 })
  .reduce((t, e) => ({ spans: t.spans + 1, ms: t.ms + e.ms }), { spans: 0, ms: 0 })
  .to(queen.queue('traces'))                      // one message per finished trace
  .run({ queryId: 'trace-merger', url, subscriptionMode: 'all' })

// Every SDK call is one event, in the partition of its trace.
const event = (span, ms) => ({ transactionId: `trace-7f3a:${span}`, data: { span, ms } })
await queen.queue('trace-events').partition('trace-7f3a').push([
  event('http', 120), event('db', 45), event('cache', 3),
])
await queen.queue('trace-events').partition('trace-7f3a')
  .push([event('db', 45)])                        // a retried call: stored once

# A trace is complete when it has been quiet for 2 seconds.
merger = await (
    Stream.from_(queen.queue("trace-events"))
    .window_session(gap=2, idle_flush_ms=500)
    .reduce(lambda t, e: {"spans": t["spans"] + 1, "ms": t["ms"] + e["ms"]},
            {"spans": 0, "ms": 0})
    .to(queen.queue("traces"))                      # one message per finished trace
    .run(query_id="trace-merger", url=url, subscription_mode="all")
)

# Every SDK call is one event, in the partition of its trace.
def event(span, ms):
    return {"transactionId": f"trace-7f3a:{span}", "data": {"span": span, "ms": ms}}

await queen.queue("trace-events").partition("trace-7f3a").push([
    event("http", 120), event("db", 45), event("cache", 3),
])
await queen.queue("trace-events").partition("trace-7f3a").push(
    [event("db", 45)]                               # a retried call: stored once
)

// A trace is complete when it has been quiet for 2 seconds.
merger, err := streams.
	From(client.Queue("trace-events").AsStreamSource()).
	WindowSession(2, streams.WithIdleFlushMs(500)).
	Reduce(func(acc, value interface{}) (interface{}, error) {
		t, e := acc.(map[string]interface{}), value.(map[string]interface{})
		return map[string]interface{}{
			"spans": t["spans"].(float64) + 1,
			"ms":    t["ms"].(float64) + e["ms"].(float64),
		}, nil
	}, map[string]interface{}{"spans": 0.0, "ms": 0.0}).
	To(client.Queue("traces")). // one message per finished trace
	Run(ctx, streams.RunOptions{QueryID: "trace-merger", URL: url,
		SubscriptionMode: queen.SubscriptionModeAll})
if err != nil {
	log.Fatal(err)
}

// Every SDK call is one event, in the partition of its trace.
// (Go's push sets the transactionId of its first message only: one push per event.)
event := func(span string, ms int) error {
	_, err := client.Queue("trace-events").Partition("trace-7f3a").
		Push(map[string]interface{}{"span": span, "ms": ms}).
		TransactionID("trace-7f3a:" + span).
		Execute(ctx)
	return err
}
if err := errors.Join(event("http", 120), event("db", 45), event("cache", 3)); err != nil {
	log.Fatal(err)
}
if err := event("db", 45); err != nil { // a retried call: stored once
	log.Fatal(err)
}

// A trace is complete when it has been quiet for 2 seconds.
let merger = Stream::from(queen.queue("trace-events"))
    .window_session(2)
    .idle_flush_ms(500)
    .reduce(json!({ "spans": 0, "ms": 0 }), |t, e| {
        json!({
            "spans": t["spans"].as_i64().unwrap_or(0) + 1,
            "ms": t["ms"].as_i64().unwrap_or(0) + e.field("ms").as_i64().unwrap_or(0),
        })
    })
    .to(queen.queue("traces"))              // one message per finished trace
    .run(&queen, RunOptions::new("trace-merger").subscription_mode(SubscriptionMode::All))
    .await?;

// Every SDK call is one event, in the partition of its trace.
let event = |span: &str, ms: i64| {
    PushItem::new("trace-events", json!({ "span": span, "ms": ms }))
        .partition("trace-7f3a")
        .transaction_id(format!("trace-7f3a:{span}"))
};
let events = queen.queue("trace-events");
events.push_items(vec![event("http", 120), event("db", 45), event("cache", 3)]).await?;
events.push_items(vec![event("db", 45)]).await?; // a retried call: stored once

Two seconds after the last event, the traces queue held one message: 3 spans and 168 ms. The retried database event was counted once.

Why

every trace is its own ordered stream, so “this partition went quiet” is exactly “this trace is finished”. The open traces live in the broker, so a restart of the merger loses nothing.

With many traces, delete the empty trace partitions quickly (Many short-lived partitions), and copy the finished traces to a bucket (the S3 sink). One thing to know: a trace that pauses longer than the gap becomes two records, and the reader merges them.

Built from: Push, Streams.

Windowed aggregation: spans merged into traces