Skip to content
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions opensearch_benchmark_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
7 changes: 5 additions & 2 deletions opensearch_example_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand All @@ -82,6 +83,7 @@ func ExampleNewClient() {
}

client, _ := opensearchapi.NewClient(cfg)
defer func() { _ = client.Close() }()
log.Print(client.Client.Transport.(*opensearchtransport.Transport).URLs())
}

Expand All @@ -99,5 +101,6 @@ func ExampleNewClient_logger() {
},
}

opensearchapi.NewClient(cfg)
client, _ := opensearchapi.NewClient(cfg)
defer func() { _ = client.Close() }()
}
10 changes: 10 additions & 0 deletions opensearch_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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{
Expand Down Expand Up @@ -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()
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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()
Expand All @@ -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
Expand Down
4 changes: 4 additions & 0 deletions opensearchapi/api_router_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
})
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand All @@ -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)
}
1 change: 1 addition & 0 deletions opensearchapi/rest_status_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"`
Expand Down
17 changes: 17 additions & 0 deletions opensearchtransport/address_resolver_internal_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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)

Expand All @@ -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)
Expand Down Expand Up @@ -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()))
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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 {
Expand All @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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())
Expand Down
31 changes: 0 additions & 31 deletions opensearchtransport/classify_extra_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading
Loading