address comments

Author: cjqzhao

internal/xds/xdsclient/xdsresource/metadata.go

@@ -27,35 +27,39 @@ import (
 )
 
 func init() {
-	Register("envoy.http11_proxy_transport_socket.proxy_address", ProxyAddressConvertor{})
+	Register("envoy.http11_proxy_transport_socket.proxy_address", proxyAddressConvertor{})
 }
 
 var (
 	// metadataregistry is a map from proto type to Converter.
-	metadataregistry = make(map[string]Converter)
+	metadataregistry = make(map[string]converter)
 )
 
 // MetadataValue is the interface for a converted metadata value. It is
 // implemented by concrete types that hold the converted metadata.
 type MetadataValue interface {
+	// Type returns the type.
 	Type() string
 }
 
-// Converter is the interface for a metadata converter. It is implemented by
+// converter is the interface for a metadata converter. It is implemented by
 // concrete types that convert raw bytes into a MetadataValue.
-type Converter interface {
-	// Convert parses the raw bytes of an Any proto into a MetadataValue.
-	Convert([]byte) (MetadataValue, error)
+type converter interface {
+	// convert parses the proto serialized bytes of an Any proto or
+	// google.protobuf.Struct into a MetadataValue. The bytes of an Any proto
+	// are from the value field, which has proto serialized bytes of the
+	// protobuf message type specified by the type_url field.
+	convert([]byte) (MetadataValue, error)
 }
 
 // Register registers the converter to the map keyed on a proto type. Must be
 // called at init time. Not thread safe.
-func Register(protoType string, c Converter) {
+func Register(protoType string, c converter) {
 	metadataregistry[protoType] = c
 }
 
 // ConverterForType retrieves a converter based on key given.
-func ConverterForType(typeURL string) Converter {
+func ConverterForType(typeURL string) converter {
 	return metadataregistry[typeURL]
 }
 
@@ -66,9 +70,11 @@ type JSONMetadata struct {
 	Data json.RawMessage
 }
 
-// ProxyAddressMetadataValue holds the address parsed from A.86 metadata.
+// ProxyAddressMetadataValue holds the address parsed from the
+// envoy.config.core.v3.Address proto message, as specified in gRFC A86.
 type ProxyAddressMetadataValue struct {
-	MetadataValue
+	// Address stores the proxy address configured (A86). It will be in the form
+	// of host:port. It has to be either IPv6 or IPv4.
 	Address string
 }
 
@@ -78,11 +84,11 @@ func (ProxyAddressMetadataValue) Type() string {
 }
 
 // ProxyAddressConvertor implements the converter for A86 (Proxy Address) metadata.
-type ProxyAddressConvertor struct{}
+type proxyAddressConvertor struct{}
 
-// Convert parses the raw bytes of an Any proto containing an Address proto into
-// a ProxyAddressMetadataValue.
-func (ProxyAddressConvertor) Convert(anyBytes []byte) (MetadataValue, error) {
+// Convert parses the bytes from the value field of an Any proto containing an
+// Address proto into a ProxyAddressMetadataValue.
+func (proxyAddressConvertor) convert(anyBytes []byte) (MetadataValue, error) {
 	addressProto := &v3corepb.Address{}
 	if err := proto.Unmarshal(anyBytes, addressProto); err != nil {
 		return nil, fmt.Errorf("failed to unmarshal resource: %v", err)
@@ -91,15 +97,16 @@ func (ProxyAddressConvertor) Convert(anyBytes []byte) (MetadataValue, error) {
 	if socketaddress == nil {
 		return nil, fmt.Errorf("no socket_address field in metadata")
 	}
+	if _, err := netip.ParseAddr(socketaddress.GetAddress()); err != nil {
+		return nil, fmt.Errorf("address field is not a valid IPv4 or IPv6 address: %q", socketaddress.GetAddress())
+	}
 	portvalue := socketaddress.GetPortValue()
 	if portvalue == 0 {
 		return nil, fmt.Errorf("port value not set in socket_address")
 	}
-	if _, err := netip.ParseAddr(socketaddress.GetAddress()); err != nil {
-		return nil, fmt.Errorf("address field is not a valid IPv4 or IPv6 address: %q", socketaddress.GetAddress())
-	}
+
 	metadata := ProxyAddressMetadataValue{
-		Address: socketaddress.Address,
+		Address: parseAddress(socketaddress),
 	}
 	return metadata, nil
 }

internal/xds/xdsclient/xdsresource/metadata_test.go

@@ -21,7 +21,7 @@ import (
 
 	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
 	"github.com/google/go-cmp/cmp"
-	"google.golang.org/protobuf/proto"
+	"google.golang.org/grpc/internal/testutils"
 )
 
 const proxyAddressFilterName = "envoy.http11_proxy_transport_socket.proxy_address"
@@ -49,7 +49,7 @@ func (s) TestProxyAddressConverterSuccess(t *testing.T) {
 				},
 			},
 			want: ProxyAddressMetadataValue{
-				Address: "192.168.1.1",
+				Address: "192.168.1.1:8080",
 			},
 		},
 		{
@@ -65,23 +65,20 @@ func (s) TestProxyAddressConverterSuccess(t *testing.T) {
 				},
 			},
 			want: ProxyAddressMetadataValue{
-				Address: "2001:0db8:85a3:0000:0000:8a2e:0370:7334",
+				Address: "[2001:0db8:85a3:0000:0000:8a2e:0370:7334]:9090",
 			},
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			anyBytes, err := proto.Marshal(tt.addr)
-			if err != nil {
-				t.Fatalf("Failed to marshal address proto: %v", err)
-			}
-			got, err := converter.Convert(anyBytes)
+			anyProto := testutils.MarshalAny(t, tt.addr)
+			got, err := converter.convert(anyProto.GetValue())
 			if err != nil {
 				t.Fatalf("convert() failed with error: %v", err)
 			}
 			if diff := cmp.Diff(tt.want, got, cmp.AllowUnexported(ProxyAddressMetadataValue{})); diff != "" {
-				t.Errorf("convert() returned unexpected value:\n%s", diff)
+				t.Errorf("convert() returned unexpected value (-want +got):\n%s", diff)
 			}
 		})
 	}
@@ -103,9 +100,6 @@ func (s) TestProxyAddressConverterFailure(t *testing.T) {
 				Address: &v3corepb.Address_SocketAddress{
 					SocketAddress: &v3corepb.SocketAddress{
 						Address: "invalid-ip",
-						PortSpecifier: &v3corepb.SocketAddress_PortValue{
-							PortValue: 8080,
-						},
 					},
 				},
 			},
@@ -147,15 +141,10 @@ func (s) TestProxyAddressConverterFailure(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			anyBytes, err := proto.Marshal(tt.addr)
-			if err != nil {
-				t.Fatalf("Failed to marshal address proto: %v", err)
-			}
-
-			// Call the convert function and check the returned error.
-			_, gotErr := converter.Convert(anyBytes)
-			if gotErr == nil || gotErr.Error() != tt.wantErr {
-				t.Errorf("convert() got error = %v, wantErr = %q", gotErr, tt.wantErr)
+			anyProto := testutils.MarshalAny(t, tt.addr)
+			_, err := converter.convert(anyProto.GetValue())
+			if err == nil || err.Error() != tt.wantErr {
+				t.Errorf("convert() got error = %v, wantErr = %q", err, tt.wantErr)
 			}
 		})
 	}

internal/xds/xdsclient/xdsresource/type_eds.go

@@ -53,6 +53,7 @@ type Endpoint struct {
 	HealthStatus EndpointHealthStatus
 	Weight       uint32
 	HashKey      string
+	Metadata     *Metadata
 }
 
 // Locality contains information of a locality.
@@ -61,6 +62,7 @@ type Locality struct {
 	ID        clients.Locality
 	Priority  uint32
 	Weight    uint32
+	Metadata  *Metadata
 }
 
 // EndpointsUpdate contains an EDS update.

internal/xds/xdsclient/xdsresource/unmarshal_eds.go

@@ -110,11 +110,20 @@ func parseEndpoints(lbEndpoints []*v3endpointpb.LbEndpoint, uniqueEndpointAddrs
 			}
 			uniqueEndpointAddrs[a] = true
 		}
+		var endpointMetadata *Metadata
+		if md := lbEndpoint.GetMetadata(); md != nil {
+			m, err := validateAndConstructMetadata(md)
+			if err != nil {
+				return nil, err
+			}
+			endpointMetadata = &m
+		}
 		endpoints = append(endpoints, Endpoint{
 			HealthStatus: EndpointHealthStatus(lbEndpoint.GetHealthStatus()),
 			Addresses:    addrs,
 			Weight:       weight,
 			HashKey:      hashKey(lbEndpoint),
+			Metadata:     endpointMetadata,
 		})
 	}
 	return endpoints, nil
@@ -192,11 +201,20 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment) (EndpointsUpdate,
 		if err != nil {
 			return EndpointsUpdate{}, err
 		}
+		var localityMetadata *Metadata
+		if md := locality.GetMetadata(); md != nil {
+			m, err := validateAndConstructMetadata(md)
+			if err != nil {
+				return EndpointsUpdate{}, err
+			}
+			localityMetadata = &m
+		}
 		ret.Localities = append(ret.Localities, Locality{
 			ID:        lid,
 			Endpoints: endpoints,
 			Weight:    weight,
 			Priority:  priority,
+			Metadata:  localityMetadata,
 		})
 	}
 	for i := 0; i < len(priorities); i++ {
@@ -209,30 +227,25 @@ func parseEDSRespProto(m *v3endpointpb.ClusterLoadAssignment) (EndpointsUpdate,
 
 func validateAndConstructMetadata(metadataProto *v3corepb.Metadata) (Metadata, error) {
 	metadata := make(map[string]MetadataValue)
-	if metadataProto == nil {
-		return Metadata{Metadata: metadata}, nil
-	}
 	// First go through TypedFilterMetadata.
 	for key, anyProto := range metadataProto.GetTypedFilterMetadata() {
-		converter := ConverterForType(anyProto.GetTypeUrl())
+		converter := ConverterForType(key)
 		// Ignore types we don't have a converter for.
 		if converter == nil {
 			continue
 		}
-
-		val, err := converter.Convert(anyProto.GetValue())
+		val, err := converter.convert(anyProto.GetValue())
 		if err != nil {
 			// If the converter fails, nack the whole resource.
-			return Metadata{}, fmt.Errorf("metadata parser for key %q and type %q failed: %v", key, anyProto.GetTypeUrl(), err)
+			return Metadata{}, fmt.Errorf("metadata converting for key %q and type %q failed: %v", key, anyProto.GetTypeUrl(), err)
 		}
 		metadata[key] = val
 	}
 
 	// Process FilterMetadata for any keys not already handled.
 	for key, structProto := range metadataProto.GetFilterMetadata() {
-		_, exists := metadata[key]
 		// Skip keys already added from TyperFilterMetadata.
-		if exists {
+		if metadata[key] != nil {
 			continue
 		}
 		b, err := protojson.Marshal(structProto)

internal/xds/xdsclient/xdsresource/unmarshal_eds_test.go

@@ -590,21 +590,6 @@ func (s) TestUnmarshalEndpoints(t *testing.T) {
 }
 
 func (s) TestValidateAndConstructMetadata(t *testing.T) {
-	originalRegistry := metadataregistry
-	// Restore the original registry.
-	defer func() {
-		metadataregistry = originalRegistry
-	}()
-	newRegistry := make(map[string]Converter)
-	metadataregistry = newRegistry
-
-	// Register a mock converter for this test. The key is the type URL for an
-	// envoy.config.core.v3.Address proto, which will be used in the success case.
-	Register("type.googleapis.com/envoy.config.core.v3.Address", testMetadataValueConvertor{})
-	metadataMap := map[string]MetadataValue{
-		"type.googleapis.com/envoy.config.core.v3.Address": testMetadataValue{Address: "1.2.3.4"},
-		"untyped-key": JSONMetadata{Data: json.RawMessage(`{"field":"value"}`)},
-	}
 	tests := []struct {
 		name          string
 		metadataProto *v3corepb.Metadata
@@ -615,23 +600,28 @@ func (s) TestValidateAndConstructMetadata(t *testing.T) {
 			name: "success-case",
 			metadataProto: &v3corepb.Metadata{
 				TypedFilterMetadata: map[string]*anypb.Any{
-					"type.googleapis.com/envoy.config.core.v3.Address": testutils.MarshalAny(t, &v3corepb.Address{
+					"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
 						Address: &v3corepb.Address_SocketAddress{
-							SocketAddress: &v3corepb.SocketAddress{Address: "1.2.3.4"},
+							SocketAddress: &v3corepb.SocketAddress{Address: "1.2.3.4", PortSpecifier: &v3corepb.SocketAddress_PortValue{
+								PortValue: 8080,
+							}},
 						},
 					}),
 				},
 				FilterMetadata: map[string]*structpb.Struct{
 					"untyped-key": {Fields: map[string]*structpb.Value{"field": structpb.NewStringValue("value")}},
 				},
 			},
-			want: Metadata{Metadata: metadataMap},
+			want: Metadata{Metadata: map[string]MetadataValue{
+				"envoy.http11_proxy_transport_socket.proxy_address": ProxyAddressMetadataValue{Address: "1.2.3.4:8080"},
+				"untyped-key": JSONMetadata{Data: json.RawMessage(`{"field":"value"}`)},
+			}},
 		},
 		{
 			name: "failure-case-converter-error",
 			metadataProto: &v3corepb.Metadata{
 				TypedFilterMetadata: map[string]*anypb.Any{
-					"type.googleapis.com/envoy.config.core.v3.Address": testutils.MarshalAny(t, &v3corepb.Address{
+					"envoy.http11_proxy_transport_socket.proxy_address": testutils.MarshalAny(t, &v3corepb.Address{
 						Address: &v3corepb.Address_SocketAddress{
 							SocketAddress: &v3corepb.SocketAddress{Address: "invalid"},
 						},
@@ -649,33 +639,13 @@ func (s) TestValidateAndConstructMetadata(t *testing.T) {
 				t.Errorf("validateAndConstructMetadata() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
-			if diff := cmp.Diff(got, tt.want, cmp.AllowUnexported(testMetadataValue{})); diff != "" {
-				t.Errorf("validateAndConstructMetadata() returned unexpected value:\n%s", diff)
+			if diff := cmp.Diff(got, tt.want, cmp.AllowUnexported(ProxyAddressMetadataValue{})); diff != "" {
+				t.Errorf("validateAndConstructMetadata() returned unexpected diff (-want +got):\n%s", diff)
 			}
 		})
 	}
 }
 
-type testMetadataValue struct {
-	MetadataValue
-	Address string
-}
-
-type testMetadataValueConvertor struct{}
-
-// testMetadataValueConvertor is a mock converter for testing purposes.
-func (testMetadataValueConvertor) Convert(anyBytes []byte) (MetadataValue, error) {
-	addrProto := &v3corepb.Address{}
-	if err := proto.Unmarshal(anyBytes, addrProto); err != nil {
-		return nil, fmt.Errorf("failed to unmarshal address proto: %v", err)
-	}
-	// For the failure case, return an error if the address is "invalid".
-	if addrProto.GetSocketAddress().GetAddress() == "invalid" {
-		return nil, fmt.Errorf("mock parser error: invalid address")
-	}
-	return testMetadataValue{Address: addrProto.GetSocketAddress().GetAddress()}, nil
-}
-
 // claBuilder builds a ClusterLoadAssignment, aka EDS
 // response.
 type claBuilder struct {