Add OpenSearch 3.3.0+ API compatibility and enhance test infrastructure

OpenSearch 3.3.0+ API Compatibility:
Add missing fields across multiple APIs for enhanced query failure
tracking and system monitoring introduced in OpenSearch 3.3.0+:

Nodes Stats API:
- Add NeuralSearch breaker for neural search circuit breaker support
- Add QueryFailed and StartreeQueryFailed for query failure tracking
- Add SystemGeneratedProcessors and SystemGeneratedFactories to search
  pipeline for system component tracking

Indices Stats API:
- Add QueryFailed and StartreeQueryFailed search failure tracking fields

Cat APIs (nodes, shards, indices):
- Add SearchQueryFailed and SearchStartreeQueryFailed fields
- Fix JSON field naming inconsistencies in cat indices API where OpenSearch
  uses mixed formats (pri.search.startree.query_* vs pri.search.startree_query_failed)

Cluster State API:
- Add IngestionStatus field to ClusterStateMetaDataIndex for tracking
  index ingestion status in cluster metadata

Security Config API:
- Add JwksURI field to JWT authentication configuration struct for
  JSON Web Key Set URI support

Field Naming Compatibility Fix:
OpenSearch 3.2.0 introduced inconsistent startree query field naming
(pri.startree.* instead of pri.search.startree.*) which was corrected
in 3.3.0+. This creates compatibility issues as applications need different
field names depending on server version.

Solution: Implement dual-field approach with automatic consolidation:
- Stable fields use corrected 3.3.0+ naming as primary
- V32 compatibility fields handle temporary 3.2.0 naming
- consolidateV320StatsFields() method provides transparent fallback logic
- Called automatically in catClient.Indices() after JSON unmarshaling

Applications can use PrimarySearchStartreeQueryCurrent,
PrimarySearchStartreeQueryTime, and PrimarySearchStartreeQueryTotal
consistently regardless of OpenSearch server version.

Test Infrastructure Enhancements:
- Add comprehensive test infrastructure with automatic cluster readiness validation
- Implement NewClient(t) with three-phase validation (health + cluster state + nodes info)
- Add version-aware client configuration supporting both secure/insecure modes
- Create testify-based OpenSearchTestSuite with built-in version detection
- Implement robust readiness polling (25 attempts × 5s intervals) to eliminate CI flakiness
- Add comprehensive test coverage for cat indices field consolidation compatibility
- Update all integration tests to use enhanced test infrastructure

These changes provide enhanced observability for query failures, neural search
circuit breaker monitoring, search pipeline system components, index ingestion
status tracking, JWT authentication configuration, field naming compatibility,
and significantly improved test reliability across all OpenSearch versions.

All compatibility code is marked for removal when 3.2.0 support ends.

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

Author: sean-

.github/workflows/test-compatibility.yml

@@ -10,7 +10,7 @@ jobs:
       matrix:
         secured: ["true", "false"]
         entry:
-          - { opensearch_version: 1.3.18 }
+          - { opensearch_version: 1.3.20 }
           - { opensearch_version: 2.0.1 }
           - { opensearch_version: 2.1.0 }
           - { opensearch_version: 2.2.1 }

CHANGELOG.md

@@ -5,6 +5,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 ## [Unreleased]
 
 ### Added
+- Enhanced cluster readiness checking for improved test reliability: `ostest.NewClient()` now includes readiness validation (health + cluster state + nodes info)
 
 ### Changed
 - Refactor Client struct to use embedded mutex pattern for improved thread safety ([#775](https://github.com/opensearch-project/opensearch-go/pull/775))
@@ -17,13 +18,13 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ### Fixed
 - Fix flaky connection integration test by replacing arbitrary sleep times with proper server readiness polling
-- Fix nodes API compatibility with OpenSearch 3.1.0+ by adding phase_results_processors field
-- Fix cluster pending tasks API compatibility with OpenSearch 3.1.0+ by adding time_in_execution fields
-- Fix nodes stats API compatibility with OpenSearch 3.2.0+ by adding max_last_index_request_timestamp and startree query fields
-- Fix indices stats API compatibility with OpenSearch 3.2.0+ by adding max_last_index_request_timestamp and startree query fields
-- Fix cat APIs compatibility with OpenSearch 3.2.0+ by adding indexing timestamp and startree query fields
+- Fix OpenSearch 3.1.0+ API compatibility by adding phase_results_processors field to nodes API and time_in_execution fields to cluster pending tasks API
+- Fix OpenSearch 3.2.0+ API compatibility by adding max_last_index_request_timestamp and startree query fields across nodes stats, indices stats, and cat APIs, plus settings field to security plugin health API
+- Fix OpenSearch 3.3.0+ API compatibility by adding neural_search breaker, query_failed and startree_query_failed search fields, search pipeline system_generated fields across multiple APIs, plus ingestion_status field to cluster state API and jwks_uri field to security config API
+- Fix OpenSearch 3.4.0+ API compatibility by adding warmer fields to merges section, parallelism field to thread pool, and status_counter field across multiple APIs
+- Fix cat indices API field naming compatibility across OpenSearch versions by using forward-compatible field names (PrimarySearchStartreeQuery*) that match the corrected 3.3.0+ naming, with fallback support for the temporary 3.2.0 field names
+- Fix cat APIs data type compatibility by changing byte fields from int to string to properly handle values like "0b"
 - Fix floating point precision loss in nodes stats concurrent_avg_slice_count field by changing from float32 to float64
-- Fix security plugin health API compatibility with OpenSearch 3.2.0+ by adding settings field
 
 ### Security
 

internal/test/config.go

@@ -0,0 +1,72 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// The OpenSearch Contributors require contributions made to
+// this file be licensed under the Apache-2.0 license or a
+// compatible open source license.
+
+package ostest
+
+import (
+	"crypto/tls"
+	"net/http"
+	"os"
+
+	"github.com/opensearch-project/opensearch-go/v4"
+	"github.com/opensearch-project/opensearch-go/v4/opensearchapi"
+)
+
+// IsSecure returns true when SECURE_INTEGRATION env is set to true
+func IsSecure() bool {
+	return os.Getenv("SECURE_INTEGRATION") == "true"
+}
+
+// ClientConfig returns an opensearchapi.Config for both secure and insecure opensearch
+func ClientConfig() (*opensearchapi.Config, error) {
+	if !IsSecure() {
+		// For insecure integration tests, explicitly use HTTP
+		return &opensearchapi.Config{
+			Client: opensearch.Config{
+				Addresses: []string{"http://localhost:9200"},
+			},
+		}, nil
+	}
+
+	password, err := GetPassword()
+	if err != nil {
+		return nil, err
+	}
+
+	return &opensearchapi.Config{
+		Client: opensearch.Config{
+			Username:  "admin",
+			Password:  password,
+			Addresses: []string{"https://localhost:9200"},
+			Transport: &http.Transport{
+				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+			},
+		},
+	}, nil
+}
+
+// GetPassword returns the password suited for the opensearch version
+func GetPassword() (string, error) {
+	var (
+		major, minor int64
+		err          error
+	)
+	password := "admin"
+	version := os.Getenv("OPENSEARCH_VERSION")
+
+	if version != "latest" && version != "" {
+		major, minor, _, err = opensearch.ParseVersion(version)
+		if err != nil {
+			return "", err
+		}
+		if version == "latest" || major > 2 || (major == 2 && minor >= 12) {
+			password = "myStrongPassword123!"
+		}
+	} else {
+		password = "myStrongPassword123!"
+	}
+	return password, nil
+}

internal/test/doc.go

@@ -0,0 +1,9 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// The OpenSearch Contributors require contributions made to
+// this file be licensed under the Apache-2.0 license or a
+// compatible open source license.
+
+// Package ostest provides OpenSearch testing utilities and infrastructure.
+// Integration-specific functionality is available only when building with integration tags.
+package ostest

internal/test/helper.go

@@ -3,18 +3,18 @@
 // The OpenSearch Contributors require contributions made to
 // this file be licensed under the Apache-2.0 license or a
 // compatible open source license.
+//go:build integration
 
 package ostest
 
 import (
 	"context"
-	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"io"
-	"net/http"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -25,71 +25,102 @@ import (
 )
 
 // NewClient returns an opensearchapi.Client that is adjusted for the wanted test case
-func NewClient() (*opensearchapi.Client, error) {
+// and ensures the OpenSearch cluster is ready for requests.
+func NewClient(t *testing.T) (*opensearchapi.Client, error) {
+	t.Helper()
 	config, err := ClientConfig()
 	if err != nil {
 		return nil, err
 	}
-	if config != nil {
-		return opensearchapi.NewClient(*config)
+
+	client, err := opensearchapi.NewClient(*config)
+	if err != nil {
+		return nil, err
 	}
-	return opensearchapi.NewDefaultClient()
-}
 
-// IsSecure returns true when SECURE_INTEGRATION env is set to true
-func IsSecure() bool {
-	return os.Getenv("SECURE_INTEGRATION") == "true"
+	// Always wait for cluster readiness
+	err = waitForClusterReady(t, client)
+	if err != nil {
+		return nil, err
+	}
+
+	return client, nil
 }
 
-// ClientConfig returns an opensearchapi.Config for secure opensearch
-func ClientConfig() (*opensearchapi.Config, error) {
-	if IsSecure() {
-		password, err := GetPassword()
-		if err != nil {
-			return nil, err
-		}
+// waitForClusterReady waits for the OpenSearch cluster to be fully ready for API calls.
+func waitForClusterReady(t *testing.T, client *opensearchapi.Client) error {
+	t.Helper()
+	const (
+		maxAttempts          = 25
+		delayBetweenAttempts = 5 * time.Second
+		requestTimeout       = 2 * time.Second
+	)
 
-		return &opensearchapi.Config{
-			Client: opensearch.Config{
-				Username:  "admin",
-				Password:  password,
-				Addresses: []string{"https://localhost:9200"},
-				Transport: &http.Transport{
-					TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-				},
-			},
-		}, nil
+	// Get version for informational logging
+	major, minor, patch, err := GetVersion(client, t)
+	if err != nil {
+		return fmt.Errorf("failed to get OpenSearch version: %w", err)
 	}
-	//nolint:nilnil // easier to test with nil rather then doing complex error handling for tests
-	return nil, nil
-}
 
-// GetPassword returns the password suited for the opensearch version
-func GetPassword() (string, error) {
-	var (
-		major, minor int64
-		err          error
-	)
-	password := "admin"
-	version := os.Getenv("OPENSEARCH_VERSION")
+	ctx, cancel := context.WithTimeout(t.Context(), requestTimeout)
+	defer cancel()
 
-	if version != "latest" && version != "" {
-		major, minor, _, err = opensearch.ParseVersion(version)
-		if err != nil {
-			return "", err
+	for attempt := range maxAttempts {
+		// Basic cluster health check
+		resp, err := client.Cluster.Health(ctx, nil)
+		if err != nil || resp == nil {
+			t.Logf("Waiting %s for cluster readiness (attempt %d/%d)...", delayBetweenAttempts, attempt+1, maxAttempts)
+			time.Sleep(delayBetweenAttempts)
+
+			// Reset context for next attempt
+			ctx, cancel = context.WithTimeout(t.Context(), requestTimeout)
+			defer cancel()
+			continue
 		}
-		if version == "latest" || major > 2 || (major == 2 && minor >= 12) {
-			password = "myStrongPassword123!"
+
+		// Extended readiness validation
+		if err := extendedReadinessCheck(ctx, client); err == nil {
+			if attempt > 0 {
+				t.Logf("Cluster ready after %d attempts (version %d.%d.%d)", attempt+1, major, minor, patch)
+			}
+			return nil
 		}
-	} else {
-		password = "myStrongPassword123!"
+
+		t.Logf("Cluster health OK but readiness validation failed (attempt %d/%d)", attempt+1, maxAttempts)
+		time.Sleep(delayBetweenAttempts)
+
+		// Reset context for next attempt
+		ctx, cancel = context.WithTimeout(t.Context(), requestTimeout)
+		defer cancel()
 	}
-	return password, nil
+
+	return fmt.Errorf("cluster not ready after %d attempts (version %d.%d.%d)", maxAttempts, major, minor, patch)
+}
+
+// extendedReadinessCheck performs validation checks to ensure the
+// cluster is ready
+func extendedReadinessCheck(ctx context.Context, client *opensearchapi.Client) error {
+	// Try a simple cluster state request - this exercises more Java serialization paths
+	_, err := client.Cluster.State(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("cluster state check failed: %w", err)
+	}
+
+	// Try a simple nodes info request - exercises node-level serialization
+	_, err = client.Nodes.Info(ctx, nil)
+	if err != nil {
+		return fmt.Errorf("nodes info check failed: %w", err)
+	}
+
+	return nil
 }
 
 // GetVersion gets cluster info and returns version as int's
-func GetVersion(client *opensearchapi.Client) (int64, int64, int64, error) {
-	resp, err := client.Info(context.Background(), nil)
+func GetVersion(client *opensearchapi.Client, t *testing.T) (int64, int64, int64, error) {
+	if client == nil {
+		return 0, 0, 0, fmt.Errorf("client cannot be nil")
+	}
+	resp, err := client.Info(t.Context(), nil)
 	if err != nil {
 		return 0, 0, 0, err
 	}
@@ -99,22 +130,22 @@ func GetVersion(client *opensearchapi.Client) (int64, int64, int64, error) {
 // SkipIfBelowVersion skips a test if the cluster version is below a given version
 func SkipIfBelowVersion(t *testing.T, client *opensearchapi.Client, majorVersion, patchVersion int64, testName string) {
 	t.Helper()
-	major, patch, _, err := GetVersion(client)
+	major, patch, _, err := GetVersion(client, t)
 	assert.Nil(t, err)
 	if major < majorVersion || (major == majorVersion && patch < patchVersion) {
-		t.Skipf("Skiping %s as version %d.%d.x does not support this endpoint", testName, major, patch)
+		t.Skipf("Skipping %s as version %d.%d.x does not support this endpoint", testName, major, patch)
 	}
 }
 
-// SkipIfNotSecure skips a test runs against an unsecure cluster
+// SkipIfNotSecure skips a test that runs against an insecure cluster
 func SkipIfNotSecure(t *testing.T) {
 	t.Helper()
 	if !IsSecure() {
-		t.Skipf("Skiping %s as it needs a secured cluster", t.Name())
+		t.Skipf("Skipping %s as it needs a secured cluster", t.Name())
 	}
 }
 
-// CompareRawJSONwithParsedJSON is a helper function to determin the difference between the parsed JSON and the raw JSON
+// CompareRawJSONwithParsedJSON is a helper function to determine the difference between the parsed JSON and the raw JSON
 // this is helpful to detect missing fields in the go structs
 func CompareRawJSONwithParsedJSON(t *testing.T, resp any, rawResp *opensearch.Response) {
 	t.Helper()