feat: stamp service.instance.id on spans and share it with the heartbeat (#24)

Author: PabloPardoGarcia

src/client.ts

@@ -1,3 +1,4 @@
+import { randomUUID } from "node:crypto";
 import { type Tracer, trace } from "@opentelemetry/api";
 // The -proto exporter (OTLP protobuf over HTTP), matching the Python SDK's
 // opentelemetry-exporter-otlp-proto-http. The Rius ingest accepts only
@@ -20,7 +21,7 @@ import { HeartbeatSender, type HeartbeatTransport, OpenRootSpanTracker } from ".
 import { enableInstrumentations } from "./instrumentation.js";
 import { MaskingSpanExporter } from "./masking.js";
 import { PendingSpanProcessor } from "./pending.js";
-import { TRACER_NAME } from "./semconv.js";
+import { SERVICE_INSTANCE_ID, TRACER_NAME } from "./semconv.js";
 import { SessionSpanProcessor } from "./session.js";
 
 /** Options accepted by {@link init}, extending the shared configuration. */
@@ -203,8 +204,16 @@ export function init(options: InitOptions = {}): RiusClient {
     processors.add(batch);
   }
 
+  // One identity per client lifetime, shared by spans (resource) and
+  // heartbeats (payload instance_id) so the backend can join them and count
+  // replicas. Workers spawned after init() (cluster/fork patterns) should
+  // init() themselves for exact per-worker span identity.
+  const instanceId = randomUUID();
   const provider = new NodeTracerProvider({
-    resource: resourceFromAttributes({ [ATTR_SERVICE_NAME]: config.serviceName }),
+    resource: resourceFromAttributes({
+      [ATTR_SERVICE_NAME]: config.serviceName,
+      [SERVICE_INSTANCE_ID]: instanceId,
+    }),
     // Always ParentBased, with no AlwaysOn shortcut at rate 1. They are not
     // equivalent: ParentBased honours a remote UNSAMPLED parent and drops,
     // while AlwaysOn records regardless, producing children of a span the
@@ -245,6 +254,7 @@ export function init(options: InitOptions = {}): RiusClient {
       headers: authHeaders,
       intervalMs: config.heartbeatIntervalMs,
       agentName: config.agentName,
+      instanceId,
       tracker,
       transport: options.heartbeatTransport,
     });

src/heartbeat.ts

@@ -1,4 +1,3 @@
-import { randomUUID } from "node:crypto";
 import { createRequire } from "node:module";
 import type { Context } from "@opentelemetry/api";
 import type { ReadableSpan, Span, SpanProcessor } from "@opentelemetry/sdk-trace-base";
@@ -122,6 +121,12 @@ export interface HeartbeatSenderOptions {
   headers: Record<string, string>;
   intervalMs: number;
   agentName: string;
+  /**
+   * Identity of one process lifetime, injected by init(): the same value
+   * rides every span as the `service.instance.id` resource attribute, which
+   * is what lets the backend join heartbeats to traces.
+   */
+  instanceId: string;
   tracker: OpenRootSpanTracker;
   /** Injected transport for tests; defaults to the fetch-based HTTP POST above. */
   transport?: HeartbeatTransport;
@@ -131,8 +136,8 @@ export interface HeartbeatSenderOptions {
 
 /** Pings the heartbeat endpoint for the process lifetime; see module docs for the contract. */
 export class HeartbeatSender {
-  /** Identity of one process lifetime, fresh per sender. */
-  readonly instanceId = randomUUID();
+  /** Identity of one process lifetime; injected, shared with span resources. */
+  readonly instanceId: string;
 
   private readonly agentName: string;
   private readonly tracker: OpenRootSpanTracker;
@@ -146,6 +151,7 @@ export class HeartbeatSender {
   private deliveryWarned = false;
 
   constructor(opts: HeartbeatSenderOptions) {
+    this.instanceId = opts.instanceId;
     this.agentName = opts.agentName;
     this.tracker = opts.tracker;
     this.intervalMs = opts.intervalMs;

src/semconv.ts

@@ -1,6 +1,12 @@
 /** Wire-visible instrumentation scope name. The backend keys on this value. */
 export const TRACER_NAME = "glassflow";
 
+// OTel standard resource attribute: identity of one process lifetime (one
+// uuid per client, minted at init). The heartbeat payload's instance_id
+// carries the SAME value, which is what lets the backend join heartbeats to
+// traces and count replicas.
+export const SERVICE_INSTANCE_ID = "service.instance.id";
+
 // OpenInference
 export const OPENINFERENCE_SPAN_KIND = "openinference.span.kind";
 export const INPUT_VALUE = "input.value";

tests/client.test.ts

@@ -4,7 +4,7 @@ import { afterEach, describe, expect, it, vi } from "vitest";
 import { type InitOptions, RiusClient, getTracer, init } from "../src/client.js";
 import type { HeartbeatTransport } from "../src/heartbeat.js";
 import { REGISTRY } from "../src/instrumentation.js";
-import { GLASSFLOW_SPAN_PENDING } from "../src/semconv.js";
+import { GLASSFLOW_SPAN_PENDING, SERVICE_INSTANCE_ID } from "../src/semconv.js";
 import { startAsCurrentSpan } from "../src/spans.js";
 
 let client: RiusClient | undefined;
@@ -313,3 +313,59 @@ describe("init: partial spans", () => {
     expect(exporter.getFinishedSpans()).toHaveLength(0);
   });
 });
+
+describe("instance identity (service.instance.id)", () => {
+  const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/;
+
+  it("stamps a uuid as service.instance.id on the resource of every span", async () => {
+    const exporter = new InMemorySpanExporter();
+    client = testInit({ serviceName: "svc", spanExporter: exporter });
+    getTracer().startSpan("s").end();
+    await client.flush();
+    const id = exporter.getFinishedSpans()[0].resource.attributes[SERVICE_INSTANCE_ID];
+    expect(id).toMatch(UUID_PATTERN);
+  });
+
+  it("sends the same id on heartbeat payloads and span resources", async () => {
+    const payloads: Record<string, unknown>[] = [];
+    const exporter = new InMemorySpanExporter();
+    client = init({
+      serviceName: "svc",
+      spanExporter: exporter,
+      heartbeatTransport: async (payload) => {
+        payloads.push(payload);
+      },
+    });
+    getTracer().startSpan("s").end();
+    await client.flush();
+    const resourceId = exporter.getFinishedSpans()[0].resource.attributes[SERVICE_INSTANCE_ID];
+    expect(payloads.length).toBeGreaterThan(0);
+    expect(payloads[0]?.instance_id).toBe(resourceId);
+  });
+
+  it("stamps the id even with the heartbeat disabled", async () => {
+    const exporter = new InMemorySpanExporter();
+    client = testInit({ serviceName: "svc", heartbeat: false, spanExporter: exporter });
+    getTracer().startSpan("s").end();
+    await client.flush();
+    const id = exporter.getFinishedSpans()[0].resource.attributes[SERVICE_INSTANCE_ID];
+    expect(id).toMatch(UUID_PATTERN);
+  });
+
+  it("gives two separately initialized clients distinct ids", async () => {
+    const exporterA = new InMemorySpanExporter();
+    client = testInit({ serviceName: "svc", spanExporter: exporterA });
+    getTracer().startSpan("a").end();
+    await client.flush();
+    const idA = exporterA.getFinishedSpans()[0].resource.attributes[SERVICE_INSTANCE_ID];
+    await client.shutdown();
+
+    const exporterB = new InMemorySpanExporter();
+    client = testInit({ serviceName: "svc", spanExporter: exporterB });
+    getTracer().startSpan("b").end();
+    await client.flush();
+    const idB = exporterB.getFinishedSpans()[0].resource.attributes[SERVICE_INSTANCE_ID];
+
+    expect(idA).not.toBe(idB);
+  });
+});

tests/heartbeat.test.ts

@@ -88,6 +88,7 @@ describe("HeartbeatSender", () => {
       headers: {},
       intervalMs: 15_000,
       agentName: "my-agent",
+      instanceId: "test-instance",
       tracker: fakeTracker(["trace-1"]),
       transport,
     });
@@ -109,25 +110,20 @@ describe("HeartbeatSender", () => {
     expect(timeoutMs).toBe(3000);
   });
 
-  it("has a fresh instanceId per sender", () => {
+  it("carries the injected instanceId, the same identity spans get as service.instance.id", () => {
     const transport = vi.fn<HeartbeatTransport>().mockResolvedValue(undefined);
-    const a = new HeartbeatSender({
+    const sender = new HeartbeatSender({
       url: "http://example.invalid",
       headers: {},
       intervalMs: 15_000,
       agentName: "a",
+      instanceId: "injected-identity",
       tracker: fakeTracker([]),
       transport,
     });
-    const b = new HeartbeatSender({
-      url: "http://example.invalid",
-      headers: {},
-      intervalMs: 15_000,
-      agentName: "b",
-      tracker: fakeTracker([]),
-      transport,
-    });
-    expect(a.instanceId).not.toBe(b.instanceId);
+    sender.start();
+    expect(sender.instanceId).toBe("injected-identity");
+    expect(transport.mock.calls[0]?.[0]?.instance_id).toBe("injected-identity");
   });
 
   it("pings again on every interval tick", () => {
@@ -137,6 +133,7 @@ describe("HeartbeatSender", () => {
       headers: {},
       intervalMs: 15_000,
       agentName: "my-agent",
+      instanceId: "test-instance",
       tracker: fakeTracker([]),
       transport,
     });
@@ -159,6 +156,7 @@ describe("HeartbeatSender", () => {
       headers: {},
       intervalMs: 15_000,
       agentName: "my-agent",
+      instanceId: "test-instance",
       tracker: fakeTracker(traceIds),
       transport,
     });
@@ -179,6 +177,7 @@ describe("HeartbeatSender", () => {
       headers: {},
       intervalMs: 15_000,
       agentName: "my-agent",
+      instanceId: "test-instance",
       tracker: fakeTracker([]),
       transport,
     });
@@ -203,6 +202,7 @@ describe("HeartbeatSender", () => {
       headers: {},
       intervalMs: 15_000,
       agentName: "my-agent",
+      instanceId: "test-instance",
       tracker: fakeTracker([]),
       transport,
     });
@@ -220,6 +220,7 @@ describe("HeartbeatSender", () => {
       headers: {},
       intervalMs: 15_000,
       agentName: "my-agent",
+      instanceId: "test-instance",
       tracker: fakeTracker([]),
       transport,
     });
@@ -240,6 +241,7 @@ describe("HeartbeatSender", () => {
       headers: {},
       intervalMs: 15_000,
       agentName: "my-agent",
+      instanceId: "test-instance",
       tracker: fakeTracker([]),
       transport,
     });
@@ -305,6 +307,7 @@ describe("default HTTP transport", () => {
         headers: { authorization: "Bearer test-key" },
         intervalMs: 15_000,
         agentName: "my-agent",
+        instanceId: "test-instance",
         tracker: fakeTracker([]),
       });
       sender.start();
@@ -332,6 +335,7 @@ describe("default HTTP transport", () => {
         headers: {},
         intervalMs: 15_000,
         agentName: "my-agent",
+        instanceId: "test-instance",
         tracker: fakeTracker([]),
       });
       sender.start();

tests/heartbeatConfigEdges.test.ts

@@ -45,6 +45,7 @@ it("(3) caps open_traces at 32 with 40 REAL root spans, count reports 40", async
     headers: {},
     intervalMs: 999999,
     agentName: "a",
+    instanceId: "test-instance",
     tracker,
     transport: async (p) => void payloads.push(p),
   });