You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/09/07 01:41:45 UTC
(skywalking-ai-sessionizer) branch main updated: Derive Claude Code's own token metric from the landed files, and push it
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-ai-sessionizer.git
The following commit(s) were added to refs/heads/main by this push:
new c7cf92c Derive Claude Code's own token metric from the landed files, and push it
c7cf92c is described below
commit c7cf92c630cbe5060f89f6ae0881450c49560a6a
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Mon Sep 7 09:41:36 2026 +0800
Derive Claude Code's own token metric from the landed files, and push it
With metrics: true on the claude-code-local adapter, the collector derives
the runtime's own metric family from what it landed, name for name and
attribute for attribute with Claude Code's OpenTelemetry exporter, so a
receiver holds one family whichever produced it. Phase one is the overlap,
claude_code.token.usage: type, model, query_source main or subagent, and
session.id, as monotonic delta sums per minute, the way the exporter's SDK
aggregates over its interval. A call is counted once from the usage its
fragments repeat, and once across a session's files: the runtime re-emits
records before a context reset and a landed file is cut at a budget, so
one call can sit in two files, and the exporter counted one API call once.
What a transcript does not carry is not derived and never estimated.
The points wait in the storage root's _metrics/ spool, one write-once file
per landed file with points, put under a sequence the spool's own state
and lock hand out. asz push sends the spool over the metrics service, on
both transports, under the same budget and once-only rule as the files,
and normalises the resource on the way out to asz's identity so the OAP
sees one service for the runtime. metrics_lookback, 24h unless set and
written in days when wanted, bounds the first derivation over a root with
history; later passes derive every new file whole. The viewer's refresh
derives too.
Every scenario now derives at each stage, as the collector does, and its
push property checks the points on the wire against the plan: every call's
usage once, under the stream that made it, and the shape the export page
states. The Collector job carries a metrics pipeline and checks the tokens
that arrive. Generated scenario roots turn the flag on with no look-back.
---
CLAUDE.md | 3 +-
asz.yaml | 9 +
cmd/asz/main.go | 49 +++-
cmd/asz/push.go | 10 +-
cmd/asz/refresh.go | 19 ++
collectorcheck | Bin 0 -> 4741170 bytes
docs/en/adapters/claude-code.md | 18 ++
docs/en/changes/changes-0.3.0.md | 11 +
docs/en/setup/command-line.md | 10 +-
docs/en/setup/configuration.md | 2 +
docs/en/setup/export-otlp.md | 27 ++
internal/config/config.go | 44 +++
internal/export/otlp/client.go | 107 ++++++--
internal/export/otlp/otlp_test.go | 56 ++++
internal/export/otlp/otlptest/receiver.go | 99 +++++--
internal/export/otlp/push.go | 107 +++++++-
internal/metrics/metrics.go | 441 ++++++++++++++++++++++++++++++
internal/metrics/metrics_test.go | 229 ++++++++++++++++
internal/scenario/build.go | 4 +
internal/scenario/expect/expect.go | 1 +
internal/scenario/run/push.go | 100 ++++++-
internal/scenario/run/run.go | 59 +++-
internal/storage/spool.go | 159 +++++++++++
tests/boundary/boundary_test.go | 2 +-
tools/collectorcheck/main.go | 57 +++-
tools/e2e-collector.sh | 3 +
26 files changed, 1564 insertions(+), 62 deletions(-)
diff --git a/CLAUDE.md b/CLAUDE.md
index aeea5c2..e520adb 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -63,7 +63,8 @@ internal/storage/ landing zone, atomic writes, cursors, session/in
internal/verify/ contiguity and digest checks over landed data
internal/view/ reads a conversation and serves it as a page
internal/view/conversation-view/ Horizon's conversation renderer, built from the pinned commit in HORIZON_COMMIT; never edited here
-internal/export/otlp/ sends landed files and rounds to an OpenTelemetry logs receiver, one record per file
+internal/export/otlp/ sends landed files and rounds to an OpenTelemetry logs receiver, one record per file, and the metrics spool
+internal/metrics/ derives the runtime's own metric family from landed Session Data into the spool; collector side
internal/repack/ re-cuts landed files into a new root under another budget
internal/adapters/claudecode/ the claude-code-local adapter
internal/adapters/mock/ the mock dialect: Session Data a scenario writes directly
diff --git a/asz.yaml b/asz.yaml
index cce858d..002690c 100644
--- a/asz.yaml
+++ b/asz.yaml
@@ -53,6 +53,15 @@ adapters:
exclude:
- /private/tmp/**
+ # The runtime's own metric family, derived from the landed files and
+ # sent by asz push: claude_code.token.usage, name for name and
+ # attribute for attribute with Claude Code's OpenTelemetry exporter, so
+ # a receiver holds one family whichever produced it. The first
+ # derivation over a root with history reaches back metrics_lookback,
+ # 24h unless set; 0 or none means everything.
+ metrics: false
+ metrics_lookback: 24h
+
collector:
# watch polls the source continuously. once makes a single pass and
# exits, which is the backfill path over history that already exists.
diff --git a/cmd/asz/main.go b/cmd/asz/main.go
index 256ca5c..f6676e4 100644
--- a/cmd/asz/main.go
+++ b/cmd/asz/main.go
@@ -32,6 +32,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
"github.com/apache/skywalking-ai-sessionizer/internal/config"
"github.com/apache/skywalking-ai-sessionizer/internal/index"
+ "github.com/apache/skywalking-ai-sessionizer/internal/metrics"
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
"github.com/apache/skywalking-ai-sessionizer/internal/verify"
"github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
@@ -648,19 +649,44 @@ func cmdCollect(cfg *config.Config, ad config.Adapter, once bool) error {
}
col := claudecode.New(root, storage.NewZone(zoneRoot), ad.Collector.MaxDeltaBytes)
match := claudecode.NewMatcher(ad.Include, ad.Exclude).Match
+ deriver, err := newDeriver(storage.NewZone(zoneRoot), ad)
+ if err != nil {
+ return err
+ }
fmt.Printf("source root : %s\nstorage root: %s\n", root, zoneRoot)
+ if deriver != nil {
+ fmt.Printf("metrics : %s, derived from the landed files, look-back %s on the first pass\n", metrics.TokenUsage, lookbackWord(deriver.Lookback))
+ }
+ first := true
pass := func() error {
start := time.Now()
st, err := col.CollectAll(match)
if err != nil {
return err
}
- fmt.Printf("[%s] sessions=%d sources=%d landed=%d records=%d bytes=%s indexed=%d gone=%d conflicts=%d busy=%d pending=%d errors=%d (%s)\n",
+ derived := ""
+ if deriver != nil {
+ // The first pass looks at every session, so history that was
+ // landed before the flag went on is derived once; later passes
+ // look only at what this pass changed.
+ var sessions []string
+ if !first {
+ sessions = st.Changed
+ }
+ ms, err := deriver.Pass(sessions)
+ if err != nil {
+ return err
+ }
+ st.Errors = append(st.Errors, ms.Errors...)
+ derived = fmt.Sprintf(" metrics=%d", ms.Requests)
+ }
+ first = false
+ fmt.Printf("[%s] sessions=%d sources=%d landed=%d records=%d bytes=%s indexed=%d gone=%d conflicts=%d busy=%d pending=%d%s errors=%d (%s)\n",
time.Now().Format("15:04:05"), st.Sessions, st.SourcesSeen, st.SourcesLanded,
st.Records, humanBytes(st.Bytes), st.Indexed, st.SourcesGone, st.Conflicts,
- st.Busy, st.Pending, len(st.Errors), time.Since(start).Round(time.Millisecond))
+ st.Busy, st.Pending, derived, len(st.Errors), time.Since(start).Round(time.Millisecond))
for _, e := range st.Errors {
fmt.Fprintf(os.Stderr, " error: %v\n", e)
}
@@ -683,6 +709,25 @@ func cmdCollect(cfg *config.Config, ad config.Adapter, once bool) error {
}
}
+// newDeriver is the metrics derivation an adapter asked for, or nil.
+func newDeriver(zone *storage.Zone, ad config.Adapter) (*metrics.Deriver, error) {
+ if !ad.Metrics {
+ return nil, nil
+ }
+ lookback, err := ad.Lookback()
+ if err != nil {
+ return nil, err
+ }
+ return &metrics.Deriver{Zone: zone, Options: metrics.Options{Lookback: lookback, Version: version}}, nil
+}
+
+func lookbackWord(d time.Duration) string {
+ if d == 0 {
+ return "none"
+ }
+ return d.String()
+}
+
func humanBytes(n int64) string {
const unit = 1024
if n < unit {
diff --git a/cmd/asz/push.go b/cmd/asz/push.go
index a5e4ad0..0fb338c 100644
--- a/cmd/asz/push.go
+++ b/cmd/asz/push.go
@@ -59,6 +59,12 @@ func cmdPush(cfg *config.Config, _ config.Adapter, once bool) error {
BatchBytes: o.BatchBytes,
MaxBytesPerMinute: o.MaxBytesPerMinute,
+ // The metrics spool concerns Claude Code whichever adapter filled it,
+ // the local derivation or the runtime's own exporter.
+ MetricsService: claudecode.RuntimeName,
+ }
+ if o.ServiceName != "" {
+ p.MetricsService = o.ServiceName
}
if err := p.Prepare(); err != nil {
return err
@@ -83,8 +89,8 @@ func cmdPush(cfg *config.Config, _ config.Adapter, once bool) error {
if err != nil {
return nil, err
}
- fmt.Printf("[%s] files=%d bytes=%s wire=%s requests=%d paused=%s errors=%d (%s)\n",
- time.Now().Format("15:04:05"), st.Files, humanBytes(st.Bytes), humanBytes(st.Wire), st.Requests,
+ fmt.Printf("[%s] files=%d metrics=%d bytes=%s wire=%s requests=%d paused=%s errors=%d (%s)\n",
+ time.Now().Format("15:04:05"), st.Files, st.Metrics, humanBytes(st.Bytes), humanBytes(st.Wire), st.Requests,
st.Paused.Round(time.Second), len(st.Errors), time.Since(start).Round(time.Millisecond))
for _, e := range st.Errors {
fmt.Fprintf(os.Stderr, " error: %v\n", e)
diff --git a/cmd/asz/refresh.go b/cmd/asz/refresh.go
index 62b1e83..22d80a7 100644
--- a/cmd/asz/refresh.go
+++ b/cmd/asz/refresh.go
@@ -25,6 +25,7 @@ import (
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
"github.com/apache/skywalking-ai-sessionizer/internal/config"
+ "github.com/apache/skywalking-ai-sessionizer/internal/metrics"
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
"github.com/apache/skywalking-ai-sessionizer/internal/view"
)
@@ -38,6 +39,7 @@ import (
type refresher struct {
srv *view.Server
zone *storage.Zone
+ deriver *metrics.Deriver
col *claudecode.Collector
match func(claudecode.Session) bool
interval time.Duration
@@ -70,9 +72,14 @@ func newRefresher(srv *view.Server, zone *storage.Zone, ad config.Adapter, maxRo
if once {
mode = config.ModeOnce
}
+ deriver, err := newDeriver(zone, ad)
+ if err != nil {
+ return nil, err
+ }
r := &refresher{
srv: srv,
zone: zone,
+ deriver: deriver,
col: claudecode.New(src, zone, ad.Collector.MaxDeltaBytes),
match: claudecode.NewMatcher(ad.Include, ad.Exclude).Match,
interval: ad.Collector.Interval,
@@ -113,6 +120,18 @@ func (r *refresher) pass() {
sessions = all
}
}
+ if r.deriver != nil {
+ // The first pass derives history once; later passes only what moved.
+ var scope []string
+ if !r.full {
+ scope = sessions
+ }
+ if ms, derr := r.deriver.Pass(scope); derr != nil {
+ errs = append(errs, derr)
+ } else {
+ errs = append(errs, ms.Errors...)
+ }
+ }
rounds := 0
for _, id := range sessions {
written := 0
diff --git a/collectorcheck b/collectorcheck
new file mode 100755
index 0000000..312a076
Binary files /dev/null and b/collectorcheck differ
diff --git a/docs/en/adapters/claude-code.md b/docs/en/adapters/claude-code.md
index 54fd2bc..19a03e4 100644
--- a/docs/en/adapters/claude-code.md
+++ b/docs/en/adapters/claude-code.md
@@ -319,6 +319,24 @@ the binary are unreachable in shipped builds.
Content defaults are also restrictive: prompts and responses are redacted unless explicitly enabled,
content is truncated, and reasoning is always redacted.
+### The same metrics from the transcripts
+
+With `metrics: true`, this adapter derives the runtime's own metric family from what it landed,
+name for name and attribute for attribute with the exporter, so a receiver holds one family
+whichever produced it. Phase one is the overlap, `claude_code.token.usage`: one count per call,
+taken from the usage the last fragment repeats, never per fragment; `query_source` is `main` for
+the session's own transcript and `subagent` for a child's; `model` is `message.model`;
+`session.id` is the session. Points are monotonic delta sums per minute, as the exporter's SDK
+aggregates over its interval, and a call is counted once however many landed files its records
+reach, since the runtime re-emits records before a context reset and the exporter counted one API
+call once.
+
+What the transcripts do not carry is not derived: cost, per-request latency, tool execution time,
+active time, lines of code, commits, pull requests, the session start type, and the tokens of the
+runtime's auxiliary calls, which never reach a transcript. Those come from the exporter alone.
+The first derivation over a root with history reaches back `metrics_lookback`, 24 hours unless
+set. See [Metrics](../setup/export-otlp.md#metrics).
+
## Local data hazards
Behaviours that will break a naive reader:
diff --git a/docs/en/changes/changes-0.3.0.md b/docs/en/changes/changes-0.3.0.md
index 501910a..973a90c 100644
--- a/docs/en/changes/changes-0.3.0.md
+++ b/docs/en/changes/changes-0.3.0.md
@@ -23,3 +23,14 @@
when it is the newest release, and nothing looser. The `major.minor` line tag, `0.2` beside
`0.2.0`, is no longer published: a reader who pulled it could not tell which release answered.
The `0.2` tag that 0.2.0 published stays on the registry and points at the same image as `0.2.0`.
+
+## Metrics
+
+- `claude-code-local` with `metrics: true` derives the runtime's own metric family from the
+ landed files, `claude_code.token.usage` in phase one, name for name and attribute for attribute
+ with Claude Code's OpenTelemetry exporter: one count per call, never per fragment, per minute,
+ by type, model, query source and session. The points wait in the storage root's `_metrics/`
+ spool and `asz push` sends them over the metrics service, under asz's identity, with the same
+ budget and once-only rule as the files. `metrics_lookback`, 24 hours unless set, bounds the
+ first derivation over a root with history. Every scenario checks the points on the wire against
+ its plan, and the Collector job verifies the tokens that arrive.
diff --git a/docs/en/setup/command-line.md b/docs/en/setup/command-line.md
index 9bdf2ff..a533984 100644
--- a/docs/en/setup/command-line.md
+++ b/docs/en/setup/command-line.md
@@ -41,7 +41,7 @@ repeats every `interval`; with `-once` or `mode: once` it makes one pass and exi
pass:
```text
-[17:14:09] sessions=44 sources=5867 landed=5867 records=359292 bytes=1.0GB indexed=359292 gone=0 conflicts=0 busy=0 pending=0 errors=0 (2m39.473s)
+[17:14:09] sessions=44 sources=5867 landed=5867 records=359292 bytes=1.0GB indexed=359292 gone=0 conflicts=0 busy=0 pending=0 metrics=5210 errors=0 (2m39.473s)
```
| Field | Meaning |
@@ -54,6 +54,7 @@ pass:
| `conflicts` | a source was rotated, truncated or rewritten behind its cursor. Collection stopped for it. |
| `busy` | a session skipped because another collector holds its lock |
| `pending` | a source still had data when the per-pass limit was reached |
+| `metrics` | metrics requests derived this pass, with `metrics: true` on the adapter; absent otherwise |
| `errors` | listed on standard error after the line |
`pending` or `errors` above zero means the pass did not collect everything, and the command exits
@@ -163,11 +164,12 @@ Sends every landed file and every round not yet sent to the OpenTelemetry logs r
`export.otlp.interval`. One line per pass:
```text
-[10:12:03] files=306 bytes=47.9MB wire=48.0MB requests=5 paused=0s errors=0 (1.1s)
+[10:12:03] files=306 metrics=41 bytes=47.9MB wire=48.0MB requests=46 paused=0s errors=0 (1.1s)
```
-`wire` is what went out, the requests as encoded, and `paused` is how long the pass waited for
-budget under `export.otlp.max_bytes_per_minute`. A pass with errors exits non-zero with `-once`;
+`metrics` counts the spooled metrics requests sent, `wire` is what went out, the requests as
+encoded, and `paused` is how long the pass waited for budget under
+`export.otlp.max_bytes_per_minute`. A pass with errors exits non-zero with `-once`;
the files whose requests failed are not recorded as sent and go again on the next pass. See
[Export over OpenTelemetry](export-otlp.md).
diff --git a/docs/en/setup/configuration.md b/docs/en/setup/configuration.md
index 776ac95..1a5c1ba 100644
--- a/docs/en/setup/configuration.md
+++ b/docs/en/setup/configuration.md
@@ -43,6 +43,8 @@ this machine. Every command runs once per enabled adapter.
| `source_root` | empty | Where Claude Code keeps its files. Empty resolves it the way Claude Code does: `CLAUDE_CONFIG_DIR`, then `XDG_CONFIG_HOME/claude`, then `~/.claude`, each followed by `projects`. Set it only to collect from a copy or a mounted directory. |
| `include` | empty | Session filters, see below. Empty means every session is a candidate. |
| `exclude` | `/private/tmp/**` | Session filters, see below. |
+| `metrics` | `false` | Derive the runtime's own metric family from the landed files, `claude_code.token.usage` in phase one, name for name with the runtime's exporter. See [Metrics](export-otlp.md#metrics). |
+| `metrics_lookback` | `24h` | How far back the first derivation over a root reaches. A duration such as `24h`, or a number of days such as `7d`; `0` or `none` derives everything. Later passes derive every new file whole. |
### Session filters
diff --git a/docs/en/setup/export-otlp.md b/docs/en/setup/export-otlp.md
index 8792b9e..6e5f753 100644
--- a/docs/en/setup/export-otlp.md
+++ b/docs/en/setup/export-otlp.md
@@ -128,6 +128,33 @@ over each transport, then checks every record against the root and rebuilds the
the Collector wrote; CI runs it on every change. Writing each record's body to `asz.file` under a
new root gives a root that `asz verify` and `asz view` read like the original.
+## Metrics
+
+`asz push` sends metrics as well, when an adapter produces them, and they are Claude Code's own
+metric family, name for name and attribute for attribute with the runtime's OpenTelemetry
+exporter, so a receiver holds one family whichever produced it. Phase one is the one measure both
+can supply exactly:
+
+| Metric | Attributes | Unit |
+| --- | --- | --- |
+| `claude_code.token.usage` | `type` (`input`, `output`, `cacheRead`, `cacheCreation`), `model`, `query_source` (`main`, `subagent`), `session.id` | `tokens`, a monotonic delta sum, one point per minute and attribute set |
+
+With `metrics: true` on the `claude-code-local` adapter, the collector derives the points from
+the landed files: one count per call, never per fragment, since a main transcript repeats the
+usage on every fragment of a call, summed per minute the way the runtime's SDK sums over its
+export interval, with `query_source` from the stream the call was made on. It cannot derive what
+a transcript does not carry: cost, latency, active time, lines of code, commits, pull requests,
+the session start type, or the tokens of the runtime's auxiliary calls, which never reach a
+transcript. Those are the runtime's exporter's alone. The first derivation over a root with
+history is bounded by `metrics_lookback`, 24 hours unless set, so switching the flag on does not
+send a year of tokens; every later pass derives each new file whole.
+
+The points wait in the storage root's `_metrics/` spool, one write-once file per landed file
+with points, and go out in order under the same budget and the same once-only rule as the files.
+On the way out the resource is normalised to asz's identity, the service, the layer, the sender,
+so the OAP holds one service for the runtime. A receiver that answers with a partial success is
+treated as having refused the request, as for logs.
+
## Rate
A first push sends the whole history of the storage root, which can be hundreds of megabytes, as
diff --git a/internal/config/config.go b/internal/config/config.go
index 4baf4c8..6460dfd 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -22,6 +22,8 @@ import (
"fmt"
"os"
"path/filepath"
+ "strconv"
+ "strings"
"time"
"gopkg.in/yaml.v3"
@@ -88,6 +90,31 @@ type OTLP struct {
Interval time.Duration `yaml:"interval"`
}
+// Lookback is the metrics look-back as a duration: 24h when unset, zero
+// for 0 or none, and a d suffix counts days, since a look-back is spoken of
+// in days.
+func (a Adapter) Lookback() (time.Duration, error) {
+ s := strings.TrimSpace(a.MetricsLookback)
+ switch s {
+ case "":
+ return 24 * time.Hour, nil
+ case "0", "none":
+ return 0, nil
+ }
+ if days, ok := strings.CutSuffix(s, "d"); ok {
+ n, err := strconv.Atoi(days)
+ if err != nil || n < 0 {
+ return 0, fmt.Errorf("config: metrics_lookback %q is not a number of days", s)
+ }
+ return time.Duration(n) * 24 * time.Hour, nil
+ }
+ d, err := time.ParseDuration(s)
+ if err != nil || d < 0 {
+ return 0, fmt.Errorf("config: metrics_lookback %q is not a duration such as 24h or 7d", s)
+ }
+ return d, nil
+}
+
// Storage locates the landing zone.
type Storage struct {
// Root is where landed data is written. Default "./data", i.e. beside the
@@ -116,6 +143,17 @@ type Adapter struct {
Include []string `yaml:"include"`
Exclude []string `yaml:"exclude"`
+ // Metrics turns on the runtime's own metric family, derived from the
+ // landed files: the same names and attributes the runtime's exporter
+ // sends, so a receiver sees one family whichever produced it. Phase one
+ // is token usage. It cannot be on while a claude-code-otlp adapter with
+ // metrics is enabled, since the two would count the same tokens twice.
+ Metrics bool `yaml:"metrics"`
+ // MetricsLookback bounds the first derivation over a root that has
+ // history: a minute older than this is not derived. A duration such as
+ // 24h or 7d; empty means 24h; 0 or none means everything.
+ MetricsLookback string `yaml:"metrics_lookback"`
+
Collector Collector `yaml:"collector"`
}
@@ -151,6 +189,9 @@ func Default() *Config {
Name: AdapterClaudeCodeLocal,
Enabled: true,
Exclude: []string{"/private/tmp/**"},
+ // The look-back is written out, as every default is, so the
+ // file says what the first derivation reaches back to.
+ MetricsLookback: "24h",
Collector: Collector{
Mode: ModeWatch,
Interval: 5 * time.Second,
@@ -259,6 +300,9 @@ func (c *Config) Validate() error {
if a.Collector.Mode != ModeWatch && a.Collector.Mode != ModeOnce {
return fmt.Errorf("config: adapter %q: unknown collector mode %q", a.Name, a.Collector.Mode)
}
+ if _, err := a.Lookback(); err != nil {
+ return fmt.Errorf("adapter %q: %w", a.Name, err)
+ }
}
if p := c.Export.OTLP.Protocol; p != "grpc" && p != "http" {
return fmt.Errorf("config: export.otlp.protocol is %q, want grpc or http", p)
diff --git a/internal/export/otlp/client.go b/internal/export/otlp/client.go
index 88be22f..1c7f86a 100644
--- a/internal/export/otlp/client.go
+++ b/internal/export/otlp/client.go
@@ -30,6 +30,7 @@ import (
"time"
collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1"
+ collmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
@@ -71,6 +72,8 @@ type Client interface {
// sent again whole on the next pass, and a receiver keeps the first
// copy of a file it already holds.
Export(req *collogspb.ExportLogsServiceRequest) error
+ // ExportMetrics sends one metrics request, with the same rules.
+ ExportMetrics(req *collmetricspb.ExportMetricsServiceRequest) error
Close() error
}
@@ -117,6 +120,7 @@ func NewClient(o Options) (Client, error) {
type grpcClient struct {
conn *grpc.ClientConn
logs collogspb.LogsServiceClient
+ metrics collmetricspb.MetricsServiceClient
headers map[string]string
timeout time.Duration
}
@@ -133,7 +137,8 @@ func newGRPC(o Options) (Client, error) {
if err != nil {
return nil, fmt.Errorf("otlp: %w", err)
}
- return &grpcClient{conn: conn, logs: collogspb.NewLogsServiceClient(conn), headers: o.Headers, timeout: o.timeout()}, nil
+ return &grpcClient{conn: conn, logs: collogspb.NewLogsServiceClient(conn), metrics: collmetricspb.NewMetricsServiceClient(conn),
+ headers: o.Headers, timeout: o.timeout()}, nil
}
func (o Options) timeout() time.Duration {
@@ -143,27 +148,49 @@ func (o Options) timeout() time.Duration {
return o.Timeout
}
-func (c *grpcClient) Export(req *collogspb.ExportLogsServiceRequest) error {
+func (c *grpcClient) context() (context.Context, context.CancelFunc) {
ctx, cancel := context.WithTimeout(context.Background(), c.timeout)
- defer cancel()
for k, v := range c.headers {
ctx = metadata.AppendToOutgoingContext(ctx, k, v)
}
+ return ctx, cancel
+}
+
+func (c *grpcClient) failure(err error) error {
+ if status.Code(err) == codes.ResourceExhausted {
+ return &Throttled{Detail: c.conn.Target() + ": " + err.Error()}
+ }
+ return fmt.Errorf("otlp: %s: %w", c.conn.Target(), err)
+}
+
+func (c *grpcClient) Export(req *collogspb.ExportLogsServiceRequest) error {
+ ctx, cancel := c.context()
+ defer cancel()
resp, err := c.logs.Export(ctx, req)
if err != nil {
- if status.Code(err) == codes.ResourceExhausted {
- return &Throttled{Detail: c.conn.Target() + ": " + err.Error()}
- }
- return fmt.Errorf("otlp: %s: %w", c.conn.Target(), err)
+ return c.failure(err)
}
return rejected(resp)
}
+func (c *grpcClient) ExportMetrics(req *collmetricspb.ExportMetricsServiceRequest) error {
+ ctx, cancel := c.context()
+ defer cancel()
+ resp, err := c.metrics.Export(ctx, req)
+ if err != nil {
+ return c.failure(err)
+ }
+ if ps := resp.GetPartialSuccess(); ps.GetRejectedDataPoints() > 0 {
+ return fmt.Errorf("otlp: the receiver rejected %d data point(s): %s", ps.GetRejectedDataPoints(), ps.GetErrorMessage())
+ }
+ return nil
+}
+
func (c *grpcClient) Close() error { return c.conn.Close() }
-// httpClient posts each request to the receiver's /v1/logs.
+// httpClient posts each request to the receiver's /v1/logs or /v1/metrics.
type httpClient struct {
- url string
+ base string
headers map[string]string
http *http.Client
}
@@ -173,21 +200,56 @@ func newHTTP(o Options) (Client, error) {
return nil, fmt.Errorf("otlp: an http endpoint is a URL such as http://127.0.0.1:12800: %s", o.Endpoint)
}
return &httpClient{
- url: strings.TrimRight(o.Endpoint, "/") + "/v1/logs",
+ base: strings.TrimRight(o.Endpoint, "/"),
headers: o.Headers,
http: &http.Client{Timeout: o.timeout()},
}, nil
}
func (c *httpClient) Export(req *collogspb.ExportLogsServiceRequest) error {
- body, err := proto.Marshal(req)
+ answer, err := c.post(c.base+"/v1/logs", req)
if err != nil {
- return fmt.Errorf("otlp: %w", err)
+ return err
+ }
+ var out collogspb.ExportLogsServiceResponse
+ if len(answer) > 0 {
+ if err := proto.Unmarshal(answer, &out); err != nil {
+ return fmt.Errorf("otlp: %s answered with a body that is not an OTLP response: %w", c.base, err)
+ }
}
- hr, err := http.NewRequest(http.MethodPost, c.url, bytes.NewReader(body))
+ return rejected(&out)
+}
+
+func (c *httpClient) ExportMetrics(req *collmetricspb.ExportMetricsServiceRequest) error {
+ answer, err := c.post(c.base+"/v1/metrics", req)
if err != nil {
return err
}
+ var out collmetricspb.ExportMetricsServiceResponse
+ if len(answer) > 0 {
+ if err := proto.Unmarshal(answer, &out); err != nil {
+ return fmt.Errorf("otlp: %s answered with a body that is not an OTLP response: %w", c.base, err)
+ }
+ }
+ if ps := out.GetPartialSuccess(); ps.GetRejectedDataPoints() > 0 {
+ return fmt.Errorf("otlp: the receiver rejected %d data point(s): %s", ps.GetRejectedDataPoints(), ps.GetErrorMessage())
+ }
+ return nil
+}
+
+// post sends one request as protobuf and returns the answer's body when it
+// is an OTLP response, or nothing when the receiver answered with anything
+// else, such as an empty body from a proxy, which is a full success as far
+// as it says.
+func (c *httpClient) post(url string, req proto.Message) ([]byte, error) {
+ body, err := proto.Marshal(req)
+ if err != nil {
+ return nil, fmt.Errorf("otlp: %w", err)
+ }
+ hr, err := http.NewRequest(http.MethodPost, url, bytes.NewReader(body))
+ if err != nil {
+ return nil, err
+ }
hr.Header.Set("Content-Type", "application/x-protobuf")
hr.Header.Set("User-Agent", "asz")
for k, v := range c.headers {
@@ -195,33 +257,28 @@ func (c *httpClient) Export(req *collogspb.ExportLogsServiceRequest) error {
}
resp, err := c.http.Do(hr)
if err != nil {
- return err
+ return nil, err
}
defer resp.Body.Close()
answer, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
if resp.StatusCode == http.StatusTooManyRequests {
- t := &Throttled{Detail: c.url + " answered " + resp.Status}
+ t := &Throttled{Detail: url + " answered " + resp.Status}
if secs, err := strconv.Atoi(strings.TrimSpace(resp.Header.Get("Retry-After"))); err == nil && secs > 0 {
t.After = time.Duration(secs) * time.Second
}
- return t
+ return nil, t
}
if resp.StatusCode < 200 || resp.StatusCode > 299 {
snippet := answer
if len(snippet) > 400 {
snippet = snippet[:400]
}
- return fmt.Errorf("otlp: %s answered %s: %s", c.url, resp.Status, strings.TrimSpace(string(snippet)))
+ return nil, fmt.Errorf("otlp: %s answered %s: %s", url, resp.Status, strings.TrimSpace(string(snippet)))
}
- // A receiver answers in the request's encoding. Anything else, such as
- // an empty body from a proxy, is a full success as far as it says.
- var out collogspb.ExportLogsServiceResponse
- if len(answer) > 0 && strings.HasPrefix(resp.Header.Get("Content-Type"), "application/x-protobuf") {
- if err := proto.Unmarshal(answer, &out); err != nil {
- return fmt.Errorf("otlp: %s answered with a body that is not an OTLP response: %w", c.url, err)
- }
+ if !strings.HasPrefix(resp.Header.Get("Content-Type"), "application/x-protobuf") {
+ return nil, nil
}
- return rejected(&out)
+ return answer, nil
}
func (c *httpClient) Close() error {
diff --git a/internal/export/otlp/otlp_test.go b/internal/export/otlp/otlp_test.go
index 7f6adf5..d0c7a0b 100644
--- a/internal/export/otlp/otlp_test.go
+++ b/internal/export/otlp/otlp_test.go
@@ -30,6 +30,11 @@ import (
"time"
collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1"
+ collmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
+ commonpb "go.opentelemetry.io/proto/otlp/common/v1"
+ metricspb "go.opentelemetry.io/proto/otlp/metrics/v1"
+ resourcepb "go.opentelemetry.io/proto/otlp/resource/v1"
+ "google.golang.org/protobuf/proto"
"github.com/apache/skywalking-ai-sessionizer/internal/export/otlp"
"github.com/apache/skywalking-ai-sessionizer/internal/export/otlp/otlptest"
@@ -542,3 +547,54 @@ func TestPassSendsSessionsOldestFirstWithTheirRounds(t *testing.T) {
t.Fatalf("sent %v, want %v: the session landed first goes first, its rounds after its files", order, want)
}
}
+
+// A spooled metrics request is sent once over either transport, under
+// asz's identity, and marked pushed like a file.
+func TestPushSendsTheMetricsSpoolUnderTheServiceIdentity(t *testing.T) {
+ for _, protocol := range protocols {
+ t.Run(protocol, func(t *testing.T) {
+ z, _ := zoneWithOneSession(t)
+ req := &collmetricspb.ExportMetricsServiceRequest{ResourceMetrics: []*metricspb.ResourceMetrics{{
+ Resource: &resourcepb.Resource{Attributes: []*commonpb.KeyValue{
+ {Key: "service.name", Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: "claude-code"}}},
+ {Key: "service.version", Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: "2.1.245"}}},
+ }},
+ ScopeMetrics: []*metricspb.ScopeMetrics{{Metrics: []*metricspb.Metric{{Name: "claude_code.token.usage", Unit: "tokens",
+ Data: &metricspb.Metric_Sum{Sum: &metricspb.Sum{DataPoints: []*metricspb.NumberDataPoint{{TimeUnixNano: 1, Value: &metricspb.NumberDataPoint_AsInt{AsInt: 7}}}}}}}}},
+ }}}
+ data, err := proto.Marshal(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := storage.NewSpool(z).Put("otlp", data, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ rcv := startReceiver(t)
+ p := &otlp.Pusher{Zone: z, Client: clientFor(t, rcv, protocol), Version: "test", ServiceName: "",
+ Runtimes: map[string]string{"test": "Test Runtime"}, MetricsService: "Claude Code", Layer: "AI_AGENT", InstanceID: "sender-1"}
+ st, err := p.Pass()
+ if err != nil || len(st.Errors) != 0 {
+ t.Fatal(err, st.Errors)
+ }
+ if st.Metrics != 1 {
+ t.Fatalf("sent %d metrics requests, want 1", st.Metrics)
+ }
+ got := rcv.MetricsRequests()
+ if len(got) != 1 {
+ t.Fatalf("the receiver holds %d metrics requests", len(got))
+ }
+ res := otlptest.Attrs(got[0].GetResourceMetrics()[0].GetResource().GetAttributes())
+ if res["service.name"] != "Claude Code" || res["service.instance.id"] != "sender-1" || res["service.layer"] != "AI_AGENT" ||
+ res["telemetry.sdk.name"] != "asz" || res["service.version"] != "2.1.245" {
+ t.Fatalf("resource on the wire: %v", res)
+ }
+ if got[0].GetResourceMetrics()[0].GetScopeMetrics()[0].GetMetrics()[0].GetSum().GetDataPoints()[0].GetAsInt() != 7 {
+ t.Fatal("the points did not travel unchanged")
+ }
+ st, err = p.Pass()
+ if err != nil || st.Metrics != 0 || len(rcv.MetricsRequests()) != 1 {
+ t.Fatalf("a second pass sent the spool again: %d, %v", st.Metrics, err)
+ }
+ })
+ }
+}
diff --git a/internal/export/otlp/otlptest/receiver.go b/internal/export/otlp/otlptest/receiver.go
index 2d7539f..f6036e6 100644
--- a/internal/export/otlp/otlptest/receiver.go
+++ b/internal/export/otlp/otlptest/receiver.go
@@ -34,6 +34,7 @@ import (
"time"
collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1"
+ collmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
commonpb "go.opentelemetry.io/proto/otlp/common/v1"
logspb "go.opentelemetry.io/proto/otlp/logs/v1"
"google.golang.org/grpc"
@@ -46,9 +47,10 @@ import (
// Receiver listens on both transports and keeps what it accepted.
type Receiver struct {
- mu sync.Mutex
- reqs []*collogspb.ExportLogsServiceRequest
- fail bool
+ mu sync.Mutex
+ reqs []*collogspb.ExportLogsServiceRequest
+ metrics []*collmetricspb.ExportMetricsServiceRequest
+ fail bool
// throttle makes the receiver ask the sender to slow down, naming
// retryAfter over HTTP when it is set.
throttle bool
@@ -70,6 +72,7 @@ func Start() (*Receiver, error) {
}
r.rpc = grpc.NewServer(grpc.MaxRecvMsgSize(64 << 20))
collogspb.RegisterLogsServiceServer(r.rpc, &logsServer{r: r})
+ collmetricspb.RegisterMetricsServiceServer(r.rpc, &metricsServer{r: r})
go func() { _ = r.rpc.Serve(lis) }()
r.addr = lis.Addr().String()
return r, nil
@@ -95,11 +98,18 @@ func (r *Receiver) Requests() []*collogspb.ExportLogsServiceRequest {
return append([]*collogspb.ExportLogsServiceRequest(nil), r.reqs...)
}
+// MetricsRequests is every metrics request accepted so far, in order.
+func (r *Receiver) MetricsRequests() []*collmetricspb.ExportMetricsServiceRequest {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ return append([]*collmetricspb.ExportMetricsServiceRequest(nil), r.metrics...)
+}
+
// Reset forgets every request accepted so far.
func (r *Receiver) Reset() {
r.mu.Lock()
defer r.mu.Unlock()
- r.reqs = nil
+ r.reqs, r.metrics = nil, nil
}
// Fail makes the receiver refuse every request, or accept them again.
@@ -130,20 +140,37 @@ func (r *Receiver) Close() {
func (r *Receiver) accept(req *collogspb.ExportLogsServiceRequest) error {
r.mu.Lock()
defer r.mu.Unlock()
+ if err := r.refusal(); err != nil {
+ return err
+ }
+ r.reqs = append(r.reqs, req)
+ return nil
+}
+
+func (r *Receiver) acceptMetrics(req *collmetricspb.ExportMetricsServiceRequest) error {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+ if err := r.refusal(); err != nil {
+ return err
+ }
+ r.metrics = append(r.metrics, req)
+ return nil
+}
+
+func (r *Receiver) refusal() error {
if r.fail {
return errors.New("down")
}
if r.throttle {
return errThrottled
}
- r.reqs = append(r.reqs, req)
return nil
}
// serveHTTP is what an OTLP/HTTP receiver expects: a protobuf body posted
-// to /v1/logs, answered with a protobuf response.
+// to /v1/logs or /v1/metrics, answered with a protobuf response.
func (r *Receiver) serveHTTP(w http.ResponseWriter, req *http.Request) {
- if req.URL.Path != "/v1/logs" || req.Header.Get("Content-Type") != "application/x-protobuf" {
+ if (req.URL.Path != "/v1/logs" && req.URL.Path != "/v1/metrics") || req.Header.Get("Content-Type") != "application/x-protobuf" {
http.Error(w, "wrong path or content type", http.StatusBadRequest)
return
}
@@ -152,23 +179,28 @@ func (r *Receiver) serveHTTP(w http.ResponseWriter, req *http.Request) {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
+ if req.URL.Path == "/v1/metrics" {
+ var msg collmetricspb.ExportMetricsServiceRequest
+ if err := proto.Unmarshal(body, &msg); err != nil {
+ http.Error(w, "not an ExportMetricsServiceRequest: "+err.Error(), http.StatusBadRequest)
+ return
+ }
+ if err := r.acceptMetrics(&msg); err != nil {
+ r.refuse(w, err)
+ return
+ }
+ out, _ := proto.Marshal(&collmetricspb.ExportMetricsServiceResponse{})
+ w.Header().Set("Content-Type", "application/x-protobuf")
+ _, _ = w.Write(out)
+ return
+ }
var msg collogspb.ExportLogsServiceRequest
if err := proto.Unmarshal(body, &msg); err != nil {
http.Error(w, "not an ExportLogsServiceRequest: "+err.Error(), http.StatusBadRequest)
return
}
if err := r.accept(&msg); err != nil {
- if errors.Is(err, errThrottled) {
- r.mu.Lock()
- after := r.retryAfter
- r.mu.Unlock()
- if after > 0 {
- w.Header().Set("Retry-After", strconv.Itoa(int(after/time.Second)))
- }
- http.Error(w, err.Error(), http.StatusTooManyRequests)
- return
- }
- http.Error(w, err.Error(), http.StatusServiceUnavailable)
+ r.refuse(w, err)
return
}
out, _ := proto.Marshal(&collogspb.ExportLogsServiceResponse{})
@@ -176,6 +208,37 @@ func (r *Receiver) serveHTTP(w http.ResponseWriter, req *http.Request) {
_, _ = w.Write(out)
}
+// refuse answers a refusal the way a receiver does: 429 with the wait it
+// names for a throttle, 503 for anything else.
+func (r *Receiver) refuse(w http.ResponseWriter, err error) {
+ if errors.Is(err, errThrottled) {
+ r.mu.Lock()
+ after := r.retryAfter
+ r.mu.Unlock()
+ if after > 0 {
+ w.Header().Set("Retry-After", strconv.Itoa(int(after/time.Second)))
+ }
+ http.Error(w, err.Error(), http.StatusTooManyRequests)
+ return
+ }
+ http.Error(w, err.Error(), http.StatusServiceUnavailable)
+}
+
+type metricsServer struct {
+ collmetricspb.UnimplementedMetricsServiceServer
+ r *Receiver
+}
+
+func (s *metricsServer) Export(_ context.Context, req *collmetricspb.ExportMetricsServiceRequest) (*collmetricspb.ExportMetricsServiceResponse, error) {
+ if err := s.r.acceptMetrics(req); err != nil {
+ if errors.Is(err, errThrottled) {
+ return nil, status.Error(codes.ResourceExhausted, err.Error())
+ }
+ return nil, status.Error(codes.Unavailable, err.Error())
+ }
+ return &collmetricspb.ExportMetricsServiceResponse{}, nil
+}
+
type logsServer struct {
collogspb.UnimplementedLogsServiceServer
r *Receiver
diff --git a/internal/export/otlp/push.go b/internal/export/otlp/push.go
index 668673a..459c643 100644
--- a/internal/export/otlp/push.go
+++ b/internal/export/otlp/push.go
@@ -36,8 +36,10 @@ import (
"time"
collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1"
+ collmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
commonpb "go.opentelemetry.io/proto/otlp/common/v1"
logspb "go.opentelemetry.io/proto/otlp/logs/v1"
+ metricspb "go.opentelemetry.io/proto/otlp/metrics/v1"
resourcepb "go.opentelemetry.io/proto/otlp/resource/v1"
"google.golang.org/protobuf/proto"
@@ -78,6 +80,12 @@ type Pusher struct {
InstanceID string
// Layer is the receiver's layer for the service, sent as service.layer.
Layer string
+ // MetricsService is the service the metrics spool is attributed to: the
+ // requests the local adapter derived and the ones the runtime's exporter
+ // sent both concern the one runtime, and they leave under its name so a
+ // receiver holds one service whichever produced the points. Empty means
+ // ServiceName.
+ MetricsService string
// BatchBytes is how many file bytes one request carries at most. A file
// larger than the budget is sent alone, in a request of its own.
BatchBytes int64
@@ -100,7 +108,9 @@ type Stats struct {
Bytes int64
Wire int64 // bytes on the wire, requests as encoded
Requests int
- Paused time.Duration // how long the pass waited for budget
+ // Metrics is how many spooled metrics requests were sent.
+ Metrics int
+ Paused time.Duration // how long the pass waited for budget
// Throttled says the receiver asked the sender to slow down and the
// pass stopped there; RetryAfter is the wait it named, if any.
Throttled bool
@@ -202,6 +212,9 @@ func (p *Pusher) Prepare() error {
}
p.InstanceID = id
}
+ if p.MetricsService == "" {
+ p.MetricsService = p.ServiceName
+ }
return nil
}
@@ -267,12 +280,104 @@ func (p *Pusher) Pass() (*Stats, error) {
st.Errors = append(st.Errors, err)
}
}
+ if b.stop == nil {
+ p.pushSpool(b)
+ }
if b.stop != nil {
st.Throttled, st.RetryAfter = true, b.stop.After
}
return st, nil
}
+// pushSpool sends the metrics spool: every request not yet sent, in the
+// order it was put, one request each, under the same budget and the same
+// once-only rule as the files. The resource is normalised on the way out
+// to asz's identity, so a receiver holds one service for the runtime
+// whether asz derived the points or the runtime's exporter sent them.
+func (p *Pusher) pushSpool(b *batch) {
+ st, state := b.st, b.state
+ files, err := storage.NewSpool(p.Zone).List()
+ if err != nil {
+ st.Errors = append(st.Errors, err)
+ return
+ }
+ for _, sf := range files {
+ rel, _ := filepath.Rel(p.Zone.Root(), sf.Path)
+ rel = filepath.ToSlash(rel)
+ if state.pushed(rel) {
+ continue
+ }
+ if p.MetricsService == "" {
+ st.Errors = append(st.Errors, fmt.Errorf("%s: no service to attribute metrics to", rel))
+ return
+ }
+ data, err := os.ReadFile(sf.Path)
+ if err != nil {
+ st.Errors = append(st.Errors, err)
+ continue
+ }
+ var req collmetricspb.ExportMetricsServiceRequest
+ if err := proto.Unmarshal(data, &req); err != nil {
+ st.Errors = append(st.Errors, fmt.Errorf("%s: not a metrics request: %w", rel, err))
+ continue
+ }
+ for _, rm := range req.ResourceMetrics {
+ p.normalise(rm)
+ }
+ size := int64(proto.Size(&req))
+ st.Paused += p.limit.take(size)
+ err = p.Client.ExportMetrics(&req)
+ st.Requests++
+ st.Wire += size
+ if err != nil {
+ st.Errors = append(st.Errors, fmt.Errorf("%s: %w", rel, err))
+ var t *Throttled
+ if errors.As(err, &t) {
+ b.stop = t
+ p.limit.drain()
+ return
+ }
+ continue
+ }
+ state.mark(rel, digestOf(data))
+ st.Metrics++
+ if err := state.save(p.statePath(), p.Now()); err != nil {
+ st.Errors = append(st.Errors, err)
+ }
+ }
+}
+
+// normalise puts asz's identity on a resource: the runtime as the service,
+// the layer, who is pushing, and the sender. Everything else the resource
+// carried, such as the runtime's own version, stays.
+func (p *Pusher) normalise(rm *metricspb.ResourceMetrics) {
+ if rm.Resource == nil {
+ rm.Resource = &resourcepb.Resource{}
+ }
+ ours := map[string]string{
+ "service.name": p.MetricsService, "service.instance.id": p.InstanceID,
+ "telemetry.sdk.name": "asz", "telemetry.sdk.version": p.Version, "telemetry.sdk.language": "go",
+ }
+ if p.Layer != "" {
+ ours["service.layer"] = p.Layer
+ }
+ kept := rm.Resource.Attributes[:0]
+ for _, kv := range rm.Resource.Attributes {
+ if _, replaced := ours[kv.GetKey()]; !replaced {
+ kept = append(kept, kv)
+ }
+ }
+ keys := make([]string, 0, len(ours))
+ for k := range ours {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ for _, k := range keys {
+ kept = append(kept, str(k, ours[k]))
+ }
+ rm.Resource.Attributes = kept
+}
+
// addRounds queues every round of a conversation not yet sent.
func (b *batch) addRounds(conv string) {
rounds, err := roundFiles(b.p.Zone.Root(), conv)
diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go
new file mode 100644
index 0000000..56e47e3
--- /dev/null
+++ b/internal/metrics/metrics.go
@@ -0,0 +1,441 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Package metrics derives the runtime's own metric family from landed
+// Session Data, so a receiver sees the same metrics whether the runtime's
+// exporter or asz produced them: the same names, the same attributes, the
+// same delta temporality, aggregated per minute as the exporter aggregates
+// over its interval. Phase one is the overlap: token usage. What the
+// transcripts do not carry, cost, latency, active time and the rest, is the
+// exporter's alone and is never estimated here.
+package metrics
+
+import (
+ "bufio"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+ "time"
+
+ collmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
+ commonpb "go.opentelemetry.io/proto/otlp/common/v1"
+ metricspb "go.opentelemetry.io/proto/otlp/metrics/v1"
+ resourcepb "go.opentelemetry.io/proto/otlp/resource/v1"
+ "google.golang.org/protobuf/proto"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// The family, as the runtime's exporter names it.
+const (
+ // TokenUsage counts tokens, by type and model, per session and per query
+ // source; the exporter's claude_code.token.usage.
+ TokenUsage = "claude_code.token.usage"
+ tokenUnit = "tokens"
+ tokenDesc = "Number of tokens used"
+
+ // RuntimeService is the service name the runtime's exporter puts on its
+ // resource. A derived request carries the same, and asz push normalises
+ // both to asz's identity on the way out.
+ RuntimeService = "claude-code"
+
+ // ScopeName identifies what derived the points.
+ ScopeName = "github.com/apache/skywalking-ai-sessionizer/metrics"
+
+ // SourceLocal names the spool files this package writes.
+ SourceLocal = "local"
+)
+
+// The token types, spelled as the exporter spells them.
+const (
+ TypeInput = "input"
+ TypeOutput = "output"
+ TypeCacheRead = "cacheRead"
+ TypeCacheCreation = "cacheCreation"
+)
+
+// The query sources the transcripts can tell apart. The exporter's third,
+// auxiliary, is a call that never reaches a transcript.
+const (
+ SourceMain = "main"
+ SourceSubagent = "subagent"
+)
+
+// Options settle a derivation.
+type Options struct {
+ // Lookback bounds the first derivation over a root: a minute that ended
+ // before now minus Lookback is not derived, so switching the flag on does
+ // not send a year of history. Zero means everything. Only the first pass
+ // is bounded; later passes derive every new file whole.
+ Lookback time.Duration
+ Version string
+ Now func() time.Time
+}
+
+// Stats reports what one pass did.
+type Stats struct {
+ // Files is how many landed files were derived this pass; Requests how
+ // many spool files were written, one per landed file with points.
+ Files int
+ Requests int
+ Points int
+ // Skipped counts files older than the look-back on the first pass.
+ Skipped int
+ Errors []error
+}
+
+// Deriver derives from the landed files of a root that no earlier pass
+// derived, and writes the requests to the root's spool. Which files were
+// derived is kept in the spool's derived.state, by path and digest, so a
+// file is derived once; which calls were counted is kept per session, so a
+// call is counted once however many files its records reach. The runtime
+// re-emits records before a context reset, and a landed file is cut at a
+// budget, so the same call can sit in two files, and the runtime's
+// exporter counted it once.
+type Deriver struct {
+ Zone *storage.Zone
+ Options
+}
+
+// Pass derives what is new in the sessions named, or in every session when
+// none is.
+func (d *Deriver) Pass(sessions []string) (*Stats, error) {
+ if d.Now == nil {
+ d.Now = time.Now
+ }
+ st := &Stats{}
+ spool := storage.NewSpool(d.Zone)
+ state, first, err := loadState(filepath.Join(spool.Dir(), "derived.state"))
+ if err != nil {
+ return nil, err
+ }
+ var since time.Time
+ if first && d.Lookback > 0 {
+ since = d.Now().Add(-d.Lookback)
+ }
+ if sessions == nil {
+ if sessions, err = sessionDirs(d.Zone.Root()); err != nil {
+ return nil, err
+ }
+ }
+ for _, session := range sessions {
+ files, err := storage.LandedFiles(d.Zone, session)
+ if err != nil {
+ st.Errors = append(st.Errors, err)
+ continue
+ }
+ callsPath := filepath.Join(spool.Dir(), "calls-"+session+".state")
+ seen, err := loadCalls(callsPath)
+ if err != nil {
+ st.Errors = append(st.Errors, err)
+ continue
+ }
+ counted := len(seen)
+ for _, lf := range files {
+ rel, _ := filepath.Rel(d.Zone.Root(), lf.Path)
+ rel = filepath.ToSlash(rel)
+ if state.has(rel) {
+ continue
+ }
+ req, points, latest, digest, err := DeriveFile(lf.Path, since, d.Version, seen)
+ if err != nil {
+ st.Errors = append(st.Errors, fmt.Errorf("%s: %w", rel, err))
+ continue
+ }
+ if !since.IsZero() && !latest.IsZero() && latest.Before(since) {
+ st.Skipped++
+ }
+ if points > 0 {
+ data, err := proto.Marshal(req)
+ if err != nil {
+ st.Errors = append(st.Errors, err)
+ continue
+ }
+ if _, err := spool.Put(SourceLocal, data, d.Now()); err != nil {
+ st.Errors = append(st.Errors, err)
+ continue
+ }
+ st.Requests++
+ st.Points += points
+ }
+ st.Files++
+ state.mark(rel, digest)
+ }
+ if len(seen) != counted {
+ if err := saveCalls(callsPath, seen, d.Now()); err != nil {
+ return nil, err
+ }
+ }
+ }
+ if err := state.save(filepath.Join(spool.Dir(), "derived.state"), d.Now()); err != nil {
+ return nil, err
+ }
+ return st, nil
+}
+
+// loadCalls reads the calls counted so far for one session, one id per
+// line after the header.
+func loadCalls(path string) (map[string]bool, error) {
+ seen := map[string]bool{}
+ f, err := os.Open(path)
+ if err != nil {
+ if os.IsNotExist(err) {
+ return seen, nil
+ }
+ return nil, err
+ }
+ defer f.Close()
+ sc := bufio.NewScanner(f)
+ for sc.Scan() {
+ if id, ok := strings.CutPrefix(sc.Text(), "call "); ok {
+ seen[id] = true
+ }
+ }
+ return seen, sc.Err()
+}
+
+func saveCalls(path string, seen map[string]bool, now time.Time) error {
+ ids := make([]string, 0, len(seen))
+ for id := range seen {
+ ids = append(ids, id)
+ }
+ sort.Strings(ids)
+ return storage.WriteAtomic(path, storage.PermState, func(w io.Writer) error {
+ bw := bufio.NewWriter(w)
+ fmt.Fprintf(bw, "schema 1\nupdated_at %s\n", now.UTC().Format(time.RFC3339Nano))
+ for _, id := range ids {
+ fmt.Fprintf(bw, "call %s\n", id)
+ }
+ return bw.Flush()
+ })
+}
+
+// point is one minute of one attribute set.
+type point struct {
+ minute time.Time
+ session string
+ model string
+ source string
+ typ string
+ value int64
+}
+
+// DeriveFile derives the token usage of one landed file: one count per
+// call, never per fragment, since a main transcript repeats the usage on
+// every fragment of a call, and never for a call in seen, which the caller
+// keeps across the files of a session; summed per minute and attribute
+// set. Minutes that ended before since are left out. It returns the
+// request, how many points it holds, the latest record time in the file,
+// and the file's digest, and adds the calls it counted to seen.
+func DeriveFile(path string, since time.Time, version string, seen map[string]bool) (*collmetricspb.ExportMetricsServiceRequest, int, time.Time, string, error) {
+ f, err := os.Open(path)
+ if err != nil {
+ return nil, 0, time.Time{}, "", err
+ }
+ defer f.Close()
+ h := sha256.New()
+ rd, err := sessiondata.NewReader(io.TeeReader(f, h))
+ if err != nil {
+ return nil, 0, time.Time{}, "", err
+ }
+ hdr := rd.Header()
+ source := SourceSubagent
+ if hdr.Stream == "main" {
+ source = SourceMain
+ }
+ sums := map[point]int64{}
+ if seen == nil {
+ seen = map[string]bool{}
+ }
+ var latest time.Time
+ for {
+ rec, err := rd.Next()
+ if errors.Is(err, io.EOF) {
+ break
+ }
+ if err != nil {
+ return nil, 0, time.Time{}, "", err
+ }
+ t, ok := recordTime(rec)
+ if ok && t.After(latest) {
+ latest = t
+ }
+ if hdr.Kind != sessiondata.KindTranscript || rec.Call == "" || rec.Usage == nil || seen[rec.Call] || !ok {
+ continue
+ }
+ seen[rec.Call] = true
+ minute := t.Truncate(time.Minute)
+ if !since.IsZero() && minute.Add(time.Minute).Before(since) {
+ continue
+ }
+ for typ, n := range map[string]int{
+ TypeInput: rec.Usage.Input, TypeOutput: rec.Usage.Output,
+ TypeCacheRead: rec.Usage.CacheRead, TypeCacheCreation: rec.Usage.CacheWrite,
+ } {
+ if n == 0 {
+ continue
+ }
+ k := point{minute: minute, session: hdr.Session, model: rec.Model, source: source, typ: typ}
+ sums[k] += int64(n)
+ }
+ }
+ // Drain what the reader did not consume, so the digest is the file's.
+ if _, err := io.Copy(io.Discard, f); err != nil {
+ return nil, 0, time.Time{}, "", err
+ }
+ digest := hex.EncodeToString(h.Sum(nil))
+ points := make([]point, 0, len(sums))
+ for k, v := range sums {
+ k.value = v
+ points = append(points, k)
+ }
+ // The same file derives the same bytes: points in one order.
+ sort.Slice(points, func(i, j int) bool {
+ a, b := points[i], points[j]
+ if !a.minute.Equal(b.minute) {
+ return a.minute.Before(b.minute)
+ }
+ if a.session != b.session {
+ return a.session < b.session
+ }
+ if a.model != b.model {
+ return a.model < b.model
+ }
+ if a.source != b.source {
+ return a.source < b.source
+ }
+ return a.typ < b.typ
+ })
+ return request(points, version), len(points), latest, digest, nil
+}
+
+// request builds the exporter's request for the points: one resource, one
+// scope, one metric, delta sums that are monotonic, as the exporter's SDK
+// writes them.
+func request(points []point, version string) *collmetricspb.ExportMetricsServiceRequest {
+ dps := make([]*metricspb.NumberDataPoint, 0, len(points))
+ for _, p := range points {
+ attrs := []*commonpb.KeyValue{
+ str("session.id", p.session),
+ str("type", p.typ),
+ str("query_source", p.source),
+ }
+ if p.model != "" {
+ attrs = append(attrs, str("model", p.model))
+ }
+ dps = append(dps, &metricspb.NumberDataPoint{
+ Attributes: attrs,
+ StartTimeUnixNano: uint64(p.minute.UnixNano()),
+ TimeUnixNano: uint64(p.minute.Add(time.Minute).UnixNano()),
+ Value: &metricspb.NumberDataPoint_AsInt{AsInt: p.value},
+ })
+ }
+ return &collmetricspb.ExportMetricsServiceRequest{ResourceMetrics: []*metricspb.ResourceMetrics{{
+ Resource: &resourcepb.Resource{Attributes: []*commonpb.KeyValue{str("service.name", RuntimeService)}},
+ ScopeMetrics: []*metricspb.ScopeMetrics{{
+ Scope: &commonpb.InstrumentationScope{Name: ScopeName, Version: version},
+ Metrics: []*metricspb.Metric{{
+ Name: TokenUsage, Description: tokenDesc, Unit: tokenUnit,
+ Data: &metricspb.Metric_Sum{Sum: &metricspb.Sum{
+ DataPoints: dps,
+ AggregationTemporality: metricspb.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA,
+ IsMonotonic: true,
+ }},
+ }},
+ }},
+ }}}
+}
+
+func str(key, val string) *commonpb.KeyValue {
+ return &commonpb.KeyValue{Key: key, Value: &commonpb.AnyValue{Value: &commonpb.AnyValue_StringValue{StringValue: val}}}
+}
+
+func recordTime(rec *sessiondata.Record) (time.Time, bool) {
+ if rec.Time == "" {
+ return time.Time{}, false
+ }
+ t, err := time.Parse(time.RFC3339Nano, rec.Time)
+ if err != nil {
+ return time.Time{}, false
+ }
+ return t, true
+}
+
+func sessionDirs(root string) ([]string, error) {
+ items, err := os.ReadDir(root)
+ if err != nil {
+ return nil, err
+ }
+ var out []string
+ for _, d := range items {
+ if d.IsDir() && !strings.HasPrefix(d.Name(), "_") {
+ out = append(out, d.Name())
+ }
+ }
+ sort.Strings(out)
+ return out, nil
+}
+
+// derivedState is the set of landed files derived so far, with the digest
+// each had.
+type derivedState struct{ files map[string]string }
+
+func loadState(path string) (*derivedState, bool, error) {
+ s := &derivedState{files: map[string]string{}}
+ f, err := os.Open(path)
+ if err != nil {
+ if os.IsNotExist(err) {
+ return s, true, nil
+ }
+ return nil, false, err
+ }
+ defer f.Close()
+ sc := bufio.NewScanner(f)
+ for sc.Scan() {
+ fields := strings.Fields(sc.Text())
+ if len(fields) == 3 && fields[0] == "derived" {
+ s.files[fields[1]] = fields[2]
+ }
+ }
+ return s, false, sc.Err()
+}
+
+func (s *derivedState) has(rel string) bool { _, ok := s.files[rel]; return ok }
+func (s *derivedState) mark(rel, digest string) { s.files[rel] = digest }
+
+func (s *derivedState) save(path string, now time.Time) error {
+ keys := make([]string, 0, len(s.files))
+ for k := range s.files {
+ keys = append(keys, k)
+ }
+ sort.Strings(keys)
+ return storage.WriteAtomic(path, storage.PermState, func(w io.Writer) error {
+ bw := bufio.NewWriter(w)
+ fmt.Fprintf(bw, "schema 1\nupdated_at %s\n", now.UTC().Format(time.RFC3339Nano))
+ for _, k := range keys {
+ fmt.Fprintf(bw, "derived %s %s\n", k, s.files[k])
+ }
+ return bw.Flush()
+ })
+}
diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go
new file mode 100644
index 0000000..9008b94
--- /dev/null
+++ b/internal/metrics/metrics_test.go
@@ -0,0 +1,229 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package metrics_test
+
+import (
+ "io"
+ "os"
+ "path/filepath"
+ "testing"
+ "time"
+
+ collmetricspb "go.opentelemetry.io/proto/otlp/collector/metrics/v1"
+ metricspb "go.opentelemetry.io/proto/otlp/metrics/v1"
+ "google.golang.org/protobuf/proto"
+
+ "github.com/apache/skywalking-ai-sessionizer/internal/export/otlp/otlptest"
+ "github.com/apache/skywalking-ai-sessionizer/internal/metrics"
+ "github.com/apache/skywalking-ai-sessionizer/internal/storage"
+ "github.com/apache/skywalking-ai-sessionizer/pkg/sessiondata"
+)
+
+// A call of the fixture: its id, model, minute and usage, written as the
+// fragments a main transcript writes, every one repeating the usage.
+type call struct {
+ id, model string
+ at time.Time
+ frags int
+ in, out int
+ read, wr int
+}
+
+func land(t *testing.T, z *storage.Zone, session, stream string, seq uint64, calls []call) string {
+ t.Helper()
+ path := filepath.Join(z.StreamDir(session, stream), storage.LandedName("transcript", storage.Stamp(time.Unix(0, 0)), seq))
+ err := storage.WriteAtomic(path, storage.PermLanded, func(w io.Writer) error {
+ hdr := &sessiondata.Header{Seq: seq, At: "2026-09-07T00:00:00Z", Kind: sessiondata.KindTranscript,
+ Adapter: "test/0", Dialect: "test/1", Src: session + "/" + stream, Session: session, Stream: stream}
+ sw, err := sessiondata.NewWriter(w, hdr)
+ if err != nil {
+ return err
+ }
+ ord := uint64(1)
+ for _, c := range calls {
+ for i := 0; i < c.frags; i++ {
+ rec := &sessiondata.Record{Ord: ord, Sha: "x", Bytes: 1, Call: c.id, Model: c.model, From: sessiondata.FromAgent,
+ Time: c.at.Add(time.Duration(i) * 100 * time.Millisecond).UTC().Format(time.RFC3339Nano),
+ Usage: &sessiondata.Usage{Input: c.in, Output: c.out, CacheRead: c.read, CacheWrite: c.wr},
+ Parts: []sessiondata.Part{{Kind: sessiondata.PartText, Text: "t", State: "available", Bytes: 1}}}
+ if err := sw.Write(rec); err != nil {
+ return err
+ }
+ ord++
+ }
+ }
+ return sw.Close()
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ return path
+}
+
+// points flattens every spooled request into "source/type/model@minute" -> value.
+func points(t *testing.T, z *storage.Zone) (map[string]int64, int) {
+ t.Helper()
+ files, err := storage.NewSpool(z).List()
+ if err != nil {
+ t.Fatal(err)
+ }
+ out := map[string]int64{}
+ for _, sf := range files {
+ data, err := os.ReadFile(sf.Path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var req collmetricspb.ExportMetricsServiceRequest
+ if err := proto.Unmarshal(data, &req); err != nil {
+ t.Fatal(err)
+ }
+ for _, rm := range req.ResourceMetrics {
+ if otlptest.Attrs(rm.GetResource().GetAttributes())["service.name"] != metrics.RuntimeService {
+ t.Fatalf("a derived request names the service %v; it must carry the runtime's own name for push to normalise", rm.GetResource().GetAttributes())
+ }
+ for _, sm := range rm.ScopeMetrics {
+ for _, m := range sm.Metrics {
+ if m.Name != metrics.TokenUsage || m.Unit != "tokens" {
+ t.Fatalf("metric %s %s", m.Name, m.Unit)
+ }
+ sum := m.GetSum()
+ if sum == nil || sum.AggregationTemporality != metricspb.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA || !sum.IsMonotonic {
+ t.Fatalf("%s is not a monotonic delta sum, as the exporter writes it", m.Name)
+ }
+ for _, dp := range sum.DataPoints {
+ a := otlptest.Attrs(dp.Attributes)
+ minute := time.Unix(0, int64(dp.StartTimeUnixNano)).UTC().Format("15:04")
+ if dp.TimeUnixNano != dp.StartTimeUnixNano+uint64(time.Minute) {
+ t.Fatalf("a point does not span one minute")
+ }
+ out[a["query_source"]+"/"+a["type"]+"/"+a["model"]+"@"+minute+"#"+a["session.id"]] += dp.GetAsInt()
+ }
+ }
+ }
+ }
+ }
+ return out, len(files)
+}
+
+// Derivation counts a call once whatever its fragments repeat, sums per
+// minute and per attribute set, tells main from a child stream, and keeps
+// state so a file is derived once and a new file is derived later.
+func TestDerivationCountsEachCallOncePerMinute(t *testing.T) {
+ z := storage.NewZone(t.TempDir())
+ base := time.Date(2026, 9, 7, 10, 0, 0, 0, time.UTC)
+ land(t, z, "s1", "main", 1, []call{
+ {id: "c1", model: "claude-opus-5", at: base.Add(10 * time.Second), frags: 3, in: 10, out: 100, read: 1000, wr: 50},
+ {id: "c2", model: "claude-opus-5", at: base.Add(40 * time.Second), frags: 2, in: 5, out: 20, read: 900},
+ {id: "c3", model: "claude-sonnet-5", at: base.Add(90 * time.Second), frags: 1, in: 7, out: 30},
+ })
+ land(t, z, "s1", "a1", 2, []call{
+ {id: "c4", model: "claude-opus-5", at: base.Add(20 * time.Second), frags: 2, in: 3, out: 8, read: 400},
+ })
+ now := base.Add(time.Hour)
+ d := &metrics.Deriver{Zone: z, Options: metrics.Options{Version: "test", Now: func() time.Time { return now }}}
+ st, err := d.Pass(nil)
+ if err != nil || len(st.Errors) != 0 {
+ t.Fatal(err, st.Errors)
+ }
+ if st.Files != 2 || st.Requests != 2 {
+ t.Fatalf("derived %d files into %d requests, want 2 and 2", st.Files, st.Requests)
+ }
+ got, n := points(t, z)
+ if n != 2 {
+ t.Fatalf("%d spool files, want one per landed file with points", n)
+ }
+ want := map[string]int64{
+ "main/input/claude-opus-5@10:00#s1": 15, "main/output/claude-opus-5@10:00#s1": 120,
+ "main/cacheRead/claude-opus-5@10:00#s1": 1900, "main/cacheCreation/claude-opus-5@10:00#s1": 50,
+ "main/input/claude-sonnet-5@10:01#s1": 7, "main/output/claude-sonnet-5@10:01#s1": 30,
+ "subagent/input/claude-opus-5@10:00#s1": 3, "subagent/output/claude-opus-5@10:00#s1": 8, "subagent/cacheRead/claude-opus-5@10:00#s1": 400,
+ }
+ for k, v := range want {
+ if got[k] != v {
+ t.Fatalf("%s = %d, want %d; all: %v", k, got[k], v, got)
+ }
+ }
+ if len(got) != len(want) {
+ t.Fatalf("%d points, want %d: %v", len(got), len(want), got)
+ }
+
+ // A second pass derives nothing; a new file is derived whole.
+ st, err = d.Pass(nil)
+ if err != nil || st.Files != 0 || st.Requests != 0 {
+ t.Fatalf("a second pass derived %d files: %v", st.Files, err)
+ }
+ land(t, z, "s1", "main", 3, []call{{id: "c5", model: "claude-opus-5", at: now, frags: 2, in: 1, out: 2}})
+ st, err = d.Pass([]string{"s1"})
+ if err != nil || st.Files != 1 || st.Requests != 1 {
+ t.Fatalf("a new file derived %d files into %d requests: %v", st.Files, st.Requests, err)
+ }
+ got, _ = points(t, z)
+ if got["main/input/claude-opus-5@11:00#s1"] != 1 {
+ t.Fatalf("the new file's call is missing: %v", got)
+ }
+
+ // A call counted in an earlier file is not counted again from a later
+ // one: the runtime re-emits records before a reset, and a file cut at
+ // a budget can hold the last fragment of a call the file before began.
+ land(t, z, "s1", "main", 4, []call{
+ {id: "c5", model: "claude-opus-5", at: now, frags: 1, in: 1, out: 2},
+ {id: "c1", model: "claude-opus-5", at: base.Add(10 * time.Second), frags: 1, in: 10, out: 100, read: 1000, wr: 50},
+ {id: "c6", model: "claude-opus-5", at: now.Add(time.Minute), frags: 1, in: 5, out: 6},
+ })
+ st, err = d.Pass([]string{"s1"})
+ if err != nil || st.Files != 1 || st.Requests != 1 {
+ t.Fatalf("the later file derived %d files into %d requests: %v", st.Files, st.Requests, err)
+ }
+ got, _ = points(t, z)
+ if got["main/input/claude-opus-5@11:00#s1"] != 1 || got["main/input/claude-opus-5@10:00#s1"] != 15 || got["main/input/claude-opus-5@11:01#s1"] != 5 {
+ t.Fatalf("a call reached from a later file was counted again: %v", got)
+ }
+}
+
+// The look-back bounds the first pass only: history older than it is
+// marked derived without points, a minute inside it is derived, and every
+// later pass derives whole files.
+func TestLookbackBoundsTheFirstPassOnly(t *testing.T) {
+ z := storage.NewZone(t.TempDir())
+ now := time.Date(2026, 9, 7, 12, 0, 0, 0, time.UTC)
+ land(t, z, "s1", "main", 1, []call{
+ {id: "old", model: "m", at: now.Add(-48 * time.Hour), frags: 1, in: 100, out: 100},
+ })
+ land(t, z, "s1", "main", 2, []call{
+ {id: "older", model: "m", at: now.Add(-25 * time.Hour), frags: 1, in: 40, out: 40},
+ {id: "recent", model: "m", at: now.Add(-2 * time.Hour), frags: 1, in: 4, out: 4},
+ })
+ d := &metrics.Deriver{Zone: z, Options: metrics.Options{Lookback: 24 * time.Hour, Version: "test", Now: func() time.Time { return now }}}
+ st, err := d.Pass(nil)
+ if err != nil || len(st.Errors) != 0 {
+ t.Fatal(err, st.Errors)
+ }
+ if st.Files != 2 || st.Skipped != 1 || st.Requests != 1 {
+ t.Fatalf("first pass: files=%d skipped=%d requests=%d, want 2 files, 1 skipped as older than the look-back, 1 request", st.Files, st.Skipped, st.Requests)
+ }
+ got, _ := points(t, z)
+ if got["main/input/m@10:00#s1"] != 4 || len(got) != 2 {
+ t.Fatalf("the first pass must derive the recent minute only: %v", got)
+ }
+ // After the first pass a file two days old is derived whole.
+ land(t, z, "s1", "main", 3, []call{{id: "late", model: "m", at: now.Add(-72 * time.Hour), frags: 1, in: 9, out: 9}})
+ st, err = d.Pass(nil)
+ if err != nil || st.Requests != 1 {
+ t.Fatalf("a later pass must derive an old file whole: requests=%d err=%v", st.Requests, err)
+ }
+}
diff --git a/internal/scenario/build.go b/internal/scenario/build.go
index ceb0c36..d004bfb 100644
--- a/internal/scenario/build.go
+++ b/internal/scenario/build.go
@@ -104,6 +104,10 @@ adapters:
- name: claude-code-local
enabled: true
source_root: %s
+ # The runtime's own metric family, derived from what is landed, with
+ # no look-back: a scenario is history by construction.
+ metrics: true
+ metrics_lookback: none
collector:
mode: once
# To export the session with asz push, name the receiver:
diff --git a/internal/scenario/expect/expect.go b/internal/scenario/expect/expect.go
index ce267cf..f0f7bb0 100644
--- a/internal/scenario/expect/expect.go
+++ b/internal/scenario/expect/expect.go
@@ -82,6 +82,7 @@ type Properties struct {
RepackKeepsStructure *bool `yaml:"repack_keeps_structure"`
PushFollowsTheWire *bool `yaml:"push_follows_the_wire"`
ViewCoversTheSession *bool `yaml:"view_covers_the_session"`
+ MetricsMatchThePlan *bool `yaml:"metrics_match_the_plan"`
}
// On reports whether a property is enabled.
diff --git a/internal/scenario/run/push.go b/internal/scenario/run/push.go
index 5ef8f34..9fcdc60 100644
--- a/internal/scenario/run/push.go
+++ b/internal/scenario/run/push.go
@@ -30,11 +30,13 @@ import (
"time"
logspb "go.opentelemetry.io/proto/otlp/logs/v1"
+ metricspb "go.opentelemetry.io/proto/otlp/metrics/v1"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/mock"
"github.com/apache/skywalking-ai-sessionizer/internal/export/otlp"
"github.com/apache/skywalking-ai-sessionizer/internal/export/otlp/otlptest"
+ "github.com/apache/skywalking-ai-sessionizer/internal/metrics"
"github.com/apache/skywalking-ai-sessionizer/internal/scenario"
"github.com/apache/skywalking-ai-sessionizer/internal/scenario/expect"
"github.com/apache/skywalking-ai-sessionizer/internal/storage"
@@ -55,10 +57,10 @@ var wireKinds = map[string]bool{
// receiver bounds a read on, delivery once and at least once, and the
// export path: writing every body back gives a root that verifies and
// folds the same.
-func pushFollowsTheWire(out, session string, f scenario.Format, want *expect.Push) ([]string, error) {
+func pushFollowsTheWire(out, session string, f scenario.Format, want *expect.Push, tokens map[string]int64, checkTokens bool) ([]string, error) {
var problems []string
for _, protocol := range []string{otlp.ProtocolGRPC, otlp.ProtocolHTTP} {
- found, err := pushOver(protocol, out, session, f, want)
+ found, err := pushOver(protocol, out, session, f, want, tokens, checkTokens)
if err != nil {
return nil, fmt.Errorf("over %s: %w", protocol, err)
}
@@ -69,7 +71,7 @@ func pushFollowsTheWire(out, session string, f scenario.Format, want *expect.Pus
return problems, nil
}
-func pushOver(protocol, out, session string, f scenario.Format, want *expect.Push) ([]string, error) {
+func pushOver(protocol, out, session string, f scenario.Format, want *expect.Push, tokens map[string]int64, checkTokens bool) ([]string, error) {
rcv, err := otlptest.Start()
if err != nil {
return nil, err
@@ -141,8 +143,8 @@ func pushOver(protocol, out, session string, f scenario.Format, want *expect.Pus
if len(st.Errors) != 0 {
return nil, fmt.Errorf("push: %v", st.Errors)
}
- if st.Files != len(files) || st.Requests != len(files) {
- bad("pushed %d files in %d requests; the session has %d files and a one-byte budget sends each alone", st.Files, st.Requests, len(files))
+ if st.Files != len(files) || st.Requests-st.Metrics != len(files) {
+ bad("pushed %d files in %d requests; the session has %d files and a one-byte budget sends each alone", st.Files, st.Requests-st.Metrics, len(files))
}
// Every request: the resource and the scope, as the page lists them.
@@ -284,8 +286,47 @@ func pushOver(protocol, out, session string, f scenario.Format, want *expect.Pus
}
}
+ // The metrics spool reached the wire under asz's identity, and the
+ // points say what the plan says: every call's usage once, under the
+ // stream that made it, name for name with the runtime's own exporter.
+ if checkTokens {
+ for _, p := range tokensOnTheWire(rcv, session) {
+ bad("%s", p)
+ }
+ got := map[string]int64{}
+ for _, req := range rcv.MetricsRequests() {
+ for _, rm := range req.GetResourceMetrics() {
+ for _, sm := range rm.GetScopeMetrics() {
+ for _, m := range sm.GetMetrics() {
+ if m.GetName() != metrics.TokenUsage {
+ continue
+ }
+ for _, dp := range m.GetSum().GetDataPoints() {
+ a := otlptest.Attrs(dp.GetAttributes())
+ got[a["query_source"]+"/"+a["type"]] += dp.GetAsInt()
+ }
+ }
+ }
+ }
+ }
+ for k, v := range tokens {
+ if got[k] != v {
+ bad("metrics: %s on the wire is %d, the plan says %d", k, got[k], v)
+ }
+ }
+ for k, v := range got {
+ if _, planned := tokens[k]; !planned {
+ bad("metrics: %s on the wire is %d, the plan has no such tokens", k, v)
+ }
+ }
+ if st.Metrics == 0 && len(tokens) > 0 {
+ bad("metrics: the plan has tokens but the push sent no metrics request")
+ }
+ }
+
// Sent once: a second pass sends nothing.
before := len(rcv.Requests())
+ beforeMetrics := len(rcv.MetricsRequests())
st, err = newPusher().Pass()
if err != nil {
return nil, err
@@ -293,6 +334,9 @@ func pushOver(protocol, out, session string, f scenario.Format, want *expect.Pus
if st.Files != 0 || len(rcv.Requests()) != before {
bad("a second pass sent %d files again", st.Files)
}
+ if st.Metrics != 0 || len(rcv.MetricsRequests()) != beforeMetrics {
+ bad("a second pass sent %d metrics requests again", st.Metrics)
+ }
// The export path: every body written to its path gives a root that
// verifies and folds the same.
@@ -328,6 +372,52 @@ func pushOver(protocol, out, session string, f scenario.Format, want *expect.Pus
return out2, nil
}
+// tokensOnTheWire checks the shape of every metrics request as the export
+// page states it: asz's identity on the resource, the runtime's metric with
+// its unit, monotonic delta sums, one-minute points that name the session.
+func tokensOnTheWire(rcv *otlptest.Receiver, session string) []string {
+ var out []string
+ bad := func(format string, a ...any) { out = append(out, "metrics: "+fmt.Sprintf(format, a...)) }
+ for i, req := range rcv.MetricsRequests() {
+ for _, rm := range req.GetResourceMetrics() {
+ res := otlptest.Attrs(rm.GetResource().GetAttributes())
+ for k, v := range map[string]string{
+ "service.name": "Scenario Check", "service.instance.id": "scenario-check", "service.layer": "AI_AGENT",
+ "telemetry.sdk.name": "asz", "telemetry.sdk.version": "check", "telemetry.sdk.language": "go",
+ } {
+ if res[k] != v {
+ bad("request %d resource %s is %q, want %q", i, k, res[k], v)
+ }
+ }
+ for _, sm := range rm.GetScopeMetrics() {
+ if sm.GetScope().GetName() != metrics.ScopeName {
+ bad("request %d scope is %q", i, sm.GetScope().GetName())
+ }
+ for _, m := range sm.GetMetrics() {
+ if m.GetName() != metrics.TokenUsage || m.GetUnit() != "tokens" {
+ bad("request %d carries metric %s %s", i, m.GetName(), m.GetUnit())
+ continue
+ }
+ sum := m.GetSum()
+ if sum == nil || sum.GetAggregationTemporality() != metricspb.AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA || !sum.GetIsMonotonic() {
+ bad("request %d: %s is not a monotonic delta sum", i, m.GetName())
+ }
+ for _, dp := range sum.GetDataPoints() {
+ a := otlptest.Attrs(dp.GetAttributes())
+ if a["session.id"] != session || a["type"] == "" || (a["query_source"] != metrics.SourceMain && a["query_source"] != metrics.SourceSubagent) || a["model"] == "" {
+ bad("request %d: a point carries %v", i, a)
+ }
+ if dp.GetTimeUnixNano() != dp.GetStartTimeUnixNano()+uint64(time.Minute) || dp.GetAsInt() <= 0 {
+ bad("request %d: a point is not one minute of a positive count", i)
+ }
+ }
+ }
+ }
+ }
+ }
+ return out
+}
+
// verifyDiffers compares what asz verify says of two roots holding the same
// session: the stream problems and the chain problems must be the same.
func verifyDiffers(a, b, session string) []string {
diff --git a/internal/scenario/run/run.go b/internal/scenario/run/run.go
index b1498aa..d5b48dd 100644
--- a/internal/scenario/run/run.go
+++ b/internal/scenario/run/run.go
@@ -30,6 +30,7 @@ import (
"time"
"github.com/apache/skywalking-ai-sessionizer/internal/adapters/claudecode"
+ "github.com/apache/skywalking-ai-sessionizer/internal/metrics"
"github.com/apache/skywalking-ai-sessionizer/internal/parse"
"github.com/apache/skywalking-ai-sessionizer/internal/repack"
"github.com/apache/skywalking-ai-sessionizer/internal/scenario"
@@ -161,6 +162,11 @@ func checkFormat(sc *scenario.Scenario, ex *expect.File, f scenario.Format, out
if err := collect(f, out); err != nil {
return nil, "", err
}
+ // The runtime's metric family, derived as the collector derives it,
+ // after each landing and before anything later happens to the root.
+ if err := derive(out); err != nil {
+ return nil, "", err
+ }
round, err = parseAll(out, built.Session, ex.Parse.MaxRoundBytes)
if err != nil {
rep.fail("%s %s: parse: %v", f, name, err)
@@ -235,7 +241,9 @@ func checkFormat(sc *scenario.Scenario, ex *expect.File, f scenario.Format, out
{"immutable_rounds", ex.Properties.ImmutableRounds, func() ([]string, error) { return expect.ImmutableRounds(out, session) }},
{"records_well_formed", ex.Properties.RecordsWellFormed, func() ([]string, error) { return expect.RecordsWellFormed(out, session) }},
{"repack_keeps_structure", ex.Properties.RepackKeepsStructure, func() ([]string, error) { return repackKeepsStructure(out, session, ex.Parse.MaxRoundBytes) }},
- {"push_follows_the_wire", ex.Properties.PushFollowsTheWire, func() ([]string, error) { return pushFollowsTheWire(out, session, f, ex.Push) }},
+ {"push_follows_the_wire", ex.Properties.PushFollowsTheWire, func() ([]string, error) {
+ return pushFollowsTheWire(out, session, f, ex.Push, expectedTokens(built.Plan), expect.On(ex.Properties.MetricsMatchThePlan))
+ }},
{"view_covers_the_session", ex.Properties.ViewCoversTheSession, func() ([]string, error) { return expect.ViewCoversTheSession(out, session) }},
{"reproducible", ex.Properties.Reproducible, func() ([]string, error) {
// Two parses of identical landed evidence must produce identical
@@ -329,6 +337,20 @@ func lose(out, session string, l expect.Lose, ctx *expect.Context) (string, erro
return filepath.ToSlash(rel), nil
}
+// derive writes the runtime's metric family for what is landed and not yet
+// derived, with no look-back: a scenario is history by construction.
+func derive(out string) error {
+ d := &metrics.Deriver{Zone: storage.NewZone(out), Options: metrics.Options{Version: "check", Now: func() time.Time { return time.Date(2026, 1, 3, 0, 0, 0, 0, time.UTC) }}}
+ st, err := d.Pass(nil)
+ if err != nil {
+ return err
+ }
+ if len(st.Errors) != 0 {
+ return fmt.Errorf("metrics: %v", st.Errors)
+ }
+ return nil
+}
+
// collect lands a runtime format's source through its adapter. An sd build
// is landed already.
func collect(f scenario.Format, out string) error {
@@ -515,3 +537,38 @@ func names(p *scenario.Plan) map[string]string {
}
return out
}
+
+// expectedTokens is what the plan says the runtime's exporter would have
+// counted: every call's usage once, whatever its fragments repeat, under
+// the stream that made it, main or a child. A lost call left no record, a
+// lost stream left no file, and a replayed copy is the same call again.
+func expectedTokens(p *scenario.Plan) map[string]int64 {
+ lost := map[string]bool{}
+ for _, s := range p.Streams {
+ if s.Lost {
+ lost[s.ID] = true
+ }
+ }
+ seen := map[string]bool{}
+ out := map[string]int64{}
+ for i := range p.Events {
+ e := &p.Events[i]
+ if e.Kind != scenario.EvFragment || e.Lost || lost[e.Stream] || seen[e.Call] {
+ continue
+ }
+ seen[e.Call] = true
+ source := metrics.SourceSubagent
+ if e.Stream == "main" {
+ source = metrics.SourceMain
+ }
+ for typ, n := range map[string]int{
+ metrics.TypeInput: e.Usage.In, metrics.TypeOutput: e.Usage.Out,
+ metrics.TypeCacheRead: e.Usage.CacheRead, metrics.TypeCacheCreation: e.Usage.CacheWrite,
+ } {
+ if n > 0 {
+ out[source+"/"+typ] += int64(n)
+ }
+ }
+ }
+ return out
+}
diff --git a/internal/storage/spool.go b/internal/storage/spool.go
new file mode 100644
index 0000000..19119d7
--- /dev/null
+++ b/internal/storage/spool.go
@@ -0,0 +1,159 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package storage
+
+import (
+ "bufio"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "sort"
+ "strconv"
+ "strings"
+ "time"
+)
+
+// SpoolDir is where OTLP metrics requests wait under a storage root: the
+// ones the local adapter derived from landed files, and the ones the
+// runtime's own exporter sent to the receiver adapter. Each is one file,
+// bytes as produced or as received, write-once like every landed file, and
+// asz push sends them in order and records which were sent.
+const SpoolDir = "_metrics"
+
+// Spool is the metrics spool of one storage root.
+type Spool struct{ dir string }
+
+// NewSpool opens the spool of a root; nothing is created until Put.
+func NewSpool(z *Zone) *Spool { return &Spool{dir: filepath.Join(z.Root(), SpoolDir)} }
+
+// Dir is the spool directory.
+func (s *Spool) Dir() string { return s.dir }
+
+// SpoolFile is one request waiting in the spool.
+type SpoolFile struct {
+ Path string
+ Seq uint64
+ Source string
+}
+
+// spoolName builds a spool filename: sortable by the time it was put and
+// then by a sequence, which keeps two requests of one nanosecond apart.
+func spoolName(stamp string, seq uint64, source string) string {
+ return fmt.Sprintf("metrics-%s-%06d-%s.pb", stamp, seq, source)
+}
+
+// Put writes one request to the spool and returns its path. The sequence
+// comes from the spool's own state file under the spool's lock, so two
+// writers in one root, the local derivation and the receiver, never take
+// the same number.
+func (s *Spool) Put(source string, data []byte, now time.Time) (string, error) {
+ if err := os.MkdirAll(s.dir, 0o755); err != nil {
+ return "", err
+ }
+ lock, err := lockDir(s.dir)
+ if err != nil {
+ return "", err
+ }
+ defer func() { _ = lock.Unlock() }()
+ seq, err := s.nextSeq()
+ if err != nil {
+ return "", err
+ }
+ path := filepath.Join(s.dir, spoolName(Stamp(now), seq, source))
+ if err := WriteAtomic(path, PermLanded, func(w io.Writer) error {
+ _, err := w.Write(data)
+ return err
+ }); err != nil {
+ return "", err
+ }
+ if err := s.saveSeq(seq+1, now); err != nil {
+ return "", err
+ }
+ return path, nil
+}
+
+func (s *Spool) statePath() string { return filepath.Join(s.dir, "spool.state") }
+
+// nextSeq reads the counter, or recovers it from the files when the state
+// is gone: the files are the authority, as they are for a session.
+func (s *Spool) nextSeq() (uint64, error) {
+ f, err := os.Open(s.statePath())
+ if err == nil {
+ defer f.Close()
+ sc := bufio.NewScanner(f)
+ for sc.Scan() {
+ if v, ok := strings.CutPrefix(sc.Text(), "next_seq "); ok {
+ return strconv.ParseUint(strings.TrimSpace(v), 10, 64)
+ }
+ }
+ return 0, errors.New("storage: spool.state carries no next_seq")
+ }
+ if !os.IsNotExist(err) {
+ return 0, err
+ }
+ files, err := s.List()
+ if err != nil {
+ return 0, err
+ }
+ var next uint64 = 1
+ for _, sf := range files {
+ if sf.Seq >= next {
+ next = sf.Seq + 1
+ }
+ }
+ return next, nil
+}
+
+func (s *Spool) saveSeq(next uint64, now time.Time) error {
+ return WriteAtomic(s.statePath(), PermState, func(w io.Writer) error {
+ _, err := fmt.Fprintf(w, "schema 1\nupdated_at %s\nnext_seq %d\n", now.UTC().Format(time.RFC3339Nano), next)
+ return err
+ })
+}
+
+// List returns every request in the spool, in the order they were put.
+func (s *Spool) List() ([]SpoolFile, error) {
+ items, err := os.ReadDir(s.dir)
+ if err != nil {
+ if os.IsNotExist(err) {
+ return nil, nil
+ }
+ return nil, err
+ }
+ var out []SpoolFile
+ for _, it := range items {
+ name := it.Name()
+ if it.IsDir() || !strings.HasPrefix(name, "metrics-") || !strings.HasSuffix(name, ".pb") {
+ continue
+ }
+ // metrics-<stamp>-<seq>-<source>.pb
+ parts := strings.Split(strings.TrimSuffix(name, ".pb"), "-")
+ if len(parts) < 4 {
+ continue
+ }
+ seq, err := strconv.ParseUint(parts[len(parts)-2], 10, 64)
+ if err != nil {
+ continue
+ }
+ out = append(out, SpoolFile{Path: filepath.Join(s.dir, name), Seq: seq, Source: parts[len(parts)-1]})
+ }
+ sort.Slice(out, func(i, j int) bool { return out[i].Seq < out[j].Seq })
+ return out, nil
+}
diff --git a/tests/boundary/boundary_test.go b/tests/boundary/boundary_test.go
index c2fcc88..fe37f81 100644
--- a/tests/boundary/boundary_test.go
+++ b/tests/boundary/boundary_test.go
@@ -40,7 +40,7 @@ var (
// The exporter is the collector side's way out: it reads the storage root
// and sends, and must not know how anything is assembled or shown. A
// scenario's writers land input, so they are collector side too.
- collectorSide = []string{"internal/adapters/", "internal/export/", "internal/scenario/"}
+ collectorSide = []string{"internal/adapters/", "internal/export/", "internal/metrics/", "internal/scenario/"}
serverSide = []string{"internal/assemble/", "internal/parse/", "internal/view/", "internal/verify/", "pkg/sessionflow/", "pkg/sessionview/", "internal/scenario/expect/"}
// The scenario runner wires both sides, as the command does. Only the
// command and the tests may import it.
diff --git a/tools/collectorcheck/main.go b/tools/collectorcheck/main.go
index 515d4bb..b48a3f9 100644
--- a/tools/collectorcheck/main.go
+++ b/tools/collectorcheck/main.go
@@ -43,8 +43,27 @@ import (
)
// The shape the Collector's file exporter writes: one JSON line per
-// request, in OTLP's JSON mapping.
+// request, in OTLP's JSON mapping; a logs request or a metrics request.
type export struct {
+ ResourceMetrics []struct {
+ Resource struct {
+ Attributes []attr `json:"attributes"`
+ } `json:"resource"`
+ ScopeMetrics []struct {
+ Metrics []struct {
+ Name string `json:"name"`
+ Unit string `json:"unit"`
+ Sum struct {
+ AggregationTemporality any `json:"aggregationTemporality"`
+ IsMonotonic bool `json:"isMonotonic"`
+ DataPoints []struct {
+ AsInt string `json:"asInt"`
+ Attributes []attr `json:"attributes"`
+ } `json:"dataPoints"`
+ } `json:"sum"`
+ } `json:"metrics"`
+ } `json:"scopeMetrics"`
+ } `json:"resourceMetrics"`
ResourceLogs []struct {
Resource struct {
Attributes []attr `json:"attributes"`
@@ -113,6 +132,10 @@ func run(root, logs string) error {
var recs []rec
services := map[string]int{}
requests := 0
+ // The token metric, as the Collector received it: points by query
+ // source and type, and how many requests carried it.
+ tokens := map[string]int64{}
+ metricRequests := 0
for _, line := range bytes.Split(raw, []byte("\n")) {
if len(bytes.TrimSpace(line)) == 0 {
continue
@@ -121,6 +144,28 @@ func run(root, logs string) error {
if err := json.Unmarshal(line, &e); err != nil {
return fmt.Errorf("a line the exporter wrote is not OTLP JSON: %w", err)
}
+ if len(e.ResourceMetrics) > 0 {
+ metricRequests++
+ for _, rm := range e.ResourceMetrics {
+ res := attrs(rm.Resource.Attributes)
+ if res["service.name"] == "" || res["service.layer"] == "" || res["telemetry.sdk.name"] != "asz" {
+ return fmt.Errorf("a metrics request carries the resource %v; asz's identity is missing", res)
+ }
+ for _, sm := range rm.ScopeMetrics {
+ for _, m := range sm.Metrics {
+ if m.Name != "claude_code.token.usage" || m.Unit != "tokens" || !m.Sum.IsMonotonic {
+ return fmt.Errorf("a metrics request carries %s %s monotonic=%v", m.Name, m.Unit, m.Sum.IsMonotonic)
+ }
+ for _, dp := range m.Sum.DataPoints {
+ a := attrs(dp.Attributes)
+ n, _ := strconv.ParseInt(dp.AsInt, 10, 64)
+ tokens[a["query_source"]+"/"+a["type"]] += n
+ }
+ }
+ }
+ }
+ continue
+ }
requests++
for _, rl := range e.ResourceLogs {
res := attrs(rl.Resource.Attributes)
@@ -257,6 +302,14 @@ func run(root, logs string) error {
talks := len(b.NodesByKind(model.KindTalk))
fmt.Printf("%s: %d talks, %d nodes, round %d, rebuilt from the Collector and verified\n", s, talks, len(b.Nodes), b.Round)
}
- fmt.Printf("ok: %d requests, %d records, kinds %v, resources %v\n", requests, len(recs), kinds, services)
+ // The metrics the local adapter derived reached the Collector too:
+ // every planned session has calls with usage, so tokens must arrive
+ // under both query sources and all four types.
+ for _, k := range []string{"main/input", "main/output", "main/cacheRead", "main/cacheCreation", "subagent/input", "subagent/output"} {
+ if tokens[k] <= 0 {
+ return fmt.Errorf("no %s tokens reached the Collector; it received %v", k, tokens)
+ }
+ }
+ fmt.Printf("ok: %d requests, %d records, kinds %v, resources %v; %d metrics requests, tokens %v\n", requests, len(recs), kinds, services, metricRequests, tokens)
return nil
}
diff --git a/tools/e2e-collector.sh b/tools/e2e-collector.sh
index e65dff1..f4a74c6 100755
--- a/tools/e2e-collector.sh
+++ b/tools/e2e-collector.sh
@@ -46,6 +46,9 @@ service:
logs:
receivers: [otlp]
exporters: [file]
+ metrics:
+ receivers: [otlp]
+ exporters: [file]
YAML
# push_over PROTOCOL ENDPOINT: build a session, start a Collector, push the