1 //
2 // Copyright 2018 gRPC authors.
3 //
4 // Licensed under the Apache License, Version 2.0 (the "License");
5 // you may not use this file except in compliance with the License.
6 // You may obtain a copy of the License at
7 //
8 // http://www.apache.org/licenses/LICENSE-2.0
9 //
10 // Unless required by applicable law or agreed to in writing, software
11 // distributed under the License is distributed on an "AS IS" BASIS,
12 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 // See the License for the specific language governing permissions and
14 // limitations under the License.
15 //
16
17 #include "src/core/client_channel/retry_service_config.h"
18
19 #include <grpc/impl/channel_arg_names.h>
20 #include <grpc/status.h>
21 #include <grpc/support/json.h>
22 #include <grpc/support/port_platform.h>
23
24 #include <map>
25 #include <string>
26 #include <utility>
27 #include <vector>
28
29 #include "absl/log/log.h"
30 #include "absl/strings/numbers.h"
31 #include "absl/strings/str_cat.h"
32 #include "absl/types/optional.h"
33 #include "src/core/config/core_configuration.h"
34 #include "src/core/lib/channel/channel_args.h"
35 #include "src/core/lib/channel/status_util.h"
36 #include "src/core/util/json/json_channel_args.h"
37
38 // As per the retry design, we do not allow more than 5 retry attempts.
39 #define MAX_MAX_RETRY_ATTEMPTS 5
40
41 namespace grpc_core {
42 namespace internal {
43
44 //
45 // RetryGlobalConfig
46 //
47
JsonLoader(const JsonArgs &)48 const JsonLoaderInterface* RetryGlobalConfig::JsonLoader(const JsonArgs&) {
49 // Note: Both fields require custom processing, so they're handled in
50 // JsonPostLoad() instead.
51 static const auto* loader = JsonObjectLoader<RetryGlobalConfig>().Finish();
52 return loader;
53 }
54
JsonPostLoad(const Json & json,const JsonArgs & args,ValidationErrors * errors)55 void RetryGlobalConfig::JsonPostLoad(const Json& json, const JsonArgs& args,
56 ValidationErrors* errors) {
57 // Parse maxTokens.
58 auto max_tokens =
59 LoadJsonObjectField<uint32_t>(json.object(), args, "maxTokens", errors);
60 if (max_tokens.has_value()) {
61 ValidationErrors::ScopedField field(errors, ".maxTokens");
62 if (*max_tokens == 0) {
63 errors->AddError("must be greater than 0");
64 } else {
65 // Multiply by 1000 to represent as milli-tokens.
66 max_milli_tokens_ = static_cast<uintptr_t>(*max_tokens) * 1000;
67 }
68 }
69 // Parse tokenRatio.
70 ValidationErrors::ScopedField field(errors, ".tokenRatio");
71 auto it = json.object().find("tokenRatio");
72 if (it == json.object().end()) {
73 errors->AddError("field not present");
74 return;
75 }
76 if (it->second.type() != Json::Type::kNumber &&
77 it->second.type() != Json::Type::kString) {
78 errors->AddError("is not a number");
79 return;
80 }
81 absl::string_view buf = it->second.string();
82 uint32_t multiplier = 1;
83 uint32_t decimal_value = 0;
84 auto decimal_point = buf.find('.');
85 if (decimal_point != absl::string_view::npos) {
86 absl::string_view after_decimal = buf.substr(decimal_point + 1);
87 buf = buf.substr(0, decimal_point);
88 // We support up to 3 decimal digits.
89 multiplier = 1000;
90 if (after_decimal.length() > 3) after_decimal = after_decimal.substr(0, 3);
91 // Parse decimal value.
92 if (!absl::SimpleAtoi(after_decimal, &decimal_value)) {
93 errors->AddError("could not parse as a number");
94 return;
95 }
96 uint32_t decimal_multiplier = 1;
97 for (size_t i = 0; i < (3 - after_decimal.length()); ++i) {
98 decimal_multiplier *= 10;
99 }
100 decimal_value *= decimal_multiplier;
101 }
102 uint32_t whole_value;
103 if (!absl::SimpleAtoi(buf, &whole_value)) {
104 errors->AddError("could not parse as a number");
105 return;
106 }
107 milli_token_ratio_ =
108 static_cast<int>((whole_value * multiplier) + decimal_value);
109 if (milli_token_ratio_ <= 0) {
110 errors->AddError("must be greater than 0");
111 }
112 }
113
114 //
115 // RetryMethodConfig
116 //
117
JsonLoader(const JsonArgs &)118 const JsonLoaderInterface* RetryMethodConfig::JsonLoader(const JsonArgs&) {
119 static const auto* loader =
120 JsonObjectLoader<RetryMethodConfig>()
121 // Note: The "retryableStatusCodes" field requires custom parsing,
122 // so it's handled in JsonPostLoad() instead.
123 .Field("maxAttempts", &RetryMethodConfig::max_attempts_)
124 .Field("initialBackoff", &RetryMethodConfig::initial_backoff_)
125 .Field("maxBackoff", &RetryMethodConfig::max_backoff_)
126 .Field("backoffMultiplier", &RetryMethodConfig::backoff_multiplier_)
127 .OptionalField("perAttemptRecvTimeout",
128 &RetryMethodConfig::per_attempt_recv_timeout_,
129 GRPC_ARG_EXPERIMENTAL_ENABLE_HEDGING)
130 .Finish();
131 return loader;
132 }
133
JsonPostLoad(const Json & json,const JsonArgs & args,ValidationErrors * errors)134 void RetryMethodConfig::JsonPostLoad(const Json& json, const JsonArgs& args,
135 ValidationErrors* errors) {
136 // Validate maxAttempts.
137 {
138 ValidationErrors::ScopedField field(errors, ".maxAttempts");
139 if (!errors->FieldHasErrors()) {
140 if (max_attempts_ <= 1) {
141 errors->AddError("must be at least 2");
142 } else if (max_attempts_ > MAX_MAX_RETRY_ATTEMPTS) {
143 LOG(ERROR) << "service config: clamped retryPolicy.maxAttempts at "
144 << MAX_MAX_RETRY_ATTEMPTS;
145 max_attempts_ = MAX_MAX_RETRY_ATTEMPTS;
146 }
147 }
148 }
149 // Validate initialBackoff.
150 {
151 ValidationErrors::ScopedField field(errors, ".initialBackoff");
152 if (!errors->FieldHasErrors() && initial_backoff_ == Duration::Zero()) {
153 errors->AddError("must be greater than 0");
154 }
155 }
156 // Validate maxBackoff.
157 {
158 ValidationErrors::ScopedField field(errors, ".maxBackoff");
159 if (!errors->FieldHasErrors() && max_backoff_ == Duration::Zero()) {
160 errors->AddError("must be greater than 0");
161 }
162 }
163 // Validate backoffMultiplier.
164 {
165 ValidationErrors::ScopedField field(errors, ".backoffMultiplier");
166 if (!errors->FieldHasErrors() && backoff_multiplier_ <= 0) {
167 errors->AddError("must be greater than 0");
168 }
169 }
170 // Parse retryableStatusCodes.
171 auto status_code_list = LoadJsonObjectField<std::vector<std::string>>(
172 json.object(), args, "retryableStatusCodes", errors,
173 /*required=*/false);
174 if (status_code_list.has_value()) {
175 for (size_t i = 0; i < status_code_list->size(); ++i) {
176 ValidationErrors::ScopedField field(
177 errors, absl::StrCat(".retryableStatusCodes[", i, "]"));
178 grpc_status_code status;
179 if (!grpc_status_code_from_string((*status_code_list)[i].c_str(),
180 &status)) {
181 errors->AddError("failed to parse status code");
182 } else {
183 retryable_status_codes_.Add(status);
184 }
185 }
186 }
187 // Validate perAttemptRecvTimeout.
188 if (args.IsEnabled(GRPC_ARG_EXPERIMENTAL_ENABLE_HEDGING)) {
189 if (per_attempt_recv_timeout_.has_value()) {
190 ValidationErrors::ScopedField field(errors, ".perAttemptRecvTimeout");
191 // TODO(roth): As part of implementing hedging, relax this check such
192 // that we allow a value of 0 if a hedging policy is specified.
193 if (!errors->FieldHasErrors() &&
194 *per_attempt_recv_timeout_ == Duration::Zero()) {
195 errors->AddError("must be greater than 0");
196 }
197 } else if (retryable_status_codes_.Empty()) {
198 // If perAttemptRecvTimeout not present, retryableStatusCodes must be
199 // non-empty.
200 ValidationErrors::ScopedField field(errors, ".retryableStatusCodes");
201 if (!errors->FieldHasErrors()) {
202 errors->AddError(
203 "must be non-empty if perAttemptRecvTimeout not present");
204 }
205 }
206 } else if (retryable_status_codes_.Empty()) {
207 // Hedging not enabled, so the error message for
208 // retryableStatusCodes unset should be different.
209 ValidationErrors::ScopedField field(errors, ".retryableStatusCodes");
210 if (!errors->FieldHasErrors()) {
211 errors->AddError("must be non-empty");
212 }
213 }
214 }
215
216 //
217 // RetryServiceConfigParser
218 //
219
ParserIndex()220 size_t RetryServiceConfigParser::ParserIndex() {
221 return CoreConfiguration::Get().service_config_parser().GetParserIndex(
222 parser_name());
223 }
224
Register(CoreConfiguration::Builder * builder)225 void RetryServiceConfigParser::Register(CoreConfiguration::Builder* builder) {
226 builder->service_config_parser()->RegisterParser(
227 std::make_unique<RetryServiceConfigParser>());
228 }
229
230 namespace {
231
232 struct GlobalConfig {
233 std::unique_ptr<RetryGlobalConfig> retry_throttling;
234
JsonLoadergrpc_core::internal::__anon89b347c30111::GlobalConfig235 static const JsonLoaderInterface* JsonLoader(const JsonArgs&) {
236 static const auto* loader =
237 JsonObjectLoader<GlobalConfig>()
238 .OptionalField("retryThrottling", &GlobalConfig::retry_throttling)
239 .Finish();
240 return loader;
241 }
242 };
243
244 } // namespace
245
246 std::unique_ptr<ServiceConfigParser::ParsedConfig>
ParseGlobalParams(const ChannelArgs &,const Json & json,ValidationErrors * errors)247 RetryServiceConfigParser::ParseGlobalParams(const ChannelArgs& /*args*/,
248 const Json& json,
249 ValidationErrors* errors) {
250 auto global_params = LoadFromJson<GlobalConfig>(json, JsonArgs(), errors);
251 return std::move(global_params.retry_throttling);
252 }
253
254 namespace {
255
256 struct MethodConfig {
257 std::unique_ptr<RetryMethodConfig> retry_policy;
258
JsonLoadergrpc_core::internal::__anon89b347c30211::MethodConfig259 static const JsonLoaderInterface* JsonLoader(const JsonArgs&) {
260 static const auto* loader =
261 JsonObjectLoader<MethodConfig>()
262 .OptionalField("retryPolicy", &MethodConfig::retry_policy)
263 .Finish();
264 return loader;
265 }
266 };
267
268 } // namespace
269
270 std::unique_ptr<ServiceConfigParser::ParsedConfig>
ParsePerMethodParams(const ChannelArgs & args,const Json & json,ValidationErrors * errors)271 RetryServiceConfigParser::ParsePerMethodParams(const ChannelArgs& args,
272 const Json& json,
273 ValidationErrors* errors) {
274 auto method_params =
275 LoadFromJson<MethodConfig>(json, JsonChannelArgs(args), errors);
276 return std::move(method_params.retry_policy);
277 }
278
279 } // namespace internal
280 } // namespace grpc_core
281