Modernize opensearchtransport with Go 1.23 idioms

Use slices.Backward for reverse iteration in pool dead-list removal.
Replace manual wg.Add/go/defer patterns with sync.WaitGroup.Go in
tests.

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

Author: sean-

opensearchapi/api_scroll_test.go

@@ -9,6 +9,7 @@
 package opensearchapi_test
 
 import (
+	"context"
 	"strings"
 	"testing"
 	"time"
@@ -29,7 +30,7 @@ func TestScrollClient(t *testing.T) {
 
 	testIndex := testutil.MustUniqueString(t, "test-scroll")
 	t.Cleanup(func() {
-		client.Indices.Delete(t.Context(), opensearchapi.IndicesDeleteReq{Indices: []string{testIndex}})
+		client.Indices.Delete(context.Background(), opensearchapi.IndicesDeleteReq{Indices: []string{testIndex}})
 	})
 
 	_, err = client.Document.Create(

opensearchapi/api_tasks_test.go

@@ -14,6 +14,7 @@ import (
 	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -28,6 +29,9 @@ func TestTasksClient(t *testing.T) {
 	t.Parallel()
 	client, err := testutil.NewClient(t)
 	require.NoError(t, err)
+	// .tasks index mapping lacked cancellation_time_millis and resource_stats fields;
+	// cancelled tasks could never persist completion (opensearch-project/OpenSearch#16201).
+	testutil.SkipIfVersion(t, client, "<", "2.18", "TestTasksClient")
 	failingClient, err := osapitest.CreateFailingClient(t)
 	require.NoError(t, err)
 
@@ -120,16 +124,6 @@ func TestTasksClient(t *testing.T) {
 
 		// Create unique indices for this test
 		destIndex := testutil.MustUniqueString(t, "test-tasks-dest")
-		testIndices := []string{sourceIndex, destIndex}
-		t.Cleanup(func() {
-			client.Indices.Delete(
-				context.Background(),
-				opensearchapi.IndicesDeleteReq{
-					Indices: testIndices,
-					Params:  opensearchapi.IndicesDeleteParams{IgnoreUnavailable: opensearchapi.ToPointer(true)},
-				},
-			)
-		})
 
 		// Create destination index
 		client.Indices.Create(
@@ -157,6 +151,21 @@ func TestTasksClient(t *testing.T) {
 		require.NotEmpty(t, resp)
 		taskID := resp.Task
 
+		t.Cleanup(func() {
+			client.Tasks.Cancel(context.Background(), opensearchapi.TasksCancelReq{TaskID: taskID})
+			require.Eventually(t, func() bool {
+				resp, err := client.Tasks.Get(context.Background(), opensearchapi.TasksGetReq{TaskID: taskID})
+				return err == nil && resp.Completed
+			}, 30*time.Second, 100*time.Millisecond, "reindex task did not complete after cancel")
+			client.Indices.Delete(
+				context.Background(),
+				opensearchapi.IndicesDeleteReq{
+					Indices: []string{destIndex},
+					Params:  opensearchapi.IndicesDeleteParams{IgnoreUnavailable: opensearchapi.ToPointer(true)},
+				},
+			)
+		})
+
 		testCases := []tasksTests{
 			{
 				Name: "with request",
@@ -204,16 +213,6 @@ func TestTasksClient(t *testing.T) {
 
 		// Create unique indices for this test
 		destIndex := testutil.MustUniqueString(t, "test-tasks-dest")
-		testIndices := []string{sourceIndex, destIndex}
-		t.Cleanup(func() {
-			client.Indices.Delete(
-				context.Background(),
-				opensearchapi.IndicesDeleteReq{
-					Indices: testIndices,
-					Params:  opensearchapi.IndicesDeleteParams{IgnoreUnavailable: opensearchapi.ToPointer(true)},
-				},
-			)
-		})
 
 		// Create destination index
 		client.Indices.Create(
@@ -241,6 +240,21 @@ func TestTasksClient(t *testing.T) {
 		require.NotEmpty(t, resp)
 		taskID := resp.Task
 
+		t.Cleanup(func() {
+			client.Tasks.Cancel(context.Background(), opensearchapi.TasksCancelReq{TaskID: taskID})
+			require.Eventually(t, func() bool {
+				resp, err := client.Tasks.Get(context.Background(), opensearchapi.TasksGetReq{TaskID: taskID})
+				return err == nil && resp.Completed
+			}, 30*time.Second, 100*time.Millisecond, "reindex task did not complete after cancel")
+			client.Indices.Delete(
+				context.Background(),
+				opensearchapi.IndicesDeleteReq{
+					Indices: []string{destIndex},
+					Params:  opensearchapi.IndicesDeleteParams{IgnoreUnavailable: opensearchapi.ToPointer(true)},
+				},
+			)
+		})
+
 		testCases := []tasksTests{
 			{
 				Name: "with request",

opensearchtransport/connection_warmup_internal_test.go

@@ -695,9 +695,7 @@ func TestWarmupConcurrentSkip(t *testing.T) {
 
 			const goroutines = 16
 			for range goroutines {
-				wg.Add(1)
-				go func() {
-					defer wg.Done()
+				wg.Go(func() {
 					for {
 						switch conn.tryWarmupSkip() {
 						case warmupSkipped:
@@ -708,7 +706,7 @@ func TestWarmupConcurrentSkip(t *testing.T) {
 							return
 						}
 					}
-				}()
+				})
 			}
 
 			wg.Wait()

opensearchtransport/opensearchtransport.go

@@ -1611,11 +1611,17 @@ func (c *Client) signRequest(req *http.Request) error {
 }
 
 func (c *Client) setReqUserAgent(req *http.Request) {
+	if req.Header == nil {
+		req.Header = make(http.Header, 1)
+	}
 	req.Header.Set("User-Agent", c.userAgent)
 }
 
 func (c *Client) setReqGlobalHeader(req *http.Request) {
 	if len(c.header) > 0 {
+		if req.Header == nil {
+			req.Header = make(http.Header, len(c.header))
+		}
 		for k, v := range c.header {
 			if req.Header.Get(k) != k {
 				for _, vv := range v {

opensearchtransport/pool_multi_server.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"math/rand/v2"
 	"net/url"
+	"slices"
 	"sort"
 	"sync"
 	"sync/atomic"
@@ -595,8 +596,8 @@ func (cp *multiServerPool) removeFromReadyWithLock(c *Connection) {
 func (cp *multiServerPool) removeFromDeadWithLock(c *Connection) {
 	idx := -1
 	// Search backward -- recently appended items are at the tail
-	for i := len(cp.mu.dead) - 1; i >= 0; i-- {
-		if cp.mu.dead[i] == c {
+	for i, v := range slices.Backward(cp.mu.dead) {
+		if v == c {
 			idx = i
 			break
 		}

opensearchtransport/selector_round_robin_internal_test.go

@@ -123,9 +123,7 @@ func TestRoundRobinSelectorConcurrency(t *testing.T) {
 
 	// Start multiple goroutines selecting connections
 	for range numGoroutines {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		wg.Go(func() {
 			for range selectionsPerGoroutine {
 				selected, err := selector.Select(connections)
 				if err != nil {
@@ -134,7 +132,7 @@ func TestRoundRobinSelectorConcurrency(t *testing.T) {
 				}
 				results <- selected
 			}
-		}()
+		})
 	}
 
 	wg.Wait()