[LB policies] fix handling of UpdateLocked() result

src/core/load_balancing/endpoint_list.cc

@@ -87,7 +87,7 @@ class EndpointList::Endpoint::Helper final
 // EndpointList::Endpoint
 //
 
-void EndpointList::Endpoint::Init(
+absl::Status EndpointList::Endpoint::Init(
     const EndpointAddresses& addresses, const ChannelArgs& args,
     std::shared_ptr<WorkSerializer> work_serializer) {
   ChannelArgs child_args =
@@ -123,9 +123,7 @@ void EndpointList::Endpoint::Init(
   update_args.addresses = std::make_shared<SingleEndpointIterator>(addresses);
   update_args.args = child_args;
   update_args.config = std::move(*config);
-  // TODO(roth): If the child reports a non-OK status with the update,
-  // we need to propagate that back to the resolver somehow.
-  (void)child_policy_->UpdateLocked(std::move(update_args));
+  return child_policy_->UpdateLocked(std::move(update_args));
 }
 
 void EndpointList::Endpoint::Orphan() {

src/core/load_balancing/endpoint_list.h

@@ -53,7 +53,8 @@ class MyEndpointList : public EndpointList {
  public:
   MyEndpointList(RefCountedPtr<MyLbPolicy> lb_policy,
                  EndpointAddressesIterator* endpoints,
-                 const ChannelArgs& args)
+                 const ChannelArgs& args,
+                 std::vector<std::string>* errors)
       : EndpointList(std::move(lb_policy),
                      GRPC_TRACE_FLAG_ENABLED(grpc_my_tracer)
                          ? "MyEndpointList"
@@ -63,18 +64,23 @@ class MyEndpointList : public EndpointList {
              const EndpointAddresses& addresses, const ChannelArgs& args) {
            return MakeOrphanable<MyEndpoint>(
                std::move(endpoint_list), addresses, args,
-               policy<MyLbPolicy>()->work_serializer());
+               policy<MyLbPolicy>()->work_serializer(), errors);
          });
   }
 
  private:
   class MyEndpoint : public Endpoint {
    public:
     MyEndpoint(RefCountedPtr<MyEndpointList> endpoint_list,
-               const EndpointAddresses& address, const ChannelArgs& args,
-               std::shared_ptr<WorkSerializer> work_serializer)
+               const EndpointAddresses& addresses, const ChannelArgs& args,
+               std::shared_ptr<WorkSerializer> work_serializer,
+               std::vector<std::string>* errors)
         : Endpoint(std::move(endpoint_list)) {
-      Init(addresses, args, std::move(work_serializer));
+      absl::Status status = Init(addresses, args, std::move(work_serializer));
+      if (!status.ok()) {
+        errors->emplace_back(absl::StrCat(
+            "endpoint ", addresses.ToString(), ": ", status.ToString()));
+      }
     }
 
    private:
@@ -120,8 +126,9 @@ class EndpointList : public InternallyRefCounted<EndpointList> {
     explicit Endpoint(RefCountedPtr<EndpointList> endpoint_list)
         : endpoint_list_(std::move(endpoint_list)) {}
 
-    void Init(const EndpointAddresses& addresses, const ChannelArgs& args,
-              std::shared_ptr<WorkSerializer> work_serializer);
+    absl::Status Init(const EndpointAddresses& addresses,
+                      const ChannelArgs& args,
+                      std::shared_ptr<WorkSerializer> work_serializer);
 
     // Templated for convenience, to provide a short-hand for
     // down-casting in the caller.

src/core/load_balancing/priority/priority.cc

@@ -415,11 +415,16 @@ void PriorityLb::ChoosePriorityLocked() {
           child_name);
       auto child_config = config_->children().find(child_name);
       GPR_DEBUG_ASSERT(child_config != config_->children().end());
-      // TODO(roth): If the child reports a non-OK status with the
-      // update, we need to propagate that back to the resolver somehow.
-      (void)child->UpdateLocked(
+      // If the child policy returns a non-OK status, request re-resolution.
+      // Note that this will initially cause fixed backoff delay in the
+      // resolver instead of exponential delay.  However, once the
+      // resolver returns the initial re-resolution, we will be able to
+      // return non-OK from UpdateLocked(), which will trigger
+      // exponential backoff instead.
+      absl::Status status = child->UpdateLocked(
           child_config->second.config,
           child_config->second.ignore_reresolution_requests);
+      if (!status.ok()) channel_control_helper()->RequestReresolution();
     } else {
       // The child already exists.  Reactivate if needed.
       child->MaybeReactivateLocked();

src/core/load_balancing/ring_hash/ring_hash.cc

@@ -169,7 +169,7 @@ class RingHash final : public LoadBalancingPolicy {
 
     size_t index() const { return index_; }
 
-    void UpdateLocked(size_t index);
+    absl::Status UpdateLocked(size_t index);
 
     grpc_connectivity_state connectivity_state() const {
       return connectivity_state_;
@@ -196,7 +196,7 @@ class RingHash final : public LoadBalancingPolicy {
     class Helper;
 
     void CreateChildPolicy();
-    void UpdateChildPolicyLocked();
+    absl::Status UpdateChildPolicyLocked();
 
     // Called when the child policy reports a connectivity state update.
     void OnStateUpdate(grpc_connectivity_state new_state,
@@ -498,9 +498,10 @@ void RingHash::RingHashEndpoint::Orphan() {
   Unref();
 }
 
-void RingHash::RingHashEndpoint::UpdateLocked(size_t index) {
+absl::Status RingHash::RingHashEndpoint::UpdateLocked(size_t index) {
   index_ = index;
-  if (child_policy_ != nullptr) UpdateChildPolicyLocked();
+  if (child_policy_ == nullptr) return absl::OkStatus();
+  return UpdateChildPolicyLocked();
 }
 
 void RingHash::RingHashEndpoint::ResetBackoffLocked() {
@@ -541,10 +542,19 @@ void RingHash::RingHashEndpoint::CreateChildPolicy() {
   // this policy, which in turn is tied to the application's call.
   grpc_pollset_set_add_pollset_set(child_policy_->interested_parties(),
                                    ring_hash_->interested_parties());
-  UpdateChildPolicyLocked();
+  // If the child policy returns a non-OK status, request re-resolution.
+  // Note that this will initially cause fixed backoff delay in the
+  // resolver instead of exponential delay.  However, once the
+  // resolver returns the initial re-resolution, we will be able to
+  // return non-OK from UpdateLocked(), which will trigger
+  // exponential backoff instead.
+  absl::Status status = UpdateChildPolicyLocked();
+  if (!status.ok()) {
+    ring_hash_->channel_control_helper()->RequestReresolution();
+  }
 }
 
-void RingHash::RingHashEndpoint::UpdateChildPolicyLocked() {
+absl::Status RingHash::RingHashEndpoint::UpdateChildPolicyLocked() {
   // Construct pick_first config.
   auto config =
       CoreConfiguration::Get().lb_policy_registry().ParseLoadBalancingConfig(
@@ -557,9 +567,7 @@ void RingHash::RingHashEndpoint::UpdateChildPolicyLocked() {
       std::make_shared<SingleEndpointIterator>(ring_hash_->endpoints_[index_]);
   update_args.args = ring_hash_->args_;
   update_args.config = std::move(*config);
-  // TODO(roth): If the child reports a non-OK status with the update,
-  // we need to propagate that back to the resolver somehow.
-  (void)child_policy_->UpdateLocked(std::move(update_args));
+  return child_policy_->UpdateLocked(std::move(update_args));
 }
 
 void RingHash::RingHashEndpoint::OnStateUpdate(
@@ -667,13 +675,18 @@ absl::Status RingHash::UpdateLocked(UpdateArgs args) {
       this, static_cast<RingHashLbConfig*>(args.config.get()));
   // Update endpoint map.
   std::map<EndpointAddressSet, OrphanablePtr<RingHashEndpoint>> endpoint_map;
+  std::vector<std::string> errors;
   for (size_t i = 0; i < endpoints_.size(); ++i) {
     const EndpointAddresses& addresses = endpoints_[i];
     const EndpointAddressSet address_set(addresses.addresses());
     // If present in old map, retain it; otherwise, create a new one.
     auto it = endpoint_map_.find(address_set);
     if (it != endpoint_map_.end()) {
-      it->second->UpdateLocked(i);
+      absl::Status status = it->second->UpdateLocked(i);
+      if (!status.ok()) {
+        errors.emplace_back(absl::StrCat("endpoint ", address_set.ToString(),
+                                         ": ", status.ToString()));
+      }
       endpoint_map.emplace(address_set, std::move(it->second));
     } else {
       endpoint_map.emplace(address_set, MakeOrphanable<RingHashEndpoint>(
@@ -695,6 +708,10 @@ absl::Status RingHash::UpdateLocked(UpdateArgs args) {
   // Return a new picker.
   UpdateAggregatedConnectivityStateLocked(/*entered_transient_failure=*/false,
                                           absl::OkStatus());
+  if (!errors.empty()) {
+    return absl::UnavailableError(absl::StrCat(
+        "errors from children: [", absl::StrJoin(errors, "; "), "]"));
+  }
   return absl::OkStatus();
 }
 

src/core/load_balancing/rls/rls.cc

@@ -1860,9 +1860,16 @@ void RlsLb::RlsRequest::OnRlsCallCompleteLocked(grpc_error_handle error) {
   // Now that we've released the lock, finish the update on any newly
   // created child policies.
   for (ChildPolicyWrapper* child : child_policies_to_finish_update) {
-    // TODO(roth): If the child reports an error with the update, we
-    // need to propagate that back to the resolver somehow.
-    (void)child->MaybeFinishUpdate();
+    // If the child policy returns a non-OK status, request re-resolution.
+    // Note that this will initially cause fixed backoff delay in the
+    // resolver instead of exponential delay.  However, once the
+    // resolver returns the initial re-resolution, we will be able to
+    // return non-OK from UpdateLocked(), which will trigger
+    // exponential backoff instead.
+    absl::Status status = child->MaybeFinishUpdate();
+    if (!status.ok()) {
+      lb_policy_->channel_control_helper()->RequestReresolution();
+    }
   }
 }
 

src/core/load_balancing/round_robin/round_robin.cc

@@ -72,7 +72,8 @@ class RoundRobin final : public LoadBalancingPolicy {
    public:
     RoundRobinEndpointList(RefCountedPtr<RoundRobin> round_robin,
                            EndpointAddressesIterator* endpoints,
-                           const ChannelArgs& args)
+                           const ChannelArgs& args,
+                           std::vector<std::string>* errors)
         : EndpointList(std::move(round_robin),
                        GRPC_TRACE_FLAG_ENABLED(grpc_lb_round_robin_trace)
                            ? "RoundRobinEndpointList"
@@ -82,7 +83,7 @@ class RoundRobin final : public LoadBalancingPolicy {
                const EndpointAddresses& addresses, const ChannelArgs& args) {
              return MakeOrphanable<RoundRobinEndpoint>(
                  std::move(endpoint_list), addresses, args,
-                 policy<RoundRobin>()->work_serializer());
+                 policy<RoundRobin>()->work_serializer(), errors);
            });
     }
 
@@ -92,9 +93,14 @@ class RoundRobin final : public LoadBalancingPolicy {
       RoundRobinEndpoint(RefCountedPtr<EndpointList> endpoint_list,
                          const EndpointAddresses& addresses,
                          const ChannelArgs& args,
-                         std::shared_ptr<WorkSerializer> work_serializer)
+                         std::shared_ptr<WorkSerializer> work_serializer,
+                         std::vector<std::string>* errors)
           : Endpoint(std::move(endpoint_list)) {
-        Init(addresses, args, std::move(work_serializer));
+        absl::Status status = Init(addresses, args, std::move(work_serializer));
+        if (!status.ok()) {
+          errors->emplace_back(absl::StrCat("endpoint ", addresses.ToString(),
+                                            ": ", status.ToString()));
+        }
       }
 
      private:
@@ -255,9 +261,10 @@ absl::Status RoundRobin::UpdateLocked(UpdateArgs args) {
     gpr_log(GPR_INFO, "[RR %p] replacing previous pending child list %p", this,
             latest_pending_endpoint_list_.get());
   }
+  std::vector<std::string> errors;
   latest_pending_endpoint_list_ = MakeOrphanable<RoundRobinEndpointList>(
       RefAsSubclass<RoundRobin>(DEBUG_LOCATION, "RoundRobinEndpointList"),
-      addresses, args.args);
+      addresses, args.args, &errors);
   // If the new list is empty, immediately promote it to
   // endpoint_list_ and report TRANSIENT_FAILURE.
   if (latest_pending_endpoint_list_->size() == 0) {
@@ -281,6 +288,10 @@ absl::Status RoundRobin::UpdateLocked(UpdateArgs args) {
   if (endpoint_list_ == nullptr) {
     endpoint_list_ = std::move(latest_pending_endpoint_list_);
   }
+  if (!errors.empty()) {
+    return absl::UnavailableError(absl::StrCat(
+        "errors from children: [", absl::StrJoin(errors, "; "), "]"));
+  }
   return absl::OkStatus();
 }
 

src/core/load_balancing/weighted_round_robin/weighted_round_robin.cc

@@ -221,11 +221,16 @@ class WeightedRoundRobin final : public LoadBalancingPolicy {
      public:
       WrrEndpoint(RefCountedPtr<EndpointList> endpoint_list,
                   const EndpointAddresses& addresses, const ChannelArgs& args,
-                  std::shared_ptr<WorkSerializer> work_serializer)
+                  std::shared_ptr<WorkSerializer> work_serializer,
+                  std::vector<std::string>* errors)
           : Endpoint(std::move(endpoint_list)),
             weight_(policy<WeightedRoundRobin>()->GetOrCreateWeight(
                 addresses.addresses())) {
-        Init(addresses, args, std::move(work_serializer));
+        absl::Status status = Init(addresses, args, std::move(work_serializer));
+        if (!status.ok()) {
+          errors->emplace_back(absl::StrCat("endpoint ", addresses.ToString(),
+                                            ": ", status.ToString()));
+        }
       }
 
       RefCountedPtr<EndpointWeight> weight() const { return weight_; }
@@ -261,7 +266,7 @@ class WeightedRoundRobin final : public LoadBalancingPolicy {
 
     WrrEndpointList(RefCountedPtr<WeightedRoundRobin> wrr,
                     EndpointAddressesIterator* endpoints,
-                    const ChannelArgs& args)
+                    const ChannelArgs& args, std::vector<std::string>* errors)
         : EndpointList(std::move(wrr),
                        GRPC_TRACE_FLAG_ENABLED(grpc_lb_wrr_trace)
                            ? "WrrEndpointList"
@@ -271,7 +276,7 @@ class WeightedRoundRobin final : public LoadBalancingPolicy {
                const EndpointAddresses& addresses, const ChannelArgs& args) {
              return MakeOrphanable<WrrEndpoint>(
                  std::move(endpoint_list), addresses, args,
-                 policy<WeightedRoundRobin>()->work_serializer());
+                 policy<WeightedRoundRobin>()->work_serializer(), errors);
            });
     }
 
@@ -767,8 +772,9 @@ absl::Status WeightedRoundRobin::UpdateLocked(UpdateArgs args) {
     gpr_log(GPR_INFO, "[WRR %p] replacing previous pending endpoint list %p",
             this, latest_pending_endpoint_list_.get());
   }
+  std::vector<std::string> errors;
   latest_pending_endpoint_list_ = MakeOrphanable<WrrEndpointList>(
-      RefAsSubclass<WeightedRoundRobin>(), addresses.get(), args.args);
+      RefAsSubclass<WeightedRoundRobin>(), addresses.get(), args.args, &errors);
   // If the new list is empty, immediately promote it to
   // endpoint_list_ and report TRANSIENT_FAILURE.
   if (latest_pending_endpoint_list_->size() == 0) {
@@ -792,6 +798,10 @@ absl::Status WeightedRoundRobin::UpdateLocked(UpdateArgs args) {
   if (endpoint_list_.get() == nullptr) {
     endpoint_list_ = std::move(latest_pending_endpoint_list_);
   }
+  if (!errors.empty()) {
+    return absl::UnavailableError(absl::StrCat(
+        "errors from children: [", absl::StrJoin(errors, "; "), "]"));
+  }
   return absl::OkStatus();
 }