merge commit

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

Author: sean-

CHANGELOG.md

@@ -5,6 +5,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 ## [Unreleased]
 
 ### Added
+- Add connection pool health probes with cluster-aware resurrection timing ([#786](https://github.com/opensearch-project/opensearch-go/pull/786))
 
 ### Changed
 - Refactor Client struct to use embedded mutex pattern for improved thread safety ([#775](https://github.com/opensearch-project/opensearch-go/pull/775))

opensearchtransport/connection.go

@@ -336,7 +336,7 @@ func (cp *statusConnectionPool) resurrectWithLock(c *Connection, removeDead bool
 				debugLogger.Logf("Health check failed for %q: %s; will retry later\n", c.URL, err)
 			}
 			// Health check failed - schedule another resurrection attempt
-			cp.scheduleResurrect(c)
+			cp.scheduleResurrect(c, c.mu.deadSince)
 			return
 		}
 		if debugLogger != nil {
@@ -398,6 +398,7 @@ func (cp *statusConnectionPool) scheduleResurrect(c *Connection, deadSince time.
 		clusterTimeout := time.Duration(float64(baseTimeout) * clusterFactor)
 
 		// Add random jitter (0 to clusterTimeout range)
+		// #nosec G404 - Non-cryptographic randomness is acceptable for connection timing jitter
 		jitter := time.Duration(rand.Float64() * float64(clusterTimeout))
 		finalTimeout = jitter
 
@@ -436,34 +437,6 @@ func (cp *statusConnectionPool) scheduleResurrect(c *Connection, deadSince time.
 			return
 		}
 
-		cp.resurrectWithLock(c, true)
-	})
-			c.URL,
-			failures,
-			factor,
-			liveNodes,
-			len(cp.dead),
-			totalNodes,
-			baseTimeout,
-			finalTimeout,
-			c.DeadSince.Add(finalTimeout).Sub(time.Now().UTC()).Truncate(time.Millisecond),
-		)
-	}
-
-	time.AfterFunc(finalTimeout, func() {
-		cp.Lock()
-		defer cp.Unlock()
-
-		c.mu.Lock()
-		defer c.mu.Unlock()
-
-		if !c.mu.isDead {
-			if debugLogger != nil {
-				debugLogger.Logf("Already resurrected %q\n", c.URL)
-			}
-			return
-		}
-
 		cp.resurrectWithLock(c, true)
 	})
 }

opensearchtransport/connection_integration_test.go

@@ -49,7 +49,28 @@ func TestStatusConnectionPool(t *testing.T) {
 		serverHosts []string
 		numServers  = 3
 
-		defaultHandler = func(w http.ResponseWriter, r *http.Request) { fmt.Fprintf(w, "OK") }
+		defaultHandler = func(w http.ResponseWriter, r *http.Request) {
+			// Return proper OpenSearch root endpoint JSON response
+			w.Header().Set("Content-Type", "application/json")
+			response := `{
+  "name": "test-node",
+  "cluster_name": "test-cluster",
+  "cluster_uuid": "test-cluster-uuid",
+  "version": {
+    "number": "3.4.0",
+    "build_type": "tar",
+    "build_hash": "test-hash",
+    "build_date": "2024-01-01T00:00:00.000Z",
+    "build_snapshot": false,
+    "lucene_version": "9.11.0",
+    "minimum_wire_compatibility_version": "7.10.0",
+    "minimum_index_compatibility_version": "7.0.0",
+    "distribution": "opensearch"
+  },
+  "tagline": "The OpenSearch Project: https://opensearch.org/"
+}`
+			fmt.Fprint(w, response)
+		}
 	)
 
 	for i := 1; i <= numServers; i++ {

opensearchtransport/connection_internal_test.go

@@ -29,53 +29,12 @@
 package opensearchtransport
 
 import (
-	"bytes"
-	"io"
-	"net/http"
 	"net/url"
 	"regexp"
 	"testing"
 	"time"
 )
 
-// mockTransport provides a mock HTTP transport that responds to health checks with valid OpenSearch responses
-type mockTransport struct{}
-
-func (t *mockTransport) RoundTrip(req *http.Request) (*http.Response, error) {
-	// Return a valid OpenSearch GET / response for health checks
-	if req.Method == "GET" && req.URL.Path == "/" {
-		body := `{
-			"name": "test-node",
-			"cluster_name": "test-cluster",
-			"cluster_uuid": "test-uuid",
-			"version": {
-				"number": "2.0.0",
-				"distribution": "opensearch",
-				"build_type": "tar",
-				"build_hash": "test-hash",
-				"build_date": "2024-01-01T00:00:00Z",
-				"build_snapshot": false,
-				"lucene_version": "9.0.0",
-				"minimum_wire_compatibility_version": "7.10.0",
-				"minimum_index_compatibility_version": "7.0.0"
-			},
-			"tagline": "The OpenSearch Project: https://opensearch.org/"
-		}`
-		return &http.Response{
-			StatusCode: 200,
-			Body:       io.NopCloser(bytes.NewBufferString(body)),
-			Header:     make(http.Header),
-		}, nil
-	}
-
-	// For other requests, return a basic response
-	return &http.Response{
-		StatusCode: 200,
-		Body:       io.NopCloser(bytes.NewBufferString("{}")),
-		Header:     make(http.Header),
-	}, nil
-}
-
 func TestSingleConnectionPoolNext(t *testing.T) {
 	t.Run("Single URL", func(t *testing.T) {
 		pool := &singleConnectionPool{
@@ -429,6 +388,10 @@ func TestStatusConnectionPoolResurrect(t *testing.T) {
 		// Channel to signal when resurrection is complete
 		done := make(chan struct{})
 
+		// Create round-robin selector
+		s := &roundRobinSelector{}
+		s.curr.Store(-1)
+
 		pool := &statusConnectionPool{
 			selector:                     s,
 			resurrectTimeoutInitial:      0,
@@ -479,7 +442,6 @@ func TestStatusConnectionPoolResurrect(t *testing.T) {
 		if len(pool.mu.dead) != 0 {
 			t.Errorf("Expected no dead connections, got: %d", len(pool.mu.dead))
 		}
-		}
 	})
 }
 

opensearchtransport/discovery.go

@@ -140,7 +140,7 @@ func (c *Client) DiscoverNodes() error {
 	}
 
 	// Set up health check function for pools that support it
-	if pool, ok := c.pool.(*statusConnectionPool); ok {
+	if pool, ok := c.mu.pool.(*statusConnectionPool); ok {
 		pool.healthCheck = c.isHealthyOpenSearchNode
 	}
 
@@ -168,7 +168,8 @@ func (c *Client) getNodesInfo() ([]nodeInfo, error) {
 			}
 
 			// Use round-robin selector to pick a startup URL
-			selector := &roundRobinSelector{curr: -1}
+			selector := &roundRobinSelector{}
+			selector.curr.Store(-1)
 			conn, err = selector.Select(startupConns)
 			if err != nil {
 				return nil, fmt.Errorf("failed to select startup URL: %w", err)

opensearchtransport/opensearchtransport.go

@@ -56,8 +56,8 @@ const (
 )
 
 var (
-	reGoVersion             = regexp.MustCompile(`go(\d+\.\d+\..+)`)
-	errHealthCheckFailed    = errors.New("connection health check error")
+	reGoVersion          = regexp.MustCompile(`go(\d+\.\d+\..+)`)
+	errHealthCheckFailed = errors.New("connection health check error")
 )
 
 // Interface defines the interface for HTTP client.
@@ -177,9 +177,6 @@ func New(cfg Config) (*Client, error) {
 	conns := make([]*Connection, len(cfg.URLs))
 	for idx, u := range cfg.URLs {
 		conn := &Connection{URL: u}
-		// Mark initial connections as dead to trigger health validation via resurrection workflow
-		// Discovery will use fallback to startup URLs when pool has no live connections
-		conn.markAsDead()
 		conns[idx] = conn
 	}
 
@@ -226,7 +223,7 @@ func New(cfg Config) (*Client, error) {
 	}
 
 	// Set up health check function for pools that support it
-	if pool, ok := client.pool.(*statusConnectionPool); ok {
+	if pool, ok := client.mu.pool.(*statusConnectionPool); ok {
 		pool.healthCheck = client.isHealthyOpenSearchNode
 	}
 
@@ -563,12 +560,12 @@ type OpenSearchInfo struct {
 	Version     struct {
 		// Permanent fields - guaranteed since OpenSearch 1.3.0
 		Number                           string `json:"number"`                              // Version number, e.g. "1.3.0"
-		BuildType                        string `json:"build_type"`                         // Build type: "tar", "docker", etc.
-		BuildHash                        string `json:"build_hash"`                         // Git commit hash
-		BuildDate                        string `json:"build_date"`                         // Build timestamp
-		BuildSnapshot                    bool   `json:"build_snapshot"`                     // Is snapshot build
-		LuceneVersion                    string `json:"lucene_version"`                     // Underlying Lucene version
-		MinimumWireCompatibilityVersion  string `json:"minimum_wire_compatibility_version"` // Minimum wire protocol version
+		BuildType                        string `json:"build_type"`                          // Build type: "tar", "docker", etc.
+		BuildHash                        string `json:"build_hash"`                          // Git commit hash
+		BuildDate                        string `json:"build_date"`                          // Build timestamp
+		BuildSnapshot                    bool   `json:"build_snapshot"`                      // Is snapshot build
+		LuceneVersion                    string `json:"lucene_version"`                      // Underlying Lucene version
+		MinimumWireCompatibilityVersion  string `json:"minimum_wire_compatibility_version"`  // Minimum wire protocol version
 		MinimumIndexCompatibilityVersion string `json:"minimum_index_compatibility_version"` // Minimum index compatibility version
 
 		// Conditional fields - may be missing in specific configurations
@@ -601,13 +598,18 @@ func (c *Client) isHealthyOpenSearchNode(url *url.URL) (string, error) {
 	if res == nil {
 		return "", fmt.Errorf("%w: nil response", errHealthCheckFailed)
 	}
-	defer res.Body.Close()
+	if res.Body != nil {
+		defer res.Body.Close()
+	}
 
 	if res.StatusCode != http.StatusOK {
 		return "", fmt.Errorf("%w: status %d", errHealthCheckFailed, res.StatusCode)
 	}
 
 	// Read and parse the response
+	if res.Body == nil {
+		return "", fmt.Errorf("%w: nil response body", errHealthCheckFailed)
+	}
 	body, err := io.ReadAll(res.Body)
 	if err != nil {
 		return "", fmt.Errorf("%w: %w", errHealthCheckFailed, err)