Extract applyRewrite helper; update context-cancellation test

Deduplicate the rewrite-detection, metrics-increment, and observer-
notification block from resolveDiscoveredNodes and
runAddressResolverRunner into (*Client).applyRewrite.

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

Author: sean-

opensearchtransport/address_resolver_internal_test.go

@@ -177,6 +177,18 @@ func TestAddressResolver(t *testing.T) {
 			wantPort: map[string]string{"rewrite-me": "9201", "keep-me": "9200"},
 			metrics:  wantMetrics{calls: 3, rewrites: 1, errors: 1},
 		},
+		{
+			name:    "malformed URL keeps default and counts an error",
+			inAddrs: map[string]string{"alpha": "10.0.0.1:9200", "beta": "10.0.0.2:9200"},
+			resolver: func(_ context.Context, _ NodeInfo) (*url.URL, error) {
+				// Empty Scheme/Host -- the kind of URL a resolver might
+				// return from a buggy parse. The client must reject it
+				// rather than enrolling a malformed connection.
+				return &url.URL{}, nil
+			},
+			wantPort: map[string]string{"alpha": "9200", "beta": "9200"},
+			metrics:  wantMetrics{calls: 2, rewrites: 0, errors: 2},
+		},
 	}
 
 	for _, tt := range protocolTests {
@@ -389,7 +401,7 @@ func TestAddressResolver(t *testing.T) {
 				ctx, cancel := context.WithCancel(t.Context())
 				cancel()
 				return ctx, func(_ context.Context, _ NodeInfo) (*url.URL, error) {
-					return nil, nil //nolint:nilnil // testing (nil, nil) protocol case
+					return nil, nil
 				}
 			},
 			wantErr: context.Canceled,
@@ -399,12 +411,13 @@ func TestAddressResolver(t *testing.T) {
 			setup: func(t *testing.T) (context.Context, AddressResolverFunc) {
 				t.Helper()
 				ctx, cancel := context.WithCancel(t.Context())
+				t.Cleanup(cancel)
 				var calls atomic.Int32
 				return ctx, func(_ context.Context, _ NodeInfo) (*url.URL, error) {
 					if calls.Add(1) == 1 {
 						cancel()
 					}
-					return nil, nil //nolint:nilnil // testing (nil, nil) protocol case
+					return nil, nil
 				}
 			},
 			maxResolvers: 1,
@@ -433,7 +446,7 @@ func TestAddressResolver(t *testing.T) {
 				return
 			}
 			require.NoError(t, err)
-			require.Equal(t, tt.wantNodes, len(nodes))
+			require.Len(t, nodes, tt.wantNodes)
 		})
 	}
 
@@ -767,42 +780,83 @@ func TestAddressResolverRunner(t *testing.T) {
 			"resolve param should be nil when AddressResolver is not set")
 	})
 
-	t.Run("runner context cancellation propagated", func(t *testing.T) {
-		t.Parallel()
-
-		ctx, cancel := context.WithCancel(t.Context())
-		cancel()
-
-		tp, err := New(Config{
-			URLs:        []*url.URL{testSeedURL(t)},
-			Transport:   newResolverTestTransport(t, nodesJSON),
-			HealthCheck: NoOpHealthCheck,
-			AddressResolverRunner: func(ctx context.Context, _ []NodeInfo, _ AddressResolverFunc) ([]ResolvedAddress, error) {
+	runnerEdgeCases := []struct {
+		name      string
+		makeCtx   func(t *testing.T) context.Context
+		runner    AddressResolverRunnerFunc
+		wantErr   error
+		wantNodes int
+	}{
+		{
+			name: "context cancellation propagated",
+			makeCtx: func(t *testing.T) context.Context {
+				t.Helper()
+				ctx, cancel := context.WithCancel(t.Context())
+				cancel()
+				return ctx
+			},
+			runner: func(ctx context.Context, _ []NodeInfo, _ AddressResolverFunc) ([]ResolvedAddress, error) {
 				return nil, ctx.Err()
 			},
-		})
-		require.NoError(t, err)
+			wantErr: context.Canceled,
+		},
+		{
+			name: "drops all returns ErrAllResolversFailed",
+			runner: func(_ context.Context, _ []NodeInfo, _ AddressResolverFunc) ([]ResolvedAddress, error) {
+				return nil, nil
+			},
+			wantErr: ErrAllResolversFailed,
+		},
+		{
+			name: "nil URLs are skipped",
+			runner: func(_ context.Context, nodes []NodeInfo, _ AddressResolverFunc) ([]ResolvedAddress, error) {
+				out := make([]ResolvedAddress, len(nodes))
+				for i, n := range nodes {
+					out[i] = ResolvedAddress{Node: n, URL: nil}
+				}
+				return out, nil
+			},
+			wantErr: ErrAllResolversFailed,
+		},
+		{
+			name: "unknown node ID is ignored",
+			runner: func(_ context.Context, nodes []NodeInfo, _ AddressResolverFunc) ([]ResolvedAddress, error) {
+				bogus := NodeInfo{ID: "bogus-id", Name: "ghost"}
+				return []ResolvedAddress{
+					{Node: nodes[0], URL: nodes[0].URL},
+					{Node: bogus, URL: nodes[0].URL},
+				}, nil
+			},
+			wantNodes: 1,
+		},
+	}
 
-		_, err = tp.getNodesInfo(ctx)
-		require.ErrorIs(t, err, context.Canceled)
-	})
+	for _, tt := range runnerEdgeCases {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
 
-	t.Run("runner drops all returns ErrAllResolversFailed", func(t *testing.T) {
-		t.Parallel()
+			ctx := t.Context()
+			if tt.makeCtx != nil {
+				ctx = tt.makeCtx(t)
+			}
 
-		tp, err := New(Config{
-			URLs:        []*url.URL{testSeedURL(t)},
-			Transport:   newResolverTestTransport(t, nodesJSON),
-			HealthCheck: NoOpHealthCheck,
-			AddressResolverRunner: func(_ context.Context, _ []NodeInfo, _ AddressResolverFunc) ([]ResolvedAddress, error) {
-				return nil, nil
-			},
-		})
-		require.NoError(t, err)
+			tp, err := New(Config{
+				URLs:                  []*url.URL{testSeedURL(t)},
+				Transport:             newResolverTestTransport(t, nodesJSON),
+				HealthCheck:           NoOpHealthCheck,
+				AddressResolverRunner: tt.runner,
+			})
+			require.NoError(t, err)
 
-		_, err = tp.getNodesInfo(t.Context())
-		require.ErrorIs(t, err, ErrAllResolversFailed)
-	})
+			nodes, err := tp.getNodesInfo(ctx)
+			if tt.wantErr != nil {
+				require.ErrorIs(t, err, tt.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			require.Len(t, nodes, tt.wantNodes)
+		})
+	}
 
 	t.Run("runner error propagated", func(t *testing.T) {
 		t.Parallel()

opensearchtransport/discovery.go

@@ -1333,30 +1333,7 @@ func (c *Client) resolveDiscoveredNodes(ctx context.Context, pending []discovery
 			}
 
 			node := p.node
-			u := p.defaultURL
-
-			if resolved != nil && resolved.String() != u.String() {
-				node.rewritten = true
-
-				if c.metrics != nil {
-					c.metrics.addressResolverRewrites.Add(1)
-				}
-
-				if obs != nil {
-					obs.OnAddressRewrite(AddressRewriteEvent{
-						ID:           node.ID,
-						Name:         node.Name,
-						Roles:        node.Roles,
-						OriginalURL:  u.String(),
-						RewrittenURL: resolved.String(),
-						Timestamp:    time.Now().UTC(),
-					})
-				}
-
-				u = resolved
-			}
-
-			node.url = u
+			node.url = c.applyRewrite(&node, p.defaultURL, resolved, obs)
 			results[i] = resolvedNode{node: node}
 		}(i, p)
 	}
@@ -1389,6 +1366,53 @@ func (c *Client) resolveDiscoveredNodes(ctx context.Context, pending []discovery
 	return out, nil
 }
 
+// applyRewrite checks whether resolved differs from defaultURL and, when it
+// does, marks the node as rewritten, increments the rewrite metric, and fires
+// the OnAddressRewrite observer event. Returns the URL to use for the node.
+//
+// A non-nil resolved URL with an empty Scheme or Host is treated as a
+// resolver error: the default URL is kept, the error counter is incremented,
+// and a debug log is emitted. Validating here surfaces the misuse at
+// resolver time rather than letting a malformed URL flow into the connection
+// pool and fail later as a confusing HTTP error.
+func (c *Client) applyRewrite(node *nodeInfo, defaultURL, resolved *url.URL, obs ConnectionObserver) *url.URL {
+	if resolved == nil || resolved.String() == defaultURL.String() {
+		return defaultURL
+	}
+
+	if resolved.Scheme == "" || resolved.Host == "" {
+		if c.metrics != nil {
+			c.metrics.addressResolverErrors.Add(1)
+		}
+		if dl := loadDebugLogger(); dl != nil {
+			dl.Logf("AddressResolver returned malformed URL for node %q (scheme=%q host=%q); keeping default %q\n",
+				node.Name, resolved.Scheme, resolved.Host, defaultURL)
+		}
+		return defaultURL
+	}
+
+	node.rewritten = true
+
+	if c.metrics != nil {
+		c.metrics.addressResolverRewrites.Add(1)
+	}
+
+	if obs != nil {
+		// Clone Roles so an observer that retains the event sees a stable
+		// copy -- matches ConnectionEvent's c.Roles.toSlice() behavior.
+		obs.OnAddressRewrite(AddressRewriteEvent{
+			ID:           node.ID,
+			Name:         node.Name,
+			Roles:        slices.Clone(node.Roles),
+			OriginalURL:  defaultURL.String(),
+			RewrittenURL: resolved.String(),
+			Timestamp:    time.Now().UTC(),
+		})
+	}
+
+	return resolved
+}
+
 // newInstrumentedResolver wraps an AddressResolverFunc with metrics
 // instrumentation. Each invocation increments addressResolverCalls, and
 // non-nil errors increment addressResolverErrors and emit a debug log.
@@ -1474,30 +1498,7 @@ func (c *Client) runAddressResolverRunner(ctx context.Context, pending []discove
 		seen[ra.Node.ID] = struct{}{}
 
 		node := p.node
-		u := p.defaultURL
-
-		if ra.URL.String() != u.String() {
-			node.rewritten = true
-
-			if c.metrics != nil {
-				c.metrics.addressResolverRewrites.Add(1)
-			}
-
-			if obs != nil {
-				obs.OnAddressRewrite(AddressRewriteEvent{
-					ID:           node.ID,
-					Name:         node.Name,
-					Roles:        node.Roles,
-					OriginalURL:  u.String(),
-					RewrittenURL: ra.URL.String(),
-					Timestamp:    time.Now().UTC(),
-				})
-			}
-
-			u = ra.URL
-		}
-
-		node.url = u
+		node.url = c.applyRewrite(&node, p.defaultURL, ra.URL, obs)
 		out = append(out, node)
 	}