src/core/ext/filters/client_channel/resolving_lb_policy.cc

   1 /*
   2  *
   3  * Copyright 2015 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #include <grpc/support/port_platform.h>
  20
  21 #include "src/core/ext/filters/client_channel/resolving_lb_policy.h"
  22
  23 #include <inttypes.h>
  24 #include <limits.h>
  25 #include <stdbool.h>
  26 #include <stdio.h>
  27 #include <string.h>
  28
  29 #include "absl/strings/str_cat.h"
  30 #include "absl/strings/str_join.h"
  31
  32 #include <grpc/support/alloc.h>
  33 #include <grpc/support/log.h>
  34 #include <grpc/support/string_util.h>
  35 #include <grpc/support/sync.h>
  36
  37 #include "src/core/ext/filters/client_channel/backup_poller.h"
  38 #include "src/core/ext/filters/client_channel/http_connect_handshaker.h"
  39 #include "src/core/ext/filters/client_channel/lb_policy/child_policy_handler.h"
  40 #include "src/core/ext/filters/client_channel/lb_policy_registry.h"
  41 #include "src/core/ext/filters/client_channel/proxy_mapper_registry.h"
  42 #include "src/core/ext/filters/client_channel/resolver_registry.h"
  43 #include "src/core/ext/filters/client_channel/retry_throttle.h"
  44 #include "src/core/ext/filters/client_channel/server_address.h"
  45 #include "src/core/ext/filters/client_channel/service_config.h"
  46 #include "src/core/ext/filters/client_channel/subchannel.h"
  47 #include "src/core/ext/filters/deadline/deadline_filter.h"
  48 #include "src/core/lib/backoff/backoff.h"
  49 #include "src/core/lib/channel/channel_args.h"
  50 #include "src/core/lib/channel/connected_channel.h"
  51 #include "src/core/lib/channel/status_util.h"
  52 #include "src/core/lib/gpr/string.h"
  53 #include "src/core/lib/gprpp/manual_constructor.h"
  54 #include "src/core/lib/gprpp/sync.h"
  55 #include "src/core/lib/iomgr/iomgr.h"
  56 #include "src/core/lib/iomgr/polling_entity.h"
  57 #include "src/core/lib/profiling/timers.h"
  58 #include "src/core/lib/slice/slice_internal.h"
  59 #include "src/core/lib/slice/slice_string_helpers.h"
  60 #include "src/core/lib/surface/channel.h"
  61 #include "src/core/lib/transport/connectivity_state.h"
  62 #include "src/core/lib/transport/error_utils.h"
  63 #include "src/core/lib/transport/metadata.h"
  64 #include "src/core/lib/transport/metadata_batch.h"
  65 #include "src/core/lib/transport/static_metadata.h"
  66 #include "src/core/lib/transport/status_metadata.h"
  67
  68 namespace grpc_core {
  69
  70 //
  71 // ResolvingLoadBalancingPolicy::ResolverResultHandler
  72 //
  73
  74 class ResolvingLoadBalancingPolicy::ResolverResultHandler
  75     : public Resolver::ResultHandler {
  76  public:
  77   explicit ResolverResultHandler(
  78       RefCountedPtr<ResolvingLoadBalancingPolicy> parent)
  79       : parent_(std::move(parent)) {}
  80
  81   ~ResolverResultHandler() {
  82     if (GRPC_TRACE_FLAG_ENABLED(*(parent_->tracer_))) {
  83       gpr_log(GPR_INFO, "resolving_lb=%p: resolver shutdown complete",
  84               parent_.get());
  85     }
  86   }
  87
  88   void ReturnResult(Resolver::Result result) override {
  89     parent_->OnResolverResultChangedLocked(std::move(result));
  90   }
  91
  92   void ReturnError(grpc_error* error) override {
  93     parent_->OnResolverError(error);
  94   }
  95
  96  private:
  97   RefCountedPtr<ResolvingLoadBalancingPolicy> parent_;
  98 };
  99
 100 //
 101 // ResolvingLoadBalancingPolicy::ResolvingControlHelper
 102 //
 103
 104 class ResolvingLoadBalancingPolicy::ResolvingControlHelper
 105     : public LoadBalancingPolicy::ChannelControlHelper {
 106  public:
 107   explicit ResolvingControlHelper(
 108       RefCountedPtr<ResolvingLoadBalancingPolicy> parent)
 109       : parent_(std::move(parent)) {}
 110
 111   RefCountedPtr<SubchannelInterface> CreateSubchannel(
 112       const grpc_channel_args& args) override {
 113     if (parent_->resolver_ == nullptr) return nullptr;  // Shutting down.
 114     return parent_->channel_control_helper()->CreateSubchannel(args);
 115   }
 116
 117   void UpdateState(grpc_connectivity_state state,
 118                    std::unique_ptr<SubchannelPicker> picker) override {
 119     if (parent_->resolver_ == nullptr) return;  // Shutting down.
 120     parent_->channel_control_helper()->UpdateState(state, std::move(picker));
 121   }
 122
 123   void RequestReresolution() override {
 124     if (parent_->resolver_ == nullptr) return;  // Shutting down.
 125     if (GRPC_TRACE_FLAG_ENABLED(*(parent_->tracer_))) {
 126       gpr_log(GPR_INFO, "resolving_lb=%p: started name re-resolving",
 127               parent_.get());
 128     }
 129     parent_->resolver_->RequestReresolutionLocked();
 130   }
 131
 132   void AddTraceEvent(TraceSeverity severity,
 133                      absl::string_view message) override {
 134     if (parent_->resolver_ == nullptr) return;  // Shutting down.
 135     parent_->channel_control_helper()->AddTraceEvent(severity, message);
 136   }
 137
 138  private:
 139   RefCountedPtr<ResolvingLoadBalancingPolicy> parent_;
 140 };
 141
 142 //
 143 // ResolvingLoadBalancingPolicy
 144 //
 145
 146 ResolvingLoadBalancingPolicy::ResolvingLoadBalancingPolicy(
 147     Args args, TraceFlag* tracer, grpc_core::UniquePtr<char> target_uri,
 148     ProcessResolverResultCallback process_resolver_result,
 149     void* process_resolver_result_user_data)
 150     : LoadBalancingPolicy(std::move(args)),
 151       tracer_(tracer),
 152       target_uri_(std::move(target_uri)),
 153       process_resolver_result_(process_resolver_result),
 154       process_resolver_result_user_data_(process_resolver_result_user_data) {
 155   GPR_ASSERT(process_resolver_result != nullptr);
 156   resolver_ = ResolverRegistry::CreateResolver(
 157       target_uri_.get(), args.args, interested_parties(), work_serializer(),
 158       absl::make_unique<ResolverResultHandler>(Ref()));
 159   // Since the validity of args has been checked when create the channel,
 160   // CreateResolver() must return a non-null result.
 161   GPR_ASSERT(resolver_ != nullptr);
 162   if (GRPC_TRACE_FLAG_ENABLED(*tracer_)) {
 163     gpr_log(GPR_INFO, "resolving_lb=%p: starting name resolution", this);
 164   }
 165   channel_control_helper()->UpdateState(GRPC_CHANNEL_CONNECTING,
 166                                         absl::make_unique<QueuePicker>(Ref()));
 167   resolver_->StartLocked();
 168 }
 169
 170 ResolvingLoadBalancingPolicy::~ResolvingLoadBalancingPolicy() {
 171   GPR_ASSERT(resolver_ == nullptr);
 172   GPR_ASSERT(lb_policy_ == nullptr);
 173 }
 174
 175 void ResolvingLoadBalancingPolicy::ShutdownLocked() {
 176   if (resolver_ != nullptr) {
 177     resolver_.reset();
 178     if (lb_policy_ != nullptr) {
 179       if (GRPC_TRACE_FLAG_ENABLED(*tracer_)) {
 180         gpr_log(GPR_INFO, "resolving_lb=%p: shutting down lb_policy=%p", this,
 181                 lb_policy_.get());
 182       }
 183       grpc_pollset_set_del_pollset_set(lb_policy_->interested_parties(),
 184                                        interested_parties());
 185       lb_policy_.reset();
 186     }
 187   }
 188 }
 189
 190 void ResolvingLoadBalancingPolicy::ExitIdleLocked() {
 191   if (lb_policy_ != nullptr) lb_policy_->ExitIdleLocked();
 192 }
 193
 194 void ResolvingLoadBalancingPolicy::ResetBackoffLocked() {
 195   if (resolver_ != nullptr) {
 196     resolver_->ResetBackoffLocked();
 197     resolver_->RequestReresolutionLocked();
 198   }
 199   if (lb_policy_ != nullptr) lb_policy_->ResetBackoffLocked();
 200 }
 201
 202 void ResolvingLoadBalancingPolicy::OnResolverError(grpc_error* error) {
 203   if (resolver_ == nullptr) {
 204     GRPC_ERROR_UNREF(error);
 205     return;
 206   }
 207   if (GRPC_TRACE_FLAG_ENABLED(*tracer_)) {
 208     gpr_log(GPR_INFO, "resolving_lb=%p: resolver transient failure: %s", this,
 209             grpc_error_string(error));
 210   }
 211   // If we already have an LB policy from a previous resolution
 212   // result, then we continue to let it set the connectivity state.
 213   // Otherwise, we go into TRANSIENT_FAILURE.
 214   if (lb_policy_ == nullptr) {
 215     grpc_error* state_error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
 216         "Resolver transient failure", &error, 1);
 217     channel_control_helper()->UpdateState(
 218         GRPC_CHANNEL_TRANSIENT_FAILURE,
 219         absl::make_unique<TransientFailurePicker>(state_error));
 220   }
 221   GRPC_ERROR_UNREF(error);
 222 }
 223
 224 void ResolvingLoadBalancingPolicy::CreateOrUpdateLbPolicyLocked(
 225     RefCountedPtr<LoadBalancingPolicy::Config> lb_policy_config,
 226     Resolver::Result result) {
 227   // Construct update.
 228   UpdateArgs update_args;
 229   update_args.addresses = std::move(result.addresses);
 230   update_args.config = std::move(lb_policy_config);
 231   // TODO(roth): Once channel args is converted to C++, use std::move() here.
 232   update_args.args = result.args;
 233   result.args = nullptr;
 234   // Create policy if needed.
 235   if (lb_policy_ == nullptr) {
 236     lb_policy_ = CreateLbPolicyLocked(*update_args.args);
 237   }
 238   // Update the policy.
 239   if (GRPC_TRACE_FLAG_ENABLED(*tracer_)) {
 240     gpr_log(GPR_INFO, "resolving_lb=%p: Updating child policy %p", this,
 241             lb_policy_.get());
 242   }
 243   lb_policy_->UpdateLocked(std::move(update_args));
 244 }
 245
 246 // Creates a new LB policy.
 247 OrphanablePtr<LoadBalancingPolicy>
 248 ResolvingLoadBalancingPolicy::CreateLbPolicyLocked(
 249     const grpc_channel_args& args) {
 250   LoadBalancingPolicy::Args lb_policy_args;
 251   lb_policy_args.work_serializer = work_serializer();
 252   lb_policy_args.channel_control_helper =
 253       absl::make_unique<ResolvingControlHelper>(Ref());
 254   lb_policy_args.args = &args;
 255   OrphanablePtr<LoadBalancingPolicy> lb_policy =
 256       MakeOrphanable<ChildPolicyHandler>(std::move(lb_policy_args), tracer_);
 257   if (GRPC_TRACE_FLAG_ENABLED(*tracer_)) {
 258     gpr_log(GPR_INFO, "resolving_lb=%p: created new LB policy %p", this,
 259             lb_policy.get());
 260   }
 261   grpc_pollset_set_add_pollset_set(lb_policy->interested_parties(),
 262                                    interested_parties());
 263   return lb_policy;
 264 }
 265
 266 void ResolvingLoadBalancingPolicy::MaybeAddTraceMessagesForAddressChangesLocked(
 267     bool resolution_contains_addresses, TraceStringVector* trace_strings) {
 268   if (!resolution_contains_addresses &&
 269       previous_resolution_contained_addresses_) {
 270     trace_strings->push_back("Address list became empty");
 271   } else if (resolution_contains_addresses &&
 272              !previous_resolution_contained_addresses_) {
 273     trace_strings->push_back("Address list became non-empty");
 274   }
 275   previous_resolution_contained_addresses_ = resolution_contains_addresses;
 276 }
 277
 278 void ResolvingLoadBalancingPolicy::ConcatenateAndAddChannelTraceLocked(
 279     const TraceStringVector& trace_strings) const {
 280   if (!trace_strings.empty()) {
 281     std::string message =
 282         absl::StrCat("Resolution event: ", absl::StrJoin(trace_strings, ", "));
 283     channel_control_helper()->AddTraceEvent(ChannelControlHelper::TRACE_INFO,
 284                                             message);
 285   }
 286 }
 287
 288 void ResolvingLoadBalancingPolicy::OnResolverResultChangedLocked(
 289     Resolver::Result result) {
 290   // Handle race conditions.
 291   if (resolver_ == nullptr) return;
 292   if (GRPC_TRACE_FLAG_ENABLED(*tracer_)) {
 293     gpr_log(GPR_INFO, "resolving_lb=%p: got resolver result", this);
 294   }
 295   // We only want to trace the address resolution in the follow cases:
 296   // (a) Address resolution resulted in service config change.
 297   // (b) Address resolution that causes number of backends to go from
 298   //     zero to non-zero.
 299   // (c) Address resolution that causes number of backends to go from
 300   //     non-zero to zero.
 301   // (d) Address resolution that causes a new LB policy to be created.
 302   //
 303   // We track a list of strings to eventually be concatenated and traced.
 304   TraceStringVector trace_strings;
 305   const bool resolution_contains_addresses = result.addresses.size() > 0;
 306   // Process the resolver result.
 307   RefCountedPtr<LoadBalancingPolicy::Config> lb_policy_config;
 308   bool service_config_changed = false;
 309   std::string service_config_error_string;
 310   if (process_resolver_result_ != nullptr) {
 311     grpc_error* service_config_error = GRPC_ERROR_NONE;
 312     bool no_valid_service_config = false;
 313     service_config_changed = process_resolver_result_(
 314         process_resolver_result_user_data_, result, &lb_policy_config,
 315         &service_config_error, &no_valid_service_config);
 316     if (service_config_error != GRPC_ERROR_NONE) {
 317       service_config_error_string = grpc_error_string(service_config_error);
 318       if (no_valid_service_config) {
 319         // We received an invalid service config and we don't have a
 320         // fallback service config.
 321         OnResolverError(service_config_error);
 322       } else {
 323         GRPC_ERROR_UNREF(service_config_error);
 324       }
 325     }
 326   } else {
 327     lb_policy_config = child_lb_config_;
 328   }
 329   if (lb_policy_config != nullptr) {
 330     // Create or update LB policy, as needed.
 331     CreateOrUpdateLbPolicyLocked(std::move(lb_policy_config),
 332                                  std::move(result));
 333   }
 334   // Add channel trace event.
 335   if (service_config_changed) {
 336     // TODO(ncteisen): might be worth somehow including a snippet of the
 337     // config in the trace, at the risk of bloating the trace logs.
 338     trace_strings.push_back("Service config changed");
 339   }
 340   if (!service_config_error_string.empty()) {
 341     trace_strings.push_back(service_config_error_string.c_str());
 342   }
 343   MaybeAddTraceMessagesForAddressChangesLocked(resolution_contains_addresses,
 344                                                &trace_strings);
 345   ConcatenateAndAddChannelTraceLocked(trace_strings);
 346 }
 347
 348 }  // namespace grpc_core