xds: Parsing xDS Cluster Metadata

xds/src/main/java/io/grpc/xds/ClusterMetadataRegistry.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.xds;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.protobuf.Any;
+import com.google.protobuf.InvalidProtocolBufferException;
+import io.grpc.xds.GcpAuthenticationFilter.AudienceMetadataParser;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Registry for parsing cluster metadata values.
+ *
+ * <p>This class maintains a mapping of type URLs to {@link ClusterMetadataValueParser} instances,
+ * allowing for the parsing of different metadata types.
+ */
+final class ClusterMetadataRegistry {
+  private static final ClusterMetadataRegistry INSTANCE = new ClusterMetadataRegistry();
+
+  private final Map<String, ClusterMetadataValueParser> supportedParsers = new HashMap<>();
+
+  private ClusterMetadataRegistry() {
+    registerParser(new AudienceMetadataParser());
+  }
+
+  static ClusterMetadataRegistry getInstance() {
+    return INSTANCE;
+  }
+
+  ClusterMetadataValueParser findParser(String typeUrl) {
+    return supportedParsers.get(typeUrl);
+  }
+
+  @VisibleForTesting
+  void registerParser(ClusterMetadataValueParser parser) {
+    supportedParsers.put(parser.getTypeUrl(), parser);
+  }
+
+  interface ClusterMetadataValueParser {
+
+    String getTypeUrl();
+
+    /**
+     * Parses the given {@link Any} object into a specific metadata value.
+     *
+     * @param any the {@link Any} object to parse.
+     * @return the parsed metadata value.
+     * @throws InvalidProtocolBufferException if the parsing fails.
+     */
+    Object parse(Any any) throws InvalidProtocolBufferException;
+  }
+}

xds/src/main/java/io/grpc/xds/GcpAuthenticationFilter.java

@@ -22,6 +22,7 @@
 import com.google.protobuf.Any;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
+import io.envoyproxy.envoy.extensions.filters.http.gcp_authn.v3.Audience;
 import io.envoyproxy.envoy.extensions.filters.http.gcp_authn.v3.GcpAuthnFilterConfig;
 import io.envoyproxy.envoy.extensions.filters.http.gcp_authn.v3.TokenCacheConfig;
 import io.grpc.CallCredentials;
@@ -35,6 +36,7 @@
 import io.grpc.MethodDescriptor;
 import io.grpc.Status;
 import io.grpc.auth.MoreCallCredentials;
+import io.grpc.xds.ClusterMetadataRegistry.ClusterMetadataValueParser;
 import io.grpc.xds.Filter.ClientInterceptorBuilder;
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -219,4 +221,32 @@
       return cache.computeIfAbsent(key, create);
     }
   }
+
+  /**
+   * Parser for Audience metadata type.
+   */
+  static class AudienceMetadataParser implements ClusterMetadataValueParser {
+
+    @Override
+    public String getTypeUrl() {
+      return "type.googleapis.com/envoy.extensions.filters.http.gcp_authn.v3.Audience";
+    }
+
+    @Override
+    public String parse(Any any) throws InvalidProtocolBufferException {
+      if (any.is(Audience.class)) {
+        Audience audience = any.unpack(Audience.class);
+        String url = audience.getUrl();
+        if (url.isEmpty()) {
+          throw new InvalidProtocolBufferException(
+              "Audience URL is empty. Metadata value must contain a valid URL.");
+        }
+        return url;
+      } else {
+        throw new InvalidProtocolBufferException(
+            String.format("Unexpected message type: %s. Expected: %s",
+                any.getTypeUrl(), Audience.getDescriptor().getFullName()));
+      }
+    }
+  }
 }

xds/src/main/java/io/grpc/xds/XdsClusterResource.java

@@ -25,13 +25,15 @@
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.protobuf.Any;
 import com.google.protobuf.Duration;
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
 import com.google.protobuf.Struct;
 import com.google.protobuf.util.Durations;
 import io.envoyproxy.envoy.config.cluster.v3.CircuitBreakers.Thresholds;
 import io.envoyproxy.envoy.config.cluster.v3.Cluster;
+import io.envoyproxy.envoy.config.core.v3.Metadata;
 import io.envoyproxy.envoy.config.core.v3.RoutingPriority;
 import io.envoyproxy.envoy.config.core.v3.SocketAddress;
 import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment;
@@ -42,14 +44,17 @@
 import io.grpc.internal.GrpcUtil;
 import io.grpc.internal.ServiceConfigUtil;
 import io.grpc.internal.ServiceConfigUtil.LbConfig;
+import io.grpc.xds.ClusterMetadataRegistry.ClusterMetadataValueParser;
 import io.grpc.xds.EnvoyServerProtoData.OutlierDetection;
 import io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext;
 import io.grpc.xds.XdsClusterResource.CdsUpdate;
 import io.grpc.xds.client.XdsClient.ResourceUpdate;
 import io.grpc.xds.client.XdsResourceType;
+import io.grpc.xds.internal.ProtobufJsonConverter;
 import io.grpc.xds.internal.security.CommonTlsContextUtil;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import java.util.Set;
 import javax.annotation.Nullable;
 
@@ -169,9 +174,50 @@
     updateBuilder.filterMetadata(
         ImmutableMap.copyOf(cluster.getMetadata().getFilterMetadataMap()));
 
+    try {
+      ImmutableMap<String, Object> parsedFilterMetadata =
+          parseClusterMetadata(cluster.getMetadata());
+      updateBuilder.parsedMetadata(parsedFilterMetadata);
+    } catch (InvalidProtocolBufferException e) {
+      throw new ResourceInvalidException(
+          "Failed to parse xDS filter metadata for cluster '" + cluster.getName() + "': "
+              + e.getMessage(), e);
+    }
+
     return updateBuilder.build();
   }
 
+  private static ImmutableMap<String, Object> parseClusterMetadata(Metadata metadata)
+      throws InvalidProtocolBufferException {
+    ImmutableMap.Builder<String, Object> parsedMetadata = ImmutableMap.builder();
+
+    ClusterMetadataRegistry registry = ClusterMetadataRegistry.getInstance();
+    // Process typed_filter_metadata
+    for (Map.Entry<String, Any> entry : metadata.getTypedFilterMetadataMap().entrySet()) {
+      String key = entry.getKey();
+      Any value = entry.getValue();
+      ClusterMetadataValueParser parser = registry.findParser(value.getTypeUrl());
+      if (parser != null) {
+        Object parsedValue = parser.parse(value);
+        parsedMetadata.put(key, parsedValue);
+      }
+    }
+    // building once to reuse in the next loop
+    ImmutableMap<String, Object> intermediateParsedMetadata = parsedMetadata.build();
+
+    // Process filter_metadata for remaining keys
+    for (Map.Entry<String, Struct> entry : metadata.getFilterMetadataMap().entrySet()) {
+      String key = entry.getKey();
+      if (!intermediateParsedMetadata.containsKey(key)) {
+        Struct structValue = entry.getValue();
+        Object jsonValue = ProtobufJsonConverter.convertToJson(structValue);
+        parsedMetadata.put(key, jsonValue);
+      }
+    }
+
+    return parsedMetadata.build();
+  }
+
   private static StructOrError<CdsUpdate.Builder> parseAggregateCluster(Cluster cluster) {
     String clusterName = cluster.getName();
     Cluster.CustomClusterType customType = cluster.getClusterType();
@@ -571,13 +617,16 @@
 
     abstract ImmutableMap<String, Struct> filterMetadata();
 
+    abstract ImmutableMap<String, Object> parsedMetadata();
+
     private static Builder newBuilder(String clusterName) {
       return new AutoValue_XdsClusterResource_CdsUpdate.Builder()
           .clusterName(clusterName)
           .minRingSize(0)
           .maxRingSize(0)
           .choiceCount(0)
-          .filterMetadata(ImmutableMap.of());
+          .filterMetadata(ImmutableMap.of())
+          .parsedMetadata(ImmutableMap.of());
     }
 
     static Builder forAggregate(String clusterName, List<String> prioritizedClusterNames) {
@@ -696,6 +745,8 @@
 
       protected abstract Builder filterMetadata(ImmutableMap<String, Struct> filterMetadata);
 
+      protected abstract Builder parsedMetadata(ImmutableMap<String, Object> parsedMetadata);
+
       abstract CdsUpdate build();
     }
   }

xds/src/main/java/io/grpc/xds/internal/ProtobufJsonConverter.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.xds.internal;
+
+import com.google.protobuf.Struct;
+import com.google.protobuf.Value;
+import io.grpc.Internal;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Converter for Protobuf {@link Struct} to JSON-like {@link Map}.
+ */
+@Internal
+public final class ProtobufJsonConverter {
+  private ProtobufJsonConverter() {}
+
+  public static Map<String, Object> convertToJson(Struct struct) {
+    Map<String, Object> result = new HashMap<>();
+    for (Map.Entry<String, Value> entry : struct.getFieldsMap().entrySet()) {
+      result.put(entry.getKey(), convertValue(entry.getValue()));
+    }
+    return result;
+  }
+
+  public static Object convertValue(Value value) {
+    switch (value.getKindCase()) {
+      case STRUCT_VALUE:
+        return convertToJson(value.getStructValue());
+      case LIST_VALUE:
+        return value.getListValue().getValuesList().stream()
+            .map(ProtobufJsonConverter::convertValue)
+            .collect(Collectors.toList());
+      case NUMBER_VALUE:
+        return value.getNumberValue();
+      case STRING_VALUE:
+        return value.getStringValue();
+      case BOOL_VALUE:
+        return value.getBoolValue();
+      case NULL_VALUE:
+        return null;
+      default:
+        throw new IllegalArgumentException("Unknown Value type: " + value.getKindCase());
+    }
+  }
+}