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 05:59:19 UTC
(skywalking-ai-sessionizer) branch main updated: Verify both metric sources through a real Collector, and run root commands without a local adapter
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 c977556 Verify both metric sources through a real Collector, and run root commands without a local adapter
c977556 is described below
commit c9775563b86c1d427880070c6efc12104ac209f8
Author: Wu Sheng <wu...@foxmail.com>
AuthorDate: Mon Sep 7 13:59:10 2026 +0800
Verify both metric sources through a real Collector, and run root commands without a local adapter
The Collector job verified one source of the token metric, the local
derivation. It now verifies the other as well: a second root runs the
receiver adapter alone, the Collector project's own OTLP exporter,
telemetrygen, sends it the metric over gRPC and then over HTTP, the spool
is pushed to the same Collector, and the check requires points from both
sources, told apart by the sender attribute the exporter is given. A
received request is forwarded as sent, so its unit is the exporter's; a
derived request must carry tokens.
That run found a gap in the command dispatch: push, parse, verify and the
other commands that read or send the root ran once per enabled local
adapter, so a root fed by the receiver alone was never pushed. They now
run once, with the local adapter's settings when one is enabled and the
defaults otherwise; sources and collect stay the local adapter's own, and
collect with receivers alone keeps the process alive for them.
---
cmd/asz/main.go | 47 +++++++++++++++++++++++++++++++++-----
collectorcheck | Bin 4741170 -> 4741170 bytes
tools/collectorcheck/main.go | 25 ++++++++++++++++++---
tools/e2e-collector.sh | 52 +++++++++++++++++++++++++++++++++++++++++++
4 files changed, 116 insertions(+), 8 deletions(-)
diff --git a/cmd/asz/main.go b/cmd/asz/main.go
index fe2e0c5..87d6bb7 100644
--- a/cmd/asz/main.go
+++ b/cmd/asz/main.go
@@ -191,14 +191,51 @@ func main() {
fmt.Fprintf(os.Stderr, "skipping unknown adapter %q\n", ad.Name)
}
}
- for _, ad := range local {
+ // sources and collect are the local adapter's own. view serves the
+ // root, with the adapter's refresh when one is enabled. Every other
+ // command reads or sends the root and runs once, with the local
+ // adapter's settings when one is enabled and the defaults otherwise, so
+ // a root fed by the receiver alone can still be parsed, verified and
+ // pushed.
+ switch cmd {
+ case "sources":
+ if len(local) == 0 {
+ fatal(fmt.Errorf("%s: no enabled %s adapter", cmd, config.AdapterClaudeCodeLocal))
+ }
+ for _, ad := range local {
+ if err := run(cfg, ad, *once); err != nil {
+ fatal(err)
+ }
+ }
+ case "collect":
+ for _, ad := range local {
+ if err := run(cfg, ad, *once); err != nil {
+ fatal(err)
+ }
+ }
+ if len(local) == 0 {
+ if receivers == 0 {
+ fatal(fmt.Errorf("%s: no enabled adapter", cmd))
+ }
+ // Nothing else keeps the process alive: the receivers do.
+ select {}
+ }
+ case "view":
+ var ad config.Adapter
+ if len(local) > 0 {
+ ad = local[0]
+ }
+ if err := run(cfg, ad, *once); err != nil {
+ fatal(err)
+ }
+ default:
+ ad := config.Default().Adapters[0]
+ if len(local) > 0 {
+ ad = local[0]
+ }
if err := run(cfg, ad, *once); err != nil {
fatal(err)
}
- }
- if len(local) == 0 && receivers > 0 {
- // Nothing else keeps the process alive: the receivers do.
- select {}
}
}
diff --git a/collectorcheck b/collectorcheck
index 312a076..e4b17ae 100755
Binary files a/collectorcheck and b/collectorcheck differ
diff --git a/tools/collectorcheck/main.go b/tools/collectorcheck/main.go
index b48a3f9..95ff84e 100644
--- a/tools/collectorcheck/main.go
+++ b/tools/collectorcheck/main.go
@@ -136,6 +136,9 @@ func run(root, logs string) error {
// source and type, and how many requests carried it.
tokens := map[string]int64{}
metricRequests := 0
+ // Points from the runtime's exporter, received by the claude-code-otlp
+ // adapter and pushed on: the exporter marks them with sender.
+ received := 0
for _, line := range bytes.Split(raw, []byte("\n")) {
if len(bytes.TrimSpace(line)) == 0 {
continue
@@ -153,12 +156,22 @@ func run(root, logs string) error {
}
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)
+ if m.Name != "claude_code.token.usage" || !m.Sum.IsMonotonic {
+ return fmt.Errorf("a metrics request carries %s monotonic=%v", m.Name, m.Sum.IsMonotonic)
}
for _, dp := range m.Sum.DataPoints {
a := attrs(dp.Attributes)
n, _ := strconv.ParseInt(dp.AsInt, 10, 64)
+ if a["sender"] == "telemetrygen" {
+ // Received from the exporter and forwarded as sent:
+ // the unit is the exporter's to set, and this one
+ // sets none.
+ received++
+ continue
+ }
+ if m.Unit != "tokens" {
+ return fmt.Errorf("a derived point of %s carries the unit %q, want tokens", m.Name, m.Unit)
+ }
tokens[a["query_source"]+"/"+a["type"]] += n
}
}
@@ -310,6 +323,12 @@ func run(root, logs string) error {
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)
+ // The other source reached the Collector too: points a real external
+ // exporter sent to asz's receiver, landed, and pushed on.
+ if received == 0 {
+ return fmt.Errorf("no point from the exporter through the receiver reached the Collector; it received %d metrics requests", metricRequests)
+ }
+ fmt.Printf("ok: %d requests, %d records, kinds %v, resources %v; %d metrics requests, derived tokens %v, %d points received from the exporter\n",
+ requests, len(recs), kinds, services, metricRequests, tokens, received)
return nil
}
diff --git a/tools/e2e-collector.sh b/tools/e2e-collector.sh
index f4a74c6..bb26cf7 100755
--- a/tools/e2e-collector.sh
+++ b/tools/e2e-collector.sh
@@ -22,8 +22,15 @@
set -eu
cd "$(dirname "$0")/.."
IMAGE="${OTELCOL_IMAGE:-otel/opentelemetry-collector-contrib:0.158.0}"
+# The Collector project's own OTLP exporter, telemetrygen: a real external
+# exporter to point at asz's receiver, in place of Claude Code's.
+GEN_IMAGE="${TELEMETRYGEN_IMAGE:-ghcr.io/open-telemetry/opentelemetry-collector-contrib/telemetrygen:v0.158.0}"
GRPC_PORT="${OTELCOL_GRPC_PORT:-14317}"
HTTP_PORT="${OTELCOL_HTTP_PORT:-14318}"
+# Where asz's receiver listens for the exporter, reachable from a container
+# as host.docker.internal. Both address families, since the name may resolve
+# to either.
+RECEIVE_PORT="${ASZ_RECEIVE_PORT:-14417}"
WORK="$(mktemp -d)"
trap 'docker rm -f asz-e2e-otelcol >/dev/null 2>&1 || true; rm -rf "$WORK"' EXIT
@@ -92,6 +99,51 @@ YAML
if [ "$i" = 5 ]; then echo "push over $protocol failed five times"; exit 1; fi
sleep 2
done
+
+ # The second source of the same metric: the runtime's own exporter. A
+ # second root runs the receiver adapter alone, a real external exporter
+ # sends it the token metric over gRPC and over HTTP, and the spool is
+ # pushed to the same Collector. The check then wants points from both
+ # sources, told apart by the sender attribute the exporter is given.
+ received="$WORK/$protocol-received"
+ mkdir -p "$received"
+ cat > "$received/asz.yaml" <<YAML
+storage:
+ root: $received
+adapters:
+ - name: claude-code-local
+ enabled: false
+ - name: claude-code-otlp
+ enabled: true
+ listen: ":$RECEIVE_PORT"
+ metrics: true
+export:
+ otlp:
+ protocol: $protocol
+ endpoint: $endpoint
+YAML
+ ./bin/asz collect -config "$received/asz.yaml" > "$received/collect.log" 2>&1 &
+ receiver_pid=$!
+ for i in $(seq 1 30); do
+ if curl -s -o /dev/null "http://127.0.0.1:$RECEIVE_PORT/v1/metrics"; then break; fi
+ sleep 1
+ done
+ echo "== the exporter sends to asz's receiver on $RECEIVE_PORT, gRPC then HTTP"
+ for transport in "" "--otlp-http"; do
+ # shellcheck disable=SC2086
+ docker run --rm --add-host=host.docker.internal:host-gateway "$GEN_IMAGE" metrics \
+ --otlp-endpoint "host.docker.internal:$RECEIVE_PORT" --otlp-insecure $transport \
+ --metrics 3 --rate 0 --metric-type Sum --aggregation-temporality delta \
+ --otlp-metric-name claude_code.token.usage --service claude-code \
+ --telemetry-attributes 'type="input"' --telemetry-attributes 'query_source="main"' \
+ --telemetry-attributes 'session.id="telemetrygen"' --telemetry-attributes 'sender="telemetrygen"' >/dev/null 2>&1 \
+ || { echo "the exporter could not send to the receiver"; cat "$received/collect.log"; exit 1; }
+ done
+ kill "$receiver_pid" >/dev/null 2>&1 || true
+ wait "$receiver_pid" 2>/dev/null || true
+ head -3 "$received/collect.log"
+ ls "$received/_metrics" | grep -c '\.pb$' | sed 's/^/spooled requests from the exporter: /'
+ ./bin/asz push -once -config "$received/asz.yaml"
# The file exporter flushes on its own schedule; give it a moment.
for i in $(seq 1 20); do
if [ -s "$WORK/otelcol/logs.json" ]; then break; fi