config: implement incremental CDS in Envoy

api/XDS_PROTOCOL.md

@@ -288,35 +288,33 @@ admin:
 
 ### Incremental xDS
 
-Incremental xDS is a separate xDS endpoint available for ADS, CDS and RDS that
-allows:
-
-  * Incremental updates of the list of tracked resources by the xDS client.
-    This supports Envoy on-demand / lazily requesting additional resources. For
-    example, this may occur when a request corresponding to an unknown cluster
-    arrives.
-  * The xDS server can incrementally update the resources on the client.
-    This supports the goal of scalability of xDS resources. Rather than deliver
-    all 100k clusters when a single cluster is modified, the management server
-    only needs to deliver the single cluster that changed.
-
-An xDS incremental session is always in the context of a gRPC bidirectional
+Incremental xDS is a separate xDS endpoint that:
+
+  * Allows the protocol to communicate on the wire in terms of resource/resource
+    name deltas ("Delta xDS"). This supports the goal of scalability of xDS
+    resources. Rather than deliver all 100k clusters when a single cluster is
+    modified, the management server only needs to deliver the single cluster
+    that changed.
+  * Allows the Envoy to on-demand / lazily request additional resources. For
+    example, requesting a cluster only when a request for that cluster arrives.
+
+An Incremental xDS session is always in the context of a gRPC bidirectional
 stream. This allows the xDS server to keep track of the state of xDS clients
-connected to it. There is no REST version of Incremental xDS.
+connected to it. There is no REST version of Incremental xDS yet.
 
-In incremental xDS the nonce field is required and used to pair a
-[`IncrementalDiscoveryResponse`](https://www.envoyproxy.io/docs/envoy/latest/api-v2/api/v2/discovery.proto#discoveryrequest)
-to a [`IncrementalDiscoveryRequest`](https://www.envoyproxy.io/docs/envoy/latest/api-v2/api/v2/discovery.proto#discoveryrequest)
+In the delta xDS wire protocol, the nonce field is required and used to pair a
+[`DeltaDiscoveryResponse`](https://www.envoyproxy.io/docs/envoy/latest/api-v2/api/v2/discovery.proto#deltadiscoveryresponse)
+to a [`DeltaDiscoveryRequest`](https://www.envoyproxy.io/docs/envoy/latest/api-v2/api/v2/discovery.proto#deltadiscoveryrequest)
 ACK or NACK.
 Optionally, a response message level system_version_info is present for
 debugging purposes only.
 
-`IncrementalDiscoveryRequest` can be sent in 3 situations:
+`DeltaDiscoveryRequest` can be sent in 3 situations:
   1. Initial message in a xDS bidirectional gRPC stream.
-  2. As an ACK or NACK response to a previous `IncrementalDiscoveryResponse`.
+  2. As an ACK or NACK response to a previous `DeltaDiscoveryResponse`.
      In this case the `response_nonce` is set to the nonce value in the Response.
      ACK or NACK is determined by the absence or presence of `error_detail`.
-  3. Spontaneous `IncrementalDiscoveryRequest` from the client.
+  3. Spontaneous `DeltaDiscoveryRequest` from the client.
      This can be done to dynamically add or remove elements from the tracked
      `resource_names` set. In this case `response_nonce` must be omitted.
 
@@ -326,8 +324,8 @@ client spontaneously requests the "wc" resource.
 
 ![Incremental session example](diagrams/incremental.svg)
 
-On reconnect the xDS Incremental client may tell the server of its known resources
-to avoid resending them over the network.
+On reconnect the Incremental xDS client may tell the server of its known
+resources to avoid resending them over the network.
 
 ![Incremental reconnect example](diagrams/incremental-reconnect.svg)
 

api/envoy/api/v2/cds.proto

@@ -36,8 +36,7 @@ service ClusterDiscoveryService {
   rpc StreamClusters(stream DiscoveryRequest) returns (stream DiscoveryResponse) {
   }
 
-  rpc IncrementalClusters(stream IncrementalDiscoveryRequest)
-      returns (stream IncrementalDiscoveryResponse) {
+  rpc DeltaClusters(stream DeltaDiscoveryRequest) returns (stream DeltaDiscoveryResponse) {
   }
 
   rpc FetchClusters(DiscoveryRequest) returns (DiscoveryResponse) {

api/envoy/api/v2/core/config_source.proto

@@ -32,6 +32,13 @@ message ApiConfigSource {
     REST = 1;
     // gRPC v2 API.
     GRPC = 2;
+    // Using the delta xDS gRPC service, i.e. DeltaDiscovery{Request,Response}
+    // rather than Discovery{Request,Response}. Rather than sending Envoy the entire state
+    // with every update, the xDS server only sends what has changed since the last update.
+    //
+    // DELTA_GRPC is not yet entirely implemented! Initially, only CDS is available.
+    // Do not use for other xDSes. TODO(fredlas) update/remove this warning when appropriate.
+    DELTA_GRPC = 3;
   }
   ApiType api_type = 1 [(validate.rules).enum.defined_only = true];
   // Cluster names should be used only with REST. If > 1

api/envoy/api/v2/discovery.proto

@@ -102,33 +102,32 @@ message DiscoveryResponse {
   core.ControlPlane control_plane = 6;
 }
 
-// IncrementalDiscoveryRequest and IncrementalDiscoveryResponse are used in a
-// new gRPC endpoint for Incremental xDS. The feature is not supported for REST
-// management servers.
+// DeltaDiscoveryRequest and DeltaDiscoveryResponse are used in a new gRPC
+// endpoint for Delta xDS.
 //
-// With Incremental xDS, the IncrementalDiscoveryResponses do not need to
-// include a full snapshot of the tracked resources. Instead
-// IncrementalDiscoveryResponses are a diff to the state of a xDS client.
-// In Incremental XDS there are per resource versions which allows to track
-// state at the resource granularity.
-// An xDS Incremental session is always in the context of a gRPC bidirectional
+// With Delta xDS, the DeltaDiscoveryResponses do not need to include a full
+// snapshot of the tracked resources. Instead, DeltaDiscoveryResponses are a
+// diff to the state of a xDS client.
+// In Delta XDS there are per resource versions, which allow tracking state at
+// the resource granularity.
+// An xDS Delta session is always in the context of a gRPC bidirectional
 // stream. This allows the xDS server to keep track of the state of xDS clients
 // connected to it.
 //
-// In Incremental xDS the nonce field is required and used to pair
-// IncrementalDiscoveryResponse to a IncrementalDiscoveryRequest ACK or NACK.
+// In Delta xDS the nonce field is required and used to pair
+// DeltaDiscoveryResponse to a DeltaDiscoveryRequest ACK or NACK.
 // Optionally, a response message level system_version_info is present for
 // debugging purposes only.
 //
-// IncrementalDiscoveryRequest can be sent in 3 situations:
+// DeltaDiscoveryRequest can be sent in 3 situations:
 //   1. Initial message in a xDS bidirectional gRPC stream.
-//   2. As a ACK or NACK response to a previous IncrementalDiscoveryResponse.
+//   2. As a ACK or NACK response to a previous DeltaDiscoveryResponse.
 //      In this case the response_nonce is set to the nonce value in the Response.
 //      ACK or NACK is determined by the absence or presence of error_detail.
-//   3. Spontaneous IncrementalDiscoveryRequest from the client.
+//   3. Spontaneous DeltaDiscoveryRequest from the client.
 //      This can be done to dynamically add or remove elements from the tracked
 //      resource_names set. In this case response_nonce must be omitted.
-message IncrementalDiscoveryRequest {
+message DeltaDiscoveryRequest {
   // The node making the request.
   core.Node node = 1;
 
@@ -138,18 +137,18 @@ message IncrementalDiscoveryRequest {
   // required for ADS.
   string type_url = 2;
 
-  // IncrementalDiscoveryRequests allow the client to add or remove individual
+  // DeltaDiscoveryRequests allow the client to add or remove individual
   // resources to the set of tracked resources in the context of a stream.
   // All resource names in the resource_names_subscribe list are added to the
   // set of tracked resources and all resource names in the resource_names_unsubscribe
   // list are removed from the set of tracked resources.
-  // Unlike in non incremental xDS, an empty resource_names_subscribe or
+  // Unlike in state-of-the-world xDS, an empty resource_names_subscribe or
   // resource_names_unsubscribe list simply means that no resources are to be
   // added or removed to the resource list.
   // The xDS server must send updates for all tracked resources but can also
   // send updates for resources the client has not subscribed to. This behavior
-  // is similar to non incremental xDS.
-  // These two fields can be set for all types of IncrementalDiscoveryRequests
+  // is similar to state-of-the-world xDS.
+  // These two fields can be set for all types of DeltaDiscoveryRequests
   // (initial, ACK/NACK or spontaneous).
   //
   // A list of Resource names to add to the list of tracked resources.
@@ -158,15 +157,17 @@ message IncrementalDiscoveryRequest {
   // A list of Resource names to remove from the list of tracked resources.
   repeated string resource_names_unsubscribe = 4;
 
-  // This map must be populated when the IncrementalDiscoveryRequest is the
-  // first in a stream. The keys are the resources names of the xDS resources
+  // This map must be populated when the DeltaDiscoveryRequest is the
+  // first in a stream (assuming there are any resources - this field's purpose is to enable
+  // a session to continue in a reconnected gRPC stream, and so will not be used in the very
+  // first stream of a session). The keys are the resources names of the xDS resources
   // known to the xDS client. The values in the map are the associated resource
   // level version info.
   map<string, string> initial_resource_versions = 5;
 
-  // When the IncrementalDiscoveryRequest is a ACK or NACK message in response
-  // to a previous IncrementalDiscoveryResponse, the response_nonce must be the
-  // nonce in the IncrementalDiscoveryResponse.
+  // When the DeltaDiscoveryRequest is a ACK or NACK message in response
+  // to a previous DeltaDiscoveryResponse, the response_nonce must be the
+  // nonce in the DeltaDiscoveryResponse.
   // Otherwise response_nonce must be omitted.
   string response_nonce = 6;
 
@@ -176,24 +177,27 @@ message IncrementalDiscoveryRequest {
   google.rpc.Status error_detail = 7;
 }
 
-message IncrementalDiscoveryResponse {
+message DeltaDiscoveryResponse {
   // The version of the response data (used for debugging).
   string system_version_info = 1;
 
   // The response resources. These are typed resources that match the type url
-  // in the IncrementalDiscoveryRequest.
+  // in the DeltaDiscoveryRequest.
   repeated Resource resources = 2 [(gogoproto.nullable) = false];
 
   // Resources names of resources that have be deleted and to be removed from the xDS Client.
   // Removed resources for missing resources can be ignored.
   repeated string removed_resources = 6;
 
-  // The nonce provides a way for IncrementalDiscoveryRequests to uniquely
-  // reference a IncrementalDiscoveryResponse. The nonce is required.
+  // The nonce provides a way for DeltaDiscoveryRequests to uniquely
+  // reference a DeltaDiscoveryResponse. The nonce is required.
   string nonce = 5;
 }
 
 message Resource {
+  // The resource's name, to distinguish it from others of the same type of resource.
+  string name = 3;
+
   // The resource level version. It allows xDS to track the state of individual
   // resources.
   string version = 1;

api/envoy/api/v2/rds.proto

@@ -33,8 +33,7 @@ service RouteDiscoveryService {
   rpc StreamRoutes(stream DiscoveryRequest) returns (stream DiscoveryResponse) {
   }
 
-  rpc IncrementalRoutes(stream IncrementalDiscoveryRequest)
-      returns (stream IncrementalDiscoveryResponse) {
+  rpc DeltaRoutes(stream DeltaDiscoveryRequest) returns (stream DeltaDiscoveryResponse) {
   }
 
   rpc FetchRoutes(DiscoveryRequest) returns (DiscoveryResponse) {

api/envoy/service/discovery/v2/ads.proto

@@ -32,7 +32,7 @@ service AggregatedDiscoveryService {
       returns (stream envoy.api.v2.DiscoveryResponse) {
   }
 
-  rpc IncrementalAggregatedResources(stream envoy.api.v2.IncrementalDiscoveryRequest)
-      returns (stream envoy.api.v2.IncrementalDiscoveryResponse) {
+  rpc DeltaAggregatedResources(stream envoy.api.v2.DeltaDiscoveryRequest)
+      returns (stream envoy.api.v2.DeltaDiscoveryResponse) {
   }
 }

include/envoy/config/BUILD

@@ -23,6 +23,7 @@ envoy_cc_library(
     deps = [
         "//include/envoy/stats:stats_macros",
         "//source/common/protobuf",
+        "@envoy_api//envoy/api/v2:discovery_cc",
     ],
 )
 

include/envoy/config/subscription.h

@@ -3,6 +3,7 @@
 #include <string>
 #include <vector>
 
+#include "envoy/api/v2/discovery.pb.h"
 #include "envoy/common/exception.h"
 #include "envoy/common/pure.h"
 #include "envoy/stats/stats_macros.h"
@@ -29,6 +30,22 @@ template <class ResourceType> class SubscriptionCallbacks {
   virtual void onConfigUpdate(const ResourceVector& resources,
                               const std::string& version_info) PURE;
 
+  // TODO(fredlas) it is a HACK that there are two of these. After delta CDS is merged,
+  //               I intend to reimplement all state-of-the-world xDSes' use of onConfigUpdate
+  //               in terms of this delta-style one (and remove the original).
+  /**
+   * Called when a delta configuration update is received.
+   * @param added_resources resources newly added since the previous fetch.
+   * @param removed_resources names of resources that this fetch instructed to be removed.
+   * @param system_version_info aggregate response data "version", for debugging.
+   * @throw EnvoyException with reason if the config changes are rejected. Otherwise the changes
+   *        are accepted. Accepted changes have their version_info reflected in subsequent requests.
+   */
+  virtual void
+  onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& added_resources,
+                 const Protobuf::RepeatedPtrField<std::string>& removed_resources,
+                 const std::string& system_version_info) PURE;
+
   /**
    * Called when either the Subscription is unable to fetch a config update or when onConfigUpdate
    * invokes an exception.

source/common/config/BUILD

@@ -97,6 +97,22 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "delta_subscription_lib",
+    hdrs = ["delta_subscription_impl.h"],
+    deps = [
+        ":grpc_stream_lib",
+        ":utility_lib",
+        "//include/envoy/config:subscription_interface",
+        "//include/envoy/grpc:async_client_interface",
+        "//include/envoy/upstream:cluster_manager_interface",
+        "//source/common/common:backoff_lib",
+        "//source/common/common:minimal_logger_lib",
+        "//source/common/common:token_bucket_impl_lib",
+        "//source/common/protobuf",
+    ],
+)
+
 envoy_cc_library(
     name = "grpc_stream_lib",
     hdrs = ["grpc_stream.h"],
@@ -303,6 +319,7 @@ envoy_cc_library(
     name = "subscription_factory_lib",
     hdrs = ["subscription_factory.h"],
     deps = [
+        ":delta_subscription_lib",
         ":filesystem_subscription_lib",
         ":grpc_mux_subscription_lib",
         ":grpc_subscription_lib",