xds: Support retrieving names from wrapped resource containers (#10975)

The xDS library only honored names retrieved from the inner resource
containers, but for wrapped resources the outer layer could contain the
required name. This commit prefers the name on the wrapped container
over the inner resource name.

Author: anicr7

xds/src/main/java/io/grpc/xds/client/XdsResourceType.java

@@ -148,15 +148,24 @@ ValidatedResourceUpdate<T> parse(Args args, List<Any> resources) {
       Any resource = resources.get(i);
 
       Message unpackedMessage;
+      String name = "";
       try {
-        resource = maybeUnwrapResources(resource);
+        if (resource.getTypeUrl().equals(TYPE_URL_RESOURCE)) {
+          Resource wrappedResource = unpackCompatibleType(resource, Resource.class,
+              TYPE_URL_RESOURCE, null);
+          resource = wrappedResource.getResource();
+          name = wrappedResource.getName();
+        } 
         unpackedMessage = unpackCompatibleType(resource, unpackedClassName(), typeUrl(), null);
       } catch (InvalidProtocolBufferException e) {
         errors.add(String.format("%s response Resource index %d - can't decode %s: %s",
                 typeName(), i, unpackedClassName().getSimpleName(), e.getMessage()));
         continue;
       }
-      String name = extractResourceName(unpackedMessage);
+      // Fallback to inner resource name if the outer resource didn't have a name.
+      if (name.isEmpty()) {
+        name = extractResourceName(unpackedMessage);
+      }
       if (name == null || !isResourceNameValid(name, resource.getTypeUrl())) {
         errors.add(
             "Unsupported resource name: " + name + " for type: " + typeName());
@@ -209,16 +218,6 @@ protected static <T extends com.google.protobuf.Message> T unpackCompatibleType(
     return any.unpack(clazz);
   }
 
-  private Any maybeUnwrapResources(Any resource)
-      throws InvalidProtocolBufferException {
-    if (resource.getTypeUrl().equals(TYPE_URL_RESOURCE)) {
-      return unpackCompatibleType(resource, Resource.class, TYPE_URL_RESOURCE,
-          null).getResource();
-    } else {
-      return resource;
-    }
-  }
-
   static final class ParsedResource<T extends ResourceUpdate> {
     private final T resourceUpdate;
     private final Any rawResource;

xds/src/test/java/io/grpc/xds/GrpcXdsClientImplTestBase.java

@@ -852,6 +852,25 @@ public void wrappedLdsResource() {
     verifySubscribedResourcesMetadataSizes(1, 0, 0, 0);
   }
 
+  @Test
+  public void wrappedLdsResource_preferWrappedName() {
+    DiscoveryRpcCall call = startResourceWatcher(XdsListenerResource.getInstance(), LDS_RESOURCE,
+        ldsResourceWatcher);
+
+    Any innerResource = Any.pack(mf.buildListenerWithApiListener("random_name" /* name */,
+            mf.buildRouteConfiguration("do not care", mf.buildOpaqueVirtualHosts(VHOST_SIZE))));
+
+    // Client sends an ACK LDS request.
+    call.sendResponse(LDS, mf.buildWrappedResourceWithName(innerResource, LDS_RESOURCE), VERSION_1,
+            "0000");
+    call.verifyRequest(LDS, LDS_RESOURCE, VERSION_1, "0000", NODE);
+    verify(ldsResourceWatcher).onChanged(ldsUpdateCaptor.capture());
+    verifyGoldenListenerVhosts(ldsUpdateCaptor.getValue());
+    assertThat(fakeClock.getPendingTasks(LDS_RESOURCE_FETCH_TIMEOUT_TASK_FILTER)).isEmpty();
+    verifyResourceMetadataAcked(LDS, LDS_RESOURCE, innerResource, VERSION_1, TIME_INCREMENT);
+    verifySubscribedResourcesMetadataSizes(1, 0, 0, 0);
+  }
+
   @Test
   public void ldsResourceFound_containsRdsName() {
     DiscoveryRpcCall call = startResourceWatcher(XdsListenerResource.getInstance(), LDS_RESOURCE,
@@ -3836,6 +3855,8 @@ protected abstract static class MessageFactory {
 
     protected abstract Any buildWrappedResource(Any originalResource);
 
+    protected abstract Any buildWrappedResourceWithName(Any originalResource, String name);
+
     protected Message buildListenerWithApiListener(String name, Message routeConfiguration) {
       return buildListenerWithApiListener(
           name, routeConfiguration, Collections.<Message>emptyList());

xds/src/test/java/io/grpc/xds/GrpcXdsClientImplV3Test.java

@@ -290,6 +290,14 @@ protected Any buildWrappedResource(Any originalResource) {
           .build());
     }
 
+    @Override
+    protected  Any buildWrappedResourceWithName(Any originalResource, String name) {
+      return Any.pack(Resource.newBuilder()
+      .setResource(originalResource)
+      .setName(name)
+      .build());
+    }
+
     @SuppressWarnings("unchecked")
     @Override
     protected Message buildListenerWithApiListener(