From 7720fe338b405b15347c772ca100b5c9dc05ded6 Mon Sep 17 00:00:00 2001 From: Pranjali-2501 Date: Sat, 6 Dec 2025 20:59:46 +0000 Subject: [PATCH 1/3] add resolver.Endpoint in Endpoint struct --- .../balancer/clusterresolver/configbuilder.go | 16 +- .../clusterresolver/configbuilder_test.go | 116 +++++++---- .../xds/xdsclient/tests/eds_watchers_test.go | 71 ++++++- .../tests/federation_watchers_test.go | 10 +- .../xdsclient/tests/resource_update_test.go | 53 +++-- .../xds/xdsclient/xdsresource/type_eds.go | 12 +- .../xdsclient/xdsresource/unmarshal_eds.go | 40 +++- .../xdsresource/unmarshal_eds_test.go | 197 ++++++++++-------- 8 files changed, 338 insertions(+), 177 deletions(-) diff --git a/internal/xds/balancer/clusterresolver/configbuilder.go b/internal/xds/balancer/clusterresolver/configbuilder.go index 3d8e08972c1d..77a3f40b01c6 100644 --- a/internal/xds/balancer/clusterresolver/configbuilder.go +++ b/internal/xds/balancer/clusterresolver/configbuilder.go @@ -265,25 +265,21 @@ func priorityLocalitiesToClusterImpl(localities []xdsresource.Locality, priority if endpoint.HealthStatus != xdsresource.EndpointHealthStatusHealthy && endpoint.HealthStatus != xdsresource.EndpointHealthStatusUnknown { continue } - resolverEndpoint := resolver.Endpoint{} - for _, as := range endpoint.Addresses { - resolverEndpoint.Addresses = append(resolverEndpoint.Addresses, resolver.Address{Addr: as}) - } - resolverEndpoint = hierarchy.SetInEndpoint(resolverEndpoint, []string{priorityName, localityStr}) - resolverEndpoint = xdsinternal.SetLocalityIDInEndpoint(resolverEndpoint, locality.ID) + endpoint.ResolverEndpoint = hierarchy.SetInEndpoint(endpoint.ResolverEndpoint, []string{priorityName, localityStr}) + endpoint.ResolverEndpoint = xdsinternal.SetLocalityIDInEndpoint(endpoint.ResolverEndpoint, locality.ID) // "To provide the xds_wrr_locality load balancer information about // locality weights received from EDS, the cluster resolver will // populate a new locality weight attribute for each address The // attribute will have the weight (as an integer) of the locality // the address is part of." - A52 - resolverEndpoint = wrrlocality.SetAddrInfoInEndpoint(resolverEndpoint, wrrlocality.AddrInfo{LocalityWeight: lw}) + endpoint.ResolverEndpoint = wrrlocality.SetAddrInfoInEndpoint(endpoint.ResolverEndpoint, wrrlocality.AddrInfo{LocalityWeight: lw}) var ew uint32 = 1 if endpoint.Weight != 0 { ew = endpoint.Weight } - resolverEndpoint = weight.Set(resolverEndpoint, weight.EndpointInfo{Weight: lw * ew}) - resolverEndpoint = ringhash.SetHashKey(resolverEndpoint, endpoint.HashKey) - retEndpoints = append(retEndpoints, resolverEndpoint) + endpoint.ResolverEndpoint = weight.Set(endpoint.ResolverEndpoint, weight.EndpointInfo{Weight: lw * ew}) + endpoint.ResolverEndpoint = ringhash.SetHashKey(endpoint.ResolverEndpoint, endpoint.HashKey) + retEndpoints = append(retEndpoints, endpoint.ResolverEndpoint) } } return &clusterimpl.LBConfig{ diff --git a/internal/xds/balancer/clusterresolver/configbuilder_test.go b/internal/xds/balancer/clusterresolver/configbuilder_test.go index 571b756484e2..5107e7587bd0 100644 --- a/internal/xds/balancer/clusterresolver/configbuilder_test.go +++ b/internal/xds/balancer/clusterresolver/configbuilder_test.go @@ -95,10 +95,12 @@ func init() { endpoints = append(endpoints, resolver.Endpoint{Addresses: []resolver.Address{{Addr: addr}}}) ends = append(ends, xdsresource.Endpoint{ HealthStatus: xdsresource.EndpointHealthStatusHealthy, - Addresses: []string{ - addr, - fmt.Sprintf("addr-%d-%d-additional-1", i, j), - fmt.Sprintf("addr-%d-%d-additional-2", i, j), + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{ + {Addr: addr}, + {Addr: fmt.Sprintf("addr-%d-%d-additional-1", i, j)}, + {Addr: fmt.Sprintf("addr-%d-%d-additional-2", i, j)}, + }, }, }) } @@ -315,8 +317,8 @@ func TestBuildClusterImplConfigForDNS(t *testing.T) { Name: "pick_first", }, } - e1 := resolver.Endpoint{Addresses: []resolver.Address{{Addr: testEndpoints[0][0].Addresses[0]}}} - e2 := resolver.Endpoint{Addresses: []resolver.Address{{Addr: testEndpoints[0][1].Addresses[0]}}} + e1 := resolver.Endpoint{Addresses: []resolver.Address{{Addr: testEndpoints[0][0].ResolverEndpoint.Addresses[0].Addr}}} + e2 := resolver.Endpoint{Addresses: []resolver.Address{{Addr: testEndpoints[0][1].ResolverEndpoint.Addresses[0].Addr}}} wantEndpoints := []resolver.Endpoint{ hierarchy.SetInEndpoint(e1, []string{"priority-3"}), hierarchy.SetInEndpoint(e2, []string{"priority-3"}), @@ -417,14 +419,14 @@ func TestBuildClusterImplConfigForEDS(t *testing.T) { }, } wantEndpoints := []resolver.Endpoint{ - testEndpointWithAttrs(testEndpoints[0][0].Addresses, 20, 1, "priority-2-0", &testLocalityIDs[0]), - testEndpointWithAttrs(testEndpoints[0][1].Addresses, 20, 1, "priority-2-0", &testLocalityIDs[0]), - testEndpointWithAttrs(testEndpoints[1][0].Addresses, 80, 1, "priority-2-0", &testLocalityIDs[1]), - testEndpointWithAttrs(testEndpoints[1][1].Addresses, 80, 1, "priority-2-0", &testLocalityIDs[1]), - testEndpointWithAttrs(testEndpoints[2][0].Addresses, 20, 1, "priority-2-1", &testLocalityIDs[2]), - testEndpointWithAttrs(testEndpoints[2][1].Addresses, 20, 1, "priority-2-1", &testLocalityIDs[2]), - testEndpointWithAttrs(testEndpoints[3][0].Addresses, 80, 1, "priority-2-1", &testLocalityIDs[3]), - testEndpointWithAttrs(testEndpoints[3][1].Addresses, 80, 1, "priority-2-1", &testLocalityIDs[3]), + testEndpointWithAttrs(testEndpoints[0][0].ResolverEndpoint, 20, 1, "priority-2-0", &testLocalityIDs[0]), + testEndpointWithAttrs(testEndpoints[0][1].ResolverEndpoint, 20, 1, "priority-2-0", &testLocalityIDs[0]), + testEndpointWithAttrs(testEndpoints[1][0].ResolverEndpoint, 80, 1, "priority-2-0", &testLocalityIDs[1]), + testEndpointWithAttrs(testEndpoints[1][1].ResolverEndpoint, 80, 1, "priority-2-0", &testLocalityIDs[1]), + testEndpointWithAttrs(testEndpoints[2][0].ResolverEndpoint, 20, 1, "priority-2-1", &testLocalityIDs[2]), + testEndpointWithAttrs(testEndpoints[2][1].ResolverEndpoint, 20, 1, "priority-2-1", &testLocalityIDs[2]), + testEndpointWithAttrs(testEndpoints[3][0].ResolverEndpoint, 80, 1, "priority-2-1", &testLocalityIDs[3]), + testEndpointWithAttrs(testEndpoints[3][1].ResolverEndpoint, 80, 1, "priority-2-1", &testLocalityIDs[3]), } if diff := cmp.Diff(gotNames, wantNames); diff != "" { @@ -547,16 +549,40 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) { localities: []xdsresource.Locality{ { Endpoints: []xdsresource.Endpoint{ - {Addresses: []string{"addr-1-1"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90}, - {Addresses: []string{"addr-1-2"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10}, + { + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr-1-1"}}, + }, + Weight: 90, + HealthStatus: xdsresource.EndpointHealthStatusHealthy, + }, + { + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr-1-2"}}, + }, + Weight: 10, + HealthStatus: xdsresource.EndpointHealthStatusHealthy, + }, }, ID: clients.Locality{Zone: "test-zone-1"}, Weight: 20, }, { Endpoints: []xdsresource.Endpoint{ - {Addresses: []string{"addr-2-1"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90}, - {Addresses: []string{"addr-2-2"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10}, + { + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr-2-1"}}, + }, + Weight: 90, + HealthStatus: xdsresource.EndpointHealthStatusHealthy, + }, + { + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr-2-2"}}, + }, + Weight: 10, + HealthStatus: xdsresource.EndpointHealthStatusHealthy, + }, }, ID: clients.Locality{Zone: "test-zone-2"}, Weight: 80, @@ -576,10 +602,10 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) { ChildPolicy: &iserviceconfig.BalancerConfig{Name: roundrobin.Name}, }, wantEndpoints: []resolver.Endpoint{ - testEndpointWithAttrs([]string{"addr-1-1"}, 20, 90, "test-priority", &clients.Locality{Zone: "test-zone-1"}), - testEndpointWithAttrs([]string{"addr-1-2"}, 20, 10, "test-priority", &clients.Locality{Zone: "test-zone-1"}), - testEndpointWithAttrs([]string{"addr-2-1"}, 80, 90, "test-priority", &clients.Locality{Zone: "test-zone-2"}), - testEndpointWithAttrs([]string{"addr-2-2"}, 80, 10, "test-priority", &clients.Locality{Zone: "test-zone-2"}), + testEndpointWithAttrs(resolver.Endpoint{Addresses: []resolver.Address{{Addr: "addr-1-1"}}}, 20, 90, "test-priority", &clients.Locality{Zone: "test-zone-1"}), + testEndpointWithAttrs(resolver.Endpoint{Addresses: []resolver.Address{{Addr: "addr-1-2"}}}, 20, 10, "test-priority", &clients.Locality{Zone: "test-zone-1"}), + testEndpointWithAttrs(resolver.Endpoint{Addresses: []resolver.Address{{Addr: "addr-2-1"}}}, 80, 90, "test-priority", &clients.Locality{Zone: "test-zone-2"}), + testEndpointWithAttrs(resolver.Endpoint{Addresses: []resolver.Address{{Addr: "addr-2-2"}}}, 80, 10, "test-priority", &clients.Locality{Zone: "test-zone-2"}), }, }, { @@ -587,16 +613,40 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) { localities: []xdsresource.Locality{ { Endpoints: []xdsresource.Endpoint{ - {Addresses: []string{"addr-1-1"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90}, - {Addresses: []string{"addr-1-2"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10}, + { + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr-1-1"}}, + }, + Weight: 90, + HealthStatus: xdsresource.EndpointHealthStatusHealthy, + }, + { + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr-1-2"}}, + }, + Weight: 10, + HealthStatus: xdsresource.EndpointHealthStatusHealthy, + }, }, ID: clients.Locality{Zone: "test-zone-1"}, Weight: 20, }, { Endpoints: []xdsresource.Endpoint{ - {Addresses: []string{"addr-2-1"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 90}, - {Addresses: []string{"addr-2-2"}, HealthStatus: xdsresource.EndpointHealthStatusHealthy, Weight: 10}, + { + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr-2-1"}}, + }, + Weight: 90, + HealthStatus: xdsresource.EndpointHealthStatusHealthy, + }, + { + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr-2-2"}}, + }, + Weight: 10, + HealthStatus: xdsresource.EndpointHealthStatusHealthy, + }, }, ID: clients.Locality{Zone: "test-zone-2"}, Weight: 80, @@ -612,10 +662,10 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) { }, }, wantEndpoints: []resolver.Endpoint{ - testEndpointWithAttrs([]string{"addr-1-1"}, 20, 90, "test-priority", &clients.Locality{Zone: "test-zone-1"}), - testEndpointWithAttrs([]string{"addr-1-2"}, 20, 10, "test-priority", &clients.Locality{Zone: "test-zone-1"}), - testEndpointWithAttrs([]string{"addr-2-1"}, 80, 90, "test-priority", &clients.Locality{Zone: "test-zone-2"}), - testEndpointWithAttrs([]string{"addr-2-2"}, 80, 10, "test-priority", &clients.Locality{Zone: "test-zone-2"}), + testEndpointWithAttrs(resolver.Endpoint{Addresses: []resolver.Address{{Addr: "addr-1-1"}}}, 20, 90, "test-priority", &clients.Locality{Zone: "test-zone-1"}), + testEndpointWithAttrs(resolver.Endpoint{Addresses: []resolver.Address{{Addr: "addr-1-2"}}}, 20, 10, "test-priority", &clients.Locality{Zone: "test-zone-1"}), + testEndpointWithAttrs(resolver.Endpoint{Addresses: []resolver.Address{{Addr: "addr-2-1"}}}, 80, 90, "test-priority", &clients.Locality{Zone: "test-zone-2"}), + testEndpointWithAttrs(resolver.Endpoint{Addresses: []resolver.Address{{Addr: "addr-2-2"}}}, 80, 10, "test-priority", &clients.Locality{Zone: "test-zone-2"}), }, }, } @@ -635,11 +685,7 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) { } } -func testEndpointWithAttrs(addrStrs []string, localityWeight, endpointWeight uint32, priority string, lID *clients.Locality) resolver.Endpoint { - endpoint := resolver.Endpoint{} - for _, a := range addrStrs { - endpoint.Addresses = append(endpoint.Addresses, resolver.Address{Addr: a}) - } +func testEndpointWithAttrs(endpoint resolver.Endpoint, localityWeight, endpointWeight uint32, priority string, lID *clients.Locality) resolver.Endpoint { path := []string{priority} if lID != nil { path = append(path, xdsinternal.LocalityString(*lID)) diff --git a/internal/xds/xdsclient/tests/eds_watchers_test.go b/internal/xds/xdsclient/tests/eds_watchers_test.go index d367e12d7a51..1192c20a7ee2 100644 --- a/internal/xds/xdsclient/tests/eds_watchers_test.go +++ b/internal/xds/xdsclient/tests/eds_watchers_test.go @@ -36,6 +36,7 @@ import ( "google.golang.org/grpc/internal/xds/clients" "google.golang.org/grpc/internal/xds/xdsclient" "google.golang.org/grpc/internal/xds/xdsclient/xdsresource" + "google.golang.org/grpc/resolver" "google.golang.org/protobuf/types/known/wrapperspb" v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3" @@ -176,7 +177,12 @@ func (s) TestEDSWatch(t *testing.T) { update: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}}, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: fmt.Sprintf("%s:%d", edsHost1, edsPort1)}}, + }, + Weight: 1, + }}, ID: clients.Locality{ Region: "region-1", Zone: "zone-1", @@ -199,7 +205,12 @@ func (s) TestEDSWatch(t *testing.T) { update: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}}, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: fmt.Sprintf("%s:%d", edsHost1, edsPort1)}}, + }, + Weight: 1, + }}, ID: clients.Locality{ Region: "region-1", Zone: "zone-1", @@ -338,7 +349,12 @@ func (s) TestEDSWatch_TwoWatchesForSameResourceName(t *testing.T) { update: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}}, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: fmt.Sprintf("%s:%d", edsHost1, edsPort1)}}, + }, + Weight: 1, + }}, ID: clients.Locality{ Region: "region-1", Zone: "zone-1", @@ -354,7 +370,12 @@ func (s) TestEDSWatch_TwoWatchesForSameResourceName(t *testing.T) { update: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost2, edsPort2)}, Weight: 1}}, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: fmt.Sprintf("%s:%d", edsHost2, edsPort2)}}, + }, + Weight: 1, + }}, ID: clients.Locality{ Region: "region-1", Zone: "zone-1", @@ -376,7 +397,12 @@ func (s) TestEDSWatch_TwoWatchesForSameResourceName(t *testing.T) { update: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}}, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: fmt.Sprintf("%s:%d", edsHost1, edsPort1)}}, + }, + Weight: 1, + }}, ID: clients.Locality{ Region: "region-1", Zone: "zone-1", @@ -392,7 +418,12 @@ func (s) TestEDSWatch_TwoWatchesForSameResourceName(t *testing.T) { update: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost2, edsPort2)}, Weight: 1}}, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: fmt.Sprintf("%s:%d", edsHost2, edsPort2)}}, + }, + Weight: 1, + }}, ID: clients.Locality{ Region: "region-1", Zone: "zone-1", @@ -590,7 +621,12 @@ func (s) TestEDSWatch_ThreeWatchesForDifferentResourceNames(t *testing.T) { update: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}}, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: fmt.Sprintf("%s:%d", edsHost1, edsPort1)}}, + }, + Weight: 1, + }}, ID: clients.Locality{ Region: "region-1", Zone: "zone-1", @@ -681,7 +717,12 @@ func (s) TestEDSWatch_ResourceCaching(t *testing.T) { update: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}}, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: fmt.Sprintf("%s:%d", edsHost1, edsPort1)}}, + }, + Weight: 1, + }}, ID: clients.Locality{ Region: "region-1", Zone: "zone-1", @@ -813,7 +854,12 @@ func (s) TestEDSWatch_ValidResponseCancelsExpiryTimerBehavior(t *testing.T) { update: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}}, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: fmt.Sprintf("%s:%d", edsHost1, edsPort1)}}, + }, + Weight: 1, + }}, ID: clients.Locality{ Region: "region-1", Zone: "zone-1", @@ -976,7 +1022,12 @@ func (s) TestEDSWatch_PartialValid(t *testing.T) { update: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{fmt.Sprintf("%s:%d", edsHost1, edsPort1)}, Weight: 1}}, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: fmt.Sprintf("%s:%d", edsHost1, edsPort1)}}, + }, + Weight: 1, + }}, ID: clients.Locality{ Region: "region-1", Zone: "zone-1", diff --git a/internal/xds/xdsclient/tests/federation_watchers_test.go b/internal/xds/xdsclient/tests/federation_watchers_test.go index 436235bfdbfd..aedf117be0e4 100644 --- a/internal/xds/xdsclient/tests/federation_watchers_test.go +++ b/internal/xds/xdsclient/tests/federation_watchers_test.go @@ -29,6 +29,7 @@ import ( "google.golang.org/grpc/internal/xds/clients" "google.golang.org/grpc/internal/xds/xdsclient" "google.golang.org/grpc/internal/xds/xdsclient/xdsresource" + "google.golang.org/grpc/resolver" v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3" v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3" @@ -291,8 +292,13 @@ func (s) TestFederation_EndpointsResourceContextParamOrder(t *testing.T) { update: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{"localhost:666"}, Weight: 1}}, - Weight: 1, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "localhost:666"}}, + }, + Weight: 1, + }}, + Weight: 1, ID: clients.Locality{ Region: "region-1", Zone: "zone-1", diff --git a/internal/xds/xdsclient/tests/resource_update_test.go b/internal/xds/xdsclient/tests/resource_update_test.go index 2a7f146f983d..87d5653fa006 100644 --- a/internal/xds/xdsclient/tests/resource_update_test.go +++ b/internal/xds/xdsclient/tests/resource_update_test.go @@ -36,6 +36,7 @@ import ( "google.golang.org/grpc/internal/xds/clients" "google.golang.org/grpc/internal/xds/xdsclient" "google.golang.org/grpc/internal/xds/xdsclient/xdsresource" + "google.golang.org/grpc/resolver" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/testing/protocmp" "google.golang.org/protobuf/types/known/anypb" @@ -1089,16 +1090,26 @@ func (s) TestHandleEndpointsResponseFromManagementServer(t *testing.T) { wantUpdate: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{"addr1:314"}, Weight: 1}}, - ID: clients.Locality{SubZone: "locality-1"}, - Priority: 1, - Weight: 1, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr1:314"}}, + }, + Weight: 1, + }}, + ID: clients.Locality{SubZone: "locality-1"}, + Priority: 1, + Weight: 1, }, { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{"addr2:159"}, Weight: 1}}, - ID: clients.Locality{SubZone: "locality-2"}, - Priority: 0, - Weight: 1, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr2:159"}}, + }, + Weight: 1, + }}, + ID: clients.Locality{SubZone: "locality-2"}, + Priority: 0, + Weight: 1, }, }, }, @@ -1123,16 +1134,26 @@ func (s) TestHandleEndpointsResponseFromManagementServer(t *testing.T) { wantUpdate: xdsresource.EndpointsUpdate{ Localities: []xdsresource.Locality{ { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{"addr1:314"}, Weight: 1}}, - ID: clients.Locality{SubZone: "locality-1"}, - Priority: 1, - Weight: 1, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr1:314"}}, + }, + Weight: 1, + }}, + ID: clients.Locality{SubZone: "locality-1"}, + Priority: 1, + Weight: 1, }, { - Endpoints: []xdsresource.Endpoint{{Addresses: []string{"addr2:159"}, Weight: 1}}, - ID: clients.Locality{SubZone: "locality-2"}, - Priority: 0, - Weight: 1, + Endpoints: []xdsresource.Endpoint{{ + ResolverEndpoint: resolver.Endpoint{ + Addresses: []resolver.Address{{Addr: "addr2:159"}}, + }, + Weight: 1, + }}, + ID: clients.Locality{SubZone: "locality-2"}, + Priority: 0, + Weight: 1, }, }, }, diff --git a/internal/xds/xdsclient/xdsresource/type_eds.go b/internal/xds/xdsclient/xdsresource/type_eds.go index 39b2ba74ddcb..830b976f4a9e 100644 --- a/internal/xds/xdsclient/xdsresource/type_eds.go +++ b/internal/xds/xdsclient/xdsresource/type_eds.go @@ -19,6 +19,7 @@ package xdsresource import ( "google.golang.org/grpc/internal/xds/clients" + "google.golang.org/grpc/resolver" "google.golang.org/protobuf/types/known/anypb" ) @@ -49,12 +50,11 @@ const ( // Endpoint contains information of an endpoint. type Endpoint struct { - Addresses []string - HealthStatus EndpointHealthStatus - Weight uint32 - HashKey string - Metadata map[string]any - Hostname string + ResolverEndpoint resolver.Endpoint + HealthStatus EndpointHealthStatus + Weight uint32 + HashKey string + Metadata map[string]any } // Locality contains information of a locality. diff --git a/internal/xds/xdsclient/xdsresource/unmarshal_eds.go b/internal/xds/xdsclient/xdsresource/unmarshal_eds.go index ebf47961c45d..8a1f2b71d43d 100644 --- a/internal/xds/xdsclient/xdsresource/unmarshal_eds.go +++ b/internal/xds/xdsclient/xdsresource/unmarshal_eds.go @@ -30,10 +30,33 @@ import ( "google.golang.org/grpc/internal/pretty" xdsinternal "google.golang.org/grpc/internal/xds" "google.golang.org/grpc/internal/xds/clients" + "google.golang.org/grpc/resolver" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" ) +type hostnameType string + +// hostnameKey is the key to store the hostname key attribute in +// a resolver.Endpoint attribute. +const hostnameKey = hostnameType("grpc.resolver.xdsresource.hostname") + +// SetHostname sets the hostname key for this resolver.Endpoint attribute. +func SetHostname(endpoint resolver.Endpoint, hostname string) resolver.Endpoint { + if hostname == "" { + return endpoint + } + endpoint.Attributes = endpoint.Attributes.WithValue(hostnameKey, hostname) + return endpoint +} + +// GetHostname returns the hostname key attribute of endpoint. If this attribute +// is not set, it returns the empty string. +func GetHostname(endpoint resolver.Endpoint) string { + hostname, _ := endpoint.Attributes.Value(hostnameKey).(string) + return hostname +} + func unmarshalEndpointsResource(r *anypb.Any) (string, EndpointsUpdate, error) { r, err := UnwrapResource(r) if err != nil { @@ -102,7 +125,9 @@ func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint, uniqueEndpointAddrs } } + address := []resolver.Address{} for _, a := range addrs { + address = append(address, resolver.Address{Addr: a}) if uniqueEndpointAddrs[a] { return nil, fmt.Errorf("duplicate endpoint with the same address %s", a) } @@ -126,13 +151,16 @@ func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint, uniqueEndpointAddrs hashKey = hashKeyFromMetadata(endpointMetadata) } } + endpoint := resolver.Endpoint{ + Addresses: address, + } + endpoint = SetHostname(endpoint, lbEndpoint.GetEndpoint().GetHostname()) endpoints = append(endpoints, Endpoint{ - HealthStatus: EndpointHealthStatus(lbEndpoint.GetHealthStatus()), - Addresses: addrs, - Weight: weight, - HashKey: hashKey, - Metadata: endpointMetadata, - Hostname: lbEndpoint.GetEndpoint().GetHostname(), + ResolverEndpoint: endpoint, + HealthStatus: EndpointHealthStatus(lbEndpoint.GetHealthStatus()), + Weight: weight, + HashKey: hashKey, + Metadata: endpointMetadata, }) } return endpoints, nil diff --git a/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go b/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go index 905cf8a20f26..33431a9a93f1 100644 --- a/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go +++ b/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go @@ -34,12 +34,25 @@ import ( "google.golang.org/grpc/internal/testutils" "google.golang.org/grpc/internal/xds/clients" "google.golang.org/grpc/internal/xds/xdsclient/xdsresource/version" + "google.golang.org/grpc/resolver" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" "google.golang.org/protobuf/types/known/structpb" "google.golang.org/protobuf/types/known/wrapperspb" ) +func getResolverEndpoint(addr []string, hostname string) resolver.Endpoint { + address := []resolver.Address{} + for _, a := range addr { + address = append(address, resolver.Address{Addr: a}) + } + resolverEndpoint := resolver.Endpoint{ + Addresses: address, + } + resolverEndpoint = SetHostname(resolverEndpoint, hostname) + return resolverEndpoint +} + func (s) TestEDSParseRespProto(t *testing.T) { tests := []struct { name string @@ -153,10 +166,9 @@ func (s) TestEDSParseRespProto(t *testing.T) { Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnhealthy, - Weight: 271, - Hostname: "addr1", + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnhealthy, + Weight: 271, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 1, @@ -164,10 +176,9 @@ func (s) TestEDSParseRespProto(t *testing.T) { }, { Endpoints: []Endpoint{{ - Addresses: []string{"addr2:159"}, - HealthStatus: EndpointHealthStatusDraining, - Weight: 828, - Hostname: "addr2", + ResolverEndpoint: getResolverEndpoint([]string{"addr2:159"}, "addr2"), + HealthStatus: EndpointHealthStatusDraining, + Weight: 828, }}, ID: clients.Locality{SubZone: "locality-2"}, Priority: 0, @@ -201,10 +212,9 @@ func (s) TestEDSParseRespProto(t *testing.T) { Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnhealthy, - Weight: 271, - Hostname: "addr1", + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnhealthy, + Weight: 271, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 1, @@ -212,10 +222,9 @@ func (s) TestEDSParseRespProto(t *testing.T) { }, { Endpoints: []Endpoint{{ - Addresses: []string{"addr2:159"}, - HealthStatus: EndpointHealthStatusDraining, - Weight: 828, - Hostname: "addr2", + ResolverEndpoint: getResolverEndpoint([]string{"addr2:159"}, "addr2"), + HealthStatus: EndpointHealthStatusDraining, + Weight: 828, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, @@ -298,13 +307,13 @@ func (s) TestEDSParseRespProtoAdditionalAddrs(t *testing.T) { name: "multiple localities", m: func() *v3endpointpb.ClusterLoadAssignment { clab0 := newClaBuilder("test", nil) - endpoints1 := []endpointOpts{{addrWithPort: "addr1:997", additionalAddrWithPorts: []string{"addr1:1000"}}} + endpoints1 := []endpointOpts{{addrWithPort: "addr1:997", additionalAddrWithPorts: []string{"addr1:1000"}, hostname: "addr1"}} locOption1 := &addLocalityOptions{ Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_UNHEALTHY}, Weight: []uint32{271}, } clab0.addLocality("locality-1", 1, 1, endpoints1, locOption1) - endpoints2 := []endpointOpts{{addrWithPort: "addr2:998", additionalAddrWithPorts: []string{"addr2:1000"}}} + endpoints2 := []endpointOpts{{addrWithPort: "addr2:998", additionalAddrWithPorts: []string{"addr2:1000"}, hostname: "addr2"}} locOption2 := &addLocalityOptions{ Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_HEALTHY}, Weight: []uint32{828}, @@ -317,9 +326,9 @@ func (s) TestEDSParseRespProtoAdditionalAddrs(t *testing.T) { Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:997", "addr1:1000"}, - HealthStatus: EndpointHealthStatusUnhealthy, - Weight: 271, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:997", "addr1:1000"}, "addr1"), + HealthStatus: EndpointHealthStatusUnhealthy, + Weight: 271, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 1, @@ -327,9 +336,9 @@ func (s) TestEDSParseRespProtoAdditionalAddrs(t *testing.T) { }, { Endpoints: []Endpoint{{ - Addresses: []string{"addr2:998", "addr2:1000"}, - HealthStatus: EndpointHealthStatusHealthy, - Weight: 828, + ResolverEndpoint: getResolverEndpoint([]string{"addr2:998", "addr2:1000"}, "addr2"), + HealthStatus: EndpointHealthStatusHealthy, + Weight: 828, }}, ID: clients.Locality{SubZone: "locality-2"}, Priority: 0, @@ -541,10 +550,9 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnhealthy, - Weight: 271, - Hostname: "addr1", + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnhealthy, + Weight: 271, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 1, @@ -552,10 +560,9 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { }, { Endpoints: []Endpoint{{ - Addresses: []string{"addr2:159"}, - HealthStatus: EndpointHealthStatusDraining, - Weight: 828, - Hostname: "addr2", + ResolverEndpoint: getResolverEndpoint([]string{"addr2:159"}, "addr2"), + HealthStatus: EndpointHealthStatusDraining, + Weight: 828, }}, ID: clients.Locality{SubZone: "locality-2"}, Priority: 0, @@ -574,10 +581,9 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnhealthy, - Weight: 271, - Hostname: "addr1", + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnhealthy, + Weight: 271, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 1, @@ -585,10 +591,9 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { }, { Endpoints: []Endpoint{{ - Addresses: []string{"addr2:159"}, - HealthStatus: EndpointHealthStatusDraining, - Weight: 828, - Hostname: "addr2", + ResolverEndpoint: getResolverEndpoint([]string{"addr2:159"}, "addr2"), + HealthStatus: EndpointHealthStatusDraining, + Weight: 828, }}, ID: clients.Locality{SubZone: "locality-2"}, Priority: 0, @@ -643,6 +648,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T }), }, }, + hostname: "addr1", }} clab0.addLocality("locality-1", 1, 0, endpoints, nil) return clab0.Build() @@ -651,9 +657,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, Metadata: map[string]any{ "test-key": ProxyAddressMetadataValue{ Address: "1.2.3.4:1111", @@ -684,6 +690,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T }, }, }, + hostname: "addr1", }} clab0.addLocality("locality-1", 1, 0, endpoints, nil) return clab0.Build() @@ -692,9 +699,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, Metadata: map[string]any{ "test-key": StructMetadataValue{Data: map[string]any{ "key": float64(123), @@ -712,7 +719,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T name: "typed_filter_metadata_in_locality", endpointProto: func() *v3endpointpb.ClusterLoadAssignment { clab0 := newClaBuilder("test", nil) - endpoints := []endpointOpts{{addrWithPort: "addr1:314"}} + endpoints := []endpointOpts{{addrWithPort: "addr1:314", hostname: "addr1"}} locOption := &addLocalityOptions{ Metadata: &v3corepb.Metadata{ TypedFilterMetadata: map[string]*anypb.Any{ @@ -735,9 +742,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, @@ -755,7 +762,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T name: "filter_metadata_in_locality", endpointProto: func() *v3endpointpb.ClusterLoadAssignment { clab0 := newClaBuilder("test", nil) - endpoints := []endpointOpts{{addrWithPort: "addr1:314"}} + endpoints := []endpointOpts{{addrWithPort: "addr1:314", hostname: "addr1"}} locOption := &addLocalityOptions{ Metadata: &v3corepb.Metadata{ FilterMetadata: map[string]*structpb.Struct{ @@ -776,9 +783,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, @@ -820,6 +827,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T }, }, }, + hostname: "addr1", }} clab0.addLocality("locality-1", 1, 0, endpoints, nil) return clab0.Build() @@ -828,9 +836,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, Metadata: map[string]any{ "test-key": ProxyAddressMetadataValue{ Address: "1.2.3.4:1111", @@ -848,7 +856,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T name: "typed_filter_metadata_over_filter_metadata_in_locality", endpointProto: func() *v3endpointpb.ClusterLoadAssignment { clab0 := newClaBuilder("test", nil) - endpoints := []endpointOpts{{addrWithPort: "addr1:314"}} + endpoints := []endpointOpts{{addrWithPort: "addr1:314", hostname: "addr1"}} locOption := &addLocalityOptions{ Metadata: &v3corepb.Metadata{ TypedFilterMetadata: map[string]*anypb.Any{ @@ -880,9 +888,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, @@ -924,6 +932,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T }, }, }, + hostname: "addr1", }} clab0.addLocality("locality-1", 1, 0, endpoints, nil) return clab0.Build() @@ -932,9 +941,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, Metadata: map[string]any{ "test-key": ProxyAddressMetadataValue{ Address: "1.2.3.4:1111", @@ -955,7 +964,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T name: "both_filter_and_typed_filter_metadata_in_locality", endpointProto: func() *v3endpointpb.ClusterLoadAssignment { clab0 := newClaBuilder("test", nil) - endpoints := []endpointOpts{{addrWithPort: "addr1:314"}} + endpoints := []endpointOpts{{addrWithPort: "addr1:314", hostname: "addr1"}} locOption := &addLocalityOptions{ Metadata: &v3corepb.Metadata{ TypedFilterMetadata: map[string]*anypb.Any{ @@ -987,9 +996,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, @@ -1048,6 +1057,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. }), }, }, + hostname: "addr1", }} clab0.addLocality("locality-1", 1, 0, endpoints, nil) return clab0.Build() @@ -1056,9 +1066,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, @@ -1084,6 +1094,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. }, }, }, + hostname: "addr1", }} clab0.addLocality("locality-1", 1, 0, endpoints, nil) return clab0.Build() @@ -1092,9 +1103,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, @@ -1107,7 +1118,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. name: "typed_filter_metadata_in_locality", endpointProto: func() *v3endpointpb.ClusterLoadAssignment { clab0 := newClaBuilder("test", nil) - endpoints := []endpointOpts{{addrWithPort: "addr1:314"}} + endpoints := []endpointOpts{{addrWithPort: "addr1:314", hostname: "addr1"}} locOption := &addLocalityOptions{ Metadata: &v3corepb.Metadata{ TypedFilterMetadata: map[string]*anypb.Any{ @@ -1130,9 +1141,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, @@ -1145,7 +1156,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. name: "filter_metadata_in_locality", endpointProto: func() *v3endpointpb.ClusterLoadAssignment { clab0 := newClaBuilder("test", nil) - endpoints := []endpointOpts{{addrWithPort: "addr1:314"}} + endpoints := []endpointOpts{{addrWithPort: "addr1:314", hostname: "addr1"}} locOption := &addLocalityOptions{ Metadata: &v3corepb.Metadata{ FilterMetadata: map[string]*structpb.Struct{ @@ -1166,9 +1177,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, @@ -1205,6 +1216,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. }, }, }, + hostname: "addr1", }} clab0.addLocality("locality-1", 1, 0, endpoints, nil) return clab0.Build() @@ -1213,9 +1225,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, @@ -1228,7 +1240,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. name: "both_filter_and_typed_filter_metadata_in_locality", endpointProto: func() *v3endpointpb.ClusterLoadAssignment { clab0 := newClaBuilder("test", nil) - endpoints := []endpointOpts{{addrWithPort: "addr1:314"}} + endpoints := []endpointOpts{{addrWithPort: "addr1:314", hostname: "addr1"}} locOption := &addLocalityOptions{ Metadata: &v3corepb.Metadata{ TypedFilterMetadata: map[string]*anypb.Any{ @@ -1260,9 +1272,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, @@ -1288,6 +1300,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. }), }, }, + hostname: "addr1", }} locOption := &addLocalityOptions{ Metadata: &v3corepb.Metadata{ @@ -1309,9 +1322,9 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - Addresses: []string{"addr1:314"}, - HealthStatus: EndpointHealthStatusUnknown, - Weight: 1, + ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + HealthStatus: EndpointHealthStatusUnknown, + Weight: 1, }}, ID: clients.Locality{SubZone: "locality-1"}, Priority: 0, From ee8e198f9fe338d244873e576d2e47ad7397f5ea Mon Sep 17 00:00:00 2001 From: Pranjali-2501 Date: Sun, 7 Dec 2025 15:16:35 +0000 Subject: [PATCH 2/3] fix race --- .../xds/balancer/clusterresolver/clusterresolver.go | 10 +++++++--- internal/xds/xdsclient/tests/loadreport_test.go | 2 +- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/internal/xds/balancer/clusterresolver/clusterresolver.go b/internal/xds/balancer/clusterresolver/clusterresolver.go index f5a30e1acbec..1958272611b7 100644 --- a/internal/xds/balancer/clusterresolver/clusterresolver.go +++ b/internal/xds/balancer/clusterresolver/clusterresolver.go @@ -248,8 +248,11 @@ func (b *clusterResolverBalancer) updateChildConfig() { b.logger.Infof("Built child policy config: %s", pretty.ToJSON(childCfg)) } + newEndpoints := make([]resolver.Endpoint, len(endpoints)) flattenedAddrs := make([]resolver.Address, len(endpoints)) - for i := range endpoints { + for i, e := range endpoints { + newEP := e + newEP.Addresses = make([]resolver.Address, len(e.Addresses)) for j := range endpoints[i].Addresses { addr := endpoints[i].Addresses[j] addr.BalancerAttributes = endpoints[i].Attributes @@ -268,12 +271,13 @@ func (b *clusterResolverBalancer) updateChildConfig() { // address used by the transport. This workaround can be removed once // the old pickfirst is removed. // See https://github.com/grpc/grpc-go/issues/7339 - endpoints[i].Addresses[j] = addr + newEP.Addresses[j] = addr } + newEndpoints[i] = newEP } if err := b.child.UpdateClientConnState(balancer.ClientConnState{ ResolverState: resolver.State{ - Endpoints: endpoints, + Endpoints: newEndpoints, Addresses: flattenedAddrs, ServiceConfig: b.configRaw, Attributes: b.attrsWithClient, diff --git a/internal/xds/xdsclient/tests/loadreport_test.go b/internal/xds/xdsclient/tests/loadreport_test.go index f43931c2301c..72ef2fe64ea4 100644 --- a/internal/xds/xdsclient/tests/loadreport_test.go +++ b/internal/xds/xdsclient/tests/loadreport_test.go @@ -491,7 +491,7 @@ func (s) TestConcurrentReportLoad(t *testing.T) { // TestConcurrentChannels verifies that we can create multiple gRPC channels // concurrently with a shared XDSClient, each of which will create a new LRS // stream without any race. -func (s) TestConcurrentChannels(t *testing.T) { +func TestConcurrentChannels(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) defer cancel() From 578caf37434ee5f9ef4866df6d530bac814488bb Mon Sep 17 00:00:00 2001 From: Pranjali-2501 Date: Tue, 9 Dec 2025 12:33:34 +0000 Subject: [PATCH 3/3] resolving comments --- .../clusterresolver/clusterresolver.go | 10 +-- .../balancer/clusterresolver/configbuilder.go | 18 +++-- .../clusterresolver/configbuilder_test.go | 16 ++--- .../xds/xdsclient/tests/loadreport_test.go | 2 +- .../xds/xdsclient/xdsresource/type_eds.go | 1 + .../xdsclient/xdsresource/unmarshal_eds.go | 30 ++++----- .../xdsresource/unmarshal_eds_test.go | 66 +++++++++---------- 7 files changed, 67 insertions(+), 76 deletions(-) diff --git a/internal/xds/balancer/clusterresolver/clusterresolver.go b/internal/xds/balancer/clusterresolver/clusterresolver.go index 1958272611b7..f5a30e1acbec 100644 --- a/internal/xds/balancer/clusterresolver/clusterresolver.go +++ b/internal/xds/balancer/clusterresolver/clusterresolver.go @@ -248,11 +248,8 @@ func (b *clusterResolverBalancer) updateChildConfig() { b.logger.Infof("Built child policy config: %s", pretty.ToJSON(childCfg)) } - newEndpoints := make([]resolver.Endpoint, len(endpoints)) flattenedAddrs := make([]resolver.Address, len(endpoints)) - for i, e := range endpoints { - newEP := e - newEP.Addresses = make([]resolver.Address, len(e.Addresses)) + for i := range endpoints { for j := range endpoints[i].Addresses { addr := endpoints[i].Addresses[j] addr.BalancerAttributes = endpoints[i].Attributes @@ -271,13 +268,12 @@ func (b *clusterResolverBalancer) updateChildConfig() { // address used by the transport. This workaround can be removed once // the old pickfirst is removed. // See https://github.com/grpc/grpc-go/issues/7339 - newEP.Addresses[j] = addr + endpoints[i].Addresses[j] = addr } - newEndpoints[i] = newEP } if err := b.child.UpdateClientConnState(balancer.ClientConnState{ ResolverState: resolver.State{ - Endpoints: newEndpoints, + Endpoints: endpoints, Addresses: flattenedAddrs, ServiceConfig: b.configRaw, Attributes: b.attrsWithClient, diff --git a/internal/xds/balancer/clusterresolver/configbuilder.go b/internal/xds/balancer/clusterresolver/configbuilder.go index 77a3f40b01c6..2037f9f71693 100644 --- a/internal/xds/balancer/clusterresolver/configbuilder.go +++ b/internal/xds/balancer/clusterresolver/configbuilder.go @@ -265,21 +265,27 @@ func priorityLocalitiesToClusterImpl(localities []xdsresource.Locality, priority if endpoint.HealthStatus != xdsresource.EndpointHealthStatusHealthy && endpoint.HealthStatus != xdsresource.EndpointHealthStatusUnknown { continue } - endpoint.ResolverEndpoint = hierarchy.SetInEndpoint(endpoint.ResolverEndpoint, []string{priorityName, localityStr}) - endpoint.ResolverEndpoint = xdsinternal.SetLocalityIDInEndpoint(endpoint.ResolverEndpoint, locality.ID) + + // Create a copy of endpoint.ResolverEndpoint to avoid race. + resolverEndpoint := endpoint.ResolverEndpoint + resolverEndpoint.Addresses = make([]resolver.Address, len(endpoint.ResolverEndpoint.Addresses)) + copy(resolverEndpoint.Addresses, endpoint.ResolverEndpoint.Addresses) + + resolverEndpoint = hierarchy.SetInEndpoint(resolverEndpoint, []string{priorityName, localityStr}) + resolverEndpoint = xdsinternal.SetLocalityIDInEndpoint(resolverEndpoint, locality.ID) // "To provide the xds_wrr_locality load balancer information about // locality weights received from EDS, the cluster resolver will // populate a new locality weight attribute for each address The // attribute will have the weight (as an integer) of the locality // the address is part of." - A52 - endpoint.ResolverEndpoint = wrrlocality.SetAddrInfoInEndpoint(endpoint.ResolverEndpoint, wrrlocality.AddrInfo{LocalityWeight: lw}) + resolverEndpoint = wrrlocality.SetAddrInfoInEndpoint(resolverEndpoint, wrrlocality.AddrInfo{LocalityWeight: lw}) var ew uint32 = 1 if endpoint.Weight != 0 { ew = endpoint.Weight } - endpoint.ResolverEndpoint = weight.Set(endpoint.ResolverEndpoint, weight.EndpointInfo{Weight: lw * ew}) - endpoint.ResolverEndpoint = ringhash.SetHashKey(endpoint.ResolverEndpoint, endpoint.HashKey) - retEndpoints = append(retEndpoints, endpoint.ResolverEndpoint) + resolverEndpoint = weight.Set(resolverEndpoint, weight.EndpointInfo{Weight: lw * ew}) + resolverEndpoint = ringhash.SetHashKey(resolverEndpoint, endpoint.HashKey) + retEndpoints = append(retEndpoints, resolverEndpoint) } } return &clusterimpl.LBConfig{ diff --git a/internal/xds/balancer/clusterresolver/configbuilder_test.go b/internal/xds/balancer/clusterresolver/configbuilder_test.go index 5107e7587bd0..af9d44422acc 100644 --- a/internal/xds/balancer/clusterresolver/configbuilder_test.go +++ b/internal/xds/balancer/clusterresolver/configbuilder_test.go @@ -550,18 +550,14 @@ func TestPriorityLocalitiesToClusterImpl(t *testing.T) { { Endpoints: []xdsresource.Endpoint{ { - ResolverEndpoint: resolver.Endpoint{ - Addresses: []resolver.Address{{Addr: "addr-1-1"}}, - }, - Weight: 90, - HealthStatus: xdsresource.EndpointHealthStatusHealthy, + ResolverEndpoint: resolver.Endpoint{Addresses: []resolver.Address{{Addr: "addr-1-1"}}}, + Weight: 90, + HealthStatus: xdsresource.EndpointHealthStatusHealthy, }, { - ResolverEndpoint: resolver.Endpoint{ - Addresses: []resolver.Address{{Addr: "addr-1-2"}}, - }, - Weight: 10, - HealthStatus: xdsresource.EndpointHealthStatusHealthy, + ResolverEndpoint: resolver.Endpoint{Addresses: []resolver.Address{{Addr: "addr-1-2"}}}, + Weight: 10, + HealthStatus: xdsresource.EndpointHealthStatusHealthy, }, }, ID: clients.Locality{Zone: "test-zone-1"}, diff --git a/internal/xds/xdsclient/tests/loadreport_test.go b/internal/xds/xdsclient/tests/loadreport_test.go index 72ef2fe64ea4..f43931c2301c 100644 --- a/internal/xds/xdsclient/tests/loadreport_test.go +++ b/internal/xds/xdsclient/tests/loadreport_test.go @@ -491,7 +491,7 @@ func (s) TestConcurrentReportLoad(t *testing.T) { // TestConcurrentChannels verifies that we can create multiple gRPC channels // concurrently with a shared XDSClient, each of which will create a new LRS // stream without any race. -func TestConcurrentChannels(t *testing.T) { +func (s) TestConcurrentChannels(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout) defer cancel() diff --git a/internal/xds/xdsclient/xdsresource/type_eds.go b/internal/xds/xdsclient/xdsresource/type_eds.go index 830b976f4a9e..c6dc0c096c19 100644 --- a/internal/xds/xdsclient/xdsresource/type_eds.go +++ b/internal/xds/xdsclient/xdsresource/type_eds.go @@ -49,6 +49,7 @@ const ( ) // Endpoint contains information of an endpoint. +// TODO(i/8757) : Replace Endpoint with resolver.Endpoint struct. type Endpoint struct { ResolverEndpoint resolver.Endpoint HealthStatus EndpointHealthStatus diff --git a/internal/xds/xdsclient/xdsresource/unmarshal_eds.go b/internal/xds/xdsclient/xdsresource/unmarshal_eds.go index 8a1f2b71d43d..93b391911589 100644 --- a/internal/xds/xdsclient/xdsresource/unmarshal_eds.go +++ b/internal/xds/xdsclient/xdsresource/unmarshal_eds.go @@ -35,25 +35,25 @@ import ( "google.golang.org/protobuf/types/known/anypb" ) -type hostnameType string +// hostnameKeyType is the key to store the hostname attribute in +// a resolver.Endpoint. +type hostnameKeyType struct{} -// hostnameKey is the key to store the hostname key attribute in -// a resolver.Endpoint attribute. -const hostnameKey = hostnameType("grpc.resolver.xdsresource.hostname") - -// SetHostname sets the hostname key for this resolver.Endpoint attribute. -func SetHostname(endpoint resolver.Endpoint, hostname string) resolver.Endpoint { +// setHostname returns a copy of the given endpoint with hostname added +// as an attribute. +func setHostname(endpoint resolver.Endpoint, hostname string) resolver.Endpoint { + // Only set if non-empty; xds_cluster_impl uses this to trigger :authority rewriting. if hostname == "" { return endpoint } - endpoint.Attributes = endpoint.Attributes.WithValue(hostnameKey, hostname) + endpoint.Attributes = endpoint.Attributes.WithValue(hostnameKeyType{}, hostname) return endpoint } -// GetHostname returns the hostname key attribute of endpoint. If this attribute -// is not set, it returns the empty string. -func GetHostname(endpoint resolver.Endpoint) string { - hostname, _ := endpoint.Attributes.Value(hostnameKey).(string) +// HostnameFromEndpoint returns the hostname attribute of endpoint. If this +// attribute is not set, it returns the empty string. +func HostnameFromEndpoint(endpoint resolver.Endpoint) string { + hostname, _ := endpoint.Attributes.Value(hostnameKeyType{}).(string) return hostname } @@ -151,10 +151,8 @@ func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint, uniqueEndpointAddrs hashKey = hashKeyFromMetadata(endpointMetadata) } } - endpoint := resolver.Endpoint{ - Addresses: address, - } - endpoint = SetHostname(endpoint, lbEndpoint.GetEndpoint().GetHostname()) + endpoint := resolver.Endpoint{Addresses: address} + endpoint = setHostname(endpoint, lbEndpoint.GetEndpoint().GetHostname()) endpoints = append(endpoints, Endpoint{ ResolverEndpoint: endpoint, HealthStatus: EndpointHealthStatus(lbEndpoint.GetHealthStatus()), diff --git a/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go b/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go index 33431a9a93f1..2a8db8a7aa33 100644 --- a/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go +++ b/internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go @@ -41,15 +41,13 @@ import ( "google.golang.org/protobuf/types/known/wrapperspb" ) -func getResolverEndpoint(addr []string, hostname string) resolver.Endpoint { +func buildResolverEndpoint(addr []string, hostname string) resolver.Endpoint { address := []resolver.Address{} for _, a := range addr { address = append(address, resolver.Address{Addr: a}) } - resolverEndpoint := resolver.Endpoint{ - Addresses: address, - } - resolverEndpoint = SetHostname(resolverEndpoint, hostname) + resolverEndpoint := resolver.Endpoint{Addresses: address} + resolverEndpoint = setHostname(resolverEndpoint, hostname) return resolverEndpoint } @@ -107,14 +105,10 @@ func (s) TestEDSParseRespProto(t *testing.T) { m: func() *v3endpointpb.ClusterLoadAssignment { clab0 := newClaBuilder("test", nil) endpoints1 := []endpointOpts{{addrWithPort: "addr1:314"}} - locOption1 := &addLocalityOptions{ - Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_HEALTHY}, - } + locOption1 := &addLocalityOptions{Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_HEALTHY}} clab0.addLocality("locality-1", 0, 1, endpoints1, locOption1) endpoints2 := []endpointOpts{{addrWithPort: "addr2:159"}} - locOption2 := &addLocalityOptions{ - Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_HEALTHY}, - } + locOption2 := &addLocalityOptions{Health: []v3corepb.HealthStatus{v3corepb.HealthStatus_HEALTHY}} clab0.addLocality("locality-2", 0, 0, endpoints2, locOption2) return clab0.Build() }(), @@ -166,7 +160,7 @@ func (s) TestEDSParseRespProto(t *testing.T) { Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnhealthy, Weight: 271, }}, @@ -176,7 +170,7 @@ func (s) TestEDSParseRespProto(t *testing.T) { }, { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr2:159"}, "addr2"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr2:159"}, "addr2"), HealthStatus: EndpointHealthStatusDraining, Weight: 828, }}, @@ -212,7 +206,7 @@ func (s) TestEDSParseRespProto(t *testing.T) { Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnhealthy, Weight: 271, }}, @@ -222,7 +216,7 @@ func (s) TestEDSParseRespProto(t *testing.T) { }, { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr2:159"}, "addr2"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr2:159"}, "addr2"), HealthStatus: EndpointHealthStatusDraining, Weight: 828, }}, @@ -326,7 +320,7 @@ func (s) TestEDSParseRespProtoAdditionalAddrs(t *testing.T) { Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:997", "addr1:1000"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:997", "addr1:1000"}, "addr1"), HealthStatus: EndpointHealthStatusUnhealthy, Weight: 271, }}, @@ -336,7 +330,7 @@ func (s) TestEDSParseRespProtoAdditionalAddrs(t *testing.T) { }, { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr2:998", "addr2:1000"}, "addr2"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr2:998", "addr2:1000"}, "addr2"), HealthStatus: EndpointHealthStatusHealthy, Weight: 828, }}, @@ -550,7 +544,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnhealthy, Weight: 271, }}, @@ -560,7 +554,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { }, { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr2:159"}, "addr2"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr2:159"}, "addr2"), HealthStatus: EndpointHealthStatusDraining, Weight: 828, }}, @@ -581,7 +575,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnhealthy, Weight: 271, }}, @@ -591,7 +585,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) { }, { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr2:159"}, "addr2"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr2:159"}, "addr2"), HealthStatus: EndpointHealthStatusDraining, Weight: 828, }}, @@ -657,7 +651,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, Metadata: map[string]any{ @@ -699,7 +693,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, Metadata: map[string]any{ @@ -742,7 +736,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, }}, @@ -783,7 +777,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, }}, @@ -836,7 +830,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, Metadata: map[string]any{ @@ -888,7 +882,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, }}, @@ -941,7 +935,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, Metadata: map[string]any{ @@ -996,7 +990,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, }}, @@ -1066,7 +1060,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, }}, @@ -1103,7 +1097,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, }}, @@ -1141,7 +1135,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, }}, @@ -1177,7 +1171,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, }}, @@ -1225,7 +1219,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, }}, @@ -1272,7 +1266,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, }}, @@ -1322,7 +1316,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing. Localities: []Locality{ { Endpoints: []Endpoint{{ - ResolverEndpoint: getResolverEndpoint([]string{"addr1:314"}, "addr1"), + ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"), HealthStatus: EndpointHealthStatusUnknown, Weight: 1, }},