Skip to content

Commit 49cc871

Browse files
sean-sean-
committed
test(benchmarks): fix goroutine leak and stateful body in transport benches
Three correctness/quality fixes on the transport-adjacent benchmarks surfaced while running locally with PGO collection. No production code changes. opensearch_benchmark_test.go (BenchmarkClient/Create client with defaults): Each iteration constructed a fresh opensearch.NewClient, which spawns the transport's per-client cluster-health and node-stats goroutines. Nothing closed them, so goroutines leaked linearly with b.N and at high iteration counts the Go runtime starved the bench loop itself. Close the underlying *opensearchtransport.Client in each iteration so the background goroutines exit. opensearchtransport/opensearchtransport_benchmark_test.go (BenchmarkTransport): - The pre-existing FakeTransport stored a single *http.Response with a strings.Reader body and returned the same pointer from every RoundTrip. strings.Reader is stateful: after the first Perform drained it the next iteration saw EOF, so the bench was measuring the EOF-handling path rather than steady-state Perform. Build a fresh response (with a fresh body) per RoundTrip. - Hoist opensearchtransport.New out of the per-iteration loop. Real callers build one transport per process; constructing one per iteration both inflates the measurement and (on this branch) leaks health-check goroutines. Disable the load-shedding poller via NodeStatsInterval = -1 so its tick rate doesn't bleed into the measurement. opensearchtransport/logger_benchmark_test.go (BenchmarkTransportLogger): Same construction-per-iteration anti-pattern across all four Text/Text-Body/JSON/JSON-Body sub-benches. Collapse the four copy-pasted bodies into a single closure, hoist New out of the loop, add b.Cleanup to close the transport, and disable the load-shedding poller. The Text-Body case had a separate bug: it called res.Body.Close() before io.ReadAll(res.Body), so the read always returned 0 bytes against a closed body and the len < 13 branch was silently flagged. Read first, then close. Drive-by: switch error format verbs from %s to %q in the touched Fatalf/Errorf sites for safer rendering of error chains. Signed-off-by: Sean Chittenden <sean.chittenden@crowdstrike.com>
1 parent 9c9060d commit 49cc871

4 files changed

Lines changed: 113 additions & 107 deletions

File tree

‎opensearch_benchmark_test.go‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ import (
3939

4040
"github.com/opensearch-project/opensearch-go/v4"
4141
"github.com/opensearch-project/opensearch-go/v4/opensearchapi"
42+
"github.com/opensearch-project/opensearch-go/v4/opensearchtransport"
4243
)
4344

4445
type FakeTransport struct {
@@ -72,6 +73,12 @@ func newFakeTransport(_ *testing.B, resp http.Response) *FakeTransport {
7273
}
7374
}
7475

76+
// BenchmarkClient measures opensearch.NewClient overhead. Each iteration
77+
// constructs a new client (and therefore spawns the transport's background
78+
// goroutines), then closes it via the transport so the goroutines exit. We
79+
// don't really care about the per-NewClient cost in steady-state usage --
80+
// real callers build one client per process -- but this measurement is kept
81+
// so changes to the construction path don't go unnoticed.
7582
func BenchmarkClient(b *testing.B) {
7683
defaultResponse := http.Response{
7784
Status: fmt.Sprintf("%d %s", http.StatusOK, http.StatusText(http.StatusOK)),
@@ -85,10 +92,15 @@ func BenchmarkClient(b *testing.B) {
8592

8693
b.Run("Create client with defaults", func(b *testing.B) {
8794
for i := 0; i < b.N; i++ {
88-
_, err := opensearch.NewClient(opensearch.Config{Transport: newFakeTransport(b, defaultResponse)})
95+
c, err := opensearch.NewClient(opensearch.Config{Transport: newFakeTransport(b, defaultResponse)})
8996
if err != nil {
90-
b.Fatalf("Unexpected error when creating a client: %s", err)
97+
b.Fatalf("Unexpected error when creating a client: %q", err)
9198
}
99+
// Close the underlying transport so the per-client background
100+
// goroutines (cluster-health/node-stats tickers) exit. Without
101+
// this the bench leaks goroutines linearly with b.N and the Go
102+
// runtime eventually starves the bench loop itself.
103+
_ = c.Transport.(*opensearchtransport.Client).Close()
92104
}
93105
})
94106
}

‎opensearchtransport/logger_benchmark_test.go‎

Lines changed: 53 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@
2929
package opensearchtransport_test
3030

3131
import (
32-
"bytes"
3332
"io"
3433
"net/http"
3534
"net/url"
@@ -38,97 +37,77 @@ import (
3837
"github.com/opensearch-project/opensearch-go/v4/opensearchtransport"
3938
)
4039

40+
// BenchmarkTransportLogger measures the per-Perform cost when a logger is
41+
// attached. The transport client is constructed once per sub-benchmark and
42+
// closed via b.Cleanup so we don't pay New() and don't leak background
43+
// goroutines across iterations.
4144
func BenchmarkTransportLogger(b *testing.B) {
4245
b.ReportAllocs()
4346

44-
b.Run("Text", func(b *testing.B) {
45-
for i := 0; i < b.N; i++ {
46-
tp, _ := opensearchtransport.New(opensearchtransport.Config{
47-
URLs: []*url.URL{{Scheme: "http", Host: "foo"}},
48-
Transport: newFakeTransport(b),
49-
Logger: &opensearchtransport.TextLogger{Output: io.Discard},
50-
})
47+
run := func(b *testing.B, cfg opensearchtransport.Config, readBody bool) {
48+
b.Helper()
49+
cfg.URLs = []*url.URL{{Scheme: "http", Host: "foo"}}
50+
cfg.Transport = newFakeTransport(b)
51+
// Disable the load-shedding poller; it isn't part of the steady-state
52+
// per-request hot path and its tick rate would otherwise bleed into
53+
// the measurement at this benchmark's iteration count.
54+
cfg.NodeStatsInterval = -1
5155

52-
req, _ := http.NewRequest(http.MethodGet, "/abc", nil)
53-
resp, err := tp.Perform(req)
54-
if err != nil {
55-
b.Fatalf("Unexpected error: %s", err)
56-
}
57-
if resp.Body != nil {
58-
resp.Body.Close()
59-
}
56+
tp, err := opensearchtransport.New(cfg)
57+
if err != nil {
58+
b.Fatalf("Unexpected error: %q", err)
6059
}
61-
})
60+
b.Cleanup(func() { _ = tp.Close() })
6261

63-
b.Run("Text-Body", func(b *testing.B) {
62+
b.ResetTimer()
6463
for i := 0; i < b.N; i++ {
65-
tp, _ := opensearchtransport.New(opensearchtransport.Config{
66-
URLs: []*url.URL{{Scheme: "http", Host: "foo"}},
67-
Transport: newFakeTransport(b),
68-
Logger: &opensearchtransport.TextLogger{
69-
Output: io.Discard,
70-
EnableRequestBody: true,
71-
EnableResponseBody: true,
72-
},
73-
})
74-
7564
req, _ := http.NewRequest(http.MethodGet, "/abc", nil)
7665
res, err := tp.Perform(req)
7766
if err != nil {
78-
b.Fatalf("Unexpected error: %s", err)
67+
b.Fatalf("Unexpected error: %q", err)
7968
}
80-
res.Body.Close()
81-
82-
body, err := io.ReadAll(res.Body)
83-
if err != nil {
84-
b.Fatalf("Error reading response body: %s", err)
85-
}
86-
res.Body = io.NopCloser(bytes.NewBuffer(body))
87-
if len(body) < 13 {
88-
b.Errorf("Error reading response body bytes, want=13, got=%d", len(body))
69+
if readBody {
70+
body, err := io.ReadAll(res.Body)
71+
if err != nil {
72+
b.Fatalf("Error reading response body: %q", err)
73+
}
74+
if len(body) < 13 {
75+
b.Errorf("Error reading response body bytes, want=13, got=%d", len(body))
76+
}
8977
}
78+
res.Body.Close()
9079
}
80+
}
81+
82+
b.Run("Text", func(b *testing.B) {
83+
run(b, opensearchtransport.Config{
84+
Logger: &opensearchtransport.TextLogger{Output: io.Discard},
85+
}, false)
9186
})
9287

93-
b.Run("JSON", func(b *testing.B) {
94-
for i := 0; i < b.N; i++ {
95-
tp, _ := opensearchtransport.New(opensearchtransport.Config{
96-
URLs: []*url.URL{{Scheme: "http", Host: "foo"}},
97-
Transport: newFakeTransport(b),
98-
Logger: &opensearchtransport.JSONLogger{Output: io.Discard},
99-
})
88+
b.Run("Text-Body", func(b *testing.B) {
89+
run(b, opensearchtransport.Config{
90+
Logger: &opensearchtransport.TextLogger{
91+
Output: io.Discard,
92+
EnableRequestBody: true,
93+
EnableResponseBody: true,
94+
},
95+
}, true)
96+
})
10097

101-
req, _ := http.NewRequest(http.MethodGet, "/abc", nil)
102-
resp, err := tp.Perform(req)
103-
if err != nil {
104-
b.Fatalf("Unexpected error: %s", err)
105-
}
106-
if resp != nil && resp.Body != nil {
107-
resp.Body.Close()
108-
}
109-
}
98+
b.Run("JSON", func(b *testing.B) {
99+
run(b, opensearchtransport.Config{
100+
Logger: &opensearchtransport.JSONLogger{Output: io.Discard},
101+
}, false)
110102
})
111103

112104
b.Run("JSON-Body", func(b *testing.B) {
113-
for i := 0; i < b.N; i++ {
114-
tp, _ := opensearchtransport.New(opensearchtransport.Config{
115-
URLs: []*url.URL{{Scheme: "http", Host: "foo"}},
116-
Transport: newFakeTransport(b),
117-
Logger: &opensearchtransport.JSONLogger{
118-
Output: io.Discard,
119-
EnableRequestBody: true,
120-
EnableResponseBody: true,
121-
},
122-
})
123-
124-
req, _ := http.NewRequest(http.MethodGet, "/abc", nil)
125-
resp, err := tp.Perform(req)
126-
if err != nil {
127-
b.Fatalf("Unexpected error: %s", err)
128-
}
129-
if resp != nil && resp.Body != nil {
130-
resp.Body.Close()
131-
}
132-
}
105+
run(b, opensearchtransport.Config{
106+
Logger: &opensearchtransport.JSONLogger{
107+
Output: io.Discard,
108+
EnableRequestBody: true,
109+
EnableResponseBody: true,
110+
},
111+
}, false)
133112
})
134113
}

‎opensearchtransport/opensearchtransport_benchmark_test.go‎

Lines changed: 45 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -39,62 +39,77 @@ import (
3939
"github.com/opensearch-project/opensearch-go/v4/opensearchtransport"
4040
)
4141

42-
type FakeTransport struct {
43-
FakeResponse *http.Response
44-
}
42+
// FakeTransport is a test http.RoundTripper that returns a fresh response per
43+
// call. We must not share the Body across iterations: a strings.Reader is
44+
// stateful, and after one Perform drains it the next iteration sees EOF.
45+
type FakeTransport struct{}
4546

4647
func (t *FakeTransport) RoundTrip(_ *http.Request) (*http.Response, error) {
47-
return t.FakeResponse, nil
48+
return &http.Response{
49+
Status: fmt.Sprintf("%d %s", http.StatusOK, http.StatusText(http.StatusOK)),
50+
StatusCode: http.StatusOK,
51+
ContentLength: 13,
52+
Header: http.Header{"Content-Type": []string{"application/json"}},
53+
Body: io.NopCloser(strings.NewReader(`{"foo":"bar"}`)),
54+
}, nil
4855
}
4956

50-
func newFakeTransport(_ *testing.B) *FakeTransport {
51-
return &FakeTransport{
52-
FakeResponse: &http.Response{
53-
Status: fmt.Sprintf("%d %s", http.StatusOK, http.StatusText(http.StatusOK)),
54-
StatusCode: http.StatusOK,
55-
ContentLength: 13,
56-
Header: http.Header(map[string][]string{"Content-Type": {"application/json"}}),
57-
Body: io.NopCloser(strings.NewReader(`{"foo":"bar"}`)),
58-
},
59-
}
60-
}
57+
func newFakeTransport(_ *testing.B) *FakeTransport { return &FakeTransport{} }
6158

59+
// BenchmarkTransport measures the per-Perform cost on a steady-state client.
60+
// opensearchtransport.New is hoisted out of the loop because it's a one-time
61+
// cost in real usage and on this branch it spawns long-lived health-check
62+
// goroutines that would otherwise leak across iterations.
6263
func BenchmarkTransport(b *testing.B) {
6364
b.ReportAllocs()
6465

6566
b.Run("Defaults", func(b *testing.B) {
66-
for i := 0; i < b.N; i++ {
67-
tp, _ := opensearchtransport.New(opensearchtransport.Config{
68-
URLs: []*url.URL{{Scheme: "http", Host: "foo"}},
69-
Transport: newFakeTransport(b),
70-
})
67+
tp, err := opensearchtransport.New(opensearchtransport.Config{
68+
URLs: []*url.URL{{Scheme: "http", Host: "foo"}},
69+
Transport: newFakeTransport(b),
70+
// Disable the load-shedding poller; it touches the fake transport
71+
// every few seconds and would skew the steady-state measurement.
72+
NodeStatsInterval: -1,
73+
})
74+
if err != nil {
75+
b.Fatalf("Unexpected error: %q", err)
76+
}
77+
b.Cleanup(func() { _ = tp.Close() })
7178

79+
b.ResetTimer()
80+
for i := 0; i < b.N; i++ {
7281
req, _ := http.NewRequest(http.MethodGet, "/abc", nil)
7382
res, err := tp.Perform(req)
7483
if err != nil {
75-
b.Fatalf("Unexpected error: %s", err)
84+
b.Fatalf("Unexpected error: %q", err)
7685
}
77-
defer res.Body.Close()
86+
res.Body.Close()
7887
}
7988
})
8089

8190
b.Run("Headers", func(b *testing.B) {
8291
hdr := http.Header{}
8392
hdr.Set("Accept", "application/yaml")
8493

85-
for i := 0; i < b.N; i++ {
86-
tp, _ := opensearchtransport.New(opensearchtransport.Config{
87-
URLs: []*url.URL{{Scheme: "http", Host: "foo"}},
88-
Header: hdr,
89-
Transport: newFakeTransport(b),
90-
})
94+
tp, err := opensearchtransport.New(opensearchtransport.Config{
95+
URLs: []*url.URL{{Scheme: "http", Host: "foo"}},
96+
Header: hdr,
97+
Transport: newFakeTransport(b),
98+
NodeStatsInterval: -1,
99+
})
100+
if err != nil {
101+
b.Fatalf("Unexpected error: %q", err)
102+
}
103+
b.Cleanup(func() { _ = tp.Close() })
91104

105+
b.ResetTimer()
106+
for i := 0; i < b.N; i++ {
92107
req, _ := http.NewRequest(http.MethodGet, "/abc", nil)
93108
res, err := tp.Perform(req)
94109
if err != nil {
95-
b.Fatalf("Unexpected error: %s", err)
110+
b.Fatalf("Unexpected error: %q", err)
96111
}
97-
defer res.Body.Close()
112+
res.Body.Close()
98113
}
99114
})
100115
}

‎opensearchutil/bulk_indexer_internal_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -539,7 +539,7 @@ func TestBulkIndexerContext(t *testing.T) {
539539
require.Equal(t, nilReturns, stats.NumAdded, "NumAdded must equal the number of Add() calls that returned nil")
540540
require.Equal(t, errReturns, stats.BulkAddFailCount, "BulkAddFailCount must equal the number of Add() calls that returned ctx.Err()")
541541
require.Equal(t, uint64(numAttempts), stats.NumAdded+stats.BulkAddFailCount, "every Add() must be accounted for exactly once")
542-
require.Greater(t, errReturns, uint64(0), "at least one Add() should fail when context is already cancelled")
542+
require.Positive(t, errReturns, "at least one Add() should fail when context is already cancelled")
543543
},
544544
},
545545
{

0 commit comments

Comments
 (0)