config: do not finish initialization on stream disconnection

include/envoy/config/BUILD

@@ -33,6 +33,7 @@ envoy_cc_library(
  name = "grpc_mux_interface",
  hdrs = ["grpc_mux.h"],
  deps = [
+ ":subscription_interface",
  "//include/envoy/stats:stats_macros",
  "//source/common/protobuf",
  ],

include/envoy/config/grpc_mux.h

@@ -2,6 +2,7 @@
 
 #include "envoy/common/exception.h"
 #include "envoy/common/pure.h"
+#include "envoy/config/subscription.h"
 #include "envoy/stats/stats_macros.h"
 
 #include "common/protobuf/protobuf.h"
@@ -42,9 +43,11 @@ class GrpcMuxCallbacks {
  /**
  * 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(const EnvoyException* e) PURE;
+ 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

include/envoy/config/subscription.h

@@ -13,6 +13,18 @@
 namespace Envoy {
 namespace Config {
 
+/**
+ * Reason that a config update is failed.
+ */
+enum class ConfigUpdateFailureReason {
+ // A connection failure took place and the update could not be fetched.
+ ConnectionFailure,
+ // Config fetch timed out.
+ FetchTimedout,
+ // Update rejected because there is a problem in applying the update.
+ UpdateRejected
+};
+
 class SubscriptionCallbacks {
 public:
  virtual ~SubscriptionCallbacks() = default;
@@ -45,9 +57,10 @@ class SubscriptionCallbacks {
  /**
  * 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(const EnvoyException* e) PURE;
+ virtual void onConfigUpdateFailed(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

source/common/config/delta_subscription_state.cc

@@ -25,7 +25,8 @@ void DeltaSubscriptionState::setInitFetchTimeout(Event::Dispatcher& dispatcher)
  if (init_fetch_timeout_.count() > 0 && !init_fetch_timeout_timer_) {
  init_fetch_timeout_timer_ = dispatcher.createTimer([this]() -> void {
  ENVOY_LOG(warn, "delta config: initial fetch timed out for {}", type_url_);
- callbacks_.onConfigUpdateFailed(nullptr);
+ callbacks_.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::FetchTimedout,
+ nullptr);
  });
  init_fetch_timeout_timer_->enableTimer(init_fetch_timeout_);
  }
@@ -145,14 +146,15 @@ void DeltaSubscriptionState::handleBadResponse(const EnvoyException& e, UpdateAc
  disableInitFetchTimeoutTimer();
  stats_.update_rejected_.inc();
  ENVOY_LOG(warn, "delta config for {} rejected: {}", type_url_, e.what());
- callbacks_.onConfigUpdateFailed(&e);
+ callbacks_.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::UpdateRejected, &e);
 }
 
 void DeltaSubscriptionState::handleEstablishmentFailure() {
  disableInitFetchTimeoutTimer();
  stats_.update_failure_.inc();
  stats_.update_attempt_.inc();
- callbacks_.onConfigUpdateFailed(nullptr);
+ callbacks_.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::ConnectionFailure,
+ nullptr);
 }
 
 envoy::api::v2::DeltaDiscoveryRequest DeltaSubscriptionState::getNextRequest() {

source/common/config/filesystem_subscription_impl.cc

@@ -49,11 +49,15 @@ void FilesystemSubscriptionImpl::refresh() {
  ENVOY_LOG(warn, "Filesystem config update rejected: {}", e.what());
  ENVOY_LOG(debug, "Failed configuration:\n{}", message.DebugString());
  stats_.update_rejected_.inc();
+ callbacks_.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::UpdateRejected, &e);
  } else {
  ENVOY_LOG(warn, "Filesystem config update failure: {}", e.what());
  stats_.update_failure_.inc();
+ // ConnectionFailure is not a meaningful error code for file system but it has been chosen so
+ // that the behaviour is uniform across all subscription types.
+ callbacks_.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::ConnectionFailure,
+ &e);
  }
- callbacks_.onConfigUpdateFailed(&e);
  }
 }
 

source/common/config/grpc_mux_impl.cc

@@ -192,7 +192,8 @@ void GrpcMuxImpl::onDiscoveryResponse(
  api_state_[type_url].request_.set_version_info(message->version_info());
  } catch (const EnvoyException& e) {
  for (auto watch : api_state_[type_url].watches_) {
- watch->callbacks_.onConfigUpdateFailed(&e);
+ watch->callbacks_.onConfigUpdateFailed(
+ Envoy::Config::ConfigUpdateFailureReason::UpdateRejected, &e);
  }
  ::google::rpc::Status* error_detail = api_state_[type_url].request_.mutable_error_detail();
  error_detail->set_code(Grpc::Status::GrpcStatus::Internal);
@@ -213,7 +214,8 @@ void GrpcMuxImpl::onStreamEstablished() {
 void GrpcMuxImpl::onEstablishmentFailure() {
  for (const auto& api_state : api_state_) {
  for (auto watch : api_state.second.watches_) {
- watch->callbacks_.onConfigUpdateFailed(nullptr);
+ watch->callbacks_.onConfigUpdateFailed(
+ Envoy::Config::ConfigUpdateFailureReason::ConnectionFailure, nullptr);
  }
  }
 }

source/common/config/grpc_mux_subscription_impl.cc

@@ -22,8 +22,8 @@ GrpcMuxSubscriptionImpl::GrpcMuxSubscriptionImpl(GrpcMux& grpc_mux,
 void GrpcMuxSubscriptionImpl::start(const std::set<std::string>& resources) {
  if (init_fetch_timeout_.count() > 0) {
  init_fetch_timeout_timer_ = dispatcher_.createTimer([this]() -> void {
- ENVOY_LOG(warn, "gRPC config: initial fetch timed out for {}", type_url_);
- callbacks_.onConfigUpdateFailed(nullptr);
+ callbacks_.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::FetchTimedout,
+  nullptr);
  });
  init_fetch_timeout_timer_->enableTimer(init_fetch_timeout_);
  }
@@ -61,18 +61,27 @@ void GrpcMuxSubscriptionImpl::onConfigUpdate(
  resources.size(), version_info);
 }
 
-void GrpcMuxSubscriptionImpl::onConfigUpdateFailed(const EnvoyException* e) {
- disableInitFetchTimeoutTimer();
- // TODO(htuch): Less fragile signal that this is failure vs. reject.
- if (e == nullptr) {
+void GrpcMuxSubscriptionImpl::onConfigUpdateFailed(ConfigUpdateFailureReason reason,
+  const EnvoyException* e) {
+ switch (reason) {
+ case Envoy::Config::ConfigUpdateFailureReason::ConnectionFailure:
  stats_.update_failure_.inc();
  ENVOY_LOG(debug, "gRPC update for {} failed", type_url_);
- } else {
+ break;
+ case Envoy::Config::ConfigUpdateFailureReason::FetchTimedout:
+ disableInitFetchTimeoutTimer();
+ ENVOY_LOG(warn, "gRPC config: initial fetch timed out for {}", type_url_);
+ break;
+ case Envoy::Config::ConfigUpdateFailureReason::UpdateRejected:
+ // We expect Envoy exception to be thrown when update is rejected.
+ ASSERT(e != nullptr);
+ disableInitFetchTimeoutTimer();
  stats_.update_rejected_.inc();
  ENVOY_LOG(warn, "gRPC config for {} rejected: {}", type_url_, e->what());
+ break;
  }
  stats_.update_attempt_.inc();
- callbacks_.onConfigUpdateFailed(e);
+ callbacks_.onConfigUpdateFailed(reason, e);
 }
 
 std::string GrpcMuxSubscriptionImpl::resourceName(const ProtobufWkt::Any& resource) {

source/common/config/grpc_mux_subscription_impl.h

@@ -29,7 +29,8 @@ class GrpcMuxSubscriptionImpl : public Subscription,
  // Config::GrpcMuxCallbacks
  void onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::Any>& resources,
  const std::string& version_info) override;
- void onConfigUpdateFailed(const EnvoyException* e) override;
+ void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,
+ const EnvoyException* e) override;
  std::string resourceName(const ProtobufWkt::Any& resource) override;
 
 private:

source/common/config/http_subscription_impl.cc

@@ -39,7 +39,8 @@ void HttpSubscriptionImpl::start(const std::set<std::string>& resource_names) {
  if (init_fetch_timeout_.count() > 0) {
  init_fetch_timeout_timer_ = dispatcher_.createTimer([this]() -> void {
  ENVOY_LOG(warn, "REST config: initial fetch timed out for", path_);
- callbacks_.onConfigUpdateFailed(nullptr);
+ callbacks_.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::FetchTimedout,
+ nullptr);
  });
  init_fetch_timeout_timer_->enableTimer(init_fetch_timeout_);
  }
@@ -87,7 +88,7 @@ void HttpSubscriptionImpl::parseResponse(const Http::Message& response) {
  } catch (const EnvoyException& e) {
  ENVOY_LOG(warn, "REST config update rejected: {}", e.what());
  stats_.update_rejected_.inc();
- callbacks_.onConfigUpdateFailed(&e);
+ callbacks_.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::UpdateRejected, &e);
  }
 }
 
@@ -101,7 +102,7 @@ void HttpSubscriptionImpl::onFetchFailure(const EnvoyException* e) {
 
 void HttpSubscriptionImpl::handleFailure(const EnvoyException* e) {
  stats_.update_failure_.inc();
- callbacks_.onConfigUpdateFailed(e);
+ callbacks_.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::ConnectionFailure, e);
 }
 
 void HttpSubscriptionImpl::disableInitFetchTimeoutTimer() {

source/common/router/rds_impl.cc

@@ -143,7 +143,8 @@ void RdsRouteConfigSubscription::onConfigUpdate(
  }
 }
 
-void RdsRouteConfigSubscription::onConfigUpdateFailed(const EnvoyException*) {
+void RdsRouteConfigSubscription::onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason,
+ const EnvoyException*) {
  // We need to allow server startup to continue, even if we have a bad
  // config.
  init_target_.ready();

source/common/router/rds_impl.h

@@ -114,7 +114,8 @@ class RdsRouteConfigSubscription : Envoy::Config::SubscriptionCallbacks,
  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& added_resources,
  const Protobuf::RepeatedPtrField<std::string>& removed_resources,
  const std::string&) override;
- void onConfigUpdateFailed(const EnvoyException* e) override;
+ void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,
+ const EnvoyException* e) override;
  std::string resourceName(const ProtobufWkt::Any& resource) override {
  return MessageUtil::anyConvert<envoy::api::v2::RouteConfiguration>(resource,
  validation_visitor_)

source/common/router/scoped_rds.h

@@ -110,8 +110,9 @@ class ScopedRdsConfigSubscription : public Envoy::Config::DeltaConfigSubscriptio
  const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override {
  NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
  }
- void onConfigUpdateFailed(const EnvoyException*) override {
- DeltaConfigSubscriptionInstance::onConfigUpdateFailed();
+ void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason,
+ const EnvoyException*) override {
+ ConfigSubscriptionCommonBase::onConfigUpdateFailed();
  }
  std::string resourceName(const ProtobufWkt::Any& resource) override {
  return MessageUtil::anyConvert<envoy::api::v2::ScopedRouteConfiguration>(resource,

source/common/router/vhds.cc

@@ -46,7 +46,8 @@ VhdsSubscription::VhdsSubscription(RouteConfigUpdatePtr& config_update_info,
  *scope_, *this);
 }
 
-void VhdsSubscription::onConfigUpdateFailed(const EnvoyException*) {
+void VhdsSubscription::onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason,
+ const EnvoyException*) {
  // We need to allow server startup to continue, even if we have a bad
  // config.
  init_target_.ready();

source/common/router/vhds.h

@@ -56,7 +56,8 @@ class VhdsSubscription : Envoy::Config::SubscriptionCallbacks,
  }
  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>&,
  const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override;
- void onConfigUpdateFailed(const EnvoyException* e) override;
+ void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,
+ const EnvoyException* e) override;
  std::string resourceName(const ProtobufWkt::Any& resource) override {
  return MessageUtil::anyConvert<envoy::api::v2::route::VirtualHost>(resource,
  validation_visitor_)

source/common/runtime/runtime_impl.cc

@@ -526,7 +526,8 @@ void RtdsSubscription::onConfigUpdate(
  onConfigUpdate(unwrapped_resource, resources[0].version());
 }
 
-void RtdsSubscription::onConfigUpdateFailed(const EnvoyException*) {
+void RtdsSubscription::onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason,
+ const EnvoyException*) {
  // We need to allow server startup to continue, even if we have a bad
  // config.
  init_target_.ready();

source/common/runtime/runtime_impl.h

@@ -205,7 +205,8 @@ struct RtdsSubscription : Config::SubscriptionCallbacks, Logger::Loggable<Logger
  const Protobuf::RepeatedPtrField<std::string>& removed_resources,
  const std::string&) override;
 
- void onConfigUpdateFailed(const EnvoyException* e) override;
+ void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,
+ const EnvoyException* e) override;
  std::string resourceName(const ProtobufWkt::Any& resource) override {
  return MessageUtil::anyConvert<envoy::service::discovery::v2::Runtime>(resource,
  validation_visitor_)

source/common/secret/sds_api.cc

@@ -59,7 +59,7 @@ void SdsApi::onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Res
  onConfigUpdate(unwrapped_resource, resources[0].version());
 }
 
-void SdsApi::onConfigUpdateFailed(const EnvoyException*) {
+void SdsApi::onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason, const EnvoyException*) {
  // We need to allow server startup to continue, even if we have a bad config.
  init_target_.ready();
 }

source/common/secret/sds_api.h

@@ -56,7 +56,8 @@ class SdsApi : public Config::SubscriptionCallbacks {
  const std::string& version_info) override;
  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>&,
  const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override;
- void onConfigUpdateFailed(const EnvoyException* e) override;
+ void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,
+ const EnvoyException* e) override;
  std::string resourceName(const ProtobufWkt::Any& resource) override {
  return MessageUtil::anyConvert<envoy::api::v2::auth::Secret>(resource, validation_visitor_)
  .name();

source/common/upstream/cds_api_impl.cc

@@ -98,7 +98,8 @@ void CdsApiImpl::onConfigUpdate(
  }
 }
 
-void CdsApiImpl::onConfigUpdateFailed(const EnvoyException*) {
+void CdsApiImpl::onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason,
+ const EnvoyException*) {
  // We need to allow server startup to continue, even if we have a bad
  // config.
  runInitializeCallbackIfAny();

source/common/upstream/cds_api_impl.h

@@ -40,7 +40,8 @@ class CdsApiImpl : public CdsApi,
  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& added_resources,
  const Protobuf::RepeatedPtrField<std::string>& removed_resources,
  const std::string& system_version_info) override;
- void onConfigUpdateFailed(const EnvoyException* e) override;
+ void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,
+ const EnvoyException* e) override;
  std::string resourceName(const ProtobufWkt::Any& resource) override {
  return MessageUtil::anyConvert<envoy::api::v2::Cluster>(resource, validation_visitor_).name();
  }

source/common/upstream/eds.cc

@@ -251,8 +251,13 @@ bool EdsClusterImpl::updateHostsPerLocality(
  return false;
 }
 
-void EdsClusterImpl::onConfigUpdateFailed(const EnvoyException* e) {
- UNREFERENCED_PARAMETER(e);
+void EdsClusterImpl::onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,
+ const EnvoyException*) {
+ // We should not call onPreInitComplete if this method is called because of stream disconnection.
+ // This might potentially hang the initialization forever, if init_fetch_timeout is disabled.
+ if (reason == Envoy::Config::ConfigUpdateFailureReason::ConnectionFailure) {
+ return;
+ }
  // We need to allow server startup to continue, even if we have a bad config.
  onPreInitComplete();
 }

source/common/upstream/eds.h

@@ -35,7 +35,8 @@ class EdsClusterImpl : public BaseDynamicClusterImpl, Config::SubscriptionCallba
  const std::string& version_info) override;
  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>&,
  const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override;
- void onConfigUpdateFailed(const EnvoyException* e) override;
+ void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,
+ const EnvoyException* e) override;
  std::string resourceName(const ProtobufWkt::Any& resource) override {
  return MessageUtil::anyConvert<envoy::api::v2::ClusterLoadAssignment>(resource,
  validation_visitor_)

source/server/lds_api.cc

@@ -108,7 +108,8 @@ void LdsApiImpl::onConfigUpdate(const Protobuf::RepeatedPtrField<ProtobufWkt::An
  onConfigUpdate(to_add_repeated, to_remove_repeated, version_info);
 }
 
-void LdsApiImpl::onConfigUpdateFailed(const EnvoyException*) {
+void LdsApiImpl::onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason,
+ const EnvoyException*) {
  // We need to allow server startup to continue, even if we have a bad
  // config.
  init_target_.ready();

source/server/lds_api.h

@@ -36,7 +36,8 @@ class LdsApiImpl : public LdsApi,
  void onConfigUpdate(const Protobuf::RepeatedPtrField<envoy::api::v2::Resource>& added_resources,
  const Protobuf::RepeatedPtrField<std::string>& removed_resources,
  const std::string& system_version_info) override;
- void onConfigUpdateFailed(const EnvoyException* e) override;
+ void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,
+ const EnvoyException* e) override;
  std::string resourceName(const ProtobufWkt::Any& resource) override {
  return MessageUtil::anyConvert<envoy::api::v2::Listener>(resource, validation_visitor_).name();
  }

test/common/config/config_provider_impl_test.cc

@@ -91,7 +91,8 @@ class DummyConfigSubscription : public ConfigSubscriptionInstance,
  }
 
  // Envoy::Config::SubscriptionCallbacks
- void onConfigUpdateFailed(const EnvoyException*) override {}
+ void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason,
+ const EnvoyException*) override {}
 
  // Envoy::Config::SubscriptionCallbacks
  std::string resourceName(const ProtobufWkt::Any&) override { return ""; }
@@ -549,7 +550,8 @@ class DeltaDummyConfigSubscription : public DeltaConfigSubscriptionInstance,
  const Protobuf::RepeatedPtrField<std::string>&, const std::string&) override {
  NOT_IMPLEMENTED_GCOVR_EXCL_LINE;
  }
- void onConfigUpdateFailed(const EnvoyException*) override {
+ void onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason,
+ const EnvoyException*) override {
  ConfigSubscriptionCommonBase::onConfigUpdateFailed();
  }
  std::string resourceName(const ProtobufWkt::Any&) override {
@@ -725,7 +727,8 @@ TEST_F(DeltaConfigProviderImplTest, DeltaSubscriptionFailure) {
  timeSystem().setSystemTime(time);
  const EnvoyException ex(fmt::format("config failure"));
  // Verify the failure updates the lastUpdated() timestamp.
- subscription.onConfigUpdateFailed(&ex);
+ subscription.onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason::ConnectionFailure,
+ &ex);
  EXPECT_EQ(std::chrono::time_point_cast<std::chrono::milliseconds>(provider->lastUpdated())
  .time_since_epoch(),
  time);