feat(opensearchtransport)!: remove EnableMetrics, reimplement deep metrics with lock-free structures (#901)

* feat(opensearchtransport)!: remove EnableMetrics, make detailed metrics always-on and lock-free (#892)

Remove EnableMetrics from the client and transport Config. Detailed
metrics are now always collected; Metrics() always returns the full
snapshot. (BREAKING)

Convert deadSince/overloadedAt from mu-guarded time.Time to lock-free
atomic.Int64 (UnixNano, 0 = unset). Writes still occur under c.mu so the
resurrection/standby read-modify-write decisions stay serialized; only
the reads went lock-free, so buildConnectionMetric no longer takes each
connection's mutex. This eliminates the #1 explicit-lock contention site
measured under concurrent Perform() load.

Signed-off-by: Ryan Yuan <ryan.yuan@crowdstrike.com>

* docs(opensearchtransport): disambiguate uninitialized-metrics error

The defensive nil-metrics path returned "transport metrics not enabled",
recycling the removed EnableMetrics concept and contradicting the Metrics()
godoc that the error is non-nil only on snapshot-callback failure. Reword the
error to "transport metrics not initialized", state both error conditions in
the godoc, and align the test comment.

Signed-off-by: Ryan Yuan <ryan.yuan@crowdstrike.com>

* test(opensearchtransport): drop EnableMetrics references from comments

The test comments narrated the removed EnableMetrics flag as history. Describe
the end-state behavior instead: callbacks register when a router has policies,
and the detailed snapshot runs unconditionally.

Signed-off-by: Ryan Yuan <ryan.yuan@crowdstrike.com>

* fix(opensearchtransport): lock pool write in RolePolicy.DiscoveryUpdate

RolePolicy.DiscoveryUpdate called recalculateWarmupParams without holding
the pool write lock, while the roundrobin and cluster_coordinator policies
took pool.Lock() for the identical call. recalculateWarmupParams writes the
pool's warmupRounds, warmupSkipCount, and activeListCap fields, which
getWarmupParams and the other DiscoveryUpdate callers read and write under
that same lock.

Two concurrent DiscoverNodes calls on a shared transport therefore raced on
those fields (observed in CI: two goroutines writing in
recalculateWarmupParams via RolePolicy.DiscoveryUpdate). This is pre-existing
on main, unrelated to the EnableMetrics removal.

Compute the projected pool size and recalculate the warmup parameters under
pool.Lock(), releasing before discoveryUpdateAdd/Remove (which acquire the
lock per-connection). Add TestRolePolicyDiscoveryUpdateConcurrent, which
reproduces the race under -race and passes with the fix.

Signed-off-by: Sean Chittenden <sean.chittenden@crowdstrike.com>

---------

Signed-off-by: Ryan Yuan <ryan.yuan@crowdstrike.com>
Signed-off-by: Sean Chittenden <sean.chittenden@crowdstrike.com>
Co-authored-by: Sean Chittenden <sean.chittenden@crowdstrike.com>

Author: ryanyuan

CHANGELOG.md

@@ -167,6 +167,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 ### Changed
 
 - **BREAKING**: Per-request transport metrics (`requests`, `failures`, responses-by-status) are now always collected via lock-free atomics, independent of `EnableMetrics`. `EnableMetrics` now gates only the detailed-metrics snapshot (per-connection, per-policy, and router state returned by `Metrics()`). The responses-by-status counter moved from a mutex-guarded map to a lock-free atomic array. `Metrics()` no longer returns an error when metrics are disabled -- it always returns the per-request counters (callers that branched on `if err != nil` for the disabled case should drop that check). See [`UPGRADING_V5.md`](UPGRADING_V5.md#metrics-error-on-disabled-removed) for migration. ([#891](https://github.com/opensearch-project/opensearch-go/issues/891))
+- Make the detailed-metrics snapshot path lock-free at call time. The per-connection `deadSince`/`overloadedAt` timestamps moved from `Connection.mu`-guarded `time.Time` fields to lock-free atomic Unix-nanosecond values, so `Metrics()` enumerates connections without taking each connection's mutex. Under concurrent request load this was the dominant lock-contention site (a mutex profile attributed ~3.85% of total contention delay to the snapshot reader taking a write lock merely to read two fields); the conversion drops that to ~0.1%. Writes still occur under `Connection.mu` so the resurrection/standby read-modify-write decisions stay serialized. Benchmarks (`BenchmarkMetrics`, `BenchmarkMetricsParallel`, `BenchmarkMetricsUnderLoad`) confirm the always-on detailed path is acceptable. ([#892](https://github.com/opensearch-project/opensearch-go/issues/892))
 - Reorganize the documentation. Split `UPGRADING.md` into a version-history index plus per-major-version guides (`UPGRADING_V5.md` through `UPGRADING_V2.md`) and rename `opensearchapi/MIGRATING.md` to `opensearchapi/UPGRADING_V4_TO_V5.md`. Group the `guides/` and `_samples/` files by subsystem (`transport-`, `indexing-`, `usage-`, `config-`) and add a `guides/README.md` index. Make `guides/usage-error_handling.md` the single source for partial-error handling and `guides/transport-retry_backoff.md` the single source for resurrection-timeout config, replacing the duplicated copies in `opensearchapi/README.md` and `guides/transport-routing.md` with links. Add package documentation (`doc.go`) for `opensearchapi`, `plugins`, `signer`, and `signer/awsv2`.
 - Trim the CI compatibility matrix to the currently-patched OpenSearch set (2.19.x and 3.x) per the 12-month support policy; older lines (1.3.x - 2.18.x) are no longer part of the tested matrix and the 4.x client remains their supported path. No client code change ([#856](https://github.com/opensearch-project/opensearch-go/issues/856))
 - **BREAKING**: Module path is now `github.com/opensearch-project/opensearch-go/v5`. Update import paths from `/v4` to `/v5`; the in-source `opensearchapi.X` package qualifier is unchanged
@@ -219,11 +220,13 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 ### Removed
 
 - Remove deprecated `(*opensearch.Client).Perform` and `(*opensearchtransport.Transport).Perform`; `Stream(*http.Request) (*http.Response, error)` is now the sole method on `opensearchtransport.Interface`. Custom transport implementations must implement `Stream` instead of `Perform`. The `opensearch.Streamer` opt-in interface and `opensearch.ErrTransportMissingMethodStream` sentinel are removed. ([#872](https://github.com/opensearch-project/opensearch-go/issues/872))
+- **BREAKING**: Remove the `EnableMetrics` config flag from `opensearch.Config` and `opensearchtransport.Config`. The detailed-metrics snapshot (per-connection enumeration, per-policy breakdowns, and router cache state) is now always available; `Metrics()` returns the full snapshot unconditionally. The flag's only remaining purpose after [#891](https://github.com/opensearch-project/opensearch-go/issues/891) was to gate the detailed path, which now does its work lazily and lock-free at call time and so costs nothing until `Metrics()` is called. Delete any `EnableMetrics` field from your config (it is a compile error otherwise); see [`UPGRADING_V5.md`](UPGRADING_V5.md#enablemetrics-removed). ([#892](https://github.com/opensearch-project/opensearch-go/issues/892))
 - Remove backport.yml and dependabot_pr.yml as we are not using backport app anymore
 - Stop emitting `opensearchapi.Client` sub-client fields that have no operations routed to them. `cmd/osgen` now emits a sub-client only when at least one operation targets it, dropping the previously-empty `Script`, `ComponentTemplate`, `IndexTemplate`, `Template`, and `DataStream` fields. Index-template and data-stream operations are reached through `client.Indices.*` (e.g. `client.Indices.PutIndexTemplate`, `client.Indices.CreateDataStream`); stored-script operations remain top-level on `Client`
 
 ### Fixed
 
+- Fix a data race on the multi-server pool's `warmupRounds`, `warmupSkipCount`, and `activeListCap` fields when two concurrent `DiscoverNodes` calls drive `RolePolicy.DiscoveryUpdate` on a shared transport. `RolePolicy` called `recalculateWarmupParams` (which writes those fields) without holding the pool write lock, while the `roundrobin` and `cluster_coordinator` policies took the lock for the identical call. `RolePolicy.DiscoveryUpdate` now computes the projected pool size and recalculates the warmup parameters under `pool.Lock()`, matching the other callers
 - Cache credentials in the `signer/awsv2` constructors. A raw `CredentialsProvider` is wrapped in an `aws.CredentialsCache` (an already-cached provider, such as one from `config.LoadDefaultConfig`, is left as-is), so SigV4 signing no longer calls `Credentials.Retrieve` on every request. For STS-backed providers (assume-role, web identity, IRSA) the previous behavior was a per-request STS call that could exhaust the account's STS rate limits under load. `signer/awsv2` shipped without this in v4.6.0.
 - Fix `cmd/osgen` silently dropping a response struct when a response schema has a `oneOf`/`anyOf` field whose parent-scoped union name collides with the parent struct's own Go name. The union registered first and the parent struct was then dropped by the type registry (its name already taken), degrading the response to raw `json.RawMessage`. Such a union is now re-keyed by its referenced schema so the parent struct survives. The generator also reports any remaining Go type name collisions to stderr at generation time instead of dropping types silently. Regenerating fixes two type families: `tasks.list`, `tasks.cancel`, and `delete_by_query_rethrottle` change from raw `Body json.RawMessage` to typed structs (`NodeFailures`, `TaskFailures`, `Nodes map[string]TasksTaskExecutingNode`, `Tasks *TasksTaskInfos`), and the `_common.mapping___DynamicTemplate.mapping` field becomes typed `*CommonMappingProperty` (accounting for the large `unions_gen.go`/`indices-put_mapping_gen.go` churn). ([#890](https://github.com/opensearch-project/opensearch-go/pull/890))
 - Fix `cmd/osgen` degrading two more schema shapes to raw `json.RawMessage`: an OpenAPI 3.1 nullable scalar (`type: ["null", "<primitive>"]`) fell through because kin-openapi's `Type.Is` matches only single-element type sets, and a response whose component schema is a bare `$ref` alias (`Foo: {$ref: Bar}`) missed the registry lookup under its alias key. Nullable scalars now resolve to the pointer primitive (`*string`/`*int`/`*bool`/`*float64`), clearing the CAT `*Record` cluster, and alias responses follow the `$ref` chain to the registered struct, fixing ISM `add`/`delete`/`get`/`remove_policy` + `retry_index` and the seven `ml.search_*` responses. ([#890](https://github.com/opensearch-project/opensearch-go/pull/890))

UPGRADING_V5.md

@@ -124,6 +124,16 @@ use(m.Requests, m.Failures)
 
 Detailed fields such as `Policies` and `Router` remain populated only when `EnableMetrics` is set; reading them without it yields nil, unchanged from v4.
 
+> Note: a later v5 change removed `EnableMetrics` entirely -- see [`EnableMetrics` removed](#enablemetrics-removed) below. The behavior described above was the intermediate state; in the shipped v5 the detailed fields are always populated.
+
+## `EnableMetrics` removed
+
+`EnableMetrics` has been removed from both `opensearch.Config` and `opensearchtransport.Config`. The detailed-metrics snapshot (per-connection enumeration, per-policy breakdowns, and router cache state) is now always available -- it is assembled lazily and lock-free at the moment you call `Metrics()`, so it adds no per-request cost. The per-request counters were already always-on.
+
+Delete any `EnableMetrics` field from your config; leaving it in place is a compile error.
+
+`Metrics()` now returns the full snapshot unconditionally, including `Connections`, `Policies`, and `Router` (the latter two populate when a router with policies is active). The returned error is still non-nil only when a snapshot callback fails.
+
 ## `opensearchtransport.Client` renamed to `opensearchtransport.Transport`
 
 The concrete `opensearchtransport.Client` type was renamed to `opensearchtransport.Transport`. The type owns HTTP round-trip concerns -- connection pooling, retries, node selection, and discovery -- so `Transport` reflects its role and avoids colliding conceptually with the API clients above it (`opensearch.Client` and `opensearchapi.Client`).

_samples/transport-discovery_demo.go

@@ -46,8 +46,7 @@ func main() {
 			{Scheme: "http", Host: "localhost:9201"},
 		},
 		DiscoverNodesInterval: 5 * time.Second, // Fast for demo
-		EnableMetrics:         true,
-		EnableDebugLogger:     !isCI(), // Enable locally, disable in CI
+		EnableDebugLogger:     !isCI(),         // Enable locally, disable in CI
 		Logger:                &opensearchtransport.ColorLogger{Output: os.Stdout, EnableRequestBody: false, EnableResponseBody: false},
 		Router:                router,
 

guides/transport-metrics.md

@@ -2,16 +2,15 @@
 
 The opensearch-go transport exposes a pull-based metrics API that returns a point-in-time snapshot of request counters, connection pool state, per-connection health, policy-level breakdowns, and router cache state. All fields are JSON-tagged for easy serialization.
 
-Metrics come in two tiers. The **per-request counters** (`requests`, `failures`, and responses-by-status) are always collected via lock-free atomics and returned by `Metrics()` regardless of configuration. The **detailed-metrics snapshot** (connection-pool state, per-connection health, per-policy breakdowns, and router cache state) is opt-in behind `EnableMetrics`.
+Metrics are collected using atomic, per-request counters (e.g. `requests`, `failures`, and responses-by-status) recorded on the request hot path. Additional detailed metrics (connection-pool state, per-connection health, per-policy breakdowns, and router cache state) are lazily accumulated and read lock-free when you call `Metrics()`, so they cost nothing until you ask for them.
 
 ## Quick Start
 
-The per-request counters require no configuration. To also populate the detailed-metrics snapshot, set `EnableMetrics: true` on `opensearch.Config`. When constructing through `opensearchapi.NewClient`, set the flag on the embedded `opensearch.Config` and reach the method via `apiClient.Client.Metrics()`.
+Metrics require no configuration. Construct a client and call `Metrics()`. When constructing through `opensearchapi.NewClient`, reach the method via `apiClient.Client.Metrics()`.
 
 ```go
 client, err := opensearch.NewClient(opensearch.Config{
-    Addresses:     []string{"https://localhost:9200"},
-    EnableMetrics: true,
+    Addresses: []string{"https://localhost:9200"},
 })
 if err != nil {
     log.Fatal(err)
@@ -26,7 +25,7 @@ data, _ := json.MarshalIndent(m, "", "  ")
 fmt.Println(string(data))
 ```
 
-The `Metrics()` method lives on `opensearch.Client`. It returns an `opensearchtransport.Metrics` struct and an error -- non-nil when a detailed-metrics snapshot callback fails. A `New()`-constructed transport always returns the per-request counters, so `Metrics()` does not error merely because `EnableMetrics` is unset.
+The `Metrics()` method lives on `opensearch.Client`. It returns an `opensearchtransport.Metrics` struct and an error -- non-nil only when a detailed-snapshot callback fails. It never errors merely because metrics are "disabled"; they are always available.
 
 ---
 

opensearch.go

@@ -188,7 +188,6 @@ type Config struct {
 	// Default: nil (no modification)
 	HealthCheckRequestModifier func(*http.Request)
 
-	EnableMetrics     bool // Enable the metrics collection.
 	EnableDebugLogger bool // Enable the debug logging.
 
 	// ActiveListCap sets the maximum number of connections in the ready list's active partition per pool.
@@ -346,7 +345,6 @@ func NewClient(cfg Config) (*Client, error) {
 
 		CompressRequestBody: cfg.CompressRequestBody,
 
-		EnableMetrics:     cfg.EnableMetrics,
 		EnableDebugLogger: cfg.EnableDebugLogger,
 
 		DiscoverNodesInterval: cfg.DiscoverNodesInterval,

opensearch_internal_test.go

@@ -569,7 +569,7 @@ func TestVersion(t *testing.T) {
 }
 
 func TestClientMetrics(t *testing.T) {
-	c, _ := NewClient(Config{EnableMetrics: true, Transport: mockhttp.NewRoundTripFunc(t, defaultRoundTripFunc)})
+	c, _ := NewClient(Config{Transport: mockhttp.NewRoundTripFunc(t, defaultRoundTripFunc)})
 
 	m, err := c.Metrics()
 	require.NoError(t, err)

opensearchtransport/address_resolver_internal_test.go

@@ -208,7 +208,6 @@ func TestAddressResolver(t *testing.T) {
 			tp, err := New(Config{
 				URLs:                []*url.URL{testSeedURL(t)},
 				Transport:           newResolverTestTransport(t, nodesJSON),
-				EnableMetrics:       true,
 				HealthCheck:         NoOpHealthCheck,
 				AddressResolver:     tt.resolver,
 				MaxAddressResolvers: 1, // serial for deterministic behavior
@@ -719,10 +718,9 @@ func TestAddressResolverRunner(t *testing.T) {
 
 		var resolverCalled atomic.Int32
 		tp, err := New(Config{
-			URLs:          []*url.URL{testSeedURL(t)},
-			Transport:     newResolverTestTransport(t, nodesJSON),
-			EnableMetrics: true,
-			HealthCheck:   NoOpHealthCheck,
+			URLs:        []*url.URL{testSeedURL(t)},
+			Transport:   newResolverTestTransport(t, nodesJSON),
+			HealthCheck: NoOpHealthCheck,
 			AddressResolver: func(_ context.Context, _ NodeInfo) (*url.URL, error) {
 				resolverCalled.Add(1)
 				return nil, nil //nolint:nilnil // testing (nil, nil) protocol case
@@ -1143,7 +1141,6 @@ func TestAddressResolverRunnerProtocol(t *testing.T) {
 					tp, err := New(Config{
 						URLs:                  []*url.URL{testSeedURL(t)},
 						Transport:             newResolverTestTransport(t, nodesJSON),
-						EnableMetrics:         true,
 						HealthCheck:           NoOpHealthCheck,
 						AddressResolver:       tt.resolver,
 						MaxAddressResolvers:   1,

opensearchtransport/cluster_health_internal_test.go

@@ -1382,7 +1382,7 @@ func TestFetchAndEvaluateNodeStats(t *testing.T) {
 		conn := &Connection{URL: serverURL}
 		conn.state.Store(int64(newConnState(lcStandby | lcOverloaded)))
 		conn.mu.Lock()
-		conn.mu.overloadedAt = time.Now()
+		conn.storeOverloadedAt(time.Now())
 		conn.mu.Unlock()
 
 		a1 := newActiveConn("a1")

opensearchtransport/connection.go

@@ -187,6 +187,14 @@ type Connection struct {
 	failures atomic.Int64
 	state    atomic.Int64 // Packed connState: connLifecycle (12b) + 2*warmupManager (26b each)
 
+	// deadSinceNano and overloadedAtNano hold Unix-nanosecond timestamps, with 0
+	// meaning "unset" (the zero time). They are read lock-free by Metrics() and
+	// written under c.mu; see c.mu for the locking protocol. Use the
+	// loadDeadSince/storeDeadSince/deadSinceIsZero accessors (and the overloadedAt
+	// equivalents) rather than touching these directly.
+	deadSinceNano    atomic.Int64
+	overloadedAtNano atomic.Int64
+
 	// drainingQuiescingRemaining counts the number of successful health checks remaining
 	// before this connection can be resurrected. Set to defaultDrainingQuiescingChecks when
 	// an HTTP/2 stream reset is observed (RST_STREAM, e.g., REFUSED_STREAM). Each successful
@@ -198,13 +206,14 @@ type Connection struct {
 	// defaultDrainingQuiescingChecks * resurrectTimeout.
 	drainingQuiescingRemaining atomic.Int64
 
+	// mu guards the fields below and serializes the resurrection/standby
+	// read-modify-write decisions. The deadSinceNano/overloadedAtNano atomics are
+	// written under mu but read lock-free.
 	mu struct {
 		sync.RWMutex
-		deadSince              time.Time
 		checkStartedAt         time.Time
 		clusterHealth          *ClusterHealthLocal // Populated when lcClusterHealthAvailable is set
 		clusterHealthCheckedAt time.Time           // When cluster health was last probed (for retry timing)
-		overloadedAt           time.Time           // When overloaded state was last set (lcOverloaded metadata bit)
 		lastBreakerTripped     map[string]int64    // Previous tripped counts for delta detection
 	}
 
@@ -226,6 +235,44 @@ type Connection struct {
 	}
 }
 
+// timeToNano converts a time.Time to its Unix-nanosecond representation. The
+// zero value is preserved to imply "unset."
+func timeToNano(t time.Time) int64 {
+	if t.IsZero() {
+		return 0
+	}
+	return t.UnixNano()
+}
+
+// nanoToTime converts a stored Unix-nanosecond value back to time.Time in UTC,
+// mapping the 0 sentinel to the zero time.
+func nanoToTime(n int64) time.Time {
+	if n == 0 {
+		return time.Time{}
+	}
+	return time.Unix(0, n).UTC()
+}
+
+// loadDeadSince returns the time the connection was marked dead, or the zero
+// time if it is alive. Lock-free.
+func (c *Connection) loadDeadSince() time.Time { return nanoToTime(c.deadSinceNano.Load()) }
+
+// storeDeadSince records (or clears, with the zero time) the dead timestamp.
+// Callers hold c.mu; see c.mu for the locking protocol.
+func (c *Connection) storeDeadSince(t time.Time) { c.deadSinceNano.Store(timeToNano(t)) }
+
+// deadSinceIsZero reports whether the connection is alive (no dead timestamp).
+// Lock-free.
+func (c *Connection) deadSinceIsZero() bool { return c.deadSinceNano.Load() == 0 }
+
+// loadOverloadedAt returns the time the connection was last marked overloaded,
+// or the zero time. Lock-free.
+func (c *Connection) loadOverloadedAt() time.Time { return nanoToTime(c.overloadedAtNano.Load()) }
+
+// storeOverloadedAt records (or clears, with the zero time) the overloaded
+// timestamp. Callers hold c.mu; see c.mu for the locking protocol.
+func (c *Connection) storeOverloadedAt(t time.Time) { c.overloadedAtNano.Store(timeToNano(t)) }
+
 // effectiveWeight returns the connection's weight for round-robin selection.
 // Returns 1 if weight is zero (default for connections created without explicit weight).
 func (c *Connection) effectiveWeight() int {
@@ -274,20 +321,20 @@ func (c *Connection) decrementDrainingQuiescing() int64 {
 
 // markAsDeadWithLock marks the connection as dead (caller must hold lock).
 func (c *Connection) markAsDeadWithLock() {
-	if c.mu.deadSince.IsZero() {
-		c.mu.deadSince = time.Now().UTC()
+	if c.deadSinceIsZero() {
+		c.storeDeadSince(time.Now().UTC())
 	}
 	c.failures.Add(1)
 }
 
 // markAsReadyWithLock marks the connection as alive (caller must hold lock).
 func (c *Connection) markAsReadyWithLock() {
-	c.mu.deadSince = time.Time{}
+	c.storeDeadSince(time.Time{})
 }
 
 // markAsHealthyWithLock marks the connection as healthy (caller must hold lock).
 func (c *Connection) markAsHealthyWithLock() {
-	c.mu.deadSince = time.Time{}
+	c.storeDeadSince(time.Time{})
 	c.failures.Store(0)
 }
 
@@ -456,9 +503,7 @@ func (c *Connection) storeMaxCwnd(poolName string, size int) {
 
 // String returns a readable connection representation.
 func (c *Connection) String() string {
-	c.mu.RLock()
-	deadAt := c.mu.deadSince
-	c.mu.RUnlock()
+	deadAt := c.loadDeadSince()
 
 	if deadAt.IsZero() {
 		return fmt.Sprintf("<%s> dead=false failures=%d", c.URL, c.failures.Load())

opensearchtransport/connection_benchmark_test.go

@@ -160,7 +160,7 @@ func createMultiServerPool(conns []*Connection) *multiServerPool {
 	for _, conn := range ready {
 		conn.state.Store(int64(newConnState(lcActive)))
 		conn.mu.Lock()
-		conn.mu.deadSince = time.Time{} // Reset from prior benchmark sub-runs
+		conn.storeDeadSince(time.Time{}) // Reset from prior benchmark sub-runs
 		conn.mu.Unlock()
 	}
 	pool.mu.ready = ready