perf(opensearchtransport): make connection dead/overloaded timestamps lock-free (#892) (#906)

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 -- the #1 explicit-lock contention site under concurrent
load. Port of the non-breaking perf work from #901; v4 retains EnableMetrics
(unchanged surface and behavior).

Also fix a pre-existing data race in RolePolicy.DiscoveryUpdate, surfaced
while auditing the pool locking. It 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, so two concurrent DiscoverNodes calls raced on those
fields. Compute the projected pool size and recalculate warmup parameters
under pool.Lock(), releasing before discoveryUpdateAdd/Remove. Add
TestRolePolicyDiscoveryUpdateConcurrent, which reproduces the race under -race.

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

Author: ryanyuan

CHANGELOG.md

@@ -164,6 +164,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ### Changed
 
+- Connection dead/overloaded timestamps (`deadSince`/`overloadedAt`) are now lock-free `atomic.Int64` values. `buildConnectionMetric` no longer locks each connection to read them, removing the dominant metrics-snapshot lock contention against the per-request writers. No public API change; `EnableMetrics` behavior is unchanged. ([#892](https://github.com/opensearch-project/opensearch-go/issues/892))
 - 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). ([#891](https://github.com/opensearch-project/opensearch-go/issues/891))
 - **BREAKING**: `opensearch.Request` interface signature changed from `GetRequest() (*http.Request, error)` to `GetRequest(method string) (*http.Request, error)`. The HTTP method is now caller-provided rather than hardcoded per operation, enabling correct method selection for operations that support multiple HTTP methods (e.g. search supports both GET and POST). This only affects code that implements or calls `GetRequest` directly; standard usage through client methods (e.g. `client.Search(ctx, req)`) is unaffected ([#650](https://github.com/opensearch-project/opensearch-go/issues/650))
 - Bump CI and developer guide OpenSearch versions: compatibility matrix to 2.19.5, default integration test version to 3.6.0 ([#810](https://github.com/opensearch-project/opensearch-go/pull/810))

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

@@ -183,6 +183,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
@@ -194,13 +202,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
 	}
 
@@ -222,6 +231,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 {
@@ -270,20 +317,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)
 }
 
@@ -452,9 +499,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

opensearchtransport/connection_internal_test.go

@@ -142,14 +142,14 @@ func TestMultiServerPoolOnSuccess(t *testing.T) {
 		pool.OnSuccess(conn)
 
 		conn.mu.Lock()
-		isDead := !conn.mu.deadSince.IsZero()
+		isDead := !conn.loadDeadSince().IsZero()
 		conn.mu.Unlock()
 		if isDead {
 			t.Errorf("Expected the connection to be ready; %s", conn)
 		}
 
 		conn.mu.Lock()
-		deadSince := conn.mu.deadSince
+		deadSince := conn.loadDeadSince()
 		conn.mu.Unlock()
 		if !deadSince.IsZero() {
 			t.Errorf("Unexpected value for DeadSince: %s", deadSince)
@@ -203,8 +203,8 @@ func TestMultiServerPoolOnFailure(t *testing.T) {
 			t.Fatalf("Unexpected error: %s", err)
 		}
 		conn.mu.Lock()
-		isDead := !conn.mu.deadSince.IsZero()
-		deadSince := conn.mu.deadSince
+		isDead := !conn.loadDeadSince().IsZero()
+		deadSince := conn.loadDeadSince()
 		conn.mu.Unlock()
 
 		if !isDead {
@@ -268,7 +268,7 @@ func TestMultiServerPoolOnFailure(t *testing.T) {
 		conn := pool.mu.ready[0]
 		conn.state.Store(int64(newConnState(lcDead)))
 		conn.mu.Lock()
-		conn.mu.deadSince = time.Now().UTC()
+		conn.storeDeadSince(time.Now().UTC())
 		conn.mu.Unlock()
 
 		if err := pool.OnFailure(conn); err != nil {
@@ -288,8 +288,8 @@ func TestConnection(t *testing.T) {
 		}
 		conn.failures.Store(10)
 		conn.mu.Lock()
-		conn.mu.deadSince = time.Now().UTC()
-		conn.mu.deadSince = time.Now().UTC()
+		conn.storeDeadSince(time.Now().UTC())
+		conn.storeDeadSince(time.Now().UTC())
 		conn.mu.Unlock()
 
 		match, err := regexp.MatchString(

opensearchtransport/coverage_critical_path_internal_test.go

@@ -43,7 +43,7 @@ func TestPolicyChainOnSuccess(t *testing.T) {
 		chain.OnSuccess(conn)
 		// Verify still alive
 		conn.mu.RLock()
-		require.True(t, conn.mu.deadSince.IsZero())
+		require.True(t, conn.loadDeadSince().IsZero())
 		conn.mu.RUnlock()
 	})
 
@@ -53,13 +53,13 @@ func TestPolicyChainOnSuccess(t *testing.T) {
 		conn := createDeadTestConnection("http://node:9200")
 
 		conn.mu.RLock()
-		require.False(t, conn.mu.deadSince.IsZero(), "precondition: must be dead")
+		require.False(t, conn.loadDeadSince().IsZero(), "precondition: must be dead")
 		conn.mu.RUnlock()
 
 		chain.OnSuccess(conn)
 
 		conn.mu.RLock()
-		require.True(t, conn.mu.deadSince.IsZero(), "OnSuccess should mark dead connection healthy")
+		require.True(t, conn.loadDeadSince().IsZero(), "OnSuccess should mark dead connection healthy")
 		conn.mu.RUnlock()
 	})
 
@@ -70,13 +70,13 @@ func TestPolicyChainOnSuccess(t *testing.T) {
 		conn.drainingQuiescingRemaining.Store(3)
 
 		conn.mu.RLock()
-		deadBefore := conn.mu.deadSince
+		deadBefore := conn.loadDeadSince()
 		conn.mu.RUnlock()
 
 		chain.OnSuccess(conn)
 
 		conn.mu.RLock()
-		require.Equal(t, deadBefore, conn.mu.deadSince, "draining connection must stay dead")
+		require.Equal(t, deadBefore, conn.loadDeadSince(), "draining connection must stay dead")
 		conn.mu.RUnlock()
 	})
 
@@ -88,13 +88,13 @@ func TestPolicyChainOnSuccess(t *testing.T) {
 		conn.state.Store(int64(newConnState(lcDead | lcOverloaded | lcNeedsWarmup)))
 
 		conn.mu.RLock()
-		deadBefore := conn.mu.deadSince
+		deadBefore := conn.loadDeadSince()
 		conn.mu.RUnlock()
 
 		chain.OnSuccess(conn)
 
 		conn.mu.RLock()
-		require.Equal(t, deadBefore, conn.mu.deadSince, "overloaded connection must stay dead")
+		require.Equal(t, deadBefore, conn.loadDeadSince(), "overloaded connection must stay dead")
 		conn.mu.RUnlock()
 	})
 
@@ -115,7 +115,7 @@ func TestPolicyChainOnSuccess(t *testing.T) {
 		wg.Wait()
 
 		conn.mu.RLock()
-		require.True(t, conn.mu.deadSince.IsZero(), "after concurrent OnSuccess, connection must be healthy")
+		require.True(t, conn.loadDeadSince().IsZero(), "after concurrent OnSuccess, connection must be healthy")
 		conn.mu.RUnlock()
 	})
 }

opensearchtransport/discovery.go

@@ -694,12 +694,12 @@ func (c *Client) updateConnectionPool(
 	for url, conn := range finalConnectionsByURL {
 		if _, isReady := readyURLs[url]; isReady {
 			conn.mu.Lock()
-			deadSince := conn.mu.deadSince
+			deadSince := conn.loadDeadSince()
 			stale := !deadSince.IsZero() && !healthCheckedAt.IsZero() && deadSince.Before(healthCheckedAt)
 			switch {
 			case stale:
 				// Dead state predates the health check -- resurrect.
-				conn.mu.deadSince = time.Time{}
+				conn.storeDeadSince(time.Time{})
 				conn.mu.Unlock()
 				conn.failures.Store(0)
 			case !deadSince.IsZero():

opensearchtransport/discovery_internal_test.go

@@ -1869,7 +1869,7 @@ func TestUpdateConnectionPool(t *testing.T) {
 
 		// Mark the connection as dead in the past
 		conn.mu.Lock()
-		conn.mu.deadSince = time.Now().Add(-5 * time.Second)
+		conn.storeDeadSince(time.Now().Add(-5 * time.Second))
 		conn.mu.Unlock()
 		conn.failures.Store(3)
 
@@ -1881,7 +1881,7 @@ func TestUpdateConnectionPool(t *testing.T) {
 
 		// The old connection's dead state should have been cleared (resurrected)
 		conn.mu.RLock()
-		deadSince := conn.mu.deadSince
+		deadSince := conn.loadDeadSince()
 		conn.mu.RUnlock()
 		require.True(t, deadSince.IsZero(), "stale dead state should be cleared")
 		require.Equal(t, int64(0), conn.failures.Load(), "failures should be reset")
@@ -1901,15 +1901,15 @@ func TestUpdateConnectionPool(t *testing.T) {
 
 		// Mark dead AFTER the health check time
 		conn.mu.Lock()
-		conn.mu.deadSince = healthCheckedAt.Add(1 * time.Second)
+		conn.storeDeadSince(healthCheckedAt.Add(1 * time.Second))
 		conn.mu.Unlock()
 
 		// Re-discover with SAME pointer: dead state is newer than healthCheckedAt -> should stay dead
 		err = client.updateConnectionPool(t.Context(), healthCheckedAt, []*Connection{conn}, nil)
 		require.NoError(t, err)
 
 		conn.mu.RLock()
-		deadSince := conn.mu.deadSince
+		deadSince := conn.loadDeadSince()
 		conn.mu.RUnlock()
 		require.False(t, deadSince.IsZero(), "concurrent dead state should be preserved")
 	})
@@ -1983,7 +1983,7 @@ func TestUpdateConnectionPool(t *testing.T) {
 
 		// Verify deadSince is initially zero
 		dead.mu.RLock()
-		require.True(t, dead.mu.deadSince.IsZero(), "deadSince should be zero before pool placement")
+		require.True(t, dead.loadDeadSince().IsZero(), "deadSince should be zero before pool placement")
 		dead.mu.RUnlock()
 
 		err := client.updateConnectionPool(t.Context(), time.Time{}, []*Connection{ready}, []*Connection{dead})
@@ -2001,7 +2001,7 @@ func TestUpdateConnectionPool(t *testing.T) {
 
 		// Verify deadSince was set by appendToDeadWithLock
 		deadConn.mu.RLock()
-		require.False(t, deadConn.mu.deadSince.IsZero(), "deadSince must be set for dead-list connections")
+		require.False(t, deadConn.loadDeadSince().IsZero(), "deadSince must be set for dead-list connections")
 		deadConn.mu.RUnlock()
 
 		// Verify lcUnknown is set

opensearchtransport/metrics.go

@@ -416,10 +416,12 @@ func buildConnectionMetric(c *Connection) ConnectionMetric {
 	state := c.loadConnState()
 	lc := state.lifecycle()
 
-	c.mu.Lock()
-	deadSince := c.mu.deadSince
-	overloadedAt := c.mu.overloadedAt
-	c.mu.Unlock()
+	// Read the dead/overloaded timestamps lock-free: they are written under c.mu
+	// but safe to read without it, so the metrics snapshot avoids the dominant
+	// per-connection lock contention against the per-request OnSuccess/OnFailure
+	// writers.
+	deadSince := c.loadDeadSince()
+	overloadedAt := c.loadOverloadedAt()
 
 	cm := ConnectionMetric{
 		URL:              c.URL.String(),