xds: decouple A76 and A86 EDS metadata parsing

internal/xds/xdsclient/xdsresource/metadata.go

@@ -22,11 +22,14 @@ import (
 	"net/netip"
 
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	"google.golang.org/grpc/internal/envconfig"
 	"google.golang.org/protobuf/types/known/anypb"
 )
 
 func init() {
-	registerMetadataConverter("type.googleapis.com/envoy.config.core.v3.Address", proxyAddressConvertor{})
+	if envconfig.XDSHTTPConnectEnabled {
+		registerMetadataConverter("type.googleapis.com/envoy.config.core.v3.Address", proxyAddressConvertor{})
+	}
 }
 
 var (
@@ -53,6 +56,12 @@ func metadataConverterForType(typeURL string) metadataConverter {
 	return metdataRegistry[typeURL]
 }
 
+// unregisterMetadataConverterForTesting removes a converter from the registry.
+// For testing only.
+func unregisterMetadataConverterForTesting(typeURL string) {
+	delete(metdataRegistry, typeURL)
+}
+
 // StructMetadataValue stores the values in a google.protobuf.Struct from
 // FilterMetadata.
 type StructMetadataValue struct {

internal/xds/xdsclient/xdsresource/metadata_test.go

@@ -26,7 +26,15 @@ import (
 
 const proxyAddressTypeURL = "type.googleapis.com/envoy.config.core.v3.Address"
 
+func setupProxyAddressConverter(t *testing.T) {
+	registerMetadataConverter(proxyAddressTypeURL, proxyAddressConvertor{})
+	t.Cleanup(func() {
+		unregisterMetadataConverterForTesting(proxyAddressTypeURL)
+	})
+}
+
 func (s) TestProxyAddressConverterSuccess(t *testing.T) {
+	setupProxyAddressConverter(t)
 	converter := metadataConverterForType(proxyAddressTypeURL)
 	if converter == nil {
 		t.Fatalf("Converter for %q not found in registry", proxyAddressTypeURL)
@@ -133,6 +141,7 @@ func (s) TestProxyAddressConverterSuccess(t *testing.T) {
 }
 
 func (s) TestProxyAddressConverterFailure(t *testing.T) {
+	setupProxyAddressConverter(t)
 	converter := metadataConverterForType(proxyAddressTypeURL)
 	if converter == nil {
 		t.Fatalf("Converter for %q not found in registry", proxyAddressTypeURL)

internal/xds/xdsclient/xdsresource/unmarshal_eds.go

@@ -278,7 +278,7 @@ func validateAndConstructMetadata(metadataProto *v3corepb.Metadata) (map[string]
 
 	// Process FilterMetadata for any keys not already handled.
 	for key, structProto := range metadataProto.GetFilterMetadata() {
-		// Skip keys already added from TyperFilterMetadata.
+		// Skip keys already added from TypedFilterMetadata.
 		if metadata[key] != nil {
 			continue
 		}

internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go

@@ -42,6 +42,25 @@ import (
 	"google.golang.org/protobuf/types/known/wrapperspb"
 )
 
+// enableA86 enables A86 support for the duration of the test by:
+// 1. Setting the GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT environment variable
+// 2. Registering the proxy address converter, since this is otherwise done in init.
+func enableA86(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
+	registerMetadataConverter(proxyAddressTypeURL, proxyAddressConvertor{})
+	t.Cleanup(func() {
+		unregisterMetadataConverterForTesting(proxyAddressTypeURL)
+	})
+}
+
+// disableA86 disables A86 support for the duration of the test by:
+// 1. Setting the GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT environment variable to false
+// 2. Unregistering the proxy address converter (in case it was registered by init or previous test)
+func disableA86(t *testing.T) {
+	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, false)
+	unregisterMetadataConverterForTesting(proxyAddressTypeURL)
+}
+
 func buildResolverEndpoint(addr []string, hostname string) resolver.Endpoint {
 	address := []resolver.Address{}
 	for _, a := range addr {
@@ -487,7 +506,7 @@ func (s) TestUnmarshalEndpointHashKey(t *testing.T) {
 }
 
 func (s) TestUnmarshalEndpoints(t *testing.T) {
-	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
+	enableA86(t)
 	var v3EndpointsAny = testutils.MarshalAny(t, func() *v3endpointpb.ClusterLoadAssignment {
 		clab0 := newClaBuilder("test", nil)
 		endpoints1 := []endpointOpts{{addrWithPort: "addr1:314", hostname: "addr1"}}
@@ -618,7 +637,7 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 // Tests custom metadata parsing for success cases when the
 // GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT environment variable is set.
 func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T) {
-	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
+	enableA86(t)
 	tests := []struct {
 		name          string
 		endpointProto *v3endpointpb.ClusterLoadAssignment
@@ -1025,9 +1044,11 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOn(t *testing.T
 }
 
 // Tests custom metadata parsing for success cases when the
-// GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT environment variable is not set.
+// GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT environment variable is not set and
+// GRPC_XDS_ENDPOINT_HASH_KEY_BACKWARD_COMPAT is set to true (disabling A76).
 func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing.T) {
-	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, false)
+	disableA86(t)
+	testutils.SetEnvConfig(t, &envconfig.XDSEndpointHashKeyBackwardCompat, true)
 	tests := []struct {
 		name          string
 		endpointProto *v3endpointpb.ClusterLoadAssignment
@@ -1345,7 +1366,7 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_EnvVarOff(t *testing.
 // Tests custom metadata parsing for converter failure cases when the
 // GRPC_EXPERIMENTAL_XDS_HTTP_CONNECT environment variable is set.
 func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_ConverterFailure(t *testing.T) {
-	testutils.SetEnvConfig(t, &envconfig.XDSHTTPConnectEnabled, true)
+	enableA86(t)
 	tests := []struct {
 		name          string
 		endpointProto *v3endpointpb.ClusterLoadAssignment
@@ -1404,6 +1425,128 @@ func (s) TestEDSParseRespProto_HTTP_Connect_CustomMetadata_ConverterFailure(t *t
 	}
 }
 
+// Tests metadata parsing when HTTP Connect is enabled but A76 hash key is
+// disabled (backward compat mode). This verifies that:
+// - Metadata parsing happens (TypedFilterMetadata + FilterMetadata)
+// - Hash key is NOT extracted from envoy.lb
+func (s) TestEDSParseRespProto_HTTP_Connect_On_HashKeyBackwardCompat_On(t *testing.T) {
+	enableA86(t)
+	testutils.SetEnvConfig(t, &envconfig.XDSEndpointHashKeyBackwardCompat, true)
+
+	clab0 := newClaBuilder("test", nil)
+	endpoints := []endpointOpts{{
+		addrWithPort: "addr1:314",
+		metadata: &v3corepb.Metadata{
+			TypedFilterMetadata: map[string]*anypb.Any{
+				"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+					Address: &v3corepb.Address_SocketAddress{
+						SocketAddress: &v3corepb.SocketAddress{
+							Address: "1.2.3.4",
+							PortSpecifier: &v3corepb.SocketAddress_PortValue{
+								PortValue: 1111,
+							},
+						},
+					},
+				}),
+			},
+			FilterMetadata: map[string]*structpb.Struct{
+				"envoy.lb": {
+					Fields: map[string]*structpb.Value{
+						"hash_key": {
+							Kind: &structpb.Value_StringValue{StringValue: "test-hash-key"},
+						},
+					},
+				},
+			},
+		},
+		hostname: "addr1",
+	}}
+	clab0.addLocality("locality-1", 1, 0, endpoints, nil)
+
+	got, err := parseEDSRespProto(clab0.Build())
+	if err != nil {
+		t.Fatalf("parseEDSRespProto() failed: %v", err)
+	}
+
+	wantEndpoint := EndpointsUpdate{
+		Localities: []Locality{
+			{
+				Endpoints: []Endpoint{{
+					ResolverEndpoint: buildResolverEndpoint([]string{"addr1:314"}, "addr1"),
+					HealthStatus:     EndpointHealthStatusUnknown,
+					Weight:           1,
+					Metadata: map[string]any{
+						"envoy.http11_proxy_transport_socket.proxy_address": ProxyAddressMetadataValue{
+							Address: "1.2.3.4:1111",
+						},
+						"envoy.lb": StructMetadataValue{Data: map[string]any{
+							"hash_key": "test-hash-key",
+						}},
+					},
+				}},
+				ID:       clients.Locality{SubZone: "locality-1"},
+				Priority: 0,
+				Weight:   1,
+			},
+		},
+	}
+
+	if diff := cmp.Diff(wantEndpoint, got, cmpopts.EquateEmpty()); diff != "" {
+		t.Errorf("parseEDSRespProto() returned unexpected diff (-want +got):\n%s", diff)
+	}
+
+	// Verify hash key is NOT extracted when backward compat is on.
+	hashKey := ringhash.HashKey(got.Localities[0].Endpoints[0].ResolverEndpoint)
+	if hashKey != "" {
+		t.Errorf("Expected empty hash key with backward compat on, got %q", hashKey)
+	}
+}
+
+// Tests that when A76 is enabled but A86 is disabled, invalid typed metadata
+// does not cause a parsing failure, and hash key is still extracted.
+func (s) TestEDSParseRespProto_HTTP_Connect_Off_HashKeyBackwardCompat_Off_InvalidTypedMetadata(t *testing.T) {
+	disableA86(t)
+	testutils.SetEnvConfig(t, &envconfig.XDSEndpointHashKeyBackwardCompat, false) // A76 on
+
+	clab0 := newClaBuilder("test", nil)
+	endpoints := []endpointOpts{{
+		addrWithPort: "addr1:314",
+		metadata: &v3corepb.Metadata{
+			TypedFilterMetadata: map[string]*anypb.Any{
+				"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
+					Address: &v3corepb.Address_SocketAddress{
+						SocketAddress: &v3corepb.SocketAddress{
+							Address: "invalid", // This would fail conversion if processed.
+						},
+					},
+				}),
+			},
+			FilterMetadata: map[string]*structpb.Struct{
+				"envoy.lb": {
+					Fields: map[string]*structpb.Value{
+						"hash_key": {
+							Kind: &structpb.Value_StringValue{StringValue: "test-hash-key"},
+						},
+					},
+				},
+			},
+		},
+		hostname: "addr1",
+	}}
+	clab0.addLocality("locality-1", 1, 0, endpoints, nil)
+
+	got, err := parseEDSRespProto(clab0.Build())
+	if err != nil {
+		t.Fatalf("parseEDSRespProto() failed unexpectedly with A76 on and A86 off: %v", err)
+	}
+
+	// Verify hash key is extracted when A76 is on.
+	hashKey := ringhash.HashKey(got.Localities[0].Endpoints[0].ResolverEndpoint)
+	if want := "test-hash-key"; hashKey != want {
+		t.Errorf("Expected hash key %q with A76 on, got %q", want, hashKey)
+	}
+}
+
 // claBuilder builds a ClusterLoadAssignment, aka EDS
 // response.
 type claBuilder struct {