Merge branch 'xds_client_fail_on_data_errors' into xds_resource_timer_change

Author: markdroth

CMakeLists.txt

@@ -36,6 +36,9 @@ namespace {
 constexpr absl::string_view kServerFeatureIgnoreResourceDeletion =
     "ignore_resource_deletion";
 
+constexpr absl::string_view kServerFeatureFailOnDataErrors =
+    "fail_on_data_errors";
+
 constexpr absl::string_view kServerFeatureResourceTimerIsTransientFailure =
     "resource_timer_is_transient_error";
 
@@ -49,6 +52,11 @@ bool GrpcXdsServer::IgnoreResourceDeletion() const {
              kServerFeatureIgnoreResourceDeletion)) != server_features_.end();
 }
 
+bool GrpcXdsServer::FailOnDataErrors() const {
+  return server_features_.find(std::string(kServerFeatureFailOnDataErrors)) !=
+         server_features_.end();
+}
+
 bool GrpcXdsServer::ResourceTimerIsTransientFailure() const {
   return server_features_.find(std::string(
              kServerFeatureResourceTimerIsTransientFailure)) !=
@@ -136,6 +144,7 @@ void GrpcXdsServer::JsonPostLoad(const Json& json, const JsonArgs& args,
           if (feature_json.type() == Json::Type::kString &&
               (feature_json.string() == kServerFeatureIgnoreResourceDeletion ||
 // FIXME: env var guard
+               feature_json.string() == kServerFeatureFailOnDataErrors ||
                feature_json.string() ==
                    kServerFeatureResourceTimerIsTransientFailure ||
                feature_json.string() == kServerFeatureTrustedXdsServer)) {

build_autogenerated.yaml

@@ -35,6 +35,7 @@ class GrpcXdsServer final : public XdsBootstrap::XdsServer {
   const std::string& server_uri() const override { return server_uri_; }
 
   bool IgnoreResourceDeletion() const override;
+  bool FailOnDataErrors() const override;
   bool ResourceTimerIsTransientFailure() const override;
 
   bool TrustedXdsServer() const;

src/core/xds/grpc/xds_server_grpc.cc

@@ -34,4 +34,13 @@ bool XdsFederationEnabled() {
   return parse_succeeded && parsed_value;
 }
 
+// TODO(roth): Remove this once the feature passes interop tests.
+bool XdsDataErrorHandlingEnabled() {
+  auto value = GetEnv("GRPC_EXPERIMENTAL_XDS_DATA_ERROR_HANDLING");
+  if (!value.has_value()) return false;
+  bool parsed_value;
+  bool parse_succeeded = gpr_parse_bool_value(value->c_str(), &parsed_value);
+  return parse_succeeded && parsed_value;
+}
+
 }  // namespace grpc_core

src/core/xds/grpc/xds_server_grpc.h

@@ -26,6 +26,7 @@
 namespace grpc_core {
 
 bool XdsFederationEnabled();
+bool XdsDataErrorHandlingEnabled();
 
 class XdsBootstrap {
  public:
@@ -47,7 +48,11 @@ class XdsBootstrap {
 
     virtual const std::string& server_uri() const = 0;
 
+    // TODO(roth): Remove this method once the data error handling
+    // feature passes interop tests.
     virtual bool IgnoreResourceDeletion() const = 0;
+
+    virtual bool FailOnDataErrors() const = 0;
     virtual bool ResourceTimerIsTransientFailure() const = 0;
 
     virtual bool Equals(const XdsServer& other) const = 0;

src/core/xds/xds_client/xds_bootstrap.cc

@@ -991,8 +991,8 @@ void XdsClient::XdsChannel::AdsCall::ParseResource(
   if (authority_it == xds_client()->authority_state_map_.end()) {
     return;  // Skip resource -- we don't have a subscription for it.
   }
-  // Found authority, so look up type.
   AuthorityState& authority_state = authority_it->second;
+  // Found authority, so look up type.
   auto type_it = authority_state.resource_map.find(context->type);
   if (type_it == authority_state.resource_map.end()) {
     return;  // Skip resource -- we don't have a subscription for it.
@@ -1021,8 +1021,15 @@ void XdsClient::XdsChannel::AdsCall::ParseResource(
   }
   // Update resource state based on whether the resource is valid.
   if (!decode_status.ok()) {
+    // If the fail_on_data_errors server feature is present, drop the
+    // existing cached resource, if any.
+    const bool drop_cached_resource = XdsDataErrorHandlingEnabled() &&
+                                      xds_channel()->server_.FailOnDataErrors();
     resource_state.SetNacked(context->version, decode_status.message(),
-                             context->update_time_);
+                             context->update_time_, drop_cached_resource);
+    // If there is no cached resource (either because we didn't have one
+    // or because we just dropped it due to fail_on_data_errors), then notify
+    // via OnResourceChanged(); otherwise, notify via OnAmbientError().
     if (!resource_state.HasResource()) {
       xds_client()->NotifyWatchersOnResourceChanged(
           resource_state.WatcherStatus(), resource_state.watchers(),
@@ -1220,7 +1227,9 @@ void XdsClient::XdsChannel::AdsCall::OnRecvMessage(absl::string_view payload) {
               // that the resource does not exist.  For that case, we rely on
               // the request timeout instead.
               if (!resource_state.HasResource()) continue;
-              if (xds_channel()->server_.IgnoreResourceDeletion()) {
+              if (XdsDataErrorHandlingEnabled()
+                      ? !xds_channel()->server_.FailOnDataErrors()
+                      : xds_channel()->server_.IgnoreResourceDeletion()) {
                 if (!resource_state.ignored_deletion()) {
                   LOG(ERROR)
                       << "[xds_client " << xds_client() << "] xds server "
@@ -1338,7 +1347,9 @@ void XdsClient::ResourceState::SetAcked(
 
 void XdsClient::ResourceState::SetNacked(const std::string& version,
                                          absl::string_view details,
-                                         Timestamp update_time) {
+                                         Timestamp update_time,
+                                         bool drop_cached_resource) {
+  if (drop_cached_resource) resource_.reset();
   client_status_ = ClientResourceStatus::NACKED;
   failed_version_ = version;
   failed_details_ = details;

src/core/xds/xds_client/xds_bootstrap.h

@@ -309,7 +309,7 @@ class XdsClient : public DualRefCounted<XdsClient> {
                   std::string serialized_proto, std::string version,
                   Timestamp update_time);
     void SetNacked(const std::string& version, absl::string_view details,
-                   Timestamp update_time);
+                   Timestamp update_time, bool drop_cached_resource);
     void SetDoesNotExist();
     void SetTransientError(const std::string& details);