Use Node instead of D2Node and D2URIMap instead of NodeMap for xDS fl…

…ow (#944) * Use Node(Map) instead of D2node(Map) The initial intent of using D2Node was to pre-deserialize the JSON, but because there is no way for the rest.li client to actually leverage this, it simply re-serializes the JSON then deserializes it again. This causes memory pressure along with validation errors as numerical typesa get mungled. * Leverage pre-deserialize JSON struct instead of switching over to raw JSON data Additionally introduce a D2URI type which has a proper schema instead of being a Struct. This will significantly decrease the resource size sent by the observer and significantly decrease the cost of deserialization on the client. * Bump minor version * Comment and simplify D2URI * Meet proto standard of snake_case naming * Update comment on D2URI * Simplify D2URI even further * Use the raw JSON Node type instead of D2Node This is causing all sorts of validation issues due to integers being interpreted as floats and vice versa * Fix some compile errors * Update changelog * Fix test failing due to missing property * Update changelog
linkedin · Nov 13, 2023 · a33a1e1 · a33a1e1
1 parent 64faf7e
commit a33a1e1
Show file tree

Hide file tree

Showing 12 changed files with 403 additions and 222 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -14,6 +14,9 @@ and what APIs have changed, if applicable.
 
 ## [Unreleased]
 
+## [29.47.0] - 2023-11-13
+- Use Node instead of D2Node and D2URIMap instead of NodeMap for xDS flow
+
 ## [29.46.9] - 2023-11-02
 - Update FieldDef so that it will lazily cache the hashCode.
 
@@ -5557,7 +5560,8 @@ patch operations can re-use these classes for generating patch messages.
 
 ## [0.14.1]
 
-[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.46.9...master
+[Unreleased]: https://github.com/linkedin/rest.li/compare/v29.47.0...master
+[29.47.0]: https://github.com/linkedin/rest.li/compare/v29.46.9...v29.47.0
 [29.46.9]: https://github.com/linkedin/rest.li/compare/v29.46.8...v29.46.9
 [29.46.8]: https://github.com/linkedin/rest.li/compare/v29.46.7...v29.46.8
 [29.46.7]: https://github.com/linkedin/rest.li/compare/v29.46.6...v29.46.7

diff --git a/d2/src/main/java/com/linkedin/d2/balancer/properties/ClusterPropertiesJsonSerializer.java b/d2/src/main/java/com/linkedin/d2/balancer/properties/ClusterPropertiesJsonSerializer.java
@@ -16,6 +16,7 @@
 
 package com.linkedin.d2.balancer.properties;
 
+import com.google.protobuf.ByteString;
 import com.linkedin.d2.balancer.properties.util.PropertyUtil;
 import com.linkedin.d2.balancer.util.JacksonUtil;
 import com.linkedin.d2.discovery.PropertyBuilder;
@@ -84,6 +85,22 @@ public ClusterProperties fromBytes(byte[] bytes, long version) throws PropertySe
     clusterProperties.setVersion(version);
     return clusterProperties;
   }
+
+  public ClusterProperties fromBytes(ByteString bytes, long version) throws PropertySerializationException
+  {
+    try
+    {
+      @SuppressWarnings("unchecked")
+      Map<String, Object> untyped = JacksonUtil.getObjectMapper().readValue(bytes.newInput(), HashMap.class);
+      ClusterProperties clusterProperties = fromMap(untyped);
+      clusterProperties.setVersion(version);
+      return clusterProperties;
+    }
+    catch (Exception e)
+    {
+      throw new PropertySerializationException(e);
+    }
+  }
 
   @SuppressWarnings("unchecked")
   private static <T> T mapGetOrDefault(Map<String, Object> map, String key, T defaultValue)

diff --git a/d2/src/main/java/com/linkedin/d2/balancer/properties/ServicePropertiesJsonSerializer.java b/d2/src/main/java/com/linkedin/d2/balancer/properties/ServicePropertiesJsonSerializer.java
@@ -17,6 +17,7 @@
 package com.linkedin.d2.balancer.properties;
 
 
+import com.google.protobuf.ByteString;
 import com.linkedin.d2.balancer.properties.util.PropertyUtil;
 import com.linkedin.d2.balancer.subsetting.SubsettingStrategy;
 import com.linkedin.d2.balancer.util.JacksonUtil;
@@ -202,6 +203,22 @@ public ServiceProperties fromBytes(byte[] bytes, long version) throws PropertySe
     return serviceProperties;
   }
 
+  public ServiceProperties fromBytes(ByteString bytes, long version) throws PropertySerializationException
+  {
+    try
+    {
+      @SuppressWarnings("unchecked")
+      Map<String, Object> untyped = JacksonUtil.getObjectMapper().readValue(bytes.newInput(), Map.class);
+      ServiceProperties serviceProperties = fromMap(untyped);
+      serviceProperties.setVersion(version);
+      return serviceProperties;
+    }
+    catch (Exception e)
+    {
+      throw new PropertySerializationException(e);
+    }
+  }
+
   /**
    * Always return the composite class {@link ServiceStoreProperties} to include ALL properties stored on service registry (like Zookeeper),
    * such as canary configs, distribution strategy, etc.

diff --git a/d2/src/main/java/com/linkedin/d2/balancer/properties/UriPropertiesJsonSerializer.java b/d2/src/main/java/com/linkedin/d2/balancer/properties/UriPropertiesJsonSerializer.java
@@ -16,20 +16,19 @@
 
 package com.linkedin.d2.balancer.properties;
 
-
 import com.linkedin.d2.balancer.properties.util.PropertyUtil;
 import com.linkedin.d2.balancer.util.JacksonUtil;
 import com.linkedin.d2.balancer.util.partitions.DefaultPartitionAccessor;
 import com.linkedin.d2.discovery.PropertyBuilder;
 import com.linkedin.d2.discovery.PropertySerializationException;
 import com.linkedin.d2.discovery.PropertySerializer;
-import java.util.Collections;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
+import indis.XdsD2;
 import java.net.URI;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class UriPropertiesJsonSerializer implements PropertySerializer<UriProperties>, PropertyBuilder<UriProperties>
 {
@@ -109,6 +108,33 @@ public UriProperties fromBytes(byte[] bytes, long version) throws PropertySerial
     return uriProperties;
   }
 
+  public UriProperties fromProto(XdsD2.D2URI protoUri) throws PropertySerializationException
+  {
+    try
+    {
+      URI uri = URI.create(protoUri.getUri());
+
+      Map<Integer, PartitionData> partitionDesc = new HashMap<>(protoUri.getPartitionDescCount());
+      for (Map.Entry<Integer, Double> partition : protoUri.getPartitionDescMap().entrySet())
+      {
+        partitionDesc.put(partition.getKey(), new PartitionData(partition.getValue()));
+      }
+
+      Map<String, Object> applicationProperties = PropertyUtil.protoStructToMap(protoUri.getUriSpecificProperties());
+
+      return new UriProperties(
+          protoUri.getClusterName(),
+          Collections.singletonMap(uri, partitionDesc),
+          Collections.singletonMap(uri, applicationProperties),
+          protoUri.getVersion()
+      );
+    }
+    catch (Exception e)
+    {
+      throw new PropertySerializationException(e);
+    }
+  }
+
   @Override
   @SuppressWarnings("unchecked")
   public UriProperties fromMap(Map<String, Object> map)

diff --git a/d2/src/main/java/com/linkedin/d2/balancer/properties/util/PropertyUtil.java b/d2/src/main/java/com/linkedin/d2/balancer/properties/util/PropertyUtil.java
@@ -16,9 +16,13 @@
 
 package com.linkedin.d2.balancer.properties.util;
 
-import com.linkedin.data.template.TemplateOutputCastException;
+import com.google.protobuf.Struct;
+import com.google.protobuf.Value;
 import com.linkedin.util.ArgumentUtil;
-
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 public class PropertyUtil
@@ -152,4 +156,63 @@ else if (value instanceof Double && clazz.equals(Integer.class))
     }
     return (T) value;
   }
+
+  /**
+   * Efficiently translates a proto JSON {@link Struct} into a {@code Map<String, Object>} without additional
+   * serialization or deserialization.
+   */
+  public static Map<String, Object> protoStructToMap(Struct struct)
+  {
+    if (struct.getFieldsCount() == 0) {
+      return Collections.emptyMap();
+    }
+    Map<String, Object> map = new HashMap<>(struct.getFieldsMap().size());
+    for (Map.Entry<String, Value> entry : struct.getFieldsMap().entrySet())
+    {
+      map.put(entry.getKey(), valueToObject(entry.getValue()));
+    }
+    return map;
+  }
+
+  private static Object valueToObject(Value value)
+  {
+    if (value.hasBoolValue())
+    {
+      return value.getBoolValue();
+    }
+    else if (value.hasStringValue())
+    {
+      return value.getStringValue();
+    }
+    else if (value.hasNumberValue())
+    {
+      return value.getNumberValue();
+    }
+    else if (value.hasNullValue())
+    {
+      return null;
+    }
+    else if (value.hasStructValue())
+    {
+      Map<String, Object> map = new HashMap<>(value.getStructValue().getFieldsCount());
+      for (Map.Entry<String, Value> entry : value.getStructValue().getFieldsMap().entrySet())
+      {
+        map.put(entry.getKey(), valueToObject(entry.getValue()));
+      }
+      return map;
+    }
+    else if (value.hasListValue())
+    {
+      List<Object> list = new ArrayList<>(value.getListValue().getValuesCount());
+      for (Value element : value.getListValue().getValuesList())
+      {
+        list.add(valueToObject(element));
+      }
+      return list;
+    }
+    else
+    {
+      throw new RuntimeException("Unexpected proto value of unknown type: " + value);
+    }
+  }
 }
diff --git a/d2/src/main/java/com/linkedin/d2/discovery/stores/zk/SymlinkAwareZooKeeper.java b/d2/src/main/java/com/linkedin/d2/discovery/stores/zk/SymlinkAwareZooKeeper.java
@@ -16,6 +16,7 @@
 
 package com.linkedin.d2.discovery.stores.zk;
 
+import com.google.protobuf.ByteString;
 import com.linkedin.d2.discovery.PropertySerializationException;
 import com.linkedin.d2.discovery.PropertySerializer;
 import org.apache.zookeeper.AsyncCallback;
@@ -601,6 +602,18 @@ public String fromBytes(byte[] bytes) throws PropertySerializationException
       }
     }
 
+    public String fromBytes(ByteString bytes) throws PropertySerializationException
+    {
+      try
+      {
+        return bytes.toString("UTF-8");
+      }
+      catch (UnsupportedEncodingException e)
+      {
+        throw new PropertySerializationException(e);
+      }
+    }
+
     @Override
     public byte[] toBytes(String property)
     {