Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Simplify service config processing and fix config selector handling.

Browse Source
pull/24114/head
Mark D. Roth 4 years ago
parent e834d0b34b
commit 84c8f7051a
6 changed files with 160 additions and 236 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 262
      src/core/ext/filters/client_channel/client_channel.cc
  2. 13
      src/core/ext/filters/client_channel/config_selector.h
  3. 86
      src/core/ext/filters/client_channel/resolving_lb_policy.cc
  4. 14
      src/core/ext/filters/client_channel/resolving_lb_policy.h
  5. 5
      src/core/ext/filters/client_channel/service_config.cc
  6. 16
      test/cpp/end2end/service_config_end2end_test.cc

262
src/core/ext/filters/client_channel/client_channel.cc
Unescape View File

@ -241,17 +241,15 @@ class ChannelData {
public: public:
explicit ChannelConfigHelper(ChannelData* chand) : chand_(chand) {} explicit ChannelConfigHelper(ChannelData* chand) : chand_(chand) {}
ApplyServiceConfigResult ApplyServiceConfig( ChooseServiceConfigResult ChooseServiceConfig(
const Resolver::Result& result) override; const Resolver::Result& result) override;
void ApplyConfigSelector( void StartUsingServiceConfigForCalls() override;
bool service_config_changed,
RefCountedPtr<ConfigSelector> config_selector) override;
void ResolverTransientFailure(grpc_error* error) override; void ResolverTransientFailure(grpc_error* error) override;
private: private:
static void ProcessLbPolicy( static void ChooseLbPolicy(
const Resolver::Result& resolver_result, const Resolver::Result& resolver_result,
const internal::ClientChannelGlobalParsedConfig* parsed_service_config, const internal::ClientChannelGlobalParsedConfig* parsed_service_config,
RefCountedPtr<LoadBalancingPolicy::Config>* lb_policy_config); RefCountedPtr<LoadBalancingPolicy::Config>* lb_policy_config);
@ -267,9 +265,12 @@ class ChannelData {
const char* reason, const char* reason,
std::unique_ptr<LoadBalancingPolicy::SubchannelPicker> picker); std::unique_ptr<LoadBalancingPolicy::SubchannelPicker> picker);
void UpdateServiceConfigInDataPlaneLocked( void UpdateServiceConfigInControlPlaneLocked(
bool service_config_changed, RefCountedPtr<ServiceConfig> service_config,
RefCountedPtr<ConfigSelector> config_selector); const internal::ClientChannelGlobalParsedConfig* parsed_service_config,
const char* lb_policy_name, const grpc_channel_args* args);
void UpdateServiceConfigInDataPlaneLocked();
void CreateResolvingLoadBalancingPolicyLocked(); void CreateResolvingLoadBalancingPolicyLocked();
@ -320,7 +321,6 @@ class ChannelData {
grpc_core::UniquePtr<char> health_check_service_name_; grpc_core::UniquePtr<char> health_check_service_name_;
RefCountedPtr<ServiceConfig> saved_service_config_; RefCountedPtr<ServiceConfig> saved_service_config_;
RefCountedPtr<ConfigSelector> saved_config_selector_; RefCountedPtr<ConfigSelector> saved_config_selector_;
bool received_first_resolver_result_ = false;
// The number of SubchannelWrapper instances referencing a given Subchannel. // The number of SubchannelWrapper instances referencing a given Subchannel.
std::map<Subchannel*, int> subchannel_refcount_map_; std::map<Subchannel*, int> subchannel_refcount_map_;
// The set of SubchannelWrappers that currently exist. // The set of SubchannelWrappers that currently exist.
@ -1433,23 +1433,18 @@ class ChannelData::ClientChannelControlHelper
// ChannelData::ChannelConfigHelper // ChannelData::ChannelConfigHelper
// //
// Synchronous callback from ResolvingLoadBalancingPolicy to process a ChannelData::ChannelConfigHelper::ChooseServiceConfigResult
// resolver result update. ChannelData::ChannelConfigHelper::ChooseServiceConfig(
ChannelData::ChannelConfigHelper::ApplyServiceConfigResult
ChannelData::ChannelConfigHelper::ApplyServiceConfig(
const Resolver::Result& result) { const Resolver::Result& result) {
ApplyServiceConfigResult service_config_result; ChooseServiceConfigResult service_config_result;
RefCountedPtr<ServiceConfig> service_config; RefCountedPtr<ServiceConfig> service_config;
// If resolver did not return a service config or returned an invalid service
// config, we need a fallback service config.
if (result.service_config_error != GRPC_ERROR_NONE) { if (result.service_config_error != GRPC_ERROR_NONE) {
if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) {
gpr_log(GPR_INFO, "chand=%p: resolver returned service config error: %s", gpr_log(GPR_INFO, "chand=%p: resolver returned service config error: %s",
chand_, grpc_error_string(result.service_config_error)); chand_, grpc_error_string(result.service_config_error));
} }
// If the service config was invalid, then fallback to the saved service // If the service config was invalid, then fallback to the
// config. If there is no saved config either, use the default service // previously returned service config.
// config.
if (chand_->saved_service_config_ != nullptr) { if (chand_->saved_service_config_ != nullptr) {
if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) {
gpr_log(GPR_INFO, gpr_log(GPR_INFO,
@ -1458,99 +1453,51 @@ ChannelData::ChannelConfigHelper::ApplyServiceConfig(
chand_); chand_);
} }
service_config = chand_->saved_service_config_; service_config = chand_->saved_service_config_;
} else if (chand_->default_service_config_ != nullptr) { } else {
if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) { // No previously returned config, so put the channel into
gpr_log(GPR_INFO, // TRANSIENT_FAILURE.
"chand=%p: resolver returned invalid service config. Using " service_config_result.no_valid_service_config = true;
"default service config provided by client API.", return service_config_result;
chand_);
}
service_config = chand_->default_service_config_;
} }
} else if (result.service_config == nullptr) { } else if (result.service_config == nullptr) {
if (chand_->default_service_config_ != nullptr) { // Resolver did not return any service config.
if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) { if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) {
gpr_log(GPR_INFO, gpr_log(GPR_INFO,
"chand=%p: resolver returned no service config. Using default " "chand=%p: resolver returned no service config. Using default "
"service config provided by client API.", "service config for channel.",
chand_); chand_);
}
service_config = chand_->default_service_config_;
} }
service_config = chand_->default_service_config_;
} else { } else {
// Use service config returned by resolver.
service_config = result.service_config; service_config = result.service_config;
} }
service_config_result.service_config_error = GPR_ASSERT(service_config != nullptr);
GRPC_ERROR_REF(result.service_config_error); // Extract global config for client channel.
if (service_config == nullptr &&
result.service_config_error != GRPC_ERROR_NONE) {
service_config_result.no_valid_service_config = true;
return service_config_result;
}
// Process service config.
grpc_core::UniquePtr<char> service_config_json;
const internal::ClientChannelGlobalParsedConfig* parsed_service_config = const internal::ClientChannelGlobalParsedConfig* parsed_service_config =
nullptr; static_cast<const internal::ClientChannelGlobalParsedConfig*>(
if (service_config != nullptr) { service_config->GetGlobalParsedConfig(
parsed_service_config = internal::ClientChannelServiceConfigParser::ParserIndex()));
static_cast<const internal::ClientChannelGlobalParsedConfig*>( // Find LB policy config.
service_config->GetGlobalParsedConfig( ChooseLbPolicy(result, parsed_service_config,
internal::ClientChannelServiceConfigParser::ParserIndex())); &service_config_result.lb_policy_config);
}
// Check if the config has changed. // Check if the config has changed.
service_config_result.service_config_changed = service_config_result.service_config_changed =
((service_config == nullptr) != chand_->saved_service_config_ == nullptr ||
(chand_->saved_service_config_ == nullptr)) || service_config->json_string() !=
(service_config != nullptr && chand_->saved_service_config_->json_string();
service_config->json_string() != // If it has, apply the global parameters now.
chand_->saved_service_config_->json_string());
if (service_config_result.service_config_changed) { if (service_config_result.service_config_changed) {
service_config_json.reset(gpr_strdup( chand_->UpdateServiceConfigInControlPlaneLocked(
service_config != nullptr ? service_config->json_string().c_str() std::move(service_config), parsed_service_config,
: "")); service_config_result.lb_policy_config->name(), result.args);
if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) {
gpr_log(GPR_INFO,
"chand=%p: resolver returned updated service config: \"%s\"",
chand_, service_config_json.get());
}
// Save health check service name.
if (service_config != nullptr) {
chand_->health_check_service_name_.reset(
gpr_strdup(parsed_service_config->health_check_service_name()));
} else {
chand_->health_check_service_name_.reset();
}
// Update health check service name used by existing subchannel wrappers.
for (auto* subchannel_wrapper : chand_->subchannel_wrappers_) {
subchannel_wrapper->UpdateHealthCheckServiceName(
grpc_core::UniquePtr<char>(
gpr_strdup(chand_->health_check_service_name_.get())));
}
// Save service config.
chand_->saved_service_config_ = std::move(service_config);
}
// Find LB policy config.
ProcessLbPolicy(result, parsed_service_config,
&service_config_result.lb_policy_config);
grpc_core::UniquePtr<char> lb_policy_name(
gpr_strdup((service_config_result.lb_policy_config)->name()));
// Swap out the data used by GetChannelInfo().
{
MutexLock lock(&chand_->info_mu_);
chand_->info_lb_policy_name_ = std::move(lb_policy_name);
if (service_config_json != nullptr) {
chand_->info_service_config_json_ = std::move(service_config_json);
}
} }
// Return results. // Return results.
return service_config_result; return service_config_result;
} }
void ChannelData::ChannelConfigHelper::ApplyConfigSelector( void ChannelData::ChannelConfigHelper::StartUsingServiceConfigForCalls() {
bool service_config_changed, chand_->UpdateServiceConfigInDataPlaneLocked();
RefCountedPtr<ConfigSelector> config_selector) {
chand_->UpdateServiceConfigInDataPlaneLocked(service_config_changed,
std::move(config_selector));
} }
void ChannelData::ChannelConfigHelper::ResolverTransientFailure( void ChannelData::ChannelConfigHelper::ResolverTransientFailure(
@ -1560,21 +1507,19 @@ void ChannelData::ChannelConfigHelper::ResolverTransientFailure(
chand_->resolver_transient_failure_error_ = error; chand_->resolver_transient_failure_error_ = error;
} }
void ChannelData::ChannelConfigHelper::ProcessLbPolicy( void ChannelData::ChannelConfigHelper::ChooseLbPolicy(
const Resolver::Result& resolver_result, const Resolver::Result& resolver_result,
const internal::ClientChannelGlobalParsedConfig* parsed_service_config, const internal::ClientChannelGlobalParsedConfig* parsed_service_config,
RefCountedPtr<LoadBalancingPolicy::Config>* lb_policy_config) { RefCountedPtr<LoadBalancingPolicy::Config>* lb_policy_config) {
// Prefer the LB policy config found in the service config. // Prefer the LB policy config found in the service config.
if (parsed_service_config != nullptr && if (parsed_service_config->parsed_lb_config() != nullptr) {
parsed_service_config->parsed_lb_config() != nullptr) {
*lb_policy_config = parsed_service_config->parsed_lb_config(); *lb_policy_config = parsed_service_config->parsed_lb_config();
return; return;
} }
// Try the deprecated LB policy name from the service config. // Try the deprecated LB policy name from the service config.
// If not, try the setting from channel args. // If not, try the setting from channel args.
const char* policy_name = nullptr; const char* policy_name = nullptr;
if (parsed_service_config != nullptr && if (!parsed_service_config->parsed_deprecated_lb_policy().empty()) {
!parsed_service_config->parsed_deprecated_lb_policy().empty()) {
policy_name = parsed_service_config->parsed_deprecated_lb_policy().c_str(); policy_name = parsed_service_config->parsed_deprecated_lb_policy().c_str();
} else { } else {
const grpc_arg* channel_arg = const grpc_arg* channel_arg =
@ -1694,16 +1639,16 @@ ChannelData::ChannelData(grpc_channel_element_args* args, grpc_error** error)
"filter"); "filter");
return; return;
} }
// Get default service config // Get default service config. If none is specified via the client API,
// we use an empty config.
const char* service_config_json = grpc_channel_arg_get_string( const char* service_config_json = grpc_channel_arg_get_string(
grpc_channel_args_find(args->channel_args, GRPC_ARG_SERVICE_CONFIG)); grpc_channel_args_find(args->channel_args, GRPC_ARG_SERVICE_CONFIG));
if (service_config_json != nullptr) { if (service_config_json == nullptr) service_config_json = "{}";
*error = GRPC_ERROR_NONE; *error = GRPC_ERROR_NONE;
default_service_config_ = ServiceConfig::Create(service_config_json, error); default_service_config_ = ServiceConfig::Create(service_config_json, error);
if (*error != GRPC_ERROR_NONE) { if (*error != GRPC_ERROR_NONE) {
default_service_config_.reset(); default_service_config_.reset();
return; return;
}
} }
grpc_uri* uri = grpc_uri_parse(server_uri, true); grpc_uri* uri = grpc_uri_parse(server_uri, true);
if (uri != nullptr && uri->path[0] != '\0') { if (uri != nullptr && uri->path[0] != '\0') {
@ -1755,7 +1700,6 @@ void ChannelData::UpdateStateAndPickerLocked(
health_check_service_name_.reset(); health_check_service_name_.reset();
saved_service_config_.reset(); saved_service_config_.reset();
saved_config_selector_.reset(); saved_config_selector_.reset();
received_first_resolver_result_ = false;
} }
// Update connectivity state. // Update connectivity state.
state_tracker_.SetState(state, status, reason); state_tracker_.SetState(state, status, reason);
@ -1823,51 +1767,56 @@ void ChannelData::UpdateStateAndPickerLocked(
pending_subchannel_updates_.clear(); pending_subchannel_updates_.clear();
} }
void ChannelData::UpdateServiceConfigInDataPlaneLocked( void ChannelData::UpdateServiceConfigInControlPlaneLocked(
bool service_config_changed, RefCountedPtr<ServiceConfig> service_config,
RefCountedPtr<ConfigSelector> config_selector) { const internal::ClientChannelGlobalParsedConfig* parsed_service_config,
// If the service config did not change and there is no new ConfigSelector, const char* lb_policy_name, const grpc_channel_args* args) {
// retain the old one (if any). grpc_core::UniquePtr<char> service_config_json(
// TODO(roth): Consider whether this is really the right way to handle gpr_strdup(service_config->json_string().c_str()));
// this. We might instead want to decide this in ApplyServiceConfig() if (GRPC_TRACE_FLAG_ENABLED(grpc_client_channel_routing_trace)) {
// where we decide whether to stick with the saved service config. gpr_log(GPR_INFO,
if (!service_config_changed && config_selector == nullptr) { "chand=%p: resolver returned updated service config: \"%s\"", this,
config_selector = saved_config_selector_; service_config_json.get());
} }
// Check if ConfigSelector has changed. // Save service config.
const bool config_selector_changed = saved_service_config_ = std::move(service_config);
saved_config_selector_ != config_selector; // Save health check service name.
saved_config_selector_ = config_selector; health_check_service_name_.reset(
// We want to set the service config at least once, even if the gpr_strdup(parsed_service_config->health_check_service_name()));
// resolver does not return a config, because that ensures that we // Update health check service name used by existing subchannel wrappers.
// disable retries if they are not enabled in the service config. for (auto* subchannel_wrapper : subchannel_wrappers_) {
// TODO(roth): Consider removing the received_first_resolver_result_ check subchannel_wrapper->UpdateHealthCheckServiceName(grpc_core::UniquePtr<char>(
// when we implement transparent retries. gpr_strdup(health_check_service_name_.get())));
if (!service_config_changed && !config_selector_changed && }
received_first_resolver_result_) { // Swap out the data used by GetChannelInfo().
return; grpc_core::UniquePtr<char> lb_policy_name_owned(gpr_strdup(lb_policy_name));
{
MutexLock lock(&info_mu_);
info_lb_policy_name_ = std::move(lb_policy_name_owned);
info_service_config_json_ = std::move(service_config_json);
} }
received_first_resolver_result_ = true; // Save config selector.
saved_config_selector_ = ConfigSelector::GetFromChannelArgs(*args);
}
void ChannelData::UpdateServiceConfigInDataPlaneLocked() {
// Get retry throttle data from service config. // Get retry throttle data from service config.
const internal::ClientChannelGlobalParsedConfig* parsed_service_config =
static_cast<const internal::ClientChannelGlobalParsedConfig*>(
saved_service_config_->GetGlobalParsedConfig(
internal::ClientChannelServiceConfigParser::ParserIndex()));
absl::optional<internal::ClientChannelGlobalParsedConfig::RetryThrottling>
retry_throttle_config = parsed_service_config->retry_throttling();
RefCountedPtr<ServerRetryThrottleData> retry_throttle_data; RefCountedPtr<ServerRetryThrottleData> retry_throttle_data;
if (saved_service_config_ != nullptr) { if (retry_throttle_config.has_value()) {
const internal::ClientChannelGlobalParsedConfig* parsed_service_config = retry_throttle_data = internal::ServerRetryThrottleMap::GetDataForServer(
static_cast<const internal::ClientChannelGlobalParsedConfig*>( server_name_.get(), retry_throttle_config.value().max_milli_tokens,
saved_service_config_->GetGlobalParsedConfig( retry_throttle_config.value().milli_token_ratio);
internal::ClientChannelServiceConfigParser::ParserIndex())); }
if (parsed_service_config != nullptr) { // Grab ref to service config.
absl::optional<internal::ClientChannelGlobalParsedConfig::RetryThrottling> RefCountedPtr<ServiceConfig> service_config = saved_service_config_;
retry_throttle_config = parsed_service_config->retry_throttling(); // Grab ref to config selector. Use default if resolver didn't supply one.
if (retry_throttle_config.has_value()) { RefCountedPtr<ConfigSelector> config_selector = saved_config_selector_;
retry_throttle_data =
internal::ServerRetryThrottleMap::GetDataForServer(
server_name_.get(),
retry_throttle_config.value().max_milli_tokens,
retry_throttle_config.value().milli_token_ratio);
}
}
}
// Create default config selector if not provided by resolver.
if (config_selector == nullptr) { if (config_selector == nullptr) {
config_selector = config_selector =
MakeRefCounted<DefaultConfigSelector>(saved_service_config_); MakeRefCounted<DefaultConfigSelector>(saved_service_config_);
@ -1876,9 +1825,6 @@ void ChannelData::UpdateServiceConfigInDataPlaneLocked(
// //
// We defer unreffing the old values (and deallocating memory) until // We defer unreffing the old values (and deallocating memory) until
// after releasing the lock to keep the critical section small. // after releasing the lock to keep the critical section small.
RefCountedPtr<ServiceConfig> service_config_to_unref = saved_service_config_;
RefCountedPtr<ConfigSelector> config_selector_to_unref =
std::move(config_selector);
{ {
MutexLock lock(&data_plane_mu_); MutexLock lock(&data_plane_mu_);
GRPC_ERROR_UNREF(resolver_transient_failure_error_); GRPC_ERROR_UNREF(resolver_transient_failure_error_);
@ -1887,8 +1833,8 @@ void ChannelData::UpdateServiceConfigInDataPlaneLocked(
received_service_config_data_ = true; received_service_config_data_ = true;
// Old values will be unreffed after lock is released. // Old values will be unreffed after lock is released.
retry_throttle_data_.swap(retry_throttle_data); retry_throttle_data_.swap(retry_throttle_data);
service_config_.swap(service_config_to_unref); service_config_.swap(service_config);
config_selector_.swap(config_selector_to_unref); config_selector_.swap(config_selector);
// Re-process queued picks. // Re-process queued picks.
for (QueuedPick* pick = queued_picks_; pick != nullptr; pick = pick->next) { for (QueuedPick* pick = queued_picks_; pick != nullptr; pick = pick->next) {
grpc_call_element* elem = pick->elem; grpc_call_element* elem = pick->elem;

13
src/core/ext/filters/client_channel/config_selector.h
Unescape View File

@ -76,14 +76,17 @@ class ConfigSelector : public RefCounted<ConfigSelector> {
class DefaultConfigSelector : public ConfigSelector { class DefaultConfigSelector : public ConfigSelector {
public: public:
explicit DefaultConfigSelector(RefCountedPtr<ServiceConfig> service_config) explicit DefaultConfigSelector(RefCountedPtr<ServiceConfig> service_config)
: service_config_(std::move(service_config)) {} : service_config_(std::move(service_config)) {
// The client channel code ensures that this will never be null.
// If neither the resolver nor the client application provide a
// config, a default empty config will be used.
GPR_DEBUG_ASSERT(service_config_ != nullptr);
}
CallConfig GetCallConfig(GetCallConfigArgs args) override { CallConfig GetCallConfig(GetCallConfigArgs args) override {
CallConfig call_config; CallConfig call_config;
if (service_config_ != nullptr) { call_config.method_configs =
call_config.method_configs = service_config_->GetMethodParsedConfigVector(*args.path);
service_config_->GetMethodParsedConfigVector(*args.path);
}
return call_config; return call_config;
} }

86
src/core/ext/filters/client_channel/resolving_lb_policy.cc
Unescape View File

@ -232,9 +232,12 @@ void ResolvingLoadBalancingPolicy::CreateOrUpdateLbPolicyLocked(
UpdateArgs update_args; UpdateArgs update_args;
update_args.addresses = std::move(result.addresses); update_args.addresses = std::move(result.addresses);
update_args.config = std::move(lb_policy_config); update_args.config = std::move(lb_policy_config);
// TODO(roth): Once channel args is converted to C++, use std::move() here. // Remove the config selector from channel args so that we're not holding
update_args.args = result.args; // unnecessary refs that cause it to be destroyed somewhere other than in the
result.args = nullptr; // WorkSerializer.
const char* arg_name = GRPC_ARG_CONFIG_SELECTOR;
update_args.args =
grpc_channel_args_copy_and_remove(result.args, &arg_name, 1);
// Create policy if needed. // Create policy if needed.
if (lb_policy_ == nullptr) { if (lb_policy_ == nullptr) {
lb_policy_ = CreateLbPolicyLocked(*update_args.args); lb_policy_ = CreateLbPolicyLocked(*update_args.args);
@ -306,61 +309,46 @@ void ResolvingLoadBalancingPolicy::OnResolverResultChangedLocked(
// //
// We track a list of strings to eventually be concatenated and traced. // We track a list of strings to eventually be concatenated and traced.
TraceStringVector trace_strings; TraceStringVector trace_strings;
const bool resolution_contains_addresses = result.addresses.size() > 0; MaybeAddTraceMessagesForAddressChangesLocked(!result.addresses.empty(),
// Process the resolver result. &trace_strings);
ChannelConfigHelper::ApplyServiceConfigResult service_config_result; // The result of grpc_error_string() is owned by the error itself.
// We're storing that string in trace_strings, so we need to make sure
// that the error lives until we're done with the string.
grpc_error* service_config_error =
GRPC_ERROR_REF(result.service_config_error);
if (service_config_error != GRPC_ERROR_NONE) {
trace_strings.push_back(grpc_error_string(service_config_error));
}
// Choose the service config.
ChannelConfigHelper::ChooseServiceConfigResult service_config_result;
if (helper_ != nullptr) { if (helper_ != nullptr) {
service_config_result = helper_->ApplyServiceConfig(result); service_config_result = helper_->ChooseServiceConfig(result);
if (service_config_result.service_config_error != GRPC_ERROR_NONE) {
if (service_config_result.no_valid_service_config) {
// We received an invalid service config and we don't have a
// fallback service config.
OnResolverError(service_config_result.service_config_error);
service_config_result.service_config_error = GRPC_ERROR_NONE;
}
}
} else { } else {
service_config_result.lb_policy_config = child_lb_config_; service_config_result.lb_policy_config = child_lb_config_;
} }
// Before we send the args to the LB policy, grab the ConfigSelector for if (service_config_result.no_valid_service_config) {
// later use. // We received an invalid service config and we don't have a
RefCountedPtr<ConfigSelector> config_selector = // previous service config to fall back to.
ConfigSelector::GetFromChannelArgs(*result.args); OnResolverError(GRPC_ERROR_REF(service_config_error));
// Remove the config selector from channel args so that we're not holding trace_strings.push_back("no valid service config");
// unnecessary refs that cause it to be destroyed somewhere other than in the } else {
// WorkSerializer. // Create or update LB policy, as needed.
const char* arg_name = GRPC_ARG_CONFIG_SELECTOR;
grpc_channel_args* new_args =
grpc_channel_args_copy_and_remove(result.args, &arg_name, 1);
grpc_channel_args_destroy(result.args);
result.args = new_args;
// Create or update LB policy, as needed.
if (service_config_result.lb_policy_config != nullptr) {
CreateOrUpdateLbPolicyLocked( CreateOrUpdateLbPolicyLocked(
std::move(service_config_result.lb_policy_config), std::move(result)); std::move(service_config_result.lb_policy_config), std::move(result));
} if (service_config_result.service_config_changed) {
// Apply ConfigSelector to channel. // Tell channel to start using new service config for calls.
// This needs to happen after the LB policy has been updated, since // This needs to happen after the LB policy has been updated, since
// the ConfigSelector may need the LB policy to know about new // the ConfigSelector may need the LB policy to know about new
// destinations before it can send RPCs to those destinations. // destinations before it can send RPCs to those destinations.
if (helper_ != nullptr) { if (helper_ != nullptr) helper_->StartUsingServiceConfigForCalls();
helper_->ApplyConfigSelector(service_config_result.service_config_changed, // TODO(ncteisen): might be worth somehow including a snippet of the
std::move(config_selector)); // config in the trace, at the risk of bloating the trace logs.
trace_strings.push_back("Service config changed");
}
} }
// Add channel trace event. // Add channel trace event.
if (service_config_result.service_config_changed) {
// TODO(ncteisen): might be worth somehow including a snippet of the
// config in the trace, at the risk of bloating the trace logs.
trace_strings.push_back("Service config changed");
}
if (service_config_result.service_config_error != GRPC_ERROR_NONE) {
trace_strings.push_back(
grpc_error_string(service_config_result.service_config_error));
}
MaybeAddTraceMessagesForAddressChangesLocked(resolution_contains_addresses,
&trace_strings);
ConcatenateAndAddChannelTraceLocked(trace_strings); ConcatenateAndAddChannelTraceLocked(trace_strings);
GRPC_ERROR_UNREF(service_config_result.service_config_error); GRPC_ERROR_UNREF(service_config_error);
} }
} // namespace grpc_core } // namespace grpc_core

14
src/core/ext/filters/client_channel/resolving_lb_policy.h
Unescape View File

@ -55,29 +55,25 @@ class ResolvingLoadBalancingPolicy : public LoadBalancingPolicy {
public: public:
class ChannelConfigHelper { class ChannelConfigHelper {
public: public:
struct ApplyServiceConfigResult { struct ChooseServiceConfigResult {
// Set to true if the service config has changed since the last result. // Set to true if the service config has changed since the last result.
bool service_config_changed = false; bool service_config_changed = false;
// Set to true if we don't have a valid service config to use. // Set to true if we don't have a valid service config to use.
// This tells the ResolvingLoadBalancingPolicy to put the channel // This tells the ResolvingLoadBalancingPolicy to put the channel
// into TRANSIENT_FAILURE. // into TRANSIENT_FAILURE.
bool no_valid_service_config = false; bool no_valid_service_config = false;
// A service config parsing error occurred.
grpc_error* service_config_error = GRPC_ERROR_NONE;
// The LB policy config to use. // The LB policy config to use.
RefCountedPtr<LoadBalancingPolicy::Config> lb_policy_config; RefCountedPtr<LoadBalancingPolicy::Config> lb_policy_config;
}; };
virtual ~ChannelConfigHelper() = default; virtual ~ChannelConfigHelper() = default;
// Applies the service config to the channel. // Chooses the service config for the channel.
virtual ApplyServiceConfigResult ApplyServiceConfig( virtual ChooseServiceConfigResult ChooseServiceConfig(
const Resolver::Result& result) = 0; const Resolver::Result& result) = 0;
// Applies the ConfigSelector to the channel. // Starts using the service config for calls.
virtual void ApplyConfigSelector( virtual void StartUsingServiceConfigForCalls() = 0;
bool service_config_changed,
RefCountedPtr<ConfigSelector> config_selector) = 0;
// Indicates a resolver transient failure. // Indicates a resolver transient failure.
virtual void ResolverTransientFailure(grpc_error* error) = 0; virtual void ResolverTransientFailure(grpc_error* error) = 0;

5
src/core/ext/filters/client_channel/service_config.cc
Unescape View File

@ -202,15 +202,16 @@ std::string ServiceConfig::ParseJsonMethodName(const Json& json,
const ServiceConfigParser::ParsedConfigVector* const ServiceConfigParser::ParsedConfigVector*
ServiceConfig::GetMethodParsedConfigVector(const grpc_slice& path) const { ServiceConfig::GetMethodParsedConfigVector(const grpc_slice& path) const {
if (parsed_method_configs_map_.empty()) return nullptr;
// Try looking up the full path in the map. // Try looking up the full path in the map.
auto it = parsed_method_configs_map_.find(path); auto it = parsed_method_configs_map_.find(path);
if (it != parsed_method_configs_map_.end()) return it->second; if (it != parsed_method_configs_map_.end()) return it->second;
// If we didn't find a match for the path, try looking for a wildcard // If we didn't find a match for the path, try looking for a wildcard
// entry (i.e., change "/service/method" to "/service/"). // entry (i.e., change "/service/method" to "/service/").
UniquePtr<char> path_str(grpc_slice_to_c_string(path)); UniquePtr<char> path_str(grpc_slice_to_c_string(path));
char* sep = strrchr(path_str.get(), '/') + 1; char* sep = strrchr(path_str.get(), '/');
if (sep == nullptr) return nullptr; // Shouldn't ever happen. if (sep == nullptr) return nullptr; // Shouldn't ever happen.
*sep = '\0'; sep[1] = '\0';
grpc_slice wildcard_path = grpc_slice_from_static_string(path_str.get()); grpc_slice wildcard_path = grpc_slice_from_static_string(path_str.get());
it = parsed_method_configs_map_.find(wildcard_path); it = parsed_method_configs_map_.find(wildcard_path);
if (it != parsed_method_configs_map_.end()) return it->second; if (it != parsed_method_configs_map_.end()) return it->second;

16
test/cpp/end2end/service_config_end2end_test.cc
Unescape View File

@ -437,7 +437,7 @@ TEST_F(ServiceConfigEnd2endTest, NoServiceConfigTest) {
auto stub = BuildStub(channel); auto stub = BuildStub(channel);
SetNextResolutionNoServiceConfig(GetServersPorts()); SetNextResolutionNoServiceConfig(GetServersPorts());
CheckRpcSendOk(stub, DEBUG_LOCATION); CheckRpcSendOk(stub, DEBUG_LOCATION);
EXPECT_STREQ("", channel->GetServiceConfigJSON().c_str()); EXPECT_STREQ("{}", channel->GetServiceConfigJSON().c_str());
} }
TEST_F(ServiceConfigEnd2endTest, NoServiceConfigWithDefaultConfigTest) { TEST_F(ServiceConfigEnd2endTest, NoServiceConfigWithDefaultConfigTest) {
@ -458,16 +458,6 @@ TEST_F(ServiceConfigEnd2endTest, InvalidServiceConfigTest) {
CheckRpcSendFailure(stub); CheckRpcSendFailure(stub);
} }
TEST_F(ServiceConfigEnd2endTest, InvalidServiceConfigWithDefaultConfigTest) {
StartServers(1);
auto channel = BuildChannelWithDefaultServiceConfig();
auto stub = BuildStub(channel);
SetNextResolutionInvalidServiceConfig(GetServersPorts());
CheckRpcSendOk(stub, DEBUG_LOCATION);
EXPECT_STREQ(ValidDefaultServiceConfig(),
channel->GetServiceConfigJSON().c_str());
}
TEST_F(ServiceConfigEnd2endTest, ValidServiceConfigUpdatesTest) { TEST_F(ServiceConfigEnd2endTest, ValidServiceConfigUpdatesTest) {
StartServers(1); StartServers(1);
auto channel = BuildChannel(); auto channel = BuildChannel();
@ -490,7 +480,7 @@ TEST_F(ServiceConfigEnd2endTest,
EXPECT_STREQ(ValidServiceConfigV1(), channel->GetServiceConfigJSON().c_str()); EXPECT_STREQ(ValidServiceConfigV1(), channel->GetServiceConfigJSON().c_str());
SetNextResolutionNoServiceConfig(GetServersPorts()); SetNextResolutionNoServiceConfig(GetServersPorts());
CheckRpcSendOk(stub, DEBUG_LOCATION); CheckRpcSendOk(stub, DEBUG_LOCATION);
EXPECT_STREQ("", channel->GetServiceConfigJSON().c_str()); EXPECT_STREQ("{}", channel->GetServiceConfigJSON().c_str());
} }
TEST_F(ServiceConfigEnd2endTest, TEST_F(ServiceConfigEnd2endTest,
@ -552,7 +542,7 @@ TEST_F(ServiceConfigEnd2endTest, NoServiceConfigAfterInvalidServiceConfigTest) {
CheckRpcSendFailure(stub); CheckRpcSendFailure(stub);
SetNextResolutionNoServiceConfig(GetServersPorts()); SetNextResolutionNoServiceConfig(GetServersPorts());
CheckRpcSendOk(stub, DEBUG_LOCATION); CheckRpcSendOk(stub, DEBUG_LOCATION);
EXPECT_STREQ("", channel->GetServiceConfigJSON().c_str()); EXPECT_STREQ("{}", channel->GetServiceConfigJSON().c_str());
} }
TEST_F(ServiceConfigEnd2endTest, TEST_F(ServiceConfigEnd2endTest,

Write Preview
Loading…
Cancel
Save