server: Cache serialized protos (#98)

Signed-off-by: Jakub Dyszkiewicz <[email protected]>

Author: jakubdyszkiewicz

server/src/main/java/io/envoyproxy/controlplane/server/DiscoveryServer.java

@@ -8,6 +8,8 @@
 import io.envoyproxy.controlplane.cache.Resources;
 import io.envoyproxy.controlplane.cache.Response;
 import io.envoyproxy.controlplane.cache.Watch;
+import io.envoyproxy.controlplane.server.serializer.DefaultProtoResourcesSerializer;
+import io.envoyproxy.controlplane.server.serializer.ProtoResourcesSerializer;
 import io.envoyproxy.envoy.api.v2.ClusterDiscoveryServiceGrpc.ClusterDiscoveryServiceImplBase;
 import io.envoyproxy.envoy.api.v2.DiscoveryRequest;
 import io.envoyproxy.envoy.api.v2.DiscoveryResponse;
@@ -20,6 +22,7 @@
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.ServerCallStreamObserver;
 import io.grpc.stub.StreamObserver;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -41,6 +44,7 @@ public class DiscoveryServer {
   private final ConfigWatcher configWatcher;
   private final ExecutorGroup executorGroup;
   private final AtomicLong streamCount = new AtomicLong();
+  private final ProtoResourcesSerializer protoResourcesSerializer;
 
   public DiscoveryServer(ConfigWatcher configWatcher) {
     this(Collections.emptyList(), configWatcher);
@@ -56,25 +60,29 @@ public DiscoveryServer(DiscoveryServerCallbacks callbacks, ConfigWatcher configW
    * @param configWatcher source of configuration updates
    */
   public DiscoveryServer(List<DiscoveryServerCallbacks> callbacks, ConfigWatcher configWatcher) {
-    this(callbacks, configWatcher, new DefaultExecutorGroup());
+    this(callbacks, configWatcher, new DefaultExecutorGroup(), new DefaultProtoResourcesSerializer());
   }
 
   /**
    * Creates the server.
    * @param callbacks server callbacks
    * @param configWatcher source of configuration updates
    * @param executorGroup executor group to use for responding stream requests
+   * @param protoResourcesSerializer serializer of proto buffer messages
    */
   public DiscoveryServer(List<DiscoveryServerCallbacks> callbacks,
                          ConfigWatcher configWatcher,
-                         ExecutorGroup executorGroup) {
+                         ExecutorGroup executorGroup,
+                         ProtoResourcesSerializer protoResourcesSerializer) {
     Preconditions.checkNotNull(callbacks, "callbacks cannot be null");
     Preconditions.checkNotNull(configWatcher, "configWatcher cannot be null");
     Preconditions.checkNotNull(executorGroup, "executorGroup cannot be null");
+    Preconditions.checkNotNull(protoResourcesSerializer, "protoResourcesSerializer cannot be null");
 
     this.callbacks = callbacks;
     this.configWatcher = configWatcher;
     this.executorGroup = executorGroup;
+    this.protoResourcesSerializer = protoResourcesSerializer;
   }
 
   /**
@@ -306,9 +314,10 @@ private void cancel() {
     private void send(Response response, String typeUrl) {
       String nonce = Long.toString(streamNonce.getAndIncrement());
 
+      Collection<Any> resources = protoResourcesSerializer.serialize(response.resources());
       DiscoveryResponse discoveryResponse = DiscoveryResponse.newBuilder()
           .setVersionInfo(response.version())
-          .addAllResources(response.resources().stream().map(Any::pack).collect(Collectors.toList()))
+          .addAllResources(resources)
           .setTypeUrl(typeUrl)
           .setNonce(nonce)
           .build();

server/src/main/java/io/envoyproxy/controlplane/server/serializer/CachedProtoResourcesSerializer.java

@@ -0,0 +1,36 @@
+package io.envoyproxy.controlplane.server.serializer;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.protobuf.Any;
+import com.google.protobuf.Message;
+import io.envoyproxy.envoy.api.v2.DiscoveryResponse;
+import java.util.concurrent.ExecutionException;
+
+/**
+ * Cached version of the {@link ProtoResourcesSerializer}. It uses Guava Cache with weak values to store the serialized
+ * messages. The weak values are used so it's possible to share the same proto instance between snapshots that are kept
+ * in the memory. The improvement especially visible when the same message is send to multiple Envoys.
+ * The message is then only serialized once as long as it's referenced anywhere else.
+ * The same instance is used not only between snapshots for different groups but also between subsequent snapshots
+ * for the same group, because the last serialized proto instance of {@link DiscoveryResponse} is set in
+ * DiscoveryRequestStreamObserver#latestResponse.
+ */
+public class CachedProtoResourcesSerializer implements ProtoResourcesSerializer {
+
+  private static final Cache<Message, Any> cache = CacheBuilder.newBuilder()
+      .weakValues()
+      .build();
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public Any serialize(Message resource) {
+    try {
+      return cache.get(resource, () -> Any.pack(resource));
+    } catch (ExecutionException e) {
+      throw new ProtoSerializerException("Error while serializing resources", e);
+    }
+  }
+}

server/src/main/java/io/envoyproxy/controlplane/server/serializer/DefaultProtoResourcesSerializer.java

@@ -0,0 +1,18 @@
+package io.envoyproxy.controlplane.server.serializer;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.Message;
+
+/**
+ * Default implementation of ProtoResourcesSerializer that uses {@link Any#pack(Message)} method on {@link Message}.
+ */
+public class DefaultProtoResourcesSerializer implements ProtoResourcesSerializer {
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public Any serialize(Message resource) {
+    return Any.pack(resource);
+  }
+}

server/src/main/java/io/envoyproxy/controlplane/server/serializer/ProtoResourcesSerializer.java

@@ -0,0 +1,35 @@
+package io.envoyproxy.controlplane.server.serializer;
+
+import com.google.protobuf.Any;
+import com.google.protobuf.Message;
+
+import java.util.Collection;
+import java.util.stream.Collectors;
+
+/**
+ * Serializer of the proto buffers resource messages.
+ */
+public interface ProtoResourcesSerializer {
+
+  /**
+   * Serialize messages to proto buffers.
+   * @param resources list of resources to serialize
+   * @return serialized resources
+   */
+  default Collection<Any> serialize(Collection<? extends Message> resources) {
+    return resources.stream().map(this::serialize).collect(Collectors.toList());
+  }
+
+  /**
+   * Serialize message to proto buffers.
+   * @param resource the resource to serialize
+   * @return serialized resource
+   */
+  Any serialize(Message resource);
+
+  class ProtoSerializerException extends RuntimeException {
+    ProtoSerializerException(String message, Throwable cause) {
+      super(message, cause);
+    }
+  }
+}

server/src/test/java/io/envoyproxy/controlplane/server/serializer/CachedProtoResourcesSerializerTest.java

@@ -0,0 +1,48 @@
+package io.envoyproxy.controlplane.server.serializer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.google.common.collect.Lists;
+import com.google.protobuf.Any;
+import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment;
+import java.util.Collection;
+import java.util.List;
+import org.junit.Test;
+
+public class CachedProtoResourcesSerializerTest {
+
+  CachedProtoResourcesSerializer serializer = new CachedProtoResourcesSerializer();
+
+  @Test
+  public void shouldKeepCachedProtoWhenSerializingSameMessage() {
+    ClusterLoadAssignment endpoint = ClusterLoadAssignment.newBuilder()
+        .setClusterName("service1")
+        .build();
+
+    Any serializedEndpoint = serializer.serialize(endpoint);
+    Any serializedSameEndpoint = serializer.serialize(endpoint);
+
+    assertThat(serializedEndpoint).isSameAs(serializedSameEndpoint);
+  }
+
+  @Test
+  public void shouldKeepCachedProtoWhenSerializingSameMessages() {
+    List<ClusterLoadAssignment> endpoints = Lists.newArrayList(
+        ClusterLoadAssignment.newBuilder()
+            .setClusterName("service1")
+            .build(),
+        ClusterLoadAssignment.newBuilder()
+            .setClusterName("service2")
+            .build()
+    );
+
+    Collection<Any> serializedEndpoints = serializer.serialize(endpoints);
+    Collection<Any> serializedSameEndpoints = serializer.serialize(endpoints);
+
+    assertThat(serializedEndpoints).isEqualTo(serializedSameEndpoints);
+    assertThat(serializedEndpoints).isNotSameAs(serializedSameEndpoints);
+    assertThat(serializedEndpoints) // elements are the same instances
+        .usingElementComparator((x, y) -> x == y ? 0 : 1)
+        .hasSameElementsAs(serializedSameEndpoints);
+  }
+}

server/src/test/java/io/envoyproxy/controlplane/server/serializer/DefaultProtoResourcesSerializerTest.java

@@ -0,0 +1,49 @@
+package io.envoyproxy.controlplane.server.serializer;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.google.common.collect.Lists;
+import com.google.protobuf.Any;
+import io.envoyproxy.envoy.api.v2.ClusterLoadAssignment;
+import java.util.Collection;
+import java.util.List;
+import org.junit.Test;
+
+public class DefaultProtoResourcesSerializerTest {
+
+  DefaultProtoResourcesSerializer serializer = new DefaultProtoResourcesSerializer();
+
+  @Test
+  public void shouldReturnDifferentInstanceOfSerializedProtoWhenResourcesAreTheSame() {
+    ClusterLoadAssignment endpoint = ClusterLoadAssignment.newBuilder()
+        .setClusterName("service1")
+        .build();
+
+    Any serializedEndpoint = serializer.serialize(endpoint);
+    Any serializedSameEndpoint = serializer.serialize(endpoint);
+
+    assertThat(serializedEndpoint).isEqualTo(serializedSameEndpoint);
+    assertThat(serializedEndpoint).isNotSameAs(serializedSameEndpoint);
+  }
+
+  @Test
+  public void shouldReturnDifferentInstancesOfSerializedProtoWhenResourcesAreTheSame() {
+    List<ClusterLoadAssignment> endpoints = Lists.newArrayList(
+        ClusterLoadAssignment.newBuilder()
+            .setClusterName("service1")
+            .build(),
+        ClusterLoadAssignment.newBuilder()
+            .setClusterName("service2")
+            .build()
+    );
+
+    Collection<Any> serializedEndpoints = serializer.serialize(endpoints);
+    Collection<Any> serializedSameEndpoints = serializer.serialize(endpoints);
+
+    assertThat(serializedEndpoints).isEqualTo(serializedSameEndpoints);
+    assertThat(serializedEndpoints).isNotSameAs(serializedSameEndpoints);
+    assertThat(serializedEndpoints) // elements are not the same instances
+        .usingElementComparator((x, y) -> x == y ? 0 : 1)
+        .doesNotContainAnyElementsOf(serializedSameEndpoints);
+  }
+}