test: release the transports and clients the tests construct (#1048)

* test: release the transports and clients the tests construct

opensearchtransport.New starts two ticker goroutines on every call: the node-stats poller (NodeStatsInterval: 0 means auto-derive, not disabled) and the cluster-health refresh loop (healthCheckRate is derived from the server core count and is never zero). Close is the only thing that stops either one, so a test that constructs a transport or client and never releases it leaves two live tickers running for the remaining life of the test binary, where they perturb process-wide measurements.

Every construction site now registers a release: t.Cleanup in tests, b.Cleanup in benchmarks, defer in examples (Example functions have no *testing.T).

TestClassify_ZeroAlloc and TestNewRequestEventZeroAlloc use testing.AllocsPerRun, a process-wide allocation differential. Neither source file carried a build tag, so both compiled into every configuration including -tags integration,core, sharing a binary with the live-cluster tests whose transports poll node stats and cluster health in the background. That is what flaked TestClassify_ZeroAlloc in CI.

New TestCloseReapsBackgroundPollers reads the goroutine dump to assert that New starts both pollers and that Close reaps them, so the leak cannot return silently. Verified to fail when Close is stubbed to skip its context cancellation. It carries the same !integration constraint, because the process-wide dump only settles in a binary where no other live transport is polling, and it omits t.Parallel for the same reason.


Signed-off-by: Ryan Yuan <ryan.yuan@crowdstrike.com>
Signed-off-by: Sean Chittenden <sean.chittenden@crowdstrike.com>
Co-authored-by: Ryan Yuan <ryan.yuan@crowdstrike.com>

Author: sean-

CHANGELOG.md

@@ -255,6 +255,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ### Fixed
 
+- Fix tests, benchmarks, and examples that construct a transport or client and never release it, each leaking two ticker goroutines that outlive the test. `opensearchtransport.New` always starts the node-stats poller and the cluster-health refresh loop: `healthCheckRate` is derived from the server core count and is never zero, and `NodeStatsInterval: 0` means auto-derive rather than disabled. `Close` is the only thing that stops either one, so a leaked poller keeps ticking for the remaining life of the test binary, where it perturbs process-wide measurements. Every site now registers a release -- `t.Cleanup` in tests, `b.Cleanup` in benchmarks, `defer` in examples -- covering 153 sites in `opensearchtransport` and 39 more across `opensearch_integration_test.go`, `opensearch_benchmark_test.go`, `opensearch_example_test.go`, `opensearchapi`, `opensearchutil`, `osprom`, and `osotel`. Three kinds of site are deliberately left alone: the process-wide shared client from `opensearchapi/testutil.NewClient`, which the package owns rather than the caller; a `New` whose construction is expected to fail and returns nothing to close; and one bulk indexer case that forces `implicitClient` to true so the indexer owns the client and `BulkIndexer.Close` releases it. A client the caller supplies is not owned, as `TestBulkIndexerOwnClientFlag` asserts, so the cases that hand the indexer a client close it themselves. The two zero-allocation assertions (`TestClassify_ZeroAlloc` and `TestNewRequestEventZeroAlloc`) move into `//go:build !integration` files: `testing.AllocsPerRun` is a process-wide allocation differential and is only sound in a binary where nothing else allocates concurrently, and sharing a binary with the live-cluster tests is what flaked `TestClassify_ZeroAlloc` in CI. New `TestCloseReapsBackgroundPollers` reads the goroutine dump to assert both pollers start with `New` and are gone after `Close`, so the leak cannot return silently; it carries the same `!integration` constraint, because the dump only settles in a binary where no other live transport is polling
 - Fix `cmd/osgen` dropping every version annotation the spec writes beside a `$ref`. kin-openapi splits a `$ref`'s siblings across two places: standard fields such as `description` are overlaid onto the resolved schema, but `x-*` keys stay on the reference and never reach the resolved schema's extensions. The generator read only the latter, so 141 annotations were lost -- 135 `x-version-added`, 5 `x-version-removed`, and 1 `x-version-deprecated`. The visible half was missing documentation: `SearchResp.PhaseTook` carries `x-version-added: '2.12'` and emitted no availability note, and no generated file mentioned that version at all. The other half is a correctness problem, since the same values feed the version filter, so those fields were tested against an empty version and could not be excluded by `-min-version` or `-max-version`. A sibling annotation now wins over one on the referenced schema, because it describes the property carrying it rather than the shared type it points at: two properties may reference one schema and have been added in different versions. Regenerating adds 138 availability notes and changes no field
 - Fix collapsed types keeping their mangled generic-instantiation name instead of the readable alias the spec provides for them. When an `allOf` adds nothing to its base the two describe one Go type and the base's name was kept, so `AsAdjacencyMatrix()` returned `CommonAggregationsMultiBucketAggregateBaseAdjacencyMatrixBucket` even though the spec supplies `AdjacencyMatrixAggregate` as a bare `allOf: [$ref]` alias precisely to name that instantiation. A post-walk pass now renames the collapsed type to its alias and rewrites every reference, including types keyed beneath it (a nested `buckets` union is registered as `<parentKey>.buckets`, so it inherited the old prefix). The rename must run after the walk rather than during it: type references are plain Go type strings, and the spec chains these collapses (`RangeAggregate` -> `RangeAggregateBase` -> `MultiBucketAggregateBaseRangeBucket`), so a mid-walk rename leaves siblings that already resolved pointing at a name that no longer exists. Two guards keep it safe: a target several aliases share keeps its own name, since no one alias is the better choice (eight schemas from `AvgAggregate` to `WeightedAvgAggregate` collapse onto `SingleMetricAggregateBase`), and a target the spec references more heavily than its alias also keeps its name, so `SearchResult` is not retired in favor of `SearchResponse`. Restores `CommonAggregationsAdjacencyMatrixAggregate`, `CommonAggregationsDateHistogramAggregate`, `CommonAggregationsGeoHashGridAggregate` and their siblings, and drops type names over 60 characters from 74 to 16 -- the remainder being genuinely descriptive nested paths rather than erasure artifacts
 - Fix `cmd/osgen` deciding union branch reachability in the wrong pipeline phase, and stop emitting wrapper structs for schemas that merely rename another. Branch deduplication ran during the Parse phase, dropping any branch whose Go type duplicated an earlier one. Whether a duplicate is dead depends on the union's decode state, which is not assigned until the IR phase: a wire-decoded union walks its branches and stops at the first that decodes, so a same-type duplicate is unreachable, but a caller-keyed lazy union retains only raw bytes and lets the caller name the branch, so every `As<Branch>()` accessor is reachable even when several decode one Go type. Deduplication moves to `dropUnreachableBranches`, which runs once every union has reached its terminal state and skips the lazy ones. `SearchResultAggregationsValue` gains back the accessors the Parse-phase drop had been silently deleting (55 -> 62), including `AsSum`, `AsMin`, `AsMax`, `AsValueCount`, `AsWeightedAvg`, `AsSimpleValue`, and `AsMedianAbsoluteDeviation` alongside `AsAvg`. With reachability now judged correctly, `collapsesToBase` also accepts a bare `allOf: [$ref]` -- the spec's way of giving a generic instantiation a friendly name -- which removes 30 further wrapper structs whose only content was the embedded base (66 such wrappers at the start of this line of work, 2 remain). Breaking: the removed wrappers are no longer distinct types, so `CommonAggregationsAvgAggregate` and its siblings are now `CommonAggregationsSingleMetricAggregateBase`, and `New...FromAvg` and friends take that type; accessor and constructor names are unchanged

opensearch_benchmark_test.go

@@ -127,6 +127,7 @@ func BenchmarkClientAPI(b *testing.B) {
 	if err != nil {
 		b.Fatalf("ERROR: %s", err)
 	}
+	b.Cleanup(func() { _ = client.Close() })
 
 	b.Run("InfoRequest{}.Do()", func(b *testing.B) {
 		b.ResetTimer()

opensearch_example_test.go

@@ -52,10 +52,11 @@ func ExampleNewDefaultClient() {
 	if err != nil {
 		log.Fatalf("Error creating the client: %s\n", err)
 	}
+	defer func() { _ = client.Close() }()
 
 	_, err = client.Info(ctx, nil)
 	if err != nil {
-		log.Fatalf("Error getting the response: %s\n", err)
+		log.Panicf("Error getting the response: %s\n", err)
 	}
 
 	log.Print(client.Client.Transport.(*opensearchtransport.Transport).URLs())
@@ -82,6 +83,7 @@ func ExampleNewClient() {
 	}
 
 	client, _ := opensearchapi.NewClient(cfg)
+	defer func() { _ = client.Close() }()
 	log.Print(client.Client.Transport.(*opensearchtransport.Transport).URLs())
 }
 
@@ -99,5 +101,6 @@ func ExampleNewClient_logger() {
 		},
 	}
 
-	opensearchapi.NewClient(cfg)
+	client, _ := opensearchapi.NewClient(cfg)
+	defer func() { _ = client.Close() }()
 }

opensearch_integration_test.go

@@ -137,6 +137,7 @@ func TestClientTransport(t *testing.T) {
 
 		client, err := opensearchapi.NewClient(cfg)
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = client.Close() })
 
 		_, err = client.Info(t.Context(), nil)
 		require.Error(t, err)
@@ -187,6 +188,7 @@ func TestClientCustomTransport(t *testing.T) {
 			}
 			client, err = opensearchapi.NewClient(*cfg)
 			require.NoError(t, err)
+			t.Cleanup(func() { _ = client.Close() })
 
 			// Wait for cluster to be ready before running tests
 			testutil.WaitForClusterReady(t, client)
@@ -220,6 +222,7 @@ func TestClientCustomTransport(t *testing.T) {
 			Password:  config.Client.Password,
 			Context:   t.Context(),
 		})
+		t.Cleanup(func() { _ = tp.Close() })
 
 		client := opensearchapi.Client{
 			Client: &opensearch.Client{
@@ -349,6 +352,7 @@ func TestClientGetConfigIntegration(t *testing.T) {
 		// Create a client with specific configuration
 		osClient, err := opensearch.NewClient(cfg.Client)
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = osClient.Close() })
 
 		// Retrieve the config
 		retrievedConfig := osClient.GetConfig()
@@ -373,6 +377,7 @@ func TestClientGetConfigIntegration(t *testing.T) {
 		// Verify we can create a new client with the retrieved config
 		newClient, err := opensearch.NewClient(*config)
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = newClient.Close() })
 		require.NotNil(t, newClient)
 
 		// Verify the new client works by making a request
@@ -394,6 +399,7 @@ func TestNewFromClientIntegration(t *testing.T) {
 		// Create an opensearchapi.Client from the shared config
 		apiClient, err := opensearchapi.NewClient(opensearchapi.Config{Client: cfg.Client})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = apiClient.Close() })
 		require.NotNil(t, apiClient)
 
 		// Verify the api client can make requests
@@ -410,9 +416,11 @@ func TestNewFromClientIntegration(t *testing.T) {
 		// Create a base opensearch.Client and an api client from the same config
 		osClient, err := opensearch.NewClient(cfg.Client)
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = osClient.Close() })
 
 		apiClient, err := opensearchapi.NewClient(opensearchapi.Config{Client: cfg.Client})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = apiClient.Close() })
 		require.NotNil(t, apiClient.Client.Transport)
 
 		// Verify both clients can make requests successfully
@@ -436,6 +444,7 @@ func TestNewFromClientIntegration(t *testing.T) {
 		// Create an opensearchapi.Client from the shared config
 		apiClient, err := opensearchapi.NewClient(opensearchapi.Config{Client: cfg.Client})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = apiClient.Close() })
 
 		// Retrieve config through the api client's wrapped opensearch client
 		retrievedConfig := apiClient.Client.GetConfig()
@@ -453,6 +462,7 @@ func TestNewFromClientIntegration(t *testing.T) {
 		// Create an opensearchapi.Client from the shared config
 		apiClient, err := opensearchapi.NewClient(opensearchapi.Config{Client: cfg.Client})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = apiClient.Close() })
 
 		// Test a few sub-clients to ensure they're properly initialized
 		// Cat client

opensearchapi/api_router_test.go

@@ -61,6 +61,7 @@ func TestNewClient_RouterInjection(t *testing.T) {
 			t.Parallel()
 			c, err := opensearchapi.NewClient(tt.cfg)
 			require.NoError(t, err)
+			t.Cleanup(func() { _ = c.Close() })
 			require.NotNil(t, c)
 			require.NotNil(t, c.Client)
 		})
@@ -110,6 +111,7 @@ func TestNewClient_RouterEnvOptOut(t *testing.T) {
 			c, err := opensearchapi.NewClient(cfg)
 			if tt.wantErrNil {
 				require.NoError(t, err)
+				t.Cleanup(func() { _ = c.Close() })
 				require.NotNil(t, c)
 			} else {
 				require.Error(t, err)
@@ -195,6 +197,7 @@ func TestNewClient_RouterTruthyEnablesDiscovery(t *testing.T) {
 			c, err := opensearchapi.NewClient(cfg)
 			if tt.wantClientBuilds {
 				require.NoError(t, err)
+				t.Cleanup(func() { _ = c.Close() })
 				require.NotNil(t, c)
 			} else {
 				require.Error(t, err)
@@ -210,5 +213,6 @@ func TestNewDefaultClient(t *testing.T) {
 	t.Parallel()
 	c, err := opensearchapi.NewDefaultClient()
 	require.NoError(t, err)
+	t.Cleanup(func() { _ = c.Close() })
 	require.NotNil(t, c)
 }

opensearchapi/rest_status_test.go

@@ -174,6 +174,7 @@ func TestRestStatusClientDecode(t *testing.T) {
 
 			client, err := opensearch.NewClient(opensearch.Config{Addresses: []string{ts.URL}})
 			require.NoError(t, err)
+			t.Cleanup(func() { _ = client.Close() })
 
 			var body struct {
 				Status *opensearchapi.RestStatus `json:"status"`

opensearchtransport/address_resolver_internal_test.go

@@ -213,6 +213,7 @@ func TestAddressResolver(t *testing.T) {
 				MaxAddressResolvers: 1, // serial for deterministic behavior
 			})
 			require.NoError(t, err)
+			t.Cleanup(func() { _ = tp.Close() })
 
 			nodes, err := tp.getNodesInfo(t.Context())
 			if tt.wantErr {
@@ -271,6 +272,7 @@ func TestAddressResolver(t *testing.T) {
 			HealthCheck: NoOpHealthCheck,
 		})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = tp.Close() })
 
 		nodes, err := tp.getNodesInfo(t.Context())
 		require.NoError(t, err)
@@ -297,6 +299,7 @@ func TestAddressResolver(t *testing.T) {
 			MaxAddressResolvers: 1,
 		})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = tp.Close() })
 
 		_, err = tp.getNodesInfo(t.Context())
 		require.NoError(t, err)
@@ -347,6 +350,7 @@ func TestAddressResolver(t *testing.T) {
 			MaxAddressResolvers: -1,
 		})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = tp.Close() })
 
 		_, err = tp.getNodesInfo(t.Context())
 		require.NoError(t, err)
@@ -379,6 +383,7 @@ func TestAddressResolver(t *testing.T) {
 			MaxAddressResolvers: 1,
 		})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = tp.Close() })
 
 		_, err = tp.getNodesInfo(t.Context())
 		require.NoError(t, err)
@@ -438,6 +443,7 @@ func TestAddressResolver(t *testing.T) {
 				MaxAddressResolvers: tt.maxResolvers,
 			})
 			require.NoError(t, err)
+			t.Cleanup(func() { _ = tp.Close() })
 
 			nodes, err := tp.getNodesInfo(ctx)
 			if tt.wantErr != nil {
@@ -470,6 +476,7 @@ func TestAddressResolver(t *testing.T) {
 			MaxAddressResolvers: -1,
 		})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = tp.Close() })
 
 		nodes, err := tp.getNodesInfo(ctx)
 
@@ -492,6 +499,7 @@ func TestAddressResolver(t *testing.T) {
 			},
 		})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = tp.Close() })
 
 		err = tp.DiscoverNodes(t.Context())
 		require.NoError(t, err)
@@ -629,6 +637,7 @@ func TestDiscoverNodes_PartialCancelDoesNotEvict(t *testing.T) {
 
 			tp, err := New(tt.configure(t, cancel))
 			require.NoError(t, err)
+			t.Cleanup(func() { _ = tp.Close() })
 
 			// Seed the pool with a clean discovery cycle.
 			require.NoError(t, tp.DiscoverNodes(t.Context()))
@@ -704,6 +713,7 @@ func TestAddressResolverRunner(t *testing.T) {
 			},
 		})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = tp.Close() })
 
 		_, err = tp.getNodesInfo(t.Context())
 		require.NoError(t, err)
@@ -736,6 +746,7 @@ func TestAddressResolverRunner(t *testing.T) {
 			},
 		})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = tp.Close() })
 
 		nodes, err := tp.getNodesInfo(t.Context())
 		require.NoError(t, err)
@@ -771,6 +782,7 @@ func TestAddressResolverRunner(t *testing.T) {
 			},
 		})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = tp.Close() })
 
 		_, err = tp.getNodesInfo(t.Context())
 		require.NoError(t, err)
@@ -845,6 +857,7 @@ func TestAddressResolverRunner(t *testing.T) {
 				AddressResolverRunner: tt.runner,
 			})
 			require.NoError(t, err)
+			t.Cleanup(func() { _ = tp.Close() })
 
 			nodes, err := tp.getNodesInfo(ctx)
 			if tt.wantErr != nil {
@@ -869,6 +882,7 @@ func TestAddressResolverRunner(t *testing.T) {
 			},
 		})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = tp.Close() })
 
 		_, err = tp.getNodesInfo(t.Context())
 		require.ErrorIs(t, err, runnerErr)
@@ -966,6 +980,7 @@ func TestAddressResolverRunner(t *testing.T) {
 					},
 				})
 				require.NoError(t, err)
+				t.Cleanup(func() { _ = tp.Close() })
 
 				nodes, err := tp.getNodesInfo(t.Context())
 				require.NoError(t, err)
@@ -1005,6 +1020,7 @@ func TestAddressResolverRunner(t *testing.T) {
 			},
 		})
 		require.NoError(t, err)
+		t.Cleanup(func() { _ = tp.Close() })
 
 		err = tp.DiscoverNodes(t.Context())
 		require.NoError(t, err)
@@ -1147,6 +1163,7 @@ func TestAddressResolverRunnerProtocol(t *testing.T) {
 						AddressResolverRunner: p.runner,
 					})
 					require.NoError(t, err)
+					t.Cleanup(func() { _ = tp.Close() })
 					tp.observer.Store(&iface)
 
 					nodes, err := tp.getNodesInfo(t.Context())

opensearchtransport/classify_extra_test.go

@@ -16,37 +16,6 @@ import (
 	"github.com/opensearch-project/opensearch-go/v5/opensearchtransport"
 )
 
-// TestClassify_ZeroAlloc guards the zero-allocation claim documented in
-// CHANGELOG: OperationClassifier.Classify must not allocate on the hot
-// path (it lives inside RoundTrip and runs once per request). A
-// regression here means a per-request heap object that compounds across
-// the cluster's RPS.
-func TestClassify_ZeroAlloc(t *testing.T) {
-	c := opensearchtransport.NewOperationClassifier()
-	// Warm any one-time setup the classifier may do.
-	_ = c.Classify(http.MethodGet, "/events/_search")
-
-	tests := []struct {
-		name   string
-		method string
-		path   string
-	}{
-		{"search hot path", http.MethodPost, "/events/_search"},
-		{"bulk hot path", http.MethodPost, "/_bulk"},
-		{"doc get hot path", http.MethodGet, "/events/_doc/abc-123"},
-		{"unknown path falls through to OpOther", http.MethodGet, "/_unknown/endpoint"},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			allocs := testing.AllocsPerRun(200, func() {
-				_ = c.Classify(tt.method, tt.path)
-			})
-			require.Zero(t, allocs, "Classify(%q, %q) must be zero-alloc, got %g", tt.method, tt.path, allocs)
-		})
-	}
-}
-
 // TestClassify_PathEdgeCases covers path-shape variants that callers
 // pass through Classify directly: trailing slashes, query strings, mixed
 // case methods. The classifier must be tolerant of common HTTP-layer