config: unification of delta and SotW gRPC xDS

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

@@ -29,15 +29,12 @@ message ApiConfigSource {
     // the v2 protos is used.
     REST = 1;
 
-    // gRPC v2 API.
+    // "State of the world" gRPC v2 API, using Discovery{Request,Response} protos.
     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 v2 API, using DeltaDiscovery{Request,Response} protos.
+    // Rather than sending Envoy the entire state with every update, the xDS server
+    // only sends what has changed since the last update.
     DELTA_GRPC = 3;
   }
 

api/envoy/api/v3alpha/core/config_source.proto

@@ -29,15 +29,12 @@ message ApiConfigSource {
     // the v2 protos is used.
     REST = 1;
 
-    // gRPC v2 API.
+    // "State of the world" gRPC v2 API, using Discovery{Request,Response} protos.
     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 v2 API, using DeltaDiscovery{Request,Response} protos.
+    // Rather than sending Envoy the entire state with every update, the xDS server
+    // only sends what has changed since the last update.
     DELTA_GRPC = 3;
   }
 

include/envoy/config/grpc_mux.h

@@ -25,48 +25,6 @@ struct ControlPlaneStats {
   ALL_CONTROL_PLANE_STATS(GENERATE_COUNTER_STRUCT, GENERATE_GAUGE_STRUCT)
 };
 
-// TODO(fredlas) redundant to SubscriptionCallbacks; remove this one.
-class GrpcMuxCallbacks {
-public:
-  virtual ~GrpcMuxCallbacks() = default;
-
-  /**
-   * Called when a configuration update is received.
-   * @param resources vector of fetched resources corresponding to the configuration update.
-   * @param version_info update version.
-   * @throw EnvoyException with reason if the configuration is rejected. Otherwise the configuration
-   *        is accepted. Accepted configurations have their version_info reflected in subsequent
-   *        requests.
-   */
-  virtual void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
-                              const std::string& version_info) PURE;
-
-  /**
-   * Called when either the subscription is unable to fetch a config update or when onConfigUpdate
-   * invokes an exception.
-   * @param reason supplies the update failure reason.
-   * @param e supplies any exception data on why the fetch failed. May be nullptr.
-   */
-  virtual void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,
-                                    const EnvoyException* e) PURE;
-
-  /**
-   * Obtain the "name" of a v2 API resource in a google.protobuf.Any, e.g. the route config name for
-   * a RouteConfiguration, based on the underlying resource type.
-   */
-  virtual std::string resourceName(const ProtobufWkt::Any& resource) PURE;
-};
-
-/**
- * Handle on an muxed gRPC subscription. The subscription is canceled on destruction.
- */
-class GrpcMuxWatch {
-public:
-  virtual ~GrpcMuxWatch() = default;
-};
-
-using GrpcMuxWatchPtr = std::unique_ptr<GrpcMuxWatch>;
-
 struct Watch;
 
 /**
@@ -82,27 +40,12 @@ class GrpcMux {
    */
   virtual void start() PURE;
 
-  /**
-   * Start a configuration subscription asynchronously for some API type and resources.
-   * @param type_url type URL corresponding to xDS API, e.g.
-   * type.googleapis.com/envoy.api.v2.Cluster.
-   * @param resources set of resource names to watch for. If this is empty, then all
-   *                  resources for type_url will result in callbacks.
-   * @param callbacks the callbacks to be notified of configuration updates. These must be valid
-   *                  until GrpcMuxWatch is destroyed.
-   * @return GrpcMuxWatchPtr a handle to cancel the subscription with. E.g. when a cluster goes
-   * away, its EDS updates should be cancelled by destroying the GrpcMuxWatchPtr.
-   */
-  virtual GrpcMuxWatchPtr subscribe(const std::string& type_url,
-                                    const std::set<std::string>& resources,
-                                    GrpcMuxCallbacks& callbacks) PURE;
-
   /**
    * Pause discovery requests for a given API type. This is useful when we're processing an update
    * for LDS or CDS and don't want a flood of updates for RDS or EDS respectively. Discovery
    * requests may later be resumed with resume().
    * @param type_url type URL corresponding to xDS API, e.g.
-   * type.googleapis.com/envoy.api.v2.Cluster.
+   * type.googleapis.com/envoy.api.v2.Cluster
    */
   virtual void pause(const std::string& type_url) PURE;
 
@@ -113,11 +56,30 @@ class GrpcMux {
    */
   virtual void resume(const std::string& type_url) PURE;
 
-  // For delta
+  /**
+   * Registers a GrpcSubscription with the GrpcMux. 'watch' may be null (meaning this is an add),
+   * or it may be the Watch* previously returned by this function (which makes it an update).
+   * @param type_url type URL corresponding to xDS API e.g. type.googleapis.com/envoy.api.v2.Cluster
+   * @param watch the Watch* to be updated, or nullptr to add one.
+   * @param resources the set of resource names for 'watch' to start out interested in. If empty,
+   *                  'watch' is treated as interested in *all* resources (of type type_url).
+   * @param callbacks the callbacks that receive updates for 'resources' when they arrive.
+   * @param init_fetch_timeout how long to wait for this new subscription's first update. Ignored
+   *                           unless the addOrUpdateWatch() call is the first for 'type_url'.
+   * @return Watch* the opaque watch token added or updated, to be used in future addOrUpdateWatch
+   *                calls.
+   */
   virtual Watch* addOrUpdateWatch(const std::string& type_url, Watch* watch,
                                   const std::set<std::string>& resources,
                                   SubscriptionCallbacks& callbacks,
                                   std::chrono::milliseconds init_fetch_timeout) PURE;
+
+  /**
+   * Cleanup of a Watch* added by addOrUpdateWatch(). Receiving a Watch* from addOrUpdateWatch()
+   * makes you responsible for eventually invoking this cleanup.
+   * @param type_url type URL corresponding to xDS API e.g. type.googleapis.com/envoy.api.v2.Cluster
+   * @param watch the watch to be cleaned up.
+   */
   virtual void removeWatch(const std::string& type_url, Watch* watch) PURE;
 
   /**
@@ -127,6 +89,12 @@ class GrpcMux {
    * @return bool whether the API is paused.
    */
   virtual bool paused(const std::string& type_url) const PURE;
+
+  /**
+   * Passes through to all multiplexed SubscriptionStates. To be called when something
+   * definitive happens with the initial fetch: either an update is successfully received,
+   * or some sort of error happened.*/
+  virtual void disableInitFetchTimeoutTimer() PURE;
 };
 
 using GrpcMuxPtr = std::unique_ptr<GrpcMux>;

include/envoy/config/subscription_factory.h

@@ -24,7 +24,7 @@ class SubscriptionFactory {
   virtual SubscriptionPtr
   subscriptionFromConfigSource(const envoy::api::v2::core::ConfigSource& config,
                                absl::string_view type_url, Stats::Scope& scope,
-                               SubscriptionCallbacks& callbacks, bool is_delta) PURE;
+                               SubscriptionCallbacks& callbacks) PURE;
 };
 
 } // namespace Config

include/envoy/router/route_config_provider_manager.h

@@ -39,7 +39,7 @@ class RouteConfigProviderManager {
   virtual RouteConfigProviderPtr createRdsRouteConfigProvider(
       const envoy::config::filter::network::http_connection_manager::v2::Rds& rds,
       Server::Configuration::FactoryContext& factory_context, const std::string& stat_prefix,
-      Init::Manager& init_manager, bool is_delta) PURE;
+      Init::Manager& init_manager) PURE;
 
   /**
    * Get a RouteConfigSharedPtr for a statically defined route. Ownership is as described for

include/envoy/server/listener_manager.h

@@ -39,8 +39,7 @@ class ListenerComponentFactory {
    * @return an LDS API provider.
    * @param lds_config supplies the management server configuration.
    */
-  virtual LdsApiPtr createLdsApi(const envoy::api::v2::core::ConfigSource& lds_config,
-                                 bool is_delta) PURE;
+  virtual LdsApiPtr createLdsApi(const envoy::api::v2::core::ConfigSource& lds_config) PURE;
 
   /**
    * Creates a socket.
@@ -129,8 +128,7 @@ class ListenerManager {
    * pieces of the server existing.
    * @param lds_config supplies the management server configuration.
    */
-  virtual void createLdsApi(const envoy::api::v2::core::ConfigSource& lds_config,
-                            bool is_delta) PURE;
+  virtual void createLdsApi(const envoy::api::v2::core::ConfigSource& lds_config) PURE;
 
   /**
    * @return std::vector<std::reference_wrapper<Network::ListenerConfig>> a list of the currently

include/envoy/upstream/cluster_manager.h

@@ -223,8 +223,6 @@ class ClusterManager {
   virtual Config::SubscriptionFactory& subscriptionFactory() PURE;
 
   virtual std::size_t warmingClusterCount() const PURE;
-
-  virtual bool xdsIsDelta() const PURE;
 };
 
 using ClusterManagerPtr = std::unique_ptr<ClusterManager>;
@@ -298,7 +296,7 @@ class ClusterManagerFactory {
   /**
    * Create a CDS API provider from configuration proto.
    */
-  virtual CdsApiPtr createCds(const envoy::api::v2::core::ConfigSource& cds_config, bool is_delta,
+  virtual CdsApiPtr createCds(const envoy::api::v2::core::ConfigSource& cds_config,
                               ClusterManager& cm) PURE;
 
   /**

source/common/config/BUILD

@@ -67,12 +67,12 @@ envoy_cc_library(
 )
 
 envoy_cc_library(
-    name = "delta_subscription_lib",
-    srcs = ["delta_subscription_impl.cc"],
-    hdrs = ["delta_subscription_impl.h"],
+    name = "grpc_subscription_lib",
+    srcs = ["grpc_subscription_impl.cc"],
+    hdrs = ["grpc_subscription_impl.h"],
     deps = [
+        ":grpc_mux_lib",
         ":grpc_stream_lib",
-        ":new_grpc_mux_lib",
         ":utility_lib",
         "//include/envoy/config:subscription_interface",
         "//include/envoy/grpc:async_client_interface",
@@ -90,16 +90,35 @@ envoy_cc_library(
     srcs = ["delta_subscription_state.cc"],
     hdrs = ["delta_subscription_state.h"],
     deps = [
-        ":pausable_ack_queue_lib",
+        ":subscription_state_lib",
+        "//source/common/grpc:common_lib",
+        "//source/common/protobuf",
+        "@envoy_api//envoy/service/discovery/v2:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_library(
+    name = "sotw_subscription_state_lib",
+    srcs = ["sotw_subscription_state.cc"],
+    hdrs = ["sotw_subscription_state.h"],
+    deps = [
+        ":subscription_state_lib",
+        "//source/common/grpc:common_lib",
+        "//source/common/protobuf",
+        "@envoy_api//envoy/service/discovery/v2:pkg_cc_proto",
+    ],
+)
+
+envoy_cc_library(
+    name = "subscription_state_lib",
+    srcs = ["subscription_state.cc"],
+    hdrs = ["subscription_state.h"],
+    deps = [
+        ":update_ack_lib",
         "//include/envoy/config:subscription_interface",
         "//include/envoy/event:dispatcher_interface",
-        "//source/common/common:assert_lib",
-        "//source/common/common:backoff_lib",
+        "//include/envoy/local_info:local_info_interface",
         "//source/common/common:minimal_logger_lib",
-        "//source/common/common:token_bucket_impl_lib",
-        "//source/common/grpc:common_lib",
-        "//source/common/protobuf",
-        "@envoy_api//envoy/api/v2:pkg_cc_proto",
     ],
 )
 
@@ -175,56 +194,13 @@ envoy_cc_library(
 
 envoy_cc_library(
     name = "grpc_mux_lib",
-    srcs = ["grpc_mux_impl.cc"],
-    hdrs = ["grpc_mux_impl.h"],
-    deps = [
-        ":grpc_stream_lib",
-        ":utility_lib",
-        "//include/envoy/config:grpc_mux_interface",
-        "//include/envoy/config:subscription_interface",
-        "//include/envoy/upstream:cluster_manager_interface",
-        "//source/common/common:minimal_logger_lib",
-        "//source/common/protobuf",
-    ],
-)
-
-envoy_cc_library(
-    name = "grpc_mux_subscription_lib",
-    srcs = ["grpc_mux_subscription_impl.cc"],
-    hdrs = ["grpc_mux_subscription_impl.h"],
-    deps = [
-        "//include/envoy/config:grpc_mux_interface",
-        "//include/envoy/config:subscription_interface",
-        "//include/envoy/event:dispatcher_interface",
-        "//source/common/common:assert_lib",
-        "//source/common/common:minimal_logger_lib",
-        "//source/common/grpc:common_lib",
-        "//source/common/protobuf",
-        "@envoy_api//envoy/api/v2:pkg_cc_proto",
-    ],
-)
-
-envoy_cc_library(
-    name = "grpc_subscription_lib",
-    hdrs = ["grpc_subscription_impl.h"],
-    deps = [
-        ":grpc_mux_lib",
-        ":grpc_mux_subscription_lib",
-        "//include/envoy/config:subscription_interface",
-        "//include/envoy/event:dispatcher_interface",
-        "//include/envoy/grpc:async_client_interface",
-        "@envoy_api//envoy/api/v2/core:pkg_cc_proto",
-    ],
-)
-
-envoy_cc_library(
-    name = "new_grpc_mux_lib",
     srcs = ["new_grpc_mux_impl.cc"],
     hdrs = ["new_grpc_mux_impl.h"],
     deps = [
         ":delta_subscription_state_lib",
         ":grpc_stream_lib",
         ":pausable_ack_queue_lib",
+        ":sotw_subscription_state_lib",
         ":watch_map_lib",
         "//include/envoy/event:dispatcher_interface",
         "//include/envoy/grpc:async_client_interface",
@@ -281,8 +257,8 @@ envoy_cc_library(
     srcs = ["pausable_ack_queue.cc"],
     hdrs = ["pausable_ack_queue.h"],
     deps = [
+        ":update_ack_lib",
         "//source/common/common:assert_lib",
-        "@envoy_api//envoy/api/v2:pkg_cc_proto",
     ],
 )
 
@@ -355,9 +331,7 @@ envoy_cc_library(
     srcs = ["subscription_factory_impl.cc"],
     hdrs = ["subscription_factory_impl.h"],
     deps = [
-        ":delta_subscription_lib",
         ":filesystem_subscription_lib",
-        ":grpc_mux_subscription_lib",
         ":grpc_subscription_lib",
         ":http_subscription_lib",
         ":type_to_endpoint_lib",
@@ -382,6 +356,14 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "update_ack_lib",
+    hdrs = ["update_ack.h"],
+    deps = [
+        "@envoy_api//envoy/service/discovery/v2:pkg_cc_proto",
+    ],
+)
+
 envoy_cc_library(
     name = "utility_lib",
     srcs = ["utility.cc"],