feat(bigtable): add sessionTracer for per-Session lifecycle + vRPC metrics - #20190

Concurrent reads and writes of interface values (such as metric.Float64Histogram) can cause a data race because interface values are two-word structures in Go. Since InitializeSessionMetrics writes to these global variables and they are read concurrently in recordOpen, recordClose, sampleUptime, and recordTransportOverhead, we should synchronize access to them.

Using atomic.Value to store and load the interface values ensures thread-safe concurrent access without the overhead of a global mutex.

var (
	sessionDurations     atomic.Value
	sessionOpenLatencies atomic.Value
	sessionUptime        atomic.Value
	transportLatencies   atomic.Value

	sessionMetricsOnce sync.Once
	sessionMetricsErr  error
)

Not a bug given the usage contract: InitializeSessionMetrics runs synchronously from the Client constructor before any Session goroutine exists, and -race on the bundled tests is clean. Same package-global pattern already merged as debugTagCounter metric.Int64Counter in debug_tracer.go:194 (#20114) — rewriting one side with atomic.Value and not the other would be asymmetric. If mutianf wants hardening we can do both files together in a follow-up using atomic.Pointer[metric.Float64Histogram] (cleaner than the atomic.Value + type-assert boilerplate). See #20190 (comment) for the batch reply.

Initialize the global histogram variables using atomic.Value.Store to prevent data races with concurrent reads in active sessions.

func InitializeSessionMetrics(meterProvider metric.MeterProvider) error {
	sessionMetricsOnce.Do(func() {
		if meterProvider == nil {
			return
		}
		meter := meterProvider.Meter(clientMeterName)

		var err error
		var durations, openLatencies, uptime, transport metric.Float64Histogram
		if durations, err = meter.Float64Histogram(
			"session.durations",
			metric.WithDescription("Duration a session was alive (startTime → close)"),
			metric.WithUnit("ms"),
		); err != nil {
			sessionMetricsErr = fmt.Errorf("create session.durations histogram: %w", err)
			return
		}
		if openLatencies, err = meter.Float64Histogram(
			"session.open_latencies",
			metric.WithDescription("Latency to open a session"),
			metric.WithUnit("ms"),
		); err != nil {
			sessionMetricsErr = fmt.Errorf("create session.open_latencies histogram: %w", err)
			return
		}
		if uptime, err = meter.Float64Histogram(
			"session.uptime",
			metric.WithDescription("Age of currently-active sessions, sampled periodically"),
			metric.WithUnit("ms"),
		); err != nil {
			sessionMetricsErr = fmt.Errorf("create session.uptime histogram: %w", err)
			return
		}
		if transport, err = meter.Float64Histogram(
			"transport_latencies",
			metric.WithDescription("The latency measured from e2e latencies minus node latencies."),
			metric.WithUnit("ms"),
			metric.WithExplicitBucketBoundaries(FineGrainLatencyBounds...),
		); err != nil {
			sessionMetricsErr = fmt.Errorf("create transport_latencies histogram: %w", err)
			return
		}
		if err = registerDebugTagCounter(meter); err != nil {
			sessionMetricsErr = err
			return
		}

		sessionDurations.Store(durations)
		sessionOpenLatencies.Store(openLatencies)
		sessionUptime.Store(uptime)
		transportLatencies.Store(transport)
	})
	return sessionMetricsErr
}

Not a bug given the usage contract: InitializeSessionMetrics runs synchronously from the Client constructor before any Session goroutine exists, and -race on the bundled tests is clean. Same package-global pattern already merged as debugTagCounter metric.Int64Counter in debug_tracer.go:194 (#20114) — rewriting one side with atomic.Value and not the other would be asymmetric. If mutianf wants hardening we can do both files together in a follow-up using atomic.Pointer[metric.Float64Histogram] (cleaner than the atomic.Value + type-assert boilerplate). See #20190 (comment) for the batch reply.

Read the global sessionOpenLatencies histogram using atomic.Value.Load to prevent data races with concurrent initialization.

func (t *sessionTracer) recordOpen(ctx context.Context, err error) {
	t.mu.Lock()
	t.opened = true
	t.mu.Unlock()

	val := sessionOpenLatencies.Load()
	if val == nil {
		return
	}
	hist, ok := val.(metric.Float64Histogram)
	if !ok || hist == nil {
		return
	}
	snap := t.snapshot()
	statusStr := "OK"
	if err != nil {
		statusStr = status.Code(err).String()
	}
	hist.Record(ctx, msSince(snap.startTime), metric.WithAttributes(
		attribute.String("transport_type", snap.transportType),
		attribute.String("status", statusStr),
		attribute.String("session_type", t.sessionType.String()),
		attribute.String("afe_location", snap.afeLocation),
		attribute.String("session_name", snap.poolName),
	))
}

Not a bug given the usage contract: InitializeSessionMetrics runs synchronously from the Client constructor before any Session goroutine exists, and -race on the bundled tests is clean. Same package-global pattern already merged as debugTagCounter metric.Int64Counter in debug_tracer.go:194 (#20114) — rewriting one side with atomic.Value and not the other would be asymmetric. If mutianf wants hardening we can do both files together in a follow-up using atomic.Pointer[metric.Float64Histogram] (cleaner than the atomic.Value + type-assert boilerplate). See #20190 (comment) for the batch reply.

Read the global sessionDurations histogram using atomic.Value.Load to prevent data races with concurrent initialization.

func (t *sessionTracer) recordClose(ctx context.Context, closingReason string, streamErr error, hadOk, hadErr bool) {
	val := sessionDurations.Load()
	if val == nil {
		return
	}
	hist, ok := val.(metric.Float64Histogram)
	if !ok || hist == nil {
		return
	}
	snap := t.snapshot()

	var elapsed float64
	if !snap.startTime.IsZero() {
		elapsed = msSince(snap.startTime)
	}

	statusStr := "OK"
	if streamErr != nil {
		statusStr = status.Code(streamErr).String()
	}

	hist.Record(ctx, elapsed, metric.WithAttributes(
		attribute.String("transport_type", snap.transportType),
		attribute.String("status", statusStr),
		attribute.String("session_type", t.sessionType.String()),
		attribute.String("closing_reason", closingReason),
		attribute.String("vrpcs", vrpcCloseState(hadOk, hadErr)),
		attribute.Bool("ready", snap.opened),
		attribute.String("afe_location", snap.afeLocation),
		attribute.String("session_name", snap.poolName),
	))
}

Not a bug given the usage contract: InitializeSessionMetrics runs synchronously from the Client constructor before any Session goroutine exists, and -race on the bundled tests is clean. Same package-global pattern already merged as debugTagCounter metric.Int64Counter in debug_tracer.go:194 (#20114) — rewriting one side with atomic.Value and not the other would be asymmetric. If mutianf wants hardening we can do both files together in a follow-up using atomic.Pointer[metric.Float64Histogram] (cleaner than the atomic.Value + type-assert boilerplate). See #20190 (comment) for the batch reply.

Read the global sessionUptime histogram using atomic.Value.Load to prevent data races with concurrent initialization.

func (t *sessionTracer) sampleUptime(ctx context.Context) {
	val := sessionUptime.Load()
	if val == nil {
		return
	}
	hist, ok := val.(metric.Float64Histogram)
	if !ok || hist == nil {
		return
	}
	snap := t.snapshot()
	if snap.startTime.IsZero() {
		return
	}
	hist.Record(ctx, msSince(snap.startTime), metric.WithAttributes(
		attribute.String("transport_type", snap.transportType),
		attribute.String("session_type", t.sessionType.String()),
		attribute.Bool("ready", snap.opened),
		attribute.String("afe_location", snap.afeLocation),
		attribute.String("session_name", snap.poolName),
	))
}

Not a bug given the usage contract: InitializeSessionMetrics runs synchronously from the Client constructor before any Session goroutine exists, and -race on the bundled tests is clean. Same package-global pattern already merged as debugTagCounter metric.Int64Counter in debug_tracer.go:194 (#20114) — rewriting one side with atomic.Value and not the other would be asymmetric. If mutianf wants hardening we can do both files together in a follow-up using atomic.Pointer[metric.Float64Histogram] (cleaner than the atomic.Value + type-assert boilerplate). See #20190 (comment) for the batch reply.

Read the global transportLatencies histogram using atomic.Value.Load to prevent data races with concurrent initialization.

func (t *sessionTracer) recordTransportOverhead(ctx context.Context, method string, overhead time.Duration) {
	if overhead <= 0 {
		return
	}
	val := transportLatencies.Load()
	if val == nil {
		return
	}
	hist, ok := val.(metric.Float64Histogram)
	if !ok || hist == nil {
		return
	}
	snap := t.snapshot()
	hist.Record(ctx, float64(overhead)/float64(time.Millisecond), metric.WithAttributes(
		attribute.String("transport_type", snap.transportType),
		attribute.String("session_type", t.sessionType.String()),
		attribute.String("afe_location", snap.afeLocation),
		attribute.String("session_name", snap.poolName),
		attribute.String("method", method),
	))
}

Not a bug given the usage contract: InitializeSessionMetrics runs synchronously from the Client constructor before any Session goroutine exists, and -race on the bundled tests is clean. Same package-global pattern already merged as debugTagCounter metric.Int64Counter in debug_tracer.go:194 (#20114) — rewriting one side with atomic.Value and not the other would be asymmetric. If mutianf wants hardening we can do both files together in a follow-up using atomic.Pointer[metric.Float64Histogram] (cleaner than the atomic.Value + type-assert boilerplate). See #20190 (comment) for the batch reply.

Synchronize the global histogram swaps in the test using atomic.Value.Store to ensure thread safety and avoid race conditions during test execution.

func TestSessionTracer_NilHistogramsAreNoOps(t *testing.T) {
	// Save and restore the package-global histograms so we can simulate
	// the pre-init state without racing other tests. This works because
	// go test runs a single package's tests serially by default within
	// a single test binary.
	origO := sessionOpenLatencies.Load()
	origD := sessionDurations.Load()
	origU := sessionUptime.Load()
	origT := transportLatencies.Load()

	sessionOpenLatencies.Store((metric.Float64Histogram)(nil))
	sessionDurations.Store((metric.Float64Histogram)(nil))
	sessionUptime.Store((metric.Float64Histogram)(nil))
	transportLatencies.Store((metric.Float64Histogram)(nil))

	t.Cleanup(func() {
		if origO != nil {
			sessionOpenLatencies.Store(origO)
		} else {
			sessionOpenLatencies.Store((metric.Float64Histogram)(nil))
		}
		if origD != nil {
			sessionDurations.Store(origD)
		} else {
			sessionDurations.Store((metric.Float64Histogram)(nil))
		}
		if origU != nil {
			sessionUptime.Store(origU)
		} else {
			sessionUptime.Store((metric.Float64Histogram)(nil))
		}
		if origT != nil {
			transportLatencies.Store(origT)
		} else {
			transportLatencies.Store((metric.Float64Histogram)(nil))
		}
	})

	tr := newSessionTracer(SessionTypeTable)
	// Each recorder MUST early-return without panic when its histogram
	// is nil (documented in-file). Simple call-and-return smoke test.
	tr.recordOpen(context.Background(), nil)
	tr.recordClose(context.Background(), "", nil, false, false)
	tr.sampleUptime(context.Background())
	tr.recordTransportOverhead(context.Background(), "Read", 1*time.Millisecond)
}

TestSessionTracer_NilHistogramsAreNoOps does not exist on this branch — see the NOTE: at session_tracer_test.go:222 explaining why it was intentionally omitted (save/restore of the package globals leaked heartbeat goroutines from other tests in this package that keep calling recordClose on the same globals). Nothing to change.

why do we need to take a snapshot and record the labels with snap.poolName and snap.afeLocation instead of just record t.afeLocation and t.PoolName? I don't think peerInfo changes once a session is created. Same question for other metrics

Refactored in 940bd44 — sessionTracer no longer holds peerInfo. Set-once on Session per SESSION_SPEC #3 (atomic.Pointer, sole writer in handleOpenSession), so shadowing it on the tracer under a second mutex was pointless duplication. tracerSnapshot / setPeerInfo / snapshot() deleted; every record*/sample* method now takes peerInfo *spb.PeerInfo as a parameter and callers pass Session.PeerInfo() at the call site. opened demoted to atomic.Bool — tracer holds no lock at all now. New peerInfoLabels(*PeerInfo) helper derives (transport_type, afe_location); nil → ("unknown", "").

there should always be a close reason.

Agree. Java parity: SessionImpl.notifyTerminalClose (SessionImpl.java:782-793) guards a missing closeReason, logs IllegalStateException, and synthesizes a fallback before calling tracer.onClose at line 819 — so at the tracer boundary the reason is always non-empty.

Tightened the doc in 940bd44 — dropped the "or empty" language; now reads "the terminal Session.CloseReason(), always non-empty — Session synthesizes a fallback if the close path forgot to set one, mirroring java-bigtable's SessionImpl.notifyTerminalClose guard (SessionImpl.java:782-793)".

Empirically load-bearing: a workload run on feat/bigtable-sessionz-debug already shows 7 hits on session_close_no_reason over ~3h, all traced to two ForceClose(nil) safety-net paths in Close() itself (drain-ctx-expired and Send-failed). Session-side fix — pass the caller's original req forward with a specific CloseAbort:* label instead of dropping it — is queued as a follow-up on that branch; will cross-link when it lands.