mirror of https://github.com/grpc/grpc.git
Juanli Shen
6 years ago
parent
fdf63a7c02
commit
8a880801ae
27 changed files with 628 additions and 467 deletions
@ -1,178 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <stdio.h> |
||||
#include <string.h> |
||||
|
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/string_util.h> |
||||
|
||||
#include "src/core/ext/filters/client_channel/method_params.h" |
||||
#include "src/core/lib/channel/status_util.h" |
||||
#include "src/core/lib/gpr/string.h" |
||||
#include "src/core/lib/gprpp/memory.h" |
||||
|
||||
// As per the retry design, we do not allow more than 5 retry attempts.
|
||||
#define MAX_MAX_RETRY_ATTEMPTS 5 |
||||
|
||||
namespace grpc_core { |
||||
namespace internal { |
||||
|
||||
namespace { |
||||
|
||||
bool ParseWaitForReady( |
||||
grpc_json* field, ClientChannelMethodParams::WaitForReady* wait_for_ready) { |
||||
if (field->type != GRPC_JSON_TRUE && field->type != GRPC_JSON_FALSE) { |
||||
return false; |
||||
} |
||||
*wait_for_ready = field->type == GRPC_JSON_TRUE |
||||
? ClientChannelMethodParams::WAIT_FOR_READY_TRUE |
||||
: ClientChannelMethodParams::WAIT_FOR_READY_FALSE; |
||||
return true; |
||||
} |
||||
|
||||
// Parses a JSON field of the form generated for a google.proto.Duration
|
||||
// proto message, as per:
|
||||
// https://developers.google.com/protocol-buffers/docs/proto3#json
|
||||
bool ParseDuration(grpc_json* field, grpc_millis* duration) { |
||||
if (field->type != GRPC_JSON_STRING) return false; |
||||
size_t len = strlen(field->value); |
||||
if (field->value[len - 1] != 's') return false; |
||||
UniquePtr<char> buf(gpr_strdup(field->value)); |
||||
*(buf.get() + len - 1) = '\0'; // Remove trailing 's'.
|
||||
char* decimal_point = strchr(buf.get(), '.'); |
||||
int nanos = 0; |
||||
if (decimal_point != nullptr) { |
||||
*decimal_point = '\0'; |
||||
nanos = gpr_parse_nonnegative_int(decimal_point + 1); |
||||
if (nanos == -1) { |
||||
return false; |
||||
} |
||||
int num_digits = static_cast<int>(strlen(decimal_point + 1)); |
||||
if (num_digits > 9) { // We don't accept greater precision than nanos.
|
||||
return false; |
||||
} |
||||
for (int i = 0; i < (9 - num_digits); ++i) { |
||||
nanos *= 10; |
||||
} |
||||
} |
||||
int seconds = |
||||
decimal_point == buf.get() ? 0 : gpr_parse_nonnegative_int(buf.get()); |
||||
if (seconds == -1) return false; |
||||
*duration = seconds * GPR_MS_PER_SEC + nanos / GPR_NS_PER_MS; |
||||
return true; |
||||
} |
||||
|
||||
UniquePtr<ClientChannelMethodParams::RetryPolicy> ParseRetryPolicy( |
||||
grpc_json* field) { |
||||
auto retry_policy = MakeUnique<ClientChannelMethodParams::RetryPolicy>(); |
||||
if (field->type != GRPC_JSON_OBJECT) return nullptr; |
||||
for (grpc_json* sub_field = field->child; sub_field != nullptr; |
||||
sub_field = sub_field->next) { |
||||
if (sub_field->key == nullptr) return nullptr; |
||||
if (strcmp(sub_field->key, "maxAttempts") == 0) { |
||||
if (retry_policy->max_attempts != 0) return nullptr; // Duplicate.
|
||||
if (sub_field->type != GRPC_JSON_NUMBER) return nullptr; |
||||
retry_policy->max_attempts = gpr_parse_nonnegative_int(sub_field->value); |
||||
if (retry_policy->max_attempts <= 1) return nullptr; |
||||
if (retry_policy->max_attempts > MAX_MAX_RETRY_ATTEMPTS) { |
||||
gpr_log(GPR_ERROR, |
||||
"service config: clamped retryPolicy.maxAttempts at %d", |
||||
MAX_MAX_RETRY_ATTEMPTS); |
||||
retry_policy->max_attempts = MAX_MAX_RETRY_ATTEMPTS; |
||||
} |
||||
} else if (strcmp(sub_field->key, "initialBackoff") == 0) { |
||||
if (retry_policy->initial_backoff > 0) return nullptr; // Duplicate.
|
||||
if (!ParseDuration(sub_field, &retry_policy->initial_backoff)) { |
||||
return nullptr; |
||||
} |
||||
if (retry_policy->initial_backoff == 0) return nullptr; |
||||
} else if (strcmp(sub_field->key, "maxBackoff") == 0) { |
||||
if (retry_policy->max_backoff > 0) return nullptr; // Duplicate.
|
||||
if (!ParseDuration(sub_field, &retry_policy->max_backoff)) { |
||||
return nullptr; |
||||
} |
||||
if (retry_policy->max_backoff == 0) return nullptr; |
||||
} else if (strcmp(sub_field->key, "backoffMultiplier") == 0) { |
||||
if (retry_policy->backoff_multiplier != 0) return nullptr; // Duplicate.
|
||||
if (sub_field->type != GRPC_JSON_NUMBER) return nullptr; |
||||
if (sscanf(sub_field->value, "%f", &retry_policy->backoff_multiplier) != |
||||
1) { |
||||
return nullptr; |
||||
} |
||||
if (retry_policy->backoff_multiplier <= 0) return nullptr; |
||||
} else if (strcmp(sub_field->key, "retryableStatusCodes") == 0) { |
||||
if (!retry_policy->retryable_status_codes.Empty()) { |
||||
return nullptr; // Duplicate.
|
||||
} |
||||
if (sub_field->type != GRPC_JSON_ARRAY) return nullptr; |
||||
for (grpc_json* element = sub_field->child; element != nullptr; |
||||
element = element->next) { |
||||
if (element->type != GRPC_JSON_STRING) return nullptr; |
||||
grpc_status_code status; |
||||
if (!grpc_status_code_from_string(element->value, &status)) { |
||||
return nullptr; |
||||
} |
||||
retry_policy->retryable_status_codes.Add(status); |
||||
} |
||||
if (retry_policy->retryable_status_codes.Empty()) return nullptr; |
||||
} |
||||
} |
||||
// Make sure required fields are set.
|
||||
if (retry_policy->max_attempts == 0 || retry_policy->initial_backoff == 0 || |
||||
retry_policy->max_backoff == 0 || retry_policy->backoff_multiplier == 0 || |
||||
retry_policy->retryable_status_codes.Empty()) { |
||||
return nullptr; |
||||
} |
||||
return retry_policy; |
||||
} |
||||
|
||||
} // namespace
|
||||
|
||||
RefCountedPtr<ClientChannelMethodParams> |
||||
ClientChannelMethodParams::CreateFromJson(const grpc_json* json) { |
||||
RefCountedPtr<ClientChannelMethodParams> method_params = |
||||
MakeRefCounted<ClientChannelMethodParams>(); |
||||
for (grpc_json* field = json->child; field != nullptr; field = field->next) { |
||||
if (field->key == nullptr) continue; |
||||
if (strcmp(field->key, "waitForReady") == 0) { |
||||
if (method_params->wait_for_ready_ != WAIT_FOR_READY_UNSET) { |
||||
return nullptr; // Duplicate.
|
||||
} |
||||
if (!ParseWaitForReady(field, &method_params->wait_for_ready_)) { |
||||
return nullptr; |
||||
} |
||||
} else if (strcmp(field->key, "timeout") == 0) { |
||||
if (method_params->timeout_ > 0) return nullptr; // Duplicate.
|
||||
if (!ParseDuration(field, &method_params->timeout_)) return nullptr; |
||||
} else if (strcmp(field->key, "retryPolicy") == 0) { |
||||
if (method_params->retry_policy_ != nullptr) { |
||||
return nullptr; // Duplicate.
|
||||
} |
||||
method_params->retry_policy_ = ParseRetryPolicy(field); |
||||
if (method_params->retry_policy_ == nullptr) return nullptr; |
||||
} |
||||
} |
||||
return method_params; |
||||
} |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc_core
|
||||
@ -1,78 +0,0 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2015 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_METHOD_PARAMS_H |
||||
#define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_METHOD_PARAMS_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/lib/channel/status_util.h" |
||||
#include "src/core/lib/gprpp/ref_counted.h" |
||||
#include "src/core/lib/gprpp/ref_counted_ptr.h" |
||||
#include "src/core/lib/iomgr/exec_ctx.h" // for grpc_millis |
||||
#include "src/core/lib/json/json.h" |
||||
|
||||
namespace grpc_core { |
||||
namespace internal { |
||||
|
||||
class ClientChannelMethodParams : public RefCounted<ClientChannelMethodParams> { |
||||
public: |
||||
enum WaitForReady { |
||||
WAIT_FOR_READY_UNSET = 0, |
||||
WAIT_FOR_READY_FALSE, |
||||
WAIT_FOR_READY_TRUE |
||||
}; |
||||
|
||||
struct RetryPolicy { |
||||
int max_attempts = 0; |
||||
grpc_millis initial_backoff = 0; |
||||
grpc_millis max_backoff = 0; |
||||
float backoff_multiplier = 0; |
||||
StatusCodeSet retryable_status_codes; |
||||
}; |
||||
|
||||
/// Creates a method_parameters object from \a json.
|
||||
/// Intended for use with ServiceConfig::CreateMethodConfigTable().
|
||||
static RefCountedPtr<ClientChannelMethodParams> CreateFromJson( |
||||
const grpc_json* json); |
||||
|
||||
grpc_millis timeout() const { return timeout_; } |
||||
WaitForReady wait_for_ready() const { return wait_for_ready_; } |
||||
const RetryPolicy* retry_policy() const { return retry_policy_.get(); } |
||||
|
||||
private: |
||||
// So New() can call our private ctor.
|
||||
template <typename T, typename... Args> |
||||
friend T* grpc_core::New(Args&&... args); |
||||
|
||||
// So Delete() can call our private dtor.
|
||||
template <typename T> |
||||
friend void grpc_core::Delete(T*); |
||||
|
||||
ClientChannelMethodParams() {} |
||||
virtual ~ClientChannelMethodParams() {} |
||||
|
||||
grpc_millis timeout_ = 0; |
||||
WaitForReady wait_for_ready_ = WAIT_FOR_READY_UNSET; |
||||
UniquePtr<RetryPolicy> retry_policy_; |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_METHOD_PARAMS_H */ |
||||
@ -0,0 +1,384 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/ext/filters/client_channel/resolver_result_parsing.h" |
||||
|
||||
#include <ctype.h> |
||||
#include <stdio.h> |
||||
#include <string.h> |
||||
|
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/string_util.h> |
||||
|
||||
#include "src/core/ext/filters/client_channel/client_channel.h" |
||||
#include "src/core/ext/filters/client_channel/lb_policy_registry.h" |
||||
#include "src/core/lib/channel/status_util.h" |
||||
#include "src/core/lib/gpr/string.h" |
||||
#include "src/core/lib/gprpp/memory.h" |
||||
|
||||
// As per the retry design, we do not allow more than 5 retry attempts.
|
||||
#define MAX_MAX_RETRY_ATTEMPTS 5 |
||||
|
||||
namespace grpc_core { |
||||
namespace internal { |
||||
|
||||
namespace { |
||||
|
||||
// Converts string format from JSON to proto.
|
||||
grpc_core::UniquePtr<char> ConvertCamelToSnake(const char* camel) { |
||||
const size_t size = strlen(camel); |
||||
char* snake = static_cast<char*>(gpr_malloc(size * 2)); |
||||
size_t j = 0; |
||||
for (size_t i = 0; i < size; ++i) { |
||||
if (isupper(camel[i])) { |
||||
snake[j++] = '_'; |
||||
snake[j++] = tolower(camel[i]); |
||||
} else { |
||||
snake[j++] = camel[i]; |
||||
} |
||||
} |
||||
snake[j] = '\0'; |
||||
return grpc_core::UniquePtr<char>(snake); |
||||
} |
||||
|
||||
} // namespace
|
||||
|
||||
ProcessedResolverResult::ProcessedResolverResult( |
||||
const grpc_channel_args* resolver_result, bool parse_retry) { |
||||
ProcessServiceConfig(resolver_result, parse_retry); |
||||
// If no LB config was found above, just find the LB policy name then.
|
||||
if (lb_policy_config_ == nullptr) ProcessLbPolicyName(resolver_result); |
||||
} |
||||
|
||||
void ProcessedResolverResult::ProcessServiceConfig( |
||||
const grpc_channel_args* resolver_result, bool parse_retry) { |
||||
const grpc_arg* channel_arg = |
||||
grpc_channel_args_find(resolver_result, GRPC_ARG_SERVICE_CONFIG); |
||||
const char* service_config_json = grpc_channel_arg_get_string(channel_arg); |
||||
if (service_config_json != nullptr) { |
||||
service_config_json_.reset(gpr_strdup(service_config_json)); |
||||
service_config_ = grpc_core::ServiceConfig::Create(service_config_json); |
||||
if (service_config_ != nullptr) { |
||||
if (parse_retry) { |
||||
channel_arg = |
||||
grpc_channel_args_find(resolver_result, GRPC_ARG_SERVER_URI); |
||||
const char* server_uri = grpc_channel_arg_get_string(channel_arg); |
||||
GPR_ASSERT(server_uri != nullptr); |
||||
grpc_uri* uri = grpc_uri_parse(server_uri, true); |
||||
GPR_ASSERT(uri->path[0] != '\0'); |
||||
server_name_ = uri->path[0] == '/' ? uri->path + 1 : uri->path; |
||||
service_config_->ParseGlobalParams(ParseServiceConfig, this); |
||||
grpc_uri_destroy(uri); |
||||
} else { |
||||
service_config_->ParseGlobalParams(ParseServiceConfig, this); |
||||
} |
||||
method_params_table_ = service_config_->CreateMethodConfigTable( |
||||
ClientChannelMethodParams::CreateFromJson); |
||||
} |
||||
} |
||||
} |
||||
|
||||
void ProcessedResolverResult::ProcessLbPolicyName( |
||||
const grpc_channel_args* resolver_result) { |
||||
const char* lb_policy_name = nullptr; |
||||
// Prefer the LB policy name found in the service config. Note that this is
|
||||
// checking the deprecated loadBalancingPolicy field, rather than the new
|
||||
// loadBalancingConfig field.
|
||||
if (service_config_ != nullptr) { |
||||
lb_policy_name = service_config_->GetLoadBalancingPolicyName(); |
||||
} |
||||
// Otherwise, find the LB policy name set by the client API.
|
||||
if (lb_policy_name == nullptr) { |
||||
const grpc_arg* channel_arg = |
||||
grpc_channel_args_find(resolver_result, GRPC_ARG_LB_POLICY_NAME); |
||||
lb_policy_name = grpc_channel_arg_get_string(channel_arg); |
||||
} |
||||
// Special case: If at least one balancer address is present, we use
|
||||
// the grpclb policy, regardless of what the resolver has returned.
|
||||
const grpc_arg* channel_arg = |
||||
grpc_channel_args_find(resolver_result, GRPC_ARG_LB_ADDRESSES); |
||||
if (channel_arg != nullptr && channel_arg->type == GRPC_ARG_POINTER) { |
||||
grpc_lb_addresses* addresses = |
||||
static_cast<grpc_lb_addresses*>(channel_arg->value.pointer.p); |
||||
if (grpc_lb_addresses_contains_balancer_address(*addresses)) { |
||||
if (lb_policy_name != nullptr && |
||||
gpr_stricmp(lb_policy_name, "grpclb") != 0) { |
||||
gpr_log(GPR_INFO, |
||||
"resolver requested LB policy %s but provided at least one " |
||||
"balancer address -- forcing use of grpclb LB policy", |
||||
lb_policy_name); |
||||
} |
||||
lb_policy_name = "grpclb"; |
||||
} |
||||
} |
||||
// Use pick_first if nothing was specified and we didn't select grpclb
|
||||
// above.
|
||||
if (lb_policy_name == nullptr) lb_policy_name = "pick_first"; |
||||
lb_policy_name_.reset(gpr_strdup(lb_policy_name)); |
||||
} |
||||
|
||||
void ProcessedResolverResult::ParseServiceConfig( |
||||
const grpc_json* field, ProcessedResolverResult* parsing_state) { |
||||
parsing_state->ParseLbConfigFromServiceConfig(field); |
||||
if (parsing_state->server_name_ != nullptr) { |
||||
parsing_state->ParseRetryThrottleParamsFromServiceConfig(field); |
||||
} |
||||
} |
||||
|
||||
void ProcessedResolverResult::ParseLbConfigFromServiceConfig( |
||||
const grpc_json* field) { |
||||
if (lb_policy_config_ != nullptr) return; // Already found.
|
||||
// Find the LB config global parameter.
|
||||
if (field->key == nullptr || strcmp(field->key, "loadBalancingConfig") != 0 || |
||||
field->type != GRPC_JSON_ARRAY) { |
||||
return; // Not valid lb config array.
|
||||
} |
||||
// Find the first LB policy that this client supports.
|
||||
for (grpc_json* lb_config = field->child; lb_config != nullptr; |
||||
lb_config = lb_config->next) { |
||||
if (lb_config->type != GRPC_JSON_OBJECT) return; |
||||
// Find the policy object.
|
||||
grpc_json* policy = nullptr; |
||||
for (grpc_json* field = lb_config->child; field != nullptr; |
||||
field = field->next) { |
||||
if (field->key == nullptr || strcmp(field->key, "policy") != 0 || |
||||
field->type != GRPC_JSON_OBJECT) { |
||||
return; |
||||
} |
||||
if (policy != nullptr) return; // Duplicate.
|
||||
policy = field; |
||||
} |
||||
// Find the specific policy content since the policy object is of type
|
||||
// "oneof".
|
||||
grpc_json* policy_content = nullptr; |
||||
for (grpc_json* field = policy->child; field != nullptr; |
||||
field = field->next) { |
||||
if (field->key == nullptr || field->type != GRPC_JSON_OBJECT) return; |
||||
if (policy_content != nullptr) return; // Violate "oneof" type.
|
||||
policy_content = field; |
||||
} |
||||
grpc_core::UniquePtr<char> lb_policy_name = |
||||
ConvertCamelToSnake(policy_content->key); |
||||
if (!grpc_core::LoadBalancingPolicyRegistry::LoadBalancingPolicyExists( |
||||
lb_policy_name.get())) { |
||||
continue; |
||||
} |
||||
lb_policy_name_ = std::move(lb_policy_name); |
||||
lb_policy_config_ = policy_content->child; |
||||
return; |
||||
} |
||||
} |
||||
|
||||
void ProcessedResolverResult::ParseRetryThrottleParamsFromServiceConfig( |
||||
const grpc_json* field) { |
||||
if (strcmp(field->key, "retryThrottling") == 0) { |
||||
if (retry_throttle_data_ != nullptr) return; // Duplicate.
|
||||
if (field->type != GRPC_JSON_OBJECT) return; |
||||
int max_milli_tokens = 0; |
||||
int milli_token_ratio = 0; |
||||
for (grpc_json* sub_field = field->child; sub_field != nullptr; |
||||
sub_field = sub_field->next) { |
||||
if (sub_field->key == nullptr) return; |
||||
if (strcmp(sub_field->key, "maxTokens") == 0) { |
||||
if (max_milli_tokens != 0) return; // Duplicate.
|
||||
if (sub_field->type != GRPC_JSON_NUMBER) return; |
||||
max_milli_tokens = gpr_parse_nonnegative_int(sub_field->value); |
||||
if (max_milli_tokens == -1) return; |
||||
max_milli_tokens *= 1000; |
||||
} else if (strcmp(sub_field->key, "tokenRatio") == 0) { |
||||
if (milli_token_ratio != 0) return; // Duplicate.
|
||||
if (sub_field->type != GRPC_JSON_NUMBER) return; |
||||
// We support up to 3 decimal digits.
|
||||
size_t whole_len = strlen(sub_field->value); |
||||
uint32_t multiplier = 1; |
||||
uint32_t decimal_value = 0; |
||||
const char* decimal_point = strchr(sub_field->value, '.'); |
||||
if (decimal_point != nullptr) { |
||||
whole_len = static_cast<size_t>(decimal_point - sub_field->value); |
||||
multiplier = 1000; |
||||
size_t decimal_len = strlen(decimal_point + 1); |
||||
if (decimal_len > 3) decimal_len = 3; |
||||
if (!gpr_parse_bytes_to_uint32(decimal_point + 1, decimal_len, |
||||
&decimal_value)) { |
||||
return; |
||||
} |
||||
uint32_t decimal_multiplier = 1; |
||||
for (size_t i = 0; i < (3 - decimal_len); ++i) { |
||||
decimal_multiplier *= 10; |
||||
} |
||||
decimal_value *= decimal_multiplier; |
||||
} |
||||
uint32_t whole_value; |
||||
if (!gpr_parse_bytes_to_uint32(sub_field->value, whole_len, |
||||
&whole_value)) { |
||||
return; |
||||
} |
||||
milli_token_ratio = |
||||
static_cast<int>((whole_value * multiplier) + decimal_value); |
||||
if (milli_token_ratio <= 0) return; |
||||
} |
||||
} |
||||
retry_throttle_data_ = |
||||
grpc_core::internal::ServerRetryThrottleMap::GetDataForServer( |
||||
server_name_, max_milli_tokens, milli_token_ratio); |
||||
} |
||||
} |
||||
|
||||
namespace { |
||||
|
||||
bool ParseWaitForReady( |
||||
grpc_json* field, ClientChannelMethodParams::WaitForReady* wait_for_ready) { |
||||
if (field->type != GRPC_JSON_TRUE && field->type != GRPC_JSON_FALSE) { |
||||
return false; |
||||
} |
||||
*wait_for_ready = field->type == GRPC_JSON_TRUE |
||||
? ClientChannelMethodParams::WAIT_FOR_READY_TRUE |
||||
: ClientChannelMethodParams::WAIT_FOR_READY_FALSE; |
||||
return true; |
||||
} |
||||
|
||||
// Parses a JSON field of the form generated for a google.proto.Duration
|
||||
// proto message, as per:
|
||||
// https://developers.google.com/protocol-buffers/docs/proto3#json
|
||||
bool ParseDuration(grpc_json* field, grpc_millis* duration) { |
||||
if (field->type != GRPC_JSON_STRING) return false; |
||||
size_t len = strlen(field->value); |
||||
if (field->value[len - 1] != 's') return false; |
||||
UniquePtr<char> buf(gpr_strdup(field->value)); |
||||
*(buf.get() + len - 1) = '\0'; // Remove trailing 's'.
|
||||
char* decimal_point = strchr(buf.get(), '.'); |
||||
int nanos = 0; |
||||
if (decimal_point != nullptr) { |
||||
*decimal_point = '\0'; |
||||
nanos = gpr_parse_nonnegative_int(decimal_point + 1); |
||||
if (nanos == -1) { |
||||
return false; |
||||
} |
||||
int num_digits = static_cast<int>(strlen(decimal_point + 1)); |
||||
if (num_digits > 9) { // We don't accept greater precision than nanos.
|
||||
return false; |
||||
} |
||||
for (int i = 0; i < (9 - num_digits); ++i) { |
||||
nanos *= 10; |
||||
} |
||||
} |
||||
int seconds = |
||||
decimal_point == buf.get() ? 0 : gpr_parse_nonnegative_int(buf.get()); |
||||
if (seconds == -1) return false; |
||||
*duration = seconds * GPR_MS_PER_SEC + nanos / GPR_NS_PER_MS; |
||||
return true; |
||||
} |
||||
|
||||
UniquePtr<ClientChannelMethodParams::RetryPolicy> ParseRetryPolicy( |
||||
grpc_json* field) { |
||||
auto retry_policy = MakeUnique<ClientChannelMethodParams::RetryPolicy>(); |
||||
if (field->type != GRPC_JSON_OBJECT) return nullptr; |
||||
for (grpc_json* sub_field = field->child; sub_field != nullptr; |
||||
sub_field = sub_field->next) { |
||||
if (sub_field->key == nullptr) return nullptr; |
||||
if (strcmp(sub_field->key, "maxAttempts") == 0) { |
||||
if (retry_policy->max_attempts != 0) return nullptr; // Duplicate.
|
||||
if (sub_field->type != GRPC_JSON_NUMBER) return nullptr; |
||||
retry_policy->max_attempts = gpr_parse_nonnegative_int(sub_field->value); |
||||
if (retry_policy->max_attempts <= 1) return nullptr; |
||||
if (retry_policy->max_attempts > MAX_MAX_RETRY_ATTEMPTS) { |
||||
gpr_log(GPR_ERROR, |
||||
"service config: clamped retryPolicy.maxAttempts at %d", |
||||
MAX_MAX_RETRY_ATTEMPTS); |
||||
retry_policy->max_attempts = MAX_MAX_RETRY_ATTEMPTS; |
||||
} |
||||
} else if (strcmp(sub_field->key, "initialBackoff") == 0) { |
||||
if (retry_policy->initial_backoff > 0) return nullptr; // Duplicate.
|
||||
if (!ParseDuration(sub_field, &retry_policy->initial_backoff)) { |
||||
return nullptr; |
||||
} |
||||
if (retry_policy->initial_backoff == 0) return nullptr; |
||||
} else if (strcmp(sub_field->key, "maxBackoff") == 0) { |
||||
if (retry_policy->max_backoff > 0) return nullptr; // Duplicate.
|
||||
if (!ParseDuration(sub_field, &retry_policy->max_backoff)) { |
||||
return nullptr; |
||||
} |
||||
if (retry_policy->max_backoff == 0) return nullptr; |
||||
} else if (strcmp(sub_field->key, "backoffMultiplier") == 0) { |
||||
if (retry_policy->backoff_multiplier != 0) return nullptr; // Duplicate.
|
||||
if (sub_field->type != GRPC_JSON_NUMBER) return nullptr; |
||||
if (sscanf(sub_field->value, "%f", &retry_policy->backoff_multiplier) != |
||||
1) { |
||||
return nullptr; |
||||
} |
||||
if (retry_policy->backoff_multiplier <= 0) return nullptr; |
||||
} else if (strcmp(sub_field->key, "retryableStatusCodes") == 0) { |
||||
if (!retry_policy->retryable_status_codes.Empty()) { |
||||
return nullptr; // Duplicate.
|
||||
} |
||||
if (sub_field->type != GRPC_JSON_ARRAY) return nullptr; |
||||
for (grpc_json* element = sub_field->child; element != nullptr; |
||||
element = element->next) { |
||||
if (element->type != GRPC_JSON_STRING) return nullptr; |
||||
grpc_status_code status; |
||||
if (!grpc_status_code_from_string(element->value, &status)) { |
||||
return nullptr; |
||||
} |
||||
retry_policy->retryable_status_codes.Add(status); |
||||
} |
||||
if (retry_policy->retryable_status_codes.Empty()) return nullptr; |
||||
} |
||||
} |
||||
// Make sure required fields are set.
|
||||
if (retry_policy->max_attempts == 0 || retry_policy->initial_backoff == 0 || |
||||
retry_policy->max_backoff == 0 || retry_policy->backoff_multiplier == 0 || |
||||
retry_policy->retryable_status_codes.Empty()) { |
||||
return nullptr; |
||||
} |
||||
return retry_policy; |
||||
} |
||||
|
||||
} // namespace
|
||||
|
||||
RefCountedPtr<ClientChannelMethodParams> |
||||
ClientChannelMethodParams::CreateFromJson(const grpc_json* json) { |
||||
RefCountedPtr<ClientChannelMethodParams> method_params = |
||||
MakeRefCounted<ClientChannelMethodParams>(); |
||||
for (grpc_json* field = json->child; field != nullptr; field = field->next) { |
||||
if (field->key == nullptr) continue; |
||||
if (strcmp(field->key, "waitForReady") == 0) { |
||||
if (method_params->wait_for_ready_ != WAIT_FOR_READY_UNSET) { |
||||
return nullptr; // Duplicate.
|
||||
} |
||||
if (!ParseWaitForReady(field, &method_params->wait_for_ready_)) { |
||||
return nullptr; |
||||
} |
||||
} else if (strcmp(field->key, "timeout") == 0) { |
||||
if (method_params->timeout_ > 0) return nullptr; // Duplicate.
|
||||
if (!ParseDuration(field, &method_params->timeout_)) return nullptr; |
||||
} else if (strcmp(field->key, "retryPolicy") == 0) { |
||||
if (method_params->retry_policy_ != nullptr) { |
||||
return nullptr; // Duplicate.
|
||||
} |
||||
method_params->retry_policy_ = ParseRetryPolicy(field); |
||||
if (method_params->retry_policy_ == nullptr) return nullptr; |
||||
} |
||||
} |
||||
return method_params; |
||||
} |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc_core
|
||||
@ -0,0 +1,146 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_RESOLVER_RESULT_PARSING_H |
||||
#define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_RESOLVER_RESULT_PARSING_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/ext/filters/client_channel/retry_throttle.h" |
||||
#include "src/core/lib/channel/status_util.h" |
||||
#include "src/core/lib/gprpp/ref_counted.h" |
||||
#include "src/core/lib/gprpp/ref_counted_ptr.h" |
||||
#include "src/core/lib/iomgr/exec_ctx.h" // for grpc_millis |
||||
#include "src/core/lib/json/json.h" |
||||
#include "src/core/lib/slice/slice_hash_table.h" |
||||
#include "src/core/lib/transport/service_config.h" |
||||
|
||||
namespace grpc_core { |
||||
namespace internal { |
||||
|
||||
class ClientChannelMethodParams; |
||||
|
||||
// A table mapping from a method name to its method parameters.
|
||||
typedef grpc_core::SliceHashTable< |
||||
grpc_core::RefCountedPtr<ClientChannelMethodParams>> |
||||
ClientChannelMethodParamsTable; |
||||
|
||||
// A container of processed fields from the resolver result. Simplifies the
|
||||
// usage of resolver result.
|
||||
class ProcessedResolverResult { |
||||
public: |
||||
// Processes the resolver result and populates the relative members
|
||||
// for later consumption. Tries to parse retry parameters only if parse_retry
|
||||
// is true.
|
||||
ProcessedResolverResult(const grpc_channel_args* resolver_result, |
||||
bool parse_retry); |
||||
|
||||
// Getters. Any managed object's ownership is transferred.
|
||||
grpc_core::UniquePtr<char> service_config_json() { |
||||
return std::move(service_config_json_); |
||||
} |
||||
grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data() { |
||||
return std::move(retry_throttle_data_); |
||||
} |
||||
grpc_core::RefCountedPtr<ClientChannelMethodParamsTable> |
||||
method_params_table() { |
||||
return std::move(method_params_table_); |
||||
} |
||||
grpc_core::UniquePtr<char> lb_policy_name() { |
||||
return std::move(lb_policy_name_); |
||||
} |
||||
grpc_json* lb_policy_config() { return lb_policy_config_; } |
||||
|
||||
private: |
||||
// Finds the service config; extracts LB config and (maybe) retry throttle
|
||||
// params from it.
|
||||
void ProcessServiceConfig(const grpc_channel_args* resolver_result, |
||||
bool parse_retry); |
||||
|
||||
// Finds the LB policy name (when no LB config was found).
|
||||
void ProcessLbPolicyName(const grpc_channel_args* resolver_result); |
||||
|
||||
// Parses the service config. Intended to be used by
|
||||
// ServiceConfig::ParseGlobalParams.
|
||||
static void ParseServiceConfig(const grpc_json* field, |
||||
ProcessedResolverResult* parsing_state); |
||||
// Parses the LB config from service config.
|
||||
void ParseLbConfigFromServiceConfig(const grpc_json* field); |
||||
// Parses the retry throttle parameters from service config.
|
||||
void ParseRetryThrottleParamsFromServiceConfig(const grpc_json* field); |
||||
|
||||
// Service config.
|
||||
grpc_core::UniquePtr<char> service_config_json_; |
||||
grpc_core::UniquePtr<grpc_core::ServiceConfig> service_config_; |
||||
// LB policy.
|
||||
grpc_json* lb_policy_config_ = nullptr; |
||||
grpc_core::UniquePtr<char> lb_policy_name_; |
||||
// Retry throttle data.
|
||||
char* server_name_ = nullptr; |
||||
grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data_; |
||||
// Method params table.
|
||||
grpc_core::RefCountedPtr<ClientChannelMethodParamsTable> method_params_table_; |
||||
}; |
||||
|
||||
// The parameters of a method.
|
||||
class ClientChannelMethodParams : public RefCounted<ClientChannelMethodParams> { |
||||
public: |
||||
enum WaitForReady { |
||||
WAIT_FOR_READY_UNSET = 0, |
||||
WAIT_FOR_READY_FALSE, |
||||
WAIT_FOR_READY_TRUE |
||||
}; |
||||
|
||||
struct RetryPolicy { |
||||
int max_attempts = 0; |
||||
grpc_millis initial_backoff = 0; |
||||
grpc_millis max_backoff = 0; |
||||
float backoff_multiplier = 0; |
||||
StatusCodeSet retryable_status_codes; |
||||
}; |
||||
|
||||
/// Creates a method_parameters object from \a json.
|
||||
/// Intended for use with ServiceConfig::CreateMethodConfigTable().
|
||||
static RefCountedPtr<ClientChannelMethodParams> CreateFromJson( |
||||
const grpc_json* json); |
||||
|
||||
grpc_millis timeout() const { return timeout_; } |
||||
WaitForReady wait_for_ready() const { return wait_for_ready_; } |
||||
const RetryPolicy* retry_policy() const { return retry_policy_.get(); } |
||||
|
||||
private: |
||||
// So New() can call our private ctor.
|
||||
template <typename T, typename... Args> |
||||
friend T* grpc_core::New(Args&&... args); |
||||
|
||||
// So Delete() can call our private dtor.
|
||||
template <typename T> |
||||
friend void grpc_core::Delete(T*); |
||||
|
||||
ClientChannelMethodParams() {} |
||||
virtual ~ClientChannelMethodParams() {} |
||||
|
||||
grpc_millis timeout_ = 0; |
||||
WaitForReady wait_for_ready_ = WAIT_FOR_READY_UNSET; |
||||
UniquePtr<RetryPolicy> retry_policy_; |
||||
}; |
||||
|
||||
} // namespace internal
|
||||
} // namespace grpc_core
|
||||
|
||||
#endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_RESOLVER_RESULT_PARSING_H */ |
||||
Loading…
Reference in new issue