fix(opensearchtransport): keep dedicated cluster managers in the connection inventory

A dedicated cluster manager (cluster_manager role with no work roles) was
filtered out of the allConns pool during discovery and promotion, while the
router received the unfiltered added/removed diffs. Because the node was absent
from allConns, the discovery reuse lookup (findConnectionByURL) never matched
it, so a new *Connection was created for it on every discovery cycle, and the
removed diff (computed against allConns) never contained it, so the prior
cycle's connection was never evicted. The round-robin fallback pool accumulated
these connections without bound, and its checkDead health checks repopulated a
per-connection poolRegistry sync.Map each cycle, growing the heap without limit.
The rate scaled with discovery frequency.

Make allConns the full connection inventory: it holds every discovered node
regardless of role, so discovery reuses and evicts connections symmetrically.
Keep dedicated cluster managers out of request routing at the point of
selection instead:

  - RoundRobinPolicy, the only policy that admits nodes irrespective of role,
    skips dedicated cluster managers in its DiscoveryUpdate add path unless
    IncludeDedicatedClusterManagers is set. The setting flows via policyConfig.
  - The allConns pool carries excludeDCM (set when IncludeDedicatedClusterManagers
    is false) so multiServerPool.Next() skips dedicated cluster managers during
    selection, including the no-router fallback path. A pool of only dedicated
    cluster managers exhausts its attempts and reports ErrNoConnections.

Discovery still bootstraps against a dedicated cluster manager seed via the
seed-fallback pool, which is unaffected.

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

Author: sean-

CHANGELOG.md

@@ -2,7 +2,11 @@
 
 Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
-## [4.7.2]
+## [4.7.3]
+
+### Fixed
+
+- Fix an unbounded connection/heap leak in node discovery when the cluster has a dedicated cluster manager (`cluster_manager` role with no work roles). The node was filtered out of the `allConns` inventory while the router received the unfiltered added/removed diffs, so `findConnectionByURL` never matched it: a new `*Connection` was created every discovery cycle and the stale one was never evicted, accumulating without bound in the round-robin fallback pool whose `checkDead` health checks repopulated a per-connection `poolRegistry` `sync.Map` each cycle (leak rate scaled with discovery frequency). `allConns` is now the full connection inventory so discovery reuses and evicts symmetrically, and dedicated cluster managers are excluded at request-routing selection instead: `RoundRobinPolicy` skips them in its `DiscoveryUpdate` add path and `multiServerPool.Next()` skips them during selection (including the no-router fallback), both gated on `IncludeDedicatedClusterManagers`. Discovery still bootstraps against a dedicated cluster manager seed via the seed-fallback pool ([#1003](https://github.com/opensearch-project/opensearch-go/pull/1003))
 
 ## [4.7.0]
 
@@ -656,6 +660,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Bumps `github.com/stretchr/testify` from 1.8.0 to 1.8.1
 - Bumps `github.com/aws/aws-sdk-go` from 1.44.45 to 1.44.132
 
+[4.7.3]: https://github.com/opensearch-project/opensearch-go/compare/v4.7.2...v4.7.3
 [4.7.2]: https://github.com/opensearch-project/opensearch-go/compare/v4.7.1...v4.7.2
 [4.7.0]: https://github.com/opensearch-project/opensearch-go/compare/v4.6.0...v4.7.0
 [4.6.0]: https://github.com/opensearch-project/opensearch-go/compare/v4.5.0...v4.6.0

internal/version/version.go

@@ -27,4 +27,4 @@
 package version
 
 // Client returns the client version as a string.
-const Client = "4.7.2"
+const Client = "4.7.3"

opensearchtransport/discovery.go

@@ -967,27 +967,16 @@ func (c *Client) createOrUpdateMultiNodePoolWithLock(readyConnections, deadConne
 		return c.promoteConnectionPoolWithLock(readyConnections, deadConnections)
 	}
 
-	// Update existing multiServerPool or create new one
-	// Apply client-level filtering for dedicated cluster managers
+	// allConns is the bookkeeping inventory of every discovered connection,
+	// including dedicated cluster managers. Discovery reuses connections by
+	// scanning allConns (findConnectionByURL) and evicts them via the removed
+	// diff computed against allConns. Query traffic is kept off dedicated
+	// cluster managers at the routing-policy layer (see RoundRobinPolicy).
 	allReadyConns := make([]*Connection, 0, len(readyConnections))
 	allDeadConns := make([]*Connection, 0, len(deadConnections))
 
-	for _, conn := range readyConnections {
-		if !c.includeDedicatedClusterManagers && conn.Roles.isDedicatedClusterManager() {
-			if dl := loadDebugLogger(); dl != nil {
-				dl.Logf("Excluding dedicated cluster manager %q from connection pool\n", conn.Name)
-			}
-			continue
-		}
-		allReadyConns = append(allReadyConns, conn)
-	}
-
-	for _, conn := range deadConnections {
-		if !c.includeDedicatedClusterManagers && conn.Roles.isDedicatedClusterManager() {
-			continue
-		}
-		allDeadConns = append(allDeadConns, conn)
-	}
+	allReadyConns = append(allReadyConns, readyConnections...)
+	allDeadConns = append(allDeadConns, deadConnections...)
 
 	// Shuffle connections for load distribution unless disabled
 	if !c.skipConnectionShuffle && len(allReadyConns) > 1 {

opensearchtransport/discovery_integration_internal_test.go

@@ -432,39 +432,43 @@ func TestIncludeDedicatedClusterManagersConfiguration(t *testing.T) {
 		name                            string
 		includeDedicatedClusterManagers bool
 		nodes                           map[string][]string // nodeName -> roles
-		expectedIncluded                []string            // nodes that should be included
-		expectedExcluded                []string            // nodes that should be excluded
+		// expectedInInventory lists nodes that must appear in the allConns pool,
+		// which holds every discovered node regardless of role.
+		expectedInInventory []string
+		// expectedNotRoutable lists dedicated cluster managers that must not be
+		// selected for request routing.
+		expectedNotRoutable []string
 	}{
 		{
-			name:                            "IncludeDedicatedClusterManagers enabled - includes all nodes",
+			name:                            "IncludeDedicatedClusterManagers enabled - all nodes routable",
 			includeDedicatedClusterManagers: true,
 			nodes: map[string][]string{
 				"cm-only":   {RoleClusterManager},
 				"data-node": {RoleData},
 			},
-			expectedIncluded: []string{"cm-only", "data-node"},
-			expectedExcluded: []string{},
+			expectedInInventory: []string{"cm-only", "data-node"},
+			expectedNotRoutable: []string{},
 		},
 		{
-			name:                            "IncludeDedicatedClusterManagers disabled (default) - excludes dedicated CM nodes",
+			name:                            "IncludeDedicatedClusterManagers disabled (default) - dedicated CM in inventory but not routable",
 			includeDedicatedClusterManagers: false,
 			nodes: map[string][]string{
 				"cm-only":   {RoleClusterManager},
 				"data-node": {RoleData},
 				"dummy":     {RoleData}, // Add second node to avoid single connection pool
 			},
-			expectedIncluded: []string{"data-node", "dummy"},
-			expectedExcluded: []string{"cm-only"},
+			expectedInInventory: []string{"cm-only", "data-node", "dummy"},
+			expectedNotRoutable: []string{"cm-only"},
 		},
 		{
-			name:                            "Mixed roles with CM always included regardless of setting",
+			name:                            "Mixed roles with CM always routable regardless of setting",
 			includeDedicatedClusterManagers: false,
 			nodes: map[string][]string{
 				"cm-data": {RoleClusterManager, RoleData},
 				"dummy":   {RoleData}, // Add second node to avoid single connection pool
 			},
-			expectedIncluded: []string{"cm-data", "dummy"},
-			expectedExcluded: []string{},
+			expectedInInventory: []string{"cm-data", "dummy"},
+			expectedNotRoutable: []string{},
 		},
 	}
 
@@ -481,41 +485,60 @@ func TestIncludeDedicatedClusterManagersConfiguration(t *testing.T) {
 			})
 			require.NoError(t, err)
 
-			// Perform discovery
-			err = c.DiscoverNodes(t.Context())
-			require.NoError(t, err)
-
-			// Verify results
-			pool, ok := c.mu.connectionPool.(*multiServerPool)
-			require.Truef(t, ok, "Expected multiServerPool but got %T with URLs: %v",
-				c.mu.connectionPool, c.mu.connectionPool.URLs())
-
-			// Check included nodes (in either ready or dead lists,
-			// since newly discovered nodes start in dead state pending health checks)
-			actualNodes := make(map[string]struct{})
-			for _, conn := range pool.mu.ready {
-				actualNodes[conn.Name] = struct{}{}
-			}
-			for _, conn := range pool.mu.dead {
-				actualNodes[conn.Name] = struct{}{}
+			// Run discovery repeatedly. The inventory must converge to exactly one
+			// connection per node and stay there: an unbounded pool that re-created
+			// connections each cycle (the dedicated-cluster-manager leak) would grow
+			// with every iteration. Asserting the exact length on every cycle is the
+			// regression guard.
+			const cycles = 5
+			var pool *multiServerPool
+			for cycle := 1; cycle <= cycles; cycle++ {
+				require.NoErrorf(t, c.DiscoverNodes(t.Context()), "discovery cycle %d", cycle)
+
+				var ok bool
+				pool, ok = c.mu.connectionPool.(*multiServerPool)
+				require.Truef(t, ok, "Expected multiServerPool but got %T with URLs: %v",
+					c.mu.connectionPool, c.mu.connectionPool.URLs())
+
+				pool.mu.RLock()
+				readyLen := len(pool.mu.ready)
+				deadLen := len(pool.mu.dead)
+				membersLen := len(pool.mu.members)
+				inventory := make(map[string]struct{}, readyLen+deadLen)
+				for _, conn := range pool.mu.ready {
+					inventory[conn.Name] = struct{}{}
+				}
+				for _, conn := range pool.mu.dead {
+					inventory[conn.Name] = struct{}{}
+				}
+				pool.mu.RUnlock()
+
+				require.Equalf(t, len(tt.nodes), readyLen+deadLen,
+					"cycle %d: inventory connection count (ready=%d dead=%d)", cycle, readyLen, deadLen)
+				require.Equalf(t, readyLen+deadLen, membersLen,
+					"cycle %d: members map must match ready+dead", cycle)
+				require.Lenf(t, inventory, len(tt.nodes),
+					"cycle %d: one connection per node (no duplicates)", cycle)
+				for _, expectedNode := range tt.expectedInInventory {
+					_, present := inventory[expectedNode]
+					require.Truef(t, present,
+						"cycle %d: expected node %q in the connection inventory", cycle, expectedNode)
+				}
 			}
 
-			for _, expectedNode := range tt.expectedIncluded {
-				_, ok := actualNodes[expectedNode]
-				require.True(t, ok,
-					"Expected node %q to be included but it wasn't", expectedNode)
+			// Dedicated cluster managers stay in the inventory but must not be
+			// handed out for routing. With no router, routing uses the inventory
+			// pool's Next(), which skips them.
+			for _, dcm := range tt.expectedNotRoutable {
+				for i := 0; i < len(tt.nodes)*4; i++ {
+					conn, nextErr := pool.Next()
+					if nextErr != nil {
+						break
+					}
+					require.NotEqual(t, dcm, conn.Name,
+						"Dedicated cluster manager %q must not be selected for routing", dcm)
+				}
 			}
-
-			for _, excludedNode := range tt.expectedExcluded {
-				_, ok := actualNodes[excludedNode]
-				require.False(t, ok,
-					"Expected node %q to be excluded but it was included", excludedNode)
-			}
-
-			// Verify total count
-			expectedTotal := len(tt.expectedIncluded)
-			require.Len(t, actualNodes, expectedTotal,
-				"Expected %d nodes but got %d", expectedTotal, len(actualNodes))
 		})
 	}
 }