feat(client): add Close() and close bulk indexer's implicit client (#932)

Backport #926 to v4 without the default-client cache.

Add Close() to opensearch.Client and opensearchapi.Client so callers can
release the transport's background goroutines and idle connections.
NewBulkIndexer now closes the client it creates when the caller supplies
none, on all Close paths including a cancelled context. Its flusher stops
via context cancellation instead of a done channel, so Close no longer
risks a deadlock when the flusher has already exited.

The process-wide client cache and its metrics-aggregation behavior are left
out, so this stays a non-breaking maintenance change.

Refs: #893
Fixes: #928

Signed-off-by: Ryan Yuan <ryan.yuan@crowdstrike.com>

Author: ryanyuan

CHANGELOG.md

@@ -6,6 +6,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ### Added
 
+- Add `Close()` to `opensearch.Client` and `opensearchapi.Client` to release background goroutines (node discovery, health/stats pollers, DNS refresh) and idle connections; `opensearchutil.NewBulkIndexer` now closes the client it implicitly creates. Backport of #926 without the default-client cache ([#928](https://github.com/opensearch-project/opensearch-go/issues/928), [#893](https://github.com/opensearch-project/opensearch-go/issues/893))
 - Add `cmd/osgen` code generator for typed path builders and API consumer files from the OpenAPI spec
 - v5preview/opensearchapi: `NewClient` and `NewDefaultClient` now inject `opensearchtransport.NewDefaultRouter` when `config.Client.Router` is nil, opting every v5preview client into intelligent request routing by default. The `OPENSEARCH_GO_ROUTER` env var preserves its v4 semantics end-to-end: `=true`/`=1` enables auto-discovery (via `DiscoverNodesOnStart`); `=false`/`=0` suppresses both Router injection and auto-discovery; unset injects the Router without auto-discovery. v4's `opensearchapi.NewClient` is unchanged. ([#816](https://github.com/opensearch-project/opensearch-go/issues/816))
 - Add `envvars.Falsy(name)` helper that distinguishes "explicitly opted out" from "unset" (Truthy collapses both into false). Used by v5preview's router injection rule.

opensearch.go

@@ -370,6 +370,17 @@ func getAddressFromEnvironment() []string {
 	return addrsFromEnvironment(envOpenSearchURL)
 }
 
+// Close releases the client's background resources by closing the underlying
+// transport if it implements io.Closer -- the built-in *opensearchtransport.Client
+// does, canceling pollers and closing idle connections -- and is a no-op for a
+// custom Interface that does not. Safe on a zero value and idempotent.
+func (c *Client) Close() error {
+	if closer, ok := c.Transport.(io.Closer); ok {
+		return closer.Close()
+	}
+	return nil
+}
+
 // ParseVersion returns an int64 representation of version.
 func ParseVersion(version string) (int64, int64, int64, error) {
 	reVersion := regexp.MustCompile(`^([0-9]+)\.([0-9]+)\.([0-9]+)`)

opensearch_internal_test.go

@@ -566,3 +566,40 @@ func TestClientGetConfig(t *testing.T) {
 		require.Equal(t, expectedConfig.EnableRetryOnTimeout, config.EnableRetryOnTimeout)
 	})
 }
+
+func TestClientClose(t *testing.T) {
+	t.Run("falls back to transport io.Closer", func(t *testing.T) {
+		tc := &stubTransportCloser{}
+		c := &Client{Transport: tc}
+		require.NoError(t, c.Close())
+		require.Equal(t, 1, tc.closed)
+	})
+
+	t.Run("no-op when transport lacks Close", func(t *testing.T) {
+		c := &Client{Transport: stubPerformOnly{}}
+		require.NoError(t, c.Close())
+	})
+
+	t.Run("idempotent", func(t *testing.T) {
+		tc := &stubTransportCloser{}
+		c := &Client{Transport: tc}
+		require.NoError(t, c.Close())
+		require.NoError(t, c.Close())
+		require.Equal(t, 2, tc.closed)
+	})
+}
+
+// stubTransportCloser implements opensearchtransport.Interface + io.Closer.
+type stubTransportCloser struct{ closed int }
+
+//nolint:nilnil // stub: Perform is never called, only Close is exercised
+func (s *stubTransportCloser) Perform(*http.Request) (*http.Response, error) { return nil, nil }
+
+//nolint:unparam // Close must return error to satisfy io.Closer; stub never fails
+func (s *stubTransportCloser) Close() error { s.closed++; return nil }
+
+// stubPerformOnly implements only opensearchtransport.Interface.
+type stubPerformOnly struct{}
+
+//nolint:nilnil // stub: Perform is never called, exists only to satisfy Interface
+func (stubPerformOnly) Perform(*http.Request) (*http.Response, error) { return nil, nil }

opensearchapi/opensearchapi.go

@@ -159,6 +159,15 @@ func NewDefaultClient() (*Client, error) {
 	return clientInit(rootClient, resolveErrorMask(Config{})), nil
 }
 
+// Close releases the client's background resources by closing the underlying
+// opensearch.Client. Safe on a zero value and idempotent.
+func (c *Client) Close() error {
+	if c.Client != nil {
+		return c.Client.Close()
+	}
+	return nil
+}
+
 // NewFromClient creates an opensearchapi client from an existing opensearch.Client.
 // In v4 this preserves the legacy "mask everything" default; use NewClient
 // with Config to enable partial-failure errors.

opensearchapi/opensearchapi_internal_test.go

@@ -0,0 +1,39 @@
+// 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 opensearchapi
+
+import (
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/opensearch-project/opensearch-go/v4"
+)
+
+func TestClientClose(t *testing.T) {
+	t.Run("delegates to embedded opensearch.Client", func(t *testing.T) {
+		tc := &stubTransportCloser{}
+		c := &Client{Client: &opensearch.Client{Transport: tc}}
+		require.NoError(t, c.Close())
+		require.Equal(t, 1, tc.closed)
+	})
+
+	t.Run("no-op when embedded client is nil", func(t *testing.T) {
+		c := &Client{}
+		require.NoError(t, c.Close())
+	})
+}
+
+// stubTransportCloser implements opensearchtransport.Interface + io.Closer.
+type stubTransportCloser struct{ closed int }
+
+//nolint:nilnil // stub: Perform is never called, only Close is exercised
+func (s *stubTransportCloser) Perform(*http.Request) (*http.Response, error) { return nil, nil }
+
+//nolint:unparam // Close must return error to satisfy io.Closer; stub never fails
+func (s *stubTransportCloser) Close() error { s.closed++; return nil }

opensearchutil/bulk_indexer.go

@@ -169,12 +169,25 @@ type bulkIndexer struct {
 	queue   chan BulkIndexerItem
 	workers []*worker
 	ticker  *time.Ticker
-	done    chan bool
-	stats   *bulkIndexerStats
+	// stopFlush cancels the flusher goroutine; flusherDone is closed when that
+	// goroutine returns. Close cancels via stopFlush (non-blocking and
+	// idempotent, so Close never deadlocks even when the flusher already
+	// returned via the construction context) and then waits on flusherDone, so
+	// the periodic flush has fully stopped before Close runs its final drain and
+	// the deferred implicit-client Close.
+	stopFlush   context.CancelFunc
+	flusherDone chan struct{}
+	stats       *bulkIndexerStats
 
 	metaPool         sync.Pool
 	metaPoolMaxBytes int
 
+	// implicitClient is true when NewBulkIndexer implicitly created the client
+	// (cfg.Client was nil). Close then closes it to release its background
+	// goroutines and connection pool; a caller-supplied client is left for its
+	// owner to close.
+	implicitClient bool
+
 	config BulkIndexerConfig
 }
 
@@ -192,12 +205,14 @@ type bulkIndexerStats struct {
 
 // NewBulkIndexer creates a new bulk indexer.
 func NewBulkIndexer(cfg BulkIndexerConfig) (BulkIndexer, error) {
+	implicitClient := false
 	if cfg.Client == nil {
 		var err error
 		cfg.Client, err = opensearchapi.NewDefaultClient()
 		if err != nil {
 			return nil, err
 		}
+		implicitClient = true
 	}
 
 	// Initialize context if not provided
@@ -223,9 +238,9 @@ func NewBulkIndexer(cfg BulkIndexerConfig) (BulkIndexer, error) {
 
 	bi := bulkIndexer{
 		config:           cfg,
-		done:             make(chan bool),
 		stats:            &bulkIndexerStats{},
 		metaPoolMaxBytes: cfg.MetaBufferPoolMaxBytes,
+		implicitClient:   implicitClient,
 		metaPool: sync.Pool{
 			New: func() any {
 				//nolint:mnd // 512B matches the original per-worker aux preallocation.
@@ -258,11 +273,27 @@ func (bi *bulkIndexer) Add(ctx context.Context, item BulkIndexerItem) error {
 }
 
 // Close stops the periodic flush, closes the indexer queue channel,
-// notifies the done channel and calls flush on all writers.
+// stops the flusher goroutine and calls flush on all writers.
 func (bi *bulkIndexer) Close(ctx context.Context) error {
 	bi.ticker.Stop()
 	close(bi.queue)
-	bi.done <- true
+	// Stop the periodic flusher and wait for it to return before the final
+	// drain below, so no auto-flush races the drain. stopFlush is non-blocking
+	// and idempotent; flusherDone is already closed if the flusher exited via
+	// the construction context, so this never blocks Close indefinitely.
+	bi.stopFlush()
+	<-bi.flusherDone
+
+	// Close the implicitly-created client on every exit path (including the
+	// ctx-cancelled early return below), or its background goroutines and
+	// connection pool would leak.
+	if bi.implicitClient {
+		defer func() {
+			if err := bi.config.Client.Close(); err != nil && bi.config.OnError != nil {
+				bi.config.OnError(ctx, err)
+			}
+		}()
+	}
 
 	select {
 	case <-ctx.Done():
@@ -324,13 +355,19 @@ func (bi *bulkIndexer) init(ctx context.Context) {
 
 	bi.ticker = time.NewTicker(bi.config.FlushInterval)
 
+	// The flusher stops on either the caller's construction context or Close's
+	// stopFlush cancel, whichever fires first. Deriving flushCtx from ctx folds
+	// both signals into one channel. Workers keep the original ctx so Close can
+	// still drive its final drain flush after stopping the periodic flusher.
+	flushCtx, stopFlush := context.WithCancel(ctx)
+	bi.stopFlush = stopFlush
+	bi.flusherDone = make(chan struct{})
+
 	go func() {
+		defer close(bi.flusherDone)
 		for {
 			select {
-			case <-bi.done:
-				return
-			case <-ctx.Done():
-				// Context cancelled, stop flusher
+			case <-flushCtx.Done():
 				return
 			case <-bi.ticker.C:
 				if bi.config.DebugLogger != nil {

opensearchutil/bulk_indexer_internal_test.go

@@ -872,3 +872,61 @@ func int64Pointer(i int64) *int64 {
 func intPointer(i int) *int {
 	return &i
 }
+
+func TestBulkIndexerImplicitClientClose(t *testing.T) {
+	t.Run("closes implicitly created client", func(t *testing.T) {
+		tc := &stubTransportCloser{}
+		bi := &bulkIndexer{
+			config:         BulkIndexerConfig{Client: &opensearchapi.Client{Client: &opensearch.Client{Transport: tc}}, FlushInterval: time.Hour},
+			implicitClient: true,
+			stats:          &bulkIndexerStats{},
+		}
+		bi.init(t.Context())
+		require.NoError(t, bi.Close(t.Context()))
+		require.Equal(t, 1, tc.closed)
+	})
+
+	t.Run("does not close caller-supplied client", func(t *testing.T) {
+		tc := &stubTransportCloser{}
+		bi := &bulkIndexer{
+			config:         BulkIndexerConfig{Client: &opensearchapi.Client{Client: &opensearch.Client{Transport: tc}}, FlushInterval: time.Hour},
+			implicitClient: false,
+			stats:          &bulkIndexerStats{},
+		}
+		bi.init(t.Context())
+		require.NoError(t, bi.Close(t.Context()))
+		require.Equal(t, 0, tc.closed)
+	})
+
+	t.Run("closes implicit client without deadlock under cancelled context", func(t *testing.T) {
+		tc := &stubTransportCloser{}
+		ctx, cancel := context.WithCancel(t.Context())
+		bi := &bulkIndexer{
+			config:         BulkIndexerConfig{Client: &opensearchapi.Client{Client: &opensearch.Client{Transport: tc}}, FlushInterval: time.Hour},
+			implicitClient: true,
+			stats:          &bulkIndexerStats{},
+		}
+		bi.init(ctx)
+		cancel()
+		done := make(chan error, 1)
+		go func() { done <- bi.Close(ctx) }()
+		select {
+		case <-done:
+			// Close returns ctx.Err() on cancelled ctx; the deferred implicit
+			// client Close still runs. Both outcomes acceptable; the point is
+			// no deadlock and the client is closed.
+		case <-time.After(5 * time.Second):
+			t.Fatal("Close deadlocked under cancelled context")
+		}
+		require.Equal(t, 1, tc.closed)
+	})
+}
+
+// stubTransportCloser implements opensearchtransport.Interface + io.Closer.
+type stubTransportCloser struct{ closed int }
+
+//nolint:nilnil // stub: Perform is never called, only Close is exercised
+func (s *stubTransportCloser) Perform(*http.Request) (*http.Response, error) { return nil, nil }
+
+//nolint:unparam // Close must return error to satisfy io.Closer; stub never fails
+func (s *stubTransportCloser) Close() error { s.closed++; return nil }