feat: setEndUser stamps enduser.id on spans and log records

README.md

@@ -203,6 +203,75 @@ What it does **not** touch:
 A `redact` that throws drops the record instead of letting it through
 unredacted — it fails closed, and it never throws into the app.
 
+## Linking telemetry to an account
+
+```dart
+observability.setEndUser('cmg2x8k4a0000abcd1234efgh'); // signed in, opted in
+observability.setEndUser(null); // signed out, or opted out
+```
+
+`OtelZone.setEndUser(String? id)` stamps the semantic convention's
+`enduser.id` on every span that starts and every log record that is emitted
+from the next one on, until it is called again. `null` stops it, and so does an
+empty string, because an attribute may not be empty. The package never sets it
+by itself: an app that does not call `setEndUser` exports no `enduser.id`
+anywhere, and whether, when and for whom to call it is the app's decision (a
+consent, a sign-in), not something this package asks.
+
+`endUserId` reads back what the next span and record will carry.
+
+**From the next one, and no further back.** A span is stamped as it starts and
+a record as it is emitted, so the identity on telemetry is the one that was
+current when it was made:
+
+- A span already running when the id is set does not gain it, and one running
+  when it is cleared keeps it. A span that wraps a sign-in, started before it,
+  stays unlinked.
+- Nothing already emitted, queued or spooled is rewritten. Clearing the id
+  stops linking; it does not recall.
+- Recovered native crash reports are not stamped. They describe the previous
+  run, and whoever is signed in now is not necessarily who was then.
+
+**The id is exempt from `redact`.** On a span it is passed through whole while
+every other attribute is scrubbed as before; a log record's is added after the
+bridge has scrubbed the record, so `redact` never meets it. The exemption is
+there because it happens: over 200,000 random ids of each kind, the unanchored
+`\d{9}` used as the example above masks about 1 in 10,000 cuids and 1 in 33
+UUIDs, and an anchored `^\+?\d{9,12}$` masks neither. A masked id unlinks its
+span with nothing to say so. The price is that the exemption trusts the app:
+pass an opaque account id, never a phone number, an e-mail address or a name,
+which is exactly what `redact` would otherwise have caught. Only the exact key
+`enduser.id` is exempt.
+
+**It never throws, and any zone may call it.** It is one field assignment with
+no lock, no timer and no I/O. The id lives on the `OtelZone` and in no `Zone`,
+so it does not matter which zone sets it or which later starts a span; the last
+call wins. It belongs to the isolate, like the rest of Dart's state: a
+background isolate has its own `OtelZone` and its own id.
+
+**It is safe at any time, `start()` included.** Before the SDK is up there is
+nothing to stamp, so the call has no visible effect; the id is kept, and the
+first span and record after a successful `start()` carry it. If the SDK never
+starts, nothing is ever stamped. Stamping itself never throws into the SDK
+either: the SDK calls its processors without awaiting them, and a failure there
+would otherwise be an uncaught error in the app's own handler. It fails towards
+an unlinked span.
+
+How it is wired, because the order is not arbitrary. Spans are stamped by a
+span processor in `onStart`, appended behind the pipeline: a span is exported
+at its end, after every processor has seen it start. Log records are stamped
+by a log processor in `onEmit`, and that one has to be **first**, because the
+batch processor queues a clone of each record and a stamp made after it lands
+on an original nobody exports. `addLogRecordProcessor` only appends, so
+`start()` hands the stamper to `OTel.initialize` and builds the rest of the
+logs pipeline behind it, with the SDK's own `LogsConfiguration` function and
+the arguments `OTel.initialize` would have passed it, so `OTEL_LOGS_EXPORTER`,
+the `OTEL_EXPORTER_OTLP_*` headers and the spool behave as they did.
+
+On sign-out, clear the id *and* the spool (see "Where the spool lives"): a
+batch spooled while the id was set keeps it, and is delivered with it on the
+next launch.
+
 ## Faults recorded offline
 
 A batch the collector refuses is retried in memory and then dropped, and one in

lib/src/config.dart

@@ -146,6 +146,10 @@ final class OtelZoneConfig {
   /// before Dart runs (in no-backup storage, deleted once acknowledged); the
   /// README's "What the platforms store before Dart runs" lists them.
   ///
+  /// It does not touch `enduser.id`, the one attribute [OtelZone.setEndUser]
+  /// stamps on purpose: a span's copy of it is exempt, and a log record's never
+  /// meets the redactor. Every other value is scrubbed exactly as before.
+  ///
   /// `null` means no scrubbing — the behaviour of every build that does not
   /// supply one. A [redact] that throws drops the record, or the span, rather
   /// than letting it through unredacted (fail closed).

lib/src/end-user.dart

@@ -0,0 +1,112 @@
+import 'package:dartastic_opentelemetry/dartastic_opentelemetry.dart';
+
+/// The attribute `OtelZone.setEndUser` stamps: the semantic convention's
+/// `enduser.id`.
+///
+/// Read from the SDK's generated registry rather than typed out, the way
+/// `app.build_id` is, and pinned to its literal by a test, so a rename upstream
+/// shows up as a failing test and not as a dashboard that quietly empties.
+final String endUserIdKey = Enduser.enduserId.key;
+
+/// Stamps [endUserIdKey] on every span as it starts.
+///
+/// Only a span that starts after the id is set carries it. The stamp is made in
+/// `onStart` and never again, so a span already running when the id is set or
+/// cleared keeps what it started with: the identity on a span is the one that
+/// was current when the work began.
+///
+/// It is appended to the tracer provider after `OTel.initialize`, behind
+/// dartastic's own processors. That is early enough: a span is queued and
+/// exported at `onEnd`, after every processor has seen it start, and
+/// `RedactingSpanExporter` exempts the key when it scrubs at export.
+///
+/// Never throws and never returns a failed future. The SDK calls `onStart`
+/// without awaiting it, so an error here would surface as an uncaught zone
+/// error, in the app's own error handler.
+final class EndUserSpanProcessor implements SpanProcessor {
+  /// Creates a processor that stamps whatever [currentId] answers when a span
+  /// starts, or nothing when it answers `null`.
+  EndUserSpanProcessor(this._currentId);
+
+  final String? Function() _currentId;
+
+  @override
+  Future<void> onStart(Span span, Context? parentContext) {
+    try {
+      final String? id = _currentId();
+      if (id != null) span.setStringAttribute<String>(endUserIdKey, id);
+    } on Object {
+      // Instrumentation is never in the functional path. A span without the
+      // id is a span that is not linked to an account, which is the safe way
+      // for this to fail.
+    }
+    return Future<void>.value();
+  }
+
+  @override
+  Future<void> onEnd(Span span) => Future<void>.value();
+
+  @override
+  Future<void> onNameUpdate(Span span, String newName) => Future<void>.value();
+
+  @override
+  Future<void> shutdown() => Future<void>.value();
+
+  @override
+  Future<void> forceFlush() => Future<void>.value();
+}
+
+/// Stamps [endUserIdKey] on every log record as it is emitted.
+///
+/// **This one has to come first.** `BatchLogRecordProcessor.onEmit` queues a
+/// *clone* of the record, so a stamp made by a processor that runs after it
+/// lands on an original nobody exports. `OtelZone.start` therefore hands this
+/// processor to `OTel.initialize` ahead of the pipeline and builds that
+/// pipeline behind it; the provider's list is append-only, so there is no
+/// adding it later. Spans do not need this, because a span is not cloned.
+///
+/// It runs after the bridge, so it never meets `OtelZoneConfig.redact`: the
+/// id goes on the record as an attribute the redactor is not handed, and a
+/// log record is exempt from scrubbing without a rule saying so.
+///
+/// An `enduser.id` the record already carries is replaced, not repeated: the
+/// key stays unique and the zone's identity is the one that is exported.
+///
+/// Never throws and never returns a failed future, for the reason
+/// [EndUserSpanProcessor] gives.
+final class EndUserLogRecordProcessor implements LogRecordProcessor {
+  /// Creates a processor that stamps whatever [currentId] answers when a
+  /// record is emitted, or nothing when it answers `null`.
+  EndUserLogRecordProcessor(this._currentId);
+
+  final String? Function() _currentId;
+
+  @override
+  Future<void> onEmit(ReadWriteLogRecord logRecord, Context? context) {
+    try {
+      final String? id = _currentId();
+      if (id != null) {
+        logRecord
+          ..removeAttribute(endUserIdKey)
+          ..addAttribute(OTel.attributeString(endUserIdKey, id));
+      }
+    } on Object {
+      // See [EndUserSpanProcessor.onStart].
+    }
+    return Future<void>.value();
+  }
+
+  @override
+  bool enabled({
+    Context? context,
+    InstrumentationScope? instrumentationScope,
+    Severity? severityNumber,
+    String? eventName,
+  }) => true;
+
+  @override
+  Future<void> shutdown() => Future<void>.value();
+
+  @override
+  Future<void> forceFlush() => Future<void>.value();
+}

lib/src/otel-zone.dart

@@ -13,6 +13,7 @@ import 'package:talker/talker.dart';
 
 import 'bridge.dart';
 import 'config.dart';
+import 'end-user.dart';
 import 'native-crash.dart';
 import 'otlp-exporter.dart';
 import 'riverpod-arguments.dart';
@@ -107,6 +108,60 @@ class OtelZone {
   /// that failed.
   bool get isReady => bridge.ready;
 
+  String? _endUserId;
+
+  /// The id the next span and the next log record will carry as `enduser.id`,
+  /// or `null` when they will carry none. See [setEndUser].
+  String? get endUserId => _endUserId;
+
+  /// Links every span started and every log record emitted from now on to the
+  /// person [id] names, as the semantic convention's `enduser.id`; `null`
+  /// stops it.
+  ///
+  /// ```dart
+  /// // Signed in, and opted in to being linked:
+  /// observability.setEndUser('cmg2x8k4a0000abcd1234efgh');
+  /// // Signed out, or opted out:
+  /// observability.setEndUser(null);
+  /// ```
+  ///
+  /// **From the next one, and no further back.** A span is stamped as it
+  /// starts and a record as it is emitted. A span already running when this is
+  /// called keeps what it started with (it is neither given the id nor stripped
+  /// of it), and so does every record already emitted, queued or spooled: the
+  /// identity on telemetry is the one that was current when it was made.
+  /// Clearing the id does not recall what was sent.
+  ///
+  /// An empty [id] clears it, since an attribute may not be an empty string.
+  ///
+  /// **The id is exempt from [OtelZoneConfig.redact]**, on spans and on log
+  /// records. It is the one attribute an app decides may leave the device, so a
+  /// redactor that masked it (a long run of digits in a cuid is enough for a
+  /// pattern written for phone numbers) would unlink every span without a
+  /// word. It follows that [id] must be an opaque identifier that is already
+  /// fit to be sent: an account id, never a phone number, an e-mail address or
+  /// a name, which `redact` would otherwise have caught. Every other attribute
+  /// is scrubbed exactly as before.
+  ///
+  /// Recovered native crash reports are not stamped. They describe the
+  /// previous run, and the person signed in now is not necessarily the person
+  /// who was then.
+  ///
+  /// **Never throws, and cheap enough for any call site**: it is one field
+  /// assignment, with no lock and no I/O. The state lives on this object and in
+  /// no `Zone`, so it makes no difference which zone calls it, or from which
+  /// zone a span is later started; the last call wins. Like the rest of Dart's
+  /// state it belongs to the isolate: a background isolate has its own
+  /// [OtelZone] and its own id.
+  ///
+  /// Safe at any time, before [start] included. Until the SDK is up there is
+  /// nothing to stamp and the call has no effect that anything can see; the id
+  /// is kept, and the first span and record after a successful [start] carry
+  /// it. If the SDK never starts, nothing is ever stamped.
+  void setEndUser(String? id) {
+    _endUserId = id == null || id.isEmpty ? null : id;
+  }
+
   /// Brings up the OpenTelemetry SDK and points it at
   /// [OtelZoneConfig.endpoint].
   ///
@@ -194,6 +249,13 @@ class OtelZone {
           ),
         );
       }
+      // The first log processor, so that it runs before the batch processor
+      // queues its clone of each record. `OTel.initialize` builds no logs
+      // pipeline of its own when it is given a processor, so the rest of it is
+      // built below, behind this one. See `EndUserLogRecordProcessor`.
+      final EndUserLogRecordProcessor? endUserLogs = config.enableLogs
+          ? EndUserLogRecordProcessor(() => _endUserId)
+          : null;
       await OTel.initialize(
         serviceName: config.serviceName,
         serviceVersion: serviceVersion,
@@ -203,12 +265,36 @@ class OtelZone {
         enableMetrics: config.enableMetrics,
         spanProcessor: redactingSpans,
         logRecordExporter: spool,
+        logRecordProcessor: endUserLogs,
         resourceAttributes: OTel.attributesFromMap(<String, String>{
           ...resourceAttributes,
           App.appBuildId.key: ?buildId,
           'deployment.environment.name': ?config.deploymentEnvironmentName,
         }),
       );
+      // The logs pipeline `initialize` would have built, by the function it
+      // would have called, with the arguments it would have passed (the
+      // environment supplies the rest, as it does there). `OTEL_SDK_DISABLED`
+      // makes `initialize` skip logs altogether, which leaves the stamper out
+      // of the provider and this with nothing to follow.
+      if (endUserLogs != null &&
+          OTel.loggerProvider().logRecordProcessors.contains(endUserLogs)) {
+        LogsConfiguration.configureLoggerProvider(
+          endpoint: config.endpoint,
+          secure: config.secure,
+          logRecordExporter: spool,
+          resource: OTel.defaultResource,
+        );
+      }
+      // Appended, where the log stamper cannot be: a span is exported when it
+      // ends, after every processor has seen it start, and is not cloned. Only
+      // behind a pipeline that exists: `OTEL_SDK_DISABLED` and
+      // `OTEL_TRACES_EXPORTER=none` build none, and a stamper alone would be a
+      // processor left running for spans nothing exports.
+      final TracerProvider tracers = OTel.tracerProvider();
+      if (tracers.hasSpanProcessors) {
+        tracers.addSpanProcessor(EndUserSpanProcessor(() => _endUserId));
+      }
       // `OTEL_SDK_DISABLED` makes `initialize` skip the trace pipeline, which
       // leaves the processor built above running its timer for nothing.
       final SpanProcessor? unused = redactingSpans;

lib/src/span-redaction.dart

@@ -1,6 +1,7 @@
 import 'package:dartastic_opentelemetry/dartastic_opentelemetry.dart';
 
 import 'config.dart';
+import 'end-user.dart';
 import 'otlp-exporter.dart';
 
 /// A [SpanExporter] that hands its delegate a scrubbed copy of every span, so
@@ -15,6 +16,15 @@ import 'otlp-exporter.dart';
 /// kind, status code, numbers) are left alone: they are the span's schema, not
 /// data an app put there.
 ///
+/// **One attribute is exempt: `enduser.id`.** It is the id `OtelZone.setEndUser`
+/// stamps, put there on purpose by an app that has decided the id may leave the
+/// device, and a redactor that masks it unlinks the span without a word. It
+/// happens: over 200,000 random ids of each kind, an unanchored `\d{9}` (the
+/// README's own example) masked about 1 in 10,000 cuids and 1 in 33 UUIDs, while an
+/// anchored whole-value pattern masked neither. Every other attribute, on a
+/// span, an event or a link, is scrubbed exactly as before, so an app that
+/// hands a phone number to `setEndUser` is the one thing this lets through.
+///
 /// ## Why an exporter, and why a view
 ///
 /// A processor cannot do this. A span is already ended when a processor's
@@ -301,11 +311,15 @@ final class _ScrubbedScope implements InstrumentationScope {
 /// [attributes] with every string value, and every element of a string-list
 /// value, run through [redact]. A value the redactor empties is dropped,
 /// because an attribute may not be an empty string.
+///
+/// The attribute named [endUserIdKey] is passed through as it is.
 Attributes _scrubAttributes(Attributes attributes, Redactor redact) {
   final List<Attribute<Object>> scrubbed = <Attribute<Object>>[];
   for (final Attribute<Object> attribute in attributes.toList()) {
     final Object value = attribute.value;
-    if (value is String) {
+    if (attribute.key == endUserIdKey) {
+      scrubbed.add(attribute);
+    } else if (value is String) {
       final String text = redact(value);
       if (text.isNotEmpty) {
         scrubbed.add(OTel.attributeString(attribute.key, text));

test/end-user-disabled_test.dart

@@ -0,0 +1,34 @@
+// `OTEL_SDK_DISABLED` with logs on: `OTel.initialize` builds neither pipeline,
+// so neither stamper may be left behind, and `setEndUser` stays a harmless
+// call. The sibling `span-redaction-disabled_test.dart` covers the trace side
+// with a `redact` configured; this is the logs side.
+//
+// In its own file: `OTel.initialize()` may only be called once per isolate.
+import 'package:dartastic_opentelemetry/dartastic_opentelemetry.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:otel_zone/otel_zone.dart';
+
+void main() {
+  test('a disabled SDK leaves no stamper in either provider', () async {
+    EnvironmentService.testOverrides = <String, String>{
+      'OTEL_SDK_DISABLED': 'true',
+    };
+    addTearDown(() => EnvironmentService.testOverrides = null);
+    final OtelZone observability = OtelZone(
+      const OtelZoneConfig(
+        serviceName: 'end-user-disabled-test',
+        endpoint: 'http://127.0.0.1:4318',
+        useConsoleLogs: false,
+      ),
+      sink: RecordingTalkerObserver(),
+    );
+
+    await observability.start(serviceVersion: '1.2.3');
+
+    expect(observability.isReady, isTrue);
+    expect(OTel.tracerProvider().spanProcessors, isEmpty);
+    expect(OTel.loggerProvider().logRecordProcessors, isEmpty);
+    expect(() => observability.setEndUser('user-1'), returnsNormally);
+    expect(observability.endUserId, 'user-1');
+  });
+}