xds: Parsing xDS Cluster Metadata (grpc#11741)

Author: shivaspeaks

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;
@@ -36,6 +37,7 @@
 import io.grpc.Status;
 import io.grpc.auth.MoreCallCredentials;
 import io.grpc.xds.Filter.ClientInterceptorBuilder;
+import io.grpc.xds.MetadataRegistry.MetadataValueParser;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.ScheduledExecutorService;
@@ -219,4 +221,23 @@ V getOrInsert(K key, Function<K, V> create) {
       return cache.computeIfAbsent(key, create);
     }
   }
+
+  static class AudienceMetadataParser implements MetadataValueParser {
+
+    @Override
+    public String getTypeUrl() {
+      return "type.googleapis.com/envoy.extensions.filters.http.gcp_authn.v3.Audience";
+    }
+
+    @Override
+    public String parse(Any any) throws InvalidProtocolBufferException {
+      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;
+    }
+  }
 }

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

@@ -0,0 +1,71 @@
+/*
+ * 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 MetadataValueParser} instances,
+ * allowing for the parsing of different metadata types.
+ */
+final class MetadataRegistry {
+  private static final MetadataRegistry INSTANCE = new MetadataRegistry();
+
+  private final Map<String, MetadataValueParser> supportedParsers = new HashMap<>();
+
+  private MetadataRegistry() {
+    registerParser(new AudienceMetadataParser());
+  }
+
+  static MetadataRegistry getInstance() {
+    return INSTANCE;
+  }
+
+  MetadataValueParser findParser(String typeUrl) {
+    return supportedParsers.get(typeUrl);
+  }
+
+  @VisibleForTesting
+  void registerParser(MetadataValueParser parser) {
+    supportedParsers.put(parser.getTypeUrl(), parser);
+  }
+
+  void removeParser(MetadataValueParser parser) {
+    supportedParsers.remove(parser.getTypeUrl());
+  }
+
+  interface MetadataValueParser {
+
+    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/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;
@@ -44,12 +46,15 @@
 import io.grpc.internal.ServiceConfigUtil.LbConfig;
 import io.grpc.xds.EnvoyServerProtoData.OutlierDetection;
 import io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext;
+import io.grpc.xds.MetadataRegistry.MetadataValueParser;
 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;
 
@@ -171,9 +176,62 @@ static CdsUpdate processCluster(Cluster cluster,
     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();
   }
 
+  /**
+   * Parses cluster metadata into a structured map.
+   *
+   * <p>Values in {@code typed_filter_metadata} take precedence over
+   * {@code filter_metadata} when keys overlap, following Envoy API behavior. See
+   * <a href="https://github.com/envoyproxy/envoy/blob/main/api/envoy/config/core/v3/base.proto#L217-L259">
+   *   Envoy metadata documentation </a> for details.
+   *
+   * @param metadata the {@link Metadata} containing the fields to parse.
+   * @return an immutable map of parsed metadata.
+   * @throws InvalidProtocolBufferException if parsing {@code typed_filter_metadata} fails.
+   */
+  private static ImmutableMap<String, Object> parseClusterMetadata(Metadata metadata)
+      throws InvalidProtocolBufferException {
+    ImmutableMap.Builder<String, Object> parsedMetadata = ImmutableMap.builder();
+
+    MetadataRegistry registry = MetadataRegistry.getInstance();
+    // Process typed_filter_metadata
+    for (Map.Entry<String, Any> entry : metadata.getTypedFilterMetadataMap().entrySet()) {
+      String key = entry.getKey();
+      Any value = entry.getValue();
+      MetadataValueParser 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();
@@ -573,13 +631,16 @@ abstract static class CdsUpdate implements ResourceUpdate {
 
     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) {
@@ -698,6 +759,8 @@ Builder leastRequestLbPolicy(Integer choiceCount) {
 
       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;
+  }
+
+  private 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());
+    }
+  }
+}