Add xds client metric for receiving invalid resource (#1005)

CHANGELOG.md

@@ -14,6 +14,9 @@ and what APIs have changed, if applicable.
 
 ## [Unreleased]
 
+## [29.57.0] - 2024-06-16
+- Add xds client metric for receiving invalid resource
+
 ## [29.56.1] - 2024-06-06
 - prevent duplicate uri property update
 
@@ -5698,7 +5701,8 @@ patch operations can re-use these classes for generating patch messages.
 
 ## [0.14.1]
 
-[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.56.1...master
+[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.57.0...master
+[29.57.0]: https://github.com/linkedin/rest.li/compare/v29.56.1...v29.57.0
 [29.56.1]: https://github.com/linkedin/rest.li/compare/v29.56.0...v29.56.1
 [29.56.0]: https://github.com/linkedin/rest.li/compare/v29.55.0...v29.56.0
 [29.55.0]: https://github.com/linkedin/rest.li/compare/v29.54.0...v29.55.0

d2/src/main/java/com/linkedin/d2/jmx/XdsClientJmx.java

@@ -29,6 +29,7 @@ public class XdsClientJmx implements XdsClientJmxMBean
 
   private final AtomicBoolean _isConnected = new AtomicBoolean();
   private final AtomicInteger _resourceNotFoundCount = new AtomicInteger();
+  private final AtomicInteger _resourceInvalidCount = new AtomicInteger();
   private final XdsServerMetricsProvider _xdsServerMetricsProvider;
 
   @Deprecated
@@ -67,6 +68,12 @@ public int getResourceNotFoundCount()
     return _resourceNotFoundCount.get();
   }
 
+  @Override
+  public int getResourceInvalidCount()
+  {
+    return _resourceInvalidCount.get();
+  }
+
   @Override
   public long getXdsServerLatencyMin() {
     return _xdsServerMetricsProvider.getLatencyMin();
@@ -130,4 +137,9 @@ public void incrementResourceNotFoundCount()
   {
     _resourceNotFoundCount.incrementAndGet();
   }
+
+  public void incrementResourceInvalidCount()
+  {
+    _resourceInvalidCount.incrementAndGet();
+  }
 }

d2/src/main/java/com/linkedin/d2/jmx/XdsClientJmxMBean.java

@@ -35,6 +35,9 @@ public interface XdsClientJmxMBean {
   // when the resource is not found.
   int getResourceNotFoundCount();
 
+  // when the resource is invalid.
+  int getResourceInvalidCount();
+
   /**
    * Get minimum of Xds server latency, which is from when the resource is updated on the Xds server to when the
    * client receives it.

d2/src/main/java/com/linkedin/d2/xds/XdsClientImpl.java

@@ -27,6 +27,7 @@
 import com.linkedin.d2.jmx.NoOpXdsServerMetricsProvider;
 import com.linkedin.d2.jmx.XdsClientJmx;
 import com.linkedin.d2.xds.GlobCollectionUtils.D2UriIdentifier;
+import com.linkedin.util.RateLimitedLogger;
 import com.linkedin.util.clock.SystemClock;
 import indis.XdsD2;
 import io.envoyproxy.envoy.service.discovery.v3.AggregatedDiscoveryServiceGrpc;
@@ -68,6 +69,8 @@
 public class XdsClientImpl extends XdsClient
 {
   private static final Logger _log = LoggerFactory.getLogger(XdsClientImpl.class);
+  private static final RateLimitedLogger RATE_LIMITED_LOGGER =
+      new RateLimitedLogger(_log, TimeUnit.MINUTES.toMillis(10), SystemClock.instance());
   public static final long DEFAULT_READY_TIMEOUT_MILLIS = 2000L;
 
   private final Map<ResourceType, Map<String, ResourceSubscriber>> _resourceSubscribers = Maps.immutableEnumMap(
@@ -136,7 +139,7 @@ void watchXdsResource(String resourceName, ResourceWatcher watcher)
       ResourceSubscriber subscriber = resourceSubscriberMap.get(resourceName);
       if (subscriber == null)
       {
-        subscriber = new ResourceSubscriber(watcher.getType(), resourceName);
+        subscriber = new ResourceSubscriber(watcher.getType(), resourceName, _xdsClientJmx);
         resourceSubscriberMap.put(resourceName, subscriber);
         ResourceType type;
         String adjustedResourceName;
@@ -506,6 +509,7 @@ static class ResourceSubscriber
     private final ResourceType _type;
     private final String _resource;
     private final Set<ResourceWatcher> _watchers = new HashSet<>();
+    private final XdsClientJmx _xdsClientJmx;
     @Nullable
     private ResourceUpdate _data;
 
@@ -522,10 +526,11 @@ public void setData(@Nullable ResourceUpdate data)
       _data = data;
     }
 
-    ResourceSubscriber(ResourceType type, String resource)
+    ResourceSubscriber(ResourceType type, String resource, XdsClientJmx xdsClientJmx)
     {
       _type = type;
       _resource = resource;
+      _xdsClientJmx = xdsClientJmx;
     }
 
     void addWatcher(ResourceWatcher watcher)
@@ -555,6 +560,19 @@ private void onData(ResourceUpdate data, XdsServerMetricsProvider metricsProvide
         trackServerLatency(data, metricsProvider); // data updated, track xds server latency
         _data = data;
       }
+      else
+      {
+        if (_type == ResourceType.D2_URI_MAP || _type == ResourceType.D2_URI)
+        {
+          RATE_LIMITED_LOGGER.warn("Received invalid data for {} {}, data: {}", _type, _resource, data);
+        }
+        else
+        {
+          _log.warn("Received invalid data for {} {}, data: {}", _type, _resource, data);
+        }
+        _xdsClientJmx.incrementResourceInvalidCount();
+      }
+
       if (_data == null)
       {
         _log.info("Initializing {} {} to empty data.", _type, _resource);

d2/src/test/java/com/linkedin/d2/xds/TestXdsClientImpl.java

@@ -447,8 +447,8 @@ private static class XdsClientImplFixture
     XdsClientImpl _xdsClientImpl;
     @Mock
     XdsClientJmx _xdsClientJmx;
-    ResourceSubscriber _nodeSubscriber = spy(new ResourceSubscriber(NODE, SERVICE_RESOURCE_NAME));
-    ResourceSubscriber _clusterSubscriber = spy(new ResourceSubscriber(D2_URI_MAP, CLUSTER_RESOURCE_NAME));
+    ResourceSubscriber _nodeSubscriber;
+    ResourceSubscriber _clusterSubscriber;
     Map<ResourceType, Map<String, ResourceSubscriber>> _subscribers = new HashMap<>();
     @Mock
     XdsClient.ResourceWatcher _resourceWatcher;
@@ -463,6 +463,8 @@ private static class XdsClientImplFixture
     XdsClientImplFixture(boolean useGlobCollections)
     {
       MockitoAnnotations.initMocks(this);
+      _nodeSubscriber = spy(new ResourceSubscriber(NODE, SERVICE_RESOURCE_NAME, _xdsClientJmx));
+      _clusterSubscriber = spy(new ResourceSubscriber(D2_URI_MAP, CLUSTER_RESOURCE_NAME, _xdsClientJmx));
 
       doNothing().when(_resourceWatcher).onChanged(any());
       for (ResourceSubscriber subscriber : Lists.newArrayList(_nodeSubscriber, _clusterSubscriber))

gradle.properties

@@ -1,4 +1,4 @@
-version=29.56.1
+version=29.57.0
 group=com.linkedin.pegasus
 org.gradle.configureondemand=true
 org.gradle.parallel=true