Skip to content

Commit c2e29fa

Browse files
feat(telemetry): include org_id in heartbeat payload (v9.1 #2277) (#171)
Brings Go SDK telemetry up to parity with the platform's startup_telemetry.go emitter. Every heartbeat body now identifies which deployment-organization emitted it via an org_id field on the wire. Three sources, precedence order: 1. ORG_ID env var when set (the operator's explicit configuration on self-hosted; the cs_<uuid> tenant identifier on Community SaaS). 2. local-dev-org sentinel when unset (default-config Community-mode developers). Always emitted. The receiver-side struct at ee/platform/checkpoint-service/pkg/telemetry/telemetry.go:454 already carries OrgID with omitempty for backward compat against pre-v9.1 SDKs; new SDKs always send it. Tests added: - TestTelemetryOrgID_HelperUnit — env→sentinel fallback (3 subtests: ORG_ID set wins, unset returns sentinel, cs_<uuid> shape passes). - TestSendTelemetryPing_OrgIDOnWire — functional E2E across 3 cases: operator-supplied, cs_<uuid> Community SaaS, sentinel. Asserts both decoded struct AND wire-literal "org_id":"..." JSON. The wire-literal assertion defends against tag-removal mutations. - TestSendTelemetryPing_OrgIDAlwaysPresent — mutation guard: with ORG_ID="" the wire MUST carry "org_id":"local-dev-org". Removing the OrgID populate line from the payload struct → org_id:"" on the wire → this test fails. - Existing TestSendTelemetryPing_Success extended to assert org_id sentinel under ORG_ID="". Mutation-tested in worktree: removing the populate line via sed flipped TestSendTelemetryPing_OrgIDAlwaysPresent to FAIL with expected wire body to ALWAYS contain org_id (sentinel when env unset), got: {... ,"org_id":""} confirming the test catches accidental regression. Output captured in /tmp/v91-org-id-telemetry-<UTC>/sdk-go/mutation-test-fails-as-expected.log. Runtime proof at runtime-e2e/v91_org_id_telemetry/main.go — in-process HTTP listener consumes the SDK's real telemetry POST and inspects the wire body for org_id. Three modes exercised at PR time: - ORG_ID=acme-corp → org_id:"acme-corp" PASS - unset ORG_ID → org_id:"local-dev-org" PASS - ORG_ID=cs_<uuid> → org_id:"cs_<uuid>" PASS (See r2-{acme-corp,sentinel,cs-uuid}.log in /tmp.) Companion wording sweep alongside the new field: - telemetry.go:279 [AxonFlow] log line "Anonymous telemetry enabled" -> "Telemetry enabled" — the operator-supplied ORG_ID on self-hosted is not anonymized. - heartbeat.go:19,68 internal comment wording softened similarly. - runtime-e2e/sandbox_telemetry_stream_tag/README.md "anonymous heartbeat" -> "heartbeat". CHANGELOG.md v8.1.0 section expanded with org_id Added bullet + the log-wording change under Changed. This SDK PR is one of 9 across all 5 SDKs + 4 plugins. Privacy.html update landed first on axonflow-landing main: see axonflow-landing#129 (merged 2026-05-21). Refs: getaxonflow/axonflow-enterprise#2230 (Epic v9 identity) Refs: getaxonflow/axonflow-enterprise#2277 (v9.1 SDK + plugin org_id) Signed-off-by: Saurabh Jain <saurabh.jain@getaxonflow.com>
1 parent 6be11d3 commit c2e29fa

7 files changed

Lines changed: 349 additions & 8 deletions

File tree

CHANGELOG.md

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,11 +21,36 @@ when no `ClientID` is configured.
2121
value, so caller-supplied values are harmless (no spoofing surface).
2222
Set in `addAuthHeaders` (`audit.go`), the canonical funnel for all
2323
SDK HTTP requests.
24+
- **`org_id` field in the telemetry heartbeat body (v9.1 preflight, #2277).**
25+
Brings SDK telemetry up to parity with the platform's `startup_telemetry.go`
26+
emitter — every heartbeat now identifies which deployment-organization
27+
emitted it. Two sources in precedence order:
28+
1. The `ORG_ID` env var when set (the operator's explicit configuration on
29+
self-hosted deployments, or the `cs_<uuid>` tenant identifier on
30+
Community SaaS).
31+
2. Otherwise the `local-dev-org` sentinel (default-config Community-mode
32+
developers).
33+
The receiver (`ee/platform/checkpoint-service/pkg/telemetry/telemetry.go`)
34+
already accepts the field with `omitempty` for backward compat with
35+
pre-v8.1 SDKs that don't send it. New SDKs always send it. Honors
36+
`AXONFLOW_TELEMETRY=off` like every other heartbeat field. See
37+
`axonflow-landing/content/privacy.html` for the customer-facing
38+
commitment that covers this field.
39+
40+
### Changed
41+
42+
- **Telemetry-enabled log line** softened from "Anonymous telemetry enabled"
43+
to "Telemetry enabled" to stay coherent with the v9.1 `org_id` addition
44+
(the operator-supplied `ORG_ID` on self-hosted is not anonymized; only
45+
the `instance_id` and `cs_<uuid>` Community SaaS identifier remain
46+
anonymous-by-design).
2447

2548
### Compatibility
2649

2750
- Backward-compatible against v8 and v9 platforms: v8 agents ignore the
2851
unknown header; v9 agents derive identity from Basic Auth regardless.
52+
- `org_id` is an additive field — the receiver's `omitempty` allows
53+
legacy SDK builds to keep working unchanged.
2954
- No SDK config changes. No removed fields. No changed defaults.
3055

3156
## [8.0.0] - 2026-05-09 — Decision History API + policy_version recorded on every decision + telemetry simplification

heartbeat.go

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,9 @@ import (
1616
// tests can override them — the test files in this package back up the
1717
// original values and restore them on cleanup.
1818
var (
19-
// heartbeatInterval bounds how often a single machine sends an anonymous
19+
// heartbeatInterval bounds how often a single machine sends a
2020
// telemetry ping. Aligned with the plugin "7-day heartbeat" contract:
21-
// "at most one anonymous heartbeat per environment every 7 days during
21+
// "at most one heartbeat per environment every 7 days during
2222
// SDK activity."
2323
heartbeatInterval = 7 * 24 * time.Hour
2424

@@ -65,8 +65,8 @@ func newHeartbeatState() *heartbeatState {
6565
// sharedHeartbeat is the process-global heartbeat singleton. ALL AxonFlow
6666
// clients in this process consult the same gate, so concurrent NewClient
6767
// calls before any stamp exists coalesce onto a single ping (per the
68-
// "at most one anonymous heartbeat per environment" contract). Tests
69-
// override via replaceHeartbeatStateForTest.
68+
// "at most one heartbeat per environment" contract). Tests override via
69+
// replaceHeartbeatStateForTest.
7070
var (
7171
sharedHeartbeat = newHeartbeatState()
7272
sharedHeartbeatMu sync.Mutex // guards swapping the singleton in tests

runtime-e2e/sandbox_telemetry_stream_tag/README.md

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
11
# Runtime proof — Sandbox-mode telemetry fires with stream=sandbox (v8)
22

3-
Verifies the v8 contract: a `Sandbox()`-constructed client produces an
4-
anonymous heartbeat ping that lands in checkpoint DynamoDB with the row
5-
tagged `stream="sandbox"`.
3+
Verifies the v8 contract: a `Sandbox()`-constructed client produces a
4+
heartbeat ping that lands in checkpoint DynamoDB with the row tagged
5+
`stream="sandbox"`.
66

77
## When to run
88

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
# Runtime proof — `org_id` in SDK telemetry payload (v9.1)
2+
3+
Verifies the v9.1 contract: every SDK telemetry ping body carries an
4+
`org_id` field, populated from the `ORG_ID` env var with a
5+
`local-dev-org` sentinel fallback. Issue #2277.
6+
7+
## When to run
8+
9+
- **Local development** — single-binary `go run`, no infrastructure
10+
prerequisites. The proof spins up its own in-process HTTP listener
11+
to receive the SDK's outgoing telemetry POST and inspects the wire
12+
body.
13+
14+
- **CI** — covered by the unit/functional tests in
15+
`telemetry_test.go::TestSendTelemetryPing_OrgIDOnWire` and
16+
`TestSendTelemetryPing_OrgIDAlwaysPresent`. This runtime proof is a
17+
redundant real-stack confirmation, not a CI gate.
18+
19+
## Usage
20+
21+
```sh
22+
# ORG_ID set — operator-supplied (self-hosted) or cs_<uuid> (Community SaaS):
23+
ORG_ID=acme-corp go run runtime-e2e/v91_org_id_telemetry/main.go
24+
25+
# ORG_ID unset — local-dev-org sentinel:
26+
unset ORG_ID && go run runtime-e2e/v91_org_id_telemetry/main.go
27+
```
28+
29+
Expected output:
30+
31+
```
32+
PASS: telemetry wire payload carries org_id="acme-corp" (expected="acme-corp")
33+
Wire body: {"telemetry_type":"sdk", ... ,"org_id":"acme-corp"}
34+
```
35+
36+
## What it asserts
37+
38+
1. The SDK constructed under any config emits a telemetry POST within
39+
2 seconds of `NewClient`.
40+
2. The POST body is valid JSON.
41+
3. The body has an `org_id` key.
42+
4. The value matches `$ORG_ID` (when set) or `local-dev-org` (when
43+
unset).
44+
45+
## Mutation proof
46+
47+
Remove the `OrgID: telemetryOrgID(),` line from `telemetry.go`'s
48+
`payload := telemetryPayload{...}` block and rerun. The proof exits 1
49+
with `FAIL: telemetry org_id = "", want "local-dev-org"`.
50+
51+
## Cross-SDK parity
52+
53+
Companion runtime-e2e tests will land in the other 4 SDKs under the
54+
same `runtime-e2e/v91_org_id_telemetry/` subdirectory as part of the
55+
#2277 cross-repo rollout:
56+
57+
- `axonflow-sdk-python/runtime-e2e/v91_org_id_telemetry/`
58+
- `axonflow-sdk-typescript/runtime-e2e/v91_org_id_telemetry/`
59+
- `axonflow-sdk-java/runtime-e2e/v91_org_id_telemetry/`
60+
- `axonflow-sdk-rust/runtime-e2e/v91_org_id_telemetry/`
61+
62+
All five SDKs send the field with the same wire name (`org_id`),
63+
same sentinel value (`local-dev-org`), and the same precedence
64+
(env > sentinel). The captured-payload transcript across the five
65+
will be posted as the delivery comment on Epic #2230.
Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
//go:build ignore
2+
3+
// runtime-e2e/v91_org_id_telemetry/main.go
4+
//
5+
// Real-wire test of the SDK's v9.1 org_id telemetry field (#2277). The
6+
// SDK does not expose its internal http.Client, so we run a tiny
7+
// in-process HTTP server that pretends to be the checkpoint receiver,
8+
// inspects the raw JSON body for the org_id field, and exits with the
9+
// verdict. Bytes flow real → real through net/http on both sides.
10+
//
11+
// Run via:
12+
//
13+
// # ORG_ID set — operator-supplied or cs_<uuid>:
14+
// ORG_ID=acme-corp go run runtime-e2e/v91_org_id_telemetry/main.go
15+
//
16+
// # ORG_ID unset — sentinel:
17+
// unset ORG_ID && go run runtime-e2e/v91_org_id_telemetry/main.go
18+
//
19+
// This proof goes alongside the unit-test wire assertions in
20+
// telemetry_test.go. Both paths run against the same SDK build; this
21+
// one additionally exercises the real net.Listen / http.Serve stack
22+
// rather than httptest.Server (which is functionally equivalent but
23+
// less faithful to a production-shaped wire path).
24+
25+
package main
26+
27+
import (
28+
"context"
29+
"encoding/json"
30+
"fmt"
31+
"io"
32+
"net"
33+
"net/http"
34+
"os"
35+
"path/filepath"
36+
"sync/atomic"
37+
"time"
38+
39+
axonflow "github.com/getaxonflow/axonflow-sdk-go/v8"
40+
)
41+
42+
func main() {
43+
expectedOrgID := os.Getenv("ORG_ID")
44+
if expectedOrgID == "" {
45+
expectedOrgID = "local-dev-org"
46+
}
47+
48+
// The SDK gates pings to once per 7 days via a stamp file at
49+
// os.UserCacheDir + axonflow/go-telemetry-last-sent. Nuke that
50+
// stamp file at the same path the SDK resolves — env var injection
51+
// here is too late (sharedHeartbeat is a package var initialized
52+
// before main() runs). This makes the proof self-contained on
53+
// any host without manual cleanup between runs.
54+
if cacheDir, cdErr := os.UserCacheDir(); cdErr == nil {
55+
stampPath := filepath.Join(cacheDir, "axonflow", "go-telemetry-last-sent")
56+
_ = os.Remove(stampPath) // ignore: stamp may not exist on first run
57+
}
58+
59+
var capturedBody atomic.Value
60+
capturedBody.Store([]byte{})
61+
62+
handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
63+
body, _ := io.ReadAll(r.Body)
64+
capturedBody.Store(body)
65+
w.Header().Set("Content-Type", "application/json")
66+
_ = json.NewEncoder(w).Encode(map[string]string{"latest_version": "8.1.0"})
67+
})
68+
69+
listener, err := net.Listen("tcp", "127.0.0.1:0")
70+
if err != nil {
71+
fmt.Fprintf(os.Stderr, "listen: %v\n", err)
72+
os.Exit(2)
73+
}
74+
defer listener.Close()
75+
76+
srv := &http.Server{Handler: handler}
77+
go func() { _ = srv.Serve(listener) }()
78+
defer srv.Shutdown(context.Background())
79+
80+
checkpointURL := "http://" + listener.Addr().String() + "/v1/ping"
81+
_ = os.Setenv("AXONFLOW_CHECKPOINT_URL", checkpointURL)
82+
_ = os.Setenv("AXONFLOW_TELEMETRY", "")
83+
84+
client := axonflow.NewClient(axonflow.AxonFlowConfig{
85+
Endpoint: "https://example.invalid",
86+
ClientID: "rt-e2e",
87+
ClientSecret: "rt-e2e",
88+
Mode: "production",
89+
})
90+
_ = client
91+
92+
// SDK fires telemetry on startup goroutine — give it a beat to land.
93+
time.Sleep(2 * time.Second)
94+
95+
got := capturedBody.Load().([]byte)
96+
if len(got) == 0 {
97+
fmt.Fprintln(os.Stderr, "FAIL: no telemetry body captured (did SDK try to ping?)")
98+
os.Exit(1)
99+
}
100+
101+
var payload map[string]any
102+
if err := json.Unmarshal(got, &payload); err != nil {
103+
fmt.Fprintf(os.Stderr, "FAIL: telemetry body not valid JSON: %v\n%s\n", err, string(got))
104+
os.Exit(1)
105+
}
106+
107+
orgID, ok := payload["org_id"].(string)
108+
if !ok {
109+
fmt.Fprintf(os.Stderr, "FAIL: telemetry body missing org_id field; body: %s\n", string(got))
110+
os.Exit(1)
111+
}
112+
if orgID != expectedOrgID {
113+
fmt.Fprintf(os.Stderr, "FAIL: telemetry org_id = %q, want %q\n", orgID, expectedOrgID)
114+
os.Exit(1)
115+
}
116+
117+
fmt.Printf("PASS: telemetry wire payload carries org_id=%q (expected=%q)\n", orgID, expectedOrgID)
118+
fmt.Printf("Wire body: %s\n", string(got))
119+
}

telemetry.go

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,13 @@ type telemetryPayload struct {
4949
// and the server defaults to "heartbeat". The wire-allowlist is enforced
5050
// server-side — see checkpoint-service IsValidIncomingStream.
5151
Stream string `json:"stream,omitempty"`
52+
// OrgID identifies the deployment's organization (#2277). Two sources,
53+
// in precedence order: the ORG_ID env var when set (the operator's
54+
// explicit configuration on self-hosted deployments, or the cs_<uuid>
55+
// tenant identifier on Community SaaS); otherwise the "local-dev-org"
56+
// sentinel. Always emitted. See axonflow-landing/content/privacy.html
57+
// for the customer-facing commitment that covers this field.
58+
OrgID string `json:"org_id"`
5259
}
5360

5461
// DeploymentMode classifications for telemetry (v1 schema, axonflow-enterprise#2008).
@@ -82,6 +89,22 @@ func ClassifyDeploymentMode(endpoint string) string {
8289
return DeploymentModeSelfHosted
8390
}
8491

92+
// OrgIDLocalDevSentinel is emitted on the telemetry wire when ORG_ID is
93+
// unset — the default-config Community-mode developer case. See #2277.
94+
const OrgIDLocalDevSentinel = "local-dev-org"
95+
96+
// telemetryOrgID returns the org_id value to emit on the next telemetry
97+
// ping. Reads ORG_ID from the environment (the operator's explicit
98+
// configuration for self-hosted deployments, or the cs_<uuid> tenant
99+
// identifier on Community SaaS) and falls back to OrgIDLocalDevSentinel
100+
// when unset. Always returns a non-empty string. See #2277.
101+
func telemetryOrgID() string {
102+
if v := os.Getenv("ORG_ID"); v != "" {
103+
return v
104+
}
105+
return OrgIDLocalDevSentinel
106+
}
107+
85108
// EndpointType classifications for telemetry.
86109
const (
87110
EndpointTypeLocalhost = "localhost"
@@ -246,7 +269,7 @@ func (c *AxonFlowClient) sendTelemetryPing() {
246269
// the total blocking time is bounded by ctx — no stacked sub-timeouts.
247270
// See issue #1693 for the original short-lived-process delivery fix.
248271
func (c *AxonFlowClient) sendTelemetryPingNow(ctx context.Context) error {
249-
log.Printf("[AxonFlow] Anonymous telemetry enabled. Opt out: AXONFLOW_TELEMETRY=off | https://docs.getaxonflow.com/docs/telemetry")
272+
log.Printf("[AxonFlow] Telemetry enabled. Opt out: AXONFLOW_TELEMETRY=off | https://docs.getaxonflow.com/docs/telemetry")
250273

251274
// Determine the checkpoint URL.
252275
checkpointURL := os.Getenv("AXONFLOW_CHECKPOINT_URL")
@@ -286,6 +309,7 @@ func (c *AxonFlowClient) sendTelemetryPingNow(ctx context.Context) error {
286309
Features: []string{},
287310
InstanceID: generateInstanceID(),
288311
Stream: stream,
312+
OrgID: telemetryOrgID(),
289313
}
290314

291315
body, err := json.Marshal(payload)

0 commit comments

Comments
 (0)