Replace WaitForAllNodesReady inline polling with readiness.Wait

The previous implementation polled cat-nodes for 60s with a single
boolean predicate, producing "Condition never satisfied" on timeout.
Switch to readiness.Wait(t, ctx, TargetClusterReady, WithCluster(c))
which observes per-node progression through LayerHTTP, LayerClusterJoin,
and LayerStatsReady, and dumps a structured per-node diagnostic with
the full last cat-nodes response on failure.

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

Author: sean-

CHANGELOG.md

@@ -177,6 +177,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ### Fixed
 
+- Replace `WaitForAllNodesReady` inline `require.Eventually` loop with a layered readiness FSM (`internal/test/readiness`) that observes per-node progression through `LayerTCP -> LayerHTTP -> LayerClusterJoin -> LayerStatsReady`, records transitions including regressions, and emits a structured per-node diagnostic with the full last cat-nodes response on timeout. Per-layer budgets are tuned for CI pessimism (cold JVM startup is the long pole); total budget for `TargetClusterReady` is 6.5 minutes. ([#650](https://github.com/opensearch-project/opensearch-go/issues/650))
 - Fix bulk indexer HTML-escaping `_id` and `routing` values containing `<`, `>`, or `&` characters, causing OpenSearch to store escaped values (e.g., `\u003croot_account\u003e` stored instead of `<root_account>`), leading to duplicate documents, unreachable data on read-by-ID paths, and potential shard routing mismatches. Present since the `json.Marshal` migration in 2021 (commit `3da59092`). Replace `json.Marshal` with `json.NewEncoder` + `SetEscapeHTML(false)` in `opensearchutil.worker.writeMeta` and `opensearchutil.JSONReader`; replace per-worker `aux []byte` with `sync.Pool`-backed `*bytes.Buffer`; add table-driven test coverage for `writeMeta` edge cases and refactor remaining `TestBulkIndexer` subtests to table-driven `require`-based style ([#824](https://github.com/opensearch-project/opensearch-go/pull/824))
 - Fix pool replacement orphaning resurrection goroutines during node discovery, causing connections to become permanently dead with no active health checker ([#786](https://github.com/opensearch-project/opensearch-go/pull/786))
 - Fix multi-to-single pool demotion leaking resurrection goroutines by giving each `multiServerPool` its own derived context and cancelling it on demotion ([#830](https://github.com/opensearch-project/opensearch-go/pull/830))

opensearchtransport/testutil/helpers.go

@@ -21,6 +21,8 @@ import (
 	"time"
 
 	"golang.org/x/mod/semver"
+
+	"github.com/opensearch-project/opensearch-go/v4/internal/test/readiness"
 )
 
 // Common timeouts for testing
@@ -337,73 +339,23 @@ func RequireMinConns(
 func WaitForCluster(t *testing.T) {
 	t.Helper()
 
-	const (
-		maxAttempts          = 25
-		delayBetweenAttempts = 5 * time.Second
-		requestTimeout       = 2 * time.Second
-	)
-
 	u := GetTestURL(t)
 	healthURL := *u
 	healthURL.Path = "/"
 
-	client := &http.Client{Transport: GetTestTransport(t)}
-
-	var eofCount int
-	for attempt := range maxAttempts {
-		ctx, cancel := context.WithTimeout(t.Context(), requestTimeout)
-
-		req, err := http.NewRequestWithContext(ctx, http.MethodGet, healthURL.String(), nil)
-		if err != nil {
-			cancel()
-			t.Fatalf("WaitForCluster: failed to create request: %v", err)
-		}
-
-		if IsSecure(t) {
-			req.SetBasicAuth("admin", GetPassword(t))
-		}
-
-		resp, err := client.Do(req)
-		cancel()
+	httpClient := &http.Client{Transport: GetTestTransport(t)}
 
-		if err != nil {
-			if strings.Contains(err.Error(), "EOF") {
-				eofCount++
-				if eofCount >= 3 {
-					t.Fatalf("WaitForCluster: cluster returned EOF on %d consecutive attempts "+
-						"(SECURE_INTEGRATION=%s); verify the cluster scheme matches this setting: %v",
-						eofCount, os.Getenv("SECURE_INTEGRATION"), err)
-				}
-			} else {
-				eofCount = 0
-			}
-			t.Logf("WaitForCluster: attempt %d/%d: %v", attempt+1, maxAttempts, err)
-			time.Sleep(delayBetweenAttempts)
-			continue
-		}
-
-		eofCount = 0
-		resp.Body.Close()
-
-		if resp.StatusCode == http.StatusOK {
-			if attempt > 0 {
-				t.Logf("WaitForCluster: cluster ready after %d attempts", attempt+1)
-			}
-			return
-		}
-
-		// Fail fast on authentication errors -- retrying with the same credentials won't help
-		if resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden {
-			t.Fatalf("WaitForCluster: cluster returned %d (SECURE_INTEGRATION=%s, OPENSEARCH_VERSION=%s); "+
-				"verify credentials are correct -- the admin password changed in OpenSearch 2.12.0+",
-				resp.StatusCode, os.Getenv("SECURE_INTEGRATION"), os.Getenv("OPENSEARCH_VERSION"))
+	var prepareReq func(*http.Request)
+	if IsSecure(t) {
+		password := GetPassword(t)
+		prepareReq = func(req *http.Request) {
+			req.SetBasicAuth("admin", password)
 		}
-
-		t.Logf("WaitForCluster: attempt %d/%d: status %d", attempt+1, maxAttempts, resp.StatusCode)
-		time.Sleep(delayBetweenAttempts)
 	}
 
-	t.Fatalf("WaitForCluster: cluster not ready after %d attempts (url=%s)", maxAttempts, healthURL.String())
+	readiness.Wait(t, t.Context(), readiness.LayerHTTP,
+		readiness.WithExpectedNodes(1),
+		readiness.WithRawHTTP(&healthURL, httpClient, prepareReq))
 }
 
 // ignoredFieldRule pairs a compiled regexp with a version expression.

osapi/testutil/helpers.go

@@ -25,6 +25,7 @@ import (
 	"golang.org/x/sync/semaphore"
 
 	"github.com/opensearch-project/opensearch-go/v4"
+	"github.com/opensearch-project/opensearch-go/v4/internal/test/readiness"
 	tptestutil "github.com/opensearch-project/opensearch-go/v4/opensearchtransport/testutil"
 	"github.com/opensearch-project/opensearch-go/v4/osapi"
 )
@@ -230,8 +231,6 @@ func InitClient(t *testing.T) (*osapi.Client, error) {
 // GET / using the layered readiness FSM in internal/test/readiness. The
 // readinessSem caps concurrent setups across tests so a stampede of
 // goroutines doesn't overload a small CI cluster.
-func WaitForClusterReady(t *testing.T, client *osapi.Client) {
-// goroutines doesn't overload a small CI cluster.
 func WaitForClusterReady(t *testing.T, client *osapi.Client) {
 	t.Helper()
 	if err := readinessSem.Acquire(t.Context(), 1); err != nil {
@@ -242,25 +241,20 @@ func WaitForClusterReady(t *testing.T, client *osapi.Client) {
 	readiness.Wait(t, t.Context(), readiness.LayerHTTP, readiness.WithCluster(client))
 }
 
-// WaitForAllNodesReady polls /_cat/nodes until every node reports non-nil cpu
-// and heap.percent metrics. This prevents flakes from nodes that haven't fully
-// initialized in CI (e.g. stats not yet collected after a fresh cluster start).
+// WaitForAllNodesReady blocks until every node in the test cluster has
+// reached LayerStatsReady — i.e. cluster-health reports the expected
+// node count AND _cat/nodes returns non-nil cpu+heap.percent for each
+// node. It uses the layered readiness FSM in internal/test/readiness so
+// that timeouts produce a structured per-node diagnostic instead of a
+// "Condition never satisfied" stub.
+//
+// Expected node count comes from OPENSEARCH_NODE_COUNT (defaults to 1).
+// Per-layer budgets are tuned for CI pessimism (cold JVM startup is the
+// long pole); see readiness.DefaultBudgets for the exact values.
 func WaitForAllNodesReady(t *testing.T, client *osapi.Client) {
 	t.Helper()
-	require.Eventually(t, func() bool {
-		resp, err := client.Cat.Nodes(t.Context(), &osapi.CatNodesReq{
-			Params: &osapi.CatNodesParams{DebugParams: osapi.DebugParams{Format: "json"}},
-		})
-		if err != nil || resp == nil || len(resp.Records) == 0 {
-			return false
-		}
-		for _, node := range resp.Records {
-			if node.Cpu == nil || node.HeapPercent == nil {
-				return false
-			}
-		}
-		return true
-	}, 60*time.Second, 1*time.Second, "not all nodes reporting stats (cpu/heap.percent nil)")
+	readiness.Wait(t, t.Context(), readiness.TargetClusterReady,
+		readiness.WithCluster(client))
 }
 
 // CompareRawJSONwithParsedJSON is a helper function to determine the difference between the parsed JSON and the raw JSON.