Merge pull request #24328 from donnadionne/timeout

Setting timeout in method config.
pull/24454/head
donnadionne 4 years ago committed by GitHub
commit ad66015c85
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 7
      CMakeLists.txt
  2. 20
      Makefile
  3. 1
      build_autogenerated.yaml
  4. 59
      src/core/ext/filters/client_channel/resolver/xds/xds_resolver.cc
  5. 4
      src/core/ext/filters/client_channel/service_config.cc
  6. 40
      src/core/ext/xds/xds_api.cc
  7. 25
      src/core/ext/xds/xds_api.h
  8. 9
      src/proto/grpc/testing/xds/v3/BUILD
  9. 6
      src/proto/grpc/testing/xds/v3/http_connection_manager.proto
  10. 27
      src/proto/grpc/testing/xds/v3/protocol.proto
  11. 23
      src/proto/grpc/testing/xds/v3/route.proto
  12. 339
      test/cpp/end2end/xds_end2end_test.cc

@ -488,6 +488,9 @@ protobuf_generate_grpc_cpp(
protobuf_generate_grpc_cpp(
src/proto/grpc/testing/xds/v3/percent.proto
)
protobuf_generate_grpc_cpp(
src/proto/grpc/testing/xds/v3/protocol.proto
)
protobuf_generate_grpc_cpp(
src/proto/grpc/testing/xds/v3/range.proto
)
@ -15254,6 +15257,10 @@ if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/xds/v3/percent.grpc.pb.cc
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/xds/v3/percent.pb.h
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/xds/v3/percent.grpc.pb.h
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/xds/v3/protocol.pb.cc
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/xds/v3/protocol.grpc.pb.cc
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/xds/v3/protocol.pb.h
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/xds/v3/protocol.grpc.pb.h
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/xds/v3/range.pb.cc
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/xds/v3/range.grpc.pb.cc
${_gRPC_PROTO_GENS_DIR}/src/proto/grpc/testing/xds/v3/range.pb.h

@ -1422,12 +1422,12 @@ $(GENDIR)/src/proto/grpc/testing/xds/v3/http_connection_manager.pb.cc: protoc_de
$(GENDIR)/src/proto/grpc/testing/xds/v3/http_connection_manager.grpc.pb.cc: protoc_dep_error
else
$(GENDIR)/src/proto/grpc/testing/xds/v3/http_connection_manager.pb.cc: src/proto/grpc/testing/xds/v3/http_connection_manager.proto $(PROTOBUF_DEP) $(PROTOC_PLUGINS) $(GENDIR)/src/proto/grpc/testing/xds/v3/config_source.pb.cc $(GENDIR)/src/proto/grpc/testing/xds/v3/route.pb.cc
$(GENDIR)/src/proto/grpc/testing/xds/v3/http_connection_manager.pb.cc: src/proto/grpc/testing/xds/v3/http_connection_manager.proto $(PROTOBUF_DEP) $(PROTOC_PLUGINS) $(GENDIR)/src/proto/grpc/testing/xds/v3/config_source.pb.cc $(GENDIR)/src/proto/grpc/testing/xds/v3/protocol.pb.cc $(GENDIR)/src/proto/grpc/testing/xds/v3/route.pb.cc
$(E) "[PROTOC] Generating protobuf CC file from $<"
$(Q) mkdir -p `dirname $@`
$(Q) $(PROTOC) -Ithird_party/protobuf/src -I. --cpp_out=$(GENDIR) $<
$(GENDIR)/src/proto/grpc/testing/xds/v3/http_connection_manager.grpc.pb.cc: src/proto/grpc/testing/xds/v3/http_connection_manager.proto $(GENDIR)/src/proto/grpc/testing/xds/v3/http_connection_manager.pb.cc $(PROTOBUF_DEP) $(PROTOC_PLUGINS) $(GENDIR)/src/proto/grpc/testing/xds/v3/config_source.pb.cc $(GENDIR)/src/proto/grpc/testing/xds/v3/config_source.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/xds/v3/route.pb.cc $(GENDIR)/src/proto/grpc/testing/xds/v3/route.grpc.pb.cc
$(GENDIR)/src/proto/grpc/testing/xds/v3/http_connection_manager.grpc.pb.cc: src/proto/grpc/testing/xds/v3/http_connection_manager.proto $(GENDIR)/src/proto/grpc/testing/xds/v3/http_connection_manager.pb.cc $(PROTOBUF_DEP) $(PROTOC_PLUGINS) $(GENDIR)/src/proto/grpc/testing/xds/v3/config_source.pb.cc $(GENDIR)/src/proto/grpc/testing/xds/v3/config_source.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/xds/v3/protocol.pb.cc $(GENDIR)/src/proto/grpc/testing/xds/v3/protocol.grpc.pb.cc $(GENDIR)/src/proto/grpc/testing/xds/v3/route.pb.cc $(GENDIR)/src/proto/grpc/testing/xds/v3/route.grpc.pb.cc
$(E) "[GRPC] Generating gRPC's protobuf service CC file from $<"
$(Q) mkdir -p `dirname $@`
$(Q) $(PROTOC) -Ithird_party/protobuf/src -I. --grpc_out=$(GENDIR) --plugin=protoc-gen-grpc=$(PROTOC_PLUGINS_DIR)/grpc_cpp_plugin$(EXECUTABLE_SUFFIX) $<
@ -1497,6 +1497,22 @@ $(GENDIR)/src/proto/grpc/testing/xds/v3/percent.grpc.pb.cc: src/proto/grpc/testi
$(Q) $(PROTOC) -Ithird_party/protobuf/src -I. --grpc_out=$(GENDIR) --plugin=protoc-gen-grpc=$(PROTOC_PLUGINS_DIR)/grpc_cpp_plugin$(EXECUTABLE_SUFFIX) $<
endif
ifeq ($(NO_PROTOC),true)
$(GENDIR)/src/proto/grpc/testing/xds/v3/protocol.pb.cc: protoc_dep_error
$(GENDIR)/src/proto/grpc/testing/xds/v3/protocol.grpc.pb.cc: protoc_dep_error
else
$(GENDIR)/src/proto/grpc/testing/xds/v3/protocol.pb.cc: src/proto/grpc/testing/xds/v3/protocol.proto $(PROTOBUF_DEP) $(PROTOC_PLUGINS)
$(E) "[PROTOC] Generating protobuf CC file from $<"
$(Q) mkdir -p `dirname $@`
$(Q) $(PROTOC) -Ithird_party/protobuf/src -I. --cpp_out=$(GENDIR) $<
$(GENDIR)/src/proto/grpc/testing/xds/v3/protocol.grpc.pb.cc: src/proto/grpc/testing/xds/v3/protocol.proto $(GENDIR)/src/proto/grpc/testing/xds/v3/protocol.pb.cc $(PROTOBUF_DEP) $(PROTOC_PLUGINS)
$(E) "[GRPC] Generating gRPC's protobuf service CC file from $<"
$(Q) mkdir -p `dirname $@`
$(Q) $(PROTOC) -Ithird_party/protobuf/src -I. --grpc_out=$(GENDIR) --plugin=protoc-gen-grpc=$(PROTOC_PLUGINS_DIR)/grpc_cpp_plugin$(EXECUTABLE_SUFFIX) $<
endif
ifeq ($(NO_PROTOC),true)
$(GENDIR)/src/proto/grpc/testing/xds/v3/range.pb.cc: protoc_dep_error
$(GENDIR)/src/proto/grpc/testing/xds/v3/range.grpc.pb.cc: protoc_dep_error

@ -7758,6 +7758,7 @@ targets:
- src/proto/grpc/testing/xds/v3/load_report.proto
- src/proto/grpc/testing/xds/v3/lrs.proto
- src/proto/grpc/testing/xds/v3/percent.proto
- src/proto/grpc/testing/xds/v3/protocol.proto
- src/proto/grpc/testing/xds/v3/range.proto
- src/proto/grpc/testing/xds/v3/regex.proto
- src/proto/grpc/testing/xds/v3/route.proto

@ -138,7 +138,8 @@ class XdsResolver : public Resolver {
class XdsConfigSelector : public ConfigSelector {
public:
XdsConfigSelector(RefCountedPtr<XdsResolver> resolver,
const std::vector<XdsApi::Route>& routes);
const std::vector<XdsApi::Route>& routes,
grpc_error* error);
~XdsConfigSelector();
const char* name() const override { return "XdsConfigSelector"; }
@ -157,6 +158,7 @@ class XdsResolver : public Resolver {
XdsApi::Route route;
absl::InlinedVector<std::pair<uint32_t, absl::string_view>, 2>
weighted_cluster_state;
RefCountedPtr<ServiceConfig> method_config;
bool operator==(const Route& other) const {
return route == other.route &&
weighted_cluster_state == other.weighted_cluster_state;
@ -165,6 +167,8 @@ class XdsResolver : public Resolver {
using RouteTable = std::vector<Route>;
void MaybeAddCluster(const std::string& name);
grpc_error* CreateMethodConfig(RefCountedPtr<ServiceConfig>* method_config,
const XdsApi::Route& route);
RefCountedPtr<XdsResolver> resolver_;
RouteTable route_table_;
@ -189,6 +193,7 @@ class XdsResolver : public Resolver {
XdsClient::RouteConfigWatcherInterface* route_config_watcher_ = nullptr;
ClusterState::ClusterStateMap cluster_state_map_;
std::vector<XdsApi::Route> current_update_;
XdsApi::Duration http_max_stream_duration_;
};
//
@ -261,7 +266,7 @@ void XdsResolver::Notifier::RunInWorkSerializer(grpc_error* error) {
XdsResolver::XdsConfigSelector::XdsConfigSelector(
RefCountedPtr<XdsResolver> resolver,
const std::vector<XdsApi::Route>& routes)
const std::vector<XdsApi::Route>& routes, grpc_error* error)
: resolver_(std::move(resolver)) {
if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_resolver_trace)) {
gpr_log(GPR_INFO, "[xds_resolver %p] creating XdsConfigSelector %p",
@ -285,6 +290,13 @@ XdsResolver::XdsConfigSelector::XdsConfigSelector(
route_table_.emplace_back();
auto& route_entry = route_table_.back();
route_entry.route = route;
// If the route doesn't specify a timeout, set its timeout to the global
// one.
if (!route.max_stream_duration.has_value()) {
route_entry.route.max_stream_duration =
resolver_->http_max_stream_duration_;
}
error = CreateMethodConfig(&route_entry.method_config, route_entry.route);
if (route.weighted_clusters.empty()) {
MaybeAddCluster(route.cluster_name);
} else {
@ -299,6 +311,34 @@ XdsResolver::XdsConfigSelector::XdsConfigSelector(
}
}
grpc_error* XdsResolver::XdsConfigSelector::CreateMethodConfig(
RefCountedPtr<ServiceConfig>* method_config, const XdsApi::Route& route) {
grpc_error* error = GRPC_ERROR_NONE;
std::vector<std::string> fields;
if (route.max_stream_duration.has_value() &&
(route.max_stream_duration->seconds != 0 ||
route.max_stream_duration->nanos != 0)) {
fields.emplace_back(absl::StrFormat(" \"timeout\": \"%d.%09ds\"",
route.max_stream_duration->seconds,
route.max_stream_duration->nanos));
}
if (!fields.empty()) {
std::string json = absl::StrCat(
"{\n"
" \"methodConfig\": [ {\n"
" \"name\": [\n"
" {}\n"
" ],\n"
" ",
absl::StrJoin(fields, ",\n"),
"\n } ]\n"
"}");
*method_config =
ServiceConfig::Create(resolver_->args_, json.c_str(), &error);
}
return error;
}
XdsResolver::XdsConfigSelector::~XdsConfigSelector() {
if (GRPC_TRACE_FLAG_ENABLED(grpc_xds_resolver_trace)) {
gpr_log(GPR_INFO, "[xds_resolver %p] destroying XdsConfigSelector %p",
@ -484,6 +524,11 @@ ConfigSelector::CallConfig XdsResolver::XdsConfigSelector::GetCallConfig(
static_cast<XdsResolver*>(resolver_->Ref().release());
ClusterState* cluster_state = it->second->Ref().release();
CallConfig call_config;
if (entry.method_config != nullptr) {
call_config.service_config = entry.method_config;
call_config.method_configs =
entry.method_config->GetMethodParsedConfigVector(grpc_empty_slice());
}
call_config.call_attributes[kXdsClusterAttribute] = it->first;
call_config.on_call_committed = [resolver, cluster_state]() {
cluster_state->Unref();
@ -586,6 +631,7 @@ void XdsResolver::OnListenerUpdate(XdsApi::LdsUpdate listener) {
xds_client_->WatchRouteConfigData(route_config_name_, std::move(watcher));
}
}
http_max_stream_duration_ = listener.http_max_stream_duration;
if (route_config_name_.empty()) {
GPR_ASSERT(listener.rds_update.has_value());
OnRouteConfigUpdate(std::move(*listener.rds_update));
@ -671,10 +717,15 @@ void XdsResolver::GenerateResult() {
if (current_update_.empty()) return;
// First create XdsConfigSelector, which may add new entries to the cluster
// state map, and then CreateServiceConfig for LB policies.
grpc_error* error = GRPC_ERROR_NONE;
auto config_selector =
MakeRefCounted<XdsConfigSelector>(Ref(), current_update_);
MakeRefCounted<XdsConfigSelector>(Ref(), current_update_, error);
if (error != GRPC_ERROR_NONE) {
OnError(error);
return;
}
Result result;
grpc_error* error = CreateServiceConfig(&result.service_config);
error = CreateServiceConfig(&result.service_config);
if (error != GRPC_ERROR_NONE) {
OnError(error);
return;

@ -204,7 +204,9 @@ std::string ServiceConfig::ParseJsonMethodName(const Json& json,
const ServiceConfigParser::ParsedConfigVector*
ServiceConfig::GetMethodParsedConfigVector(const grpc_slice& path) const {
if (parsed_method_configs_map_.empty()) return nullptr;
if (parsed_method_configs_map_.empty()) {
return default_method_config_vector_;
}
// Try looking up the full path in the map.
auto it = parsed_method_configs_map_.find(path);
if (it != parsed_method_configs_map_.end()) return it->second;

@ -49,6 +49,7 @@
#include "envoy/config/core/v3/base.upb.h"
#include "envoy/config/core/v3/config_source.upb.h"
#include "envoy/config/core/v3/health_check.upb.h"
#include "envoy/config/core/v3/protocol.upb.h"
#include "envoy/config/endpoint/v3/endpoint.upb.h"
#include "envoy/config/endpoint/v3/endpoint.upbdefs.h"
#include "envoy/config/endpoint/v3/endpoint_components.upb.h"
@ -273,6 +274,9 @@ std::string XdsApi::Route::ToString() const {
for (const ClusterWeight& cluster_weight : weighted_clusters) {
contents.push_back(cluster_weight.ToString());
}
if (max_stream_duration.has_value()) {
contents.push_back(max_stream_duration->ToString());
}
return absl::StrJoin(contents, "\n");
}
@ -1121,7 +1125,27 @@ grpc_error* RouteActionParse(const envoy_config_route_v3_Route* route_msg,
} else {
// No cluster or weighted_clusters found in RouteAction, ignore this route.
*ignore_route = true;
return GRPC_ERROR_NONE;
}
if (!*ignore_route) {
const envoy_config_route_v3_RouteAction_MaxStreamDuration*
max_stream_duration =
envoy_config_route_v3_RouteAction_max_stream_duration(route_action);
if (max_stream_duration != nullptr) {
const google_protobuf_Duration* duration =
envoy_config_route_v3_RouteAction_MaxStreamDuration_grpc_timeout_header_max(
max_stream_duration);
if (duration == nullptr) {
duration =
envoy_config_route_v3_RouteAction_MaxStreamDuration_max_stream_duration(
max_stream_duration);
}
if (duration != nullptr) {
XdsApi::Duration duration_in_route;
duration_in_route.seconds = google_protobuf_Duration_seconds(duration);
duration_in_route.nanos = google_protobuf_Duration_nanos(duration);
route->max_stream_duration = duration_in_route;
}
}
}
return GRPC_ERROR_NONE;
}
@ -1251,6 +1275,20 @@ grpc_error* LdsResponseParse(
return GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"Could not parse HttpConnectionManager config from ApiListener");
}
// Obtain max_stream_duration from Http Protocol Options.
const envoy_config_core_v3_HttpProtocolOptions* options =
envoy_extensions_filters_network_http_connection_manager_v3_HttpConnectionManager_common_http_protocol_options(
http_connection_manager);
if (options != nullptr) {
const google_protobuf_Duration* duration =
envoy_config_core_v3_HttpProtocolOptions_max_stream_duration(options);
if (duration != nullptr) {
lds_update.http_max_stream_duration.seconds =
google_protobuf_Duration_seconds(duration);
lds_update.http_max_stream_duration.nanos =
google_protobuf_Duration_nanos(duration);
}
}
// Found inlined route_config. Parse it to find the cluster_name.
if (envoy_extensions_filters_network_http_connection_manager_v3_HttpConnectionManager_has_route_config(
http_connection_manager)) {

@ -48,6 +48,17 @@ class XdsApi {
static const char* kCdsTypeUrl;
static const char* kEdsTypeUrl;
struct Duration {
int64_t seconds = 0;
int32_t nanos = 0;
bool operator==(const Duration& other) const {
return (seconds == other.seconds && nanos == other.nanos);
}
std::string ToString() const {
return absl::StrFormat("Duration seconds: %ld, nanos %d", seconds, nanos);
}
};
// TODO(donnadionne): When we can use absl::variant<>, consider using that
// for: PathMatcher, HeaderMatcher, cluster_name and weighted_clusters
struct Route {
@ -125,11 +136,17 @@ class XdsApi {
std::string ToString() const;
};
std::vector<ClusterWeight> weighted_clusters;
// Storing the timeout duration from route action:
// RouteAction.max_stream_duration.grpc_timeout_header_max or
// RouteAction.max_stream_duration.max_stream_duration if the former is
// not set.
absl::optional<Duration> max_stream_duration;
bool operator==(const Route& other) const {
return (matchers == other.matchers &&
cluster_name == other.cluster_name &&
weighted_clusters == other.weighted_clusters);
weighted_clusters == other.weighted_clusters &&
max_stream_duration == other.max_stream_duration);
}
std::string ToString() const;
};
@ -207,13 +224,17 @@ class XdsApi {
struct LdsUpdate {
// The name to use in the RDS request.
std::string route_config_name;
// Storing the Http Connection Manager Common Http Protocol Option
// max_stream_duration
Duration http_max_stream_duration;
// The RouteConfiguration to use for this listener.
// Present only if it is inlined in the LDS response.
absl::optional<RdsUpdate> rds_update;
bool operator==(const LdsUpdate& other) const {
return route_config_name == other.route_config_name &&
rds_update == other.rds_update;
rds_update == other.rds_update &&
http_max_stream_duration == other.http_max_stream_duration;
}
};

@ -140,6 +140,14 @@ grpc_proto_library(
],
)
grpc_proto_library(
name = "protocol_proto",
srcs = [
"protocol.proto",
],
well_known_protos = True,
)
grpc_proto_library(
name = "range_proto",
srcs = [
@ -175,6 +183,7 @@ grpc_proto_library(
],
deps = [
"config_source_proto",
"protocol_proto",
"route_proto",
],
)

@ -19,13 +19,13 @@ syntax = "proto3";
package envoy.extensions.filters.network.http_connection_manager.v3;
import "src/proto/grpc/testing/xds/v3/config_source.proto";
import "src/proto/grpc/testing/xds/v3/protocol.proto";
import "src/proto/grpc/testing/xds/v3/route.proto";
// [#protodoc-title: HTTP connection manager]
// HTTP connection manager :ref:`configuration overview <config_http_conn_man>`.
// [#extension: envoy.filters.network.http_connection_manager]
// [#next-free-field: 38]
message HttpConnectionManager {
oneof route_specifier {
// The connection managers route table will be dynamically loaded via the RDS API.
@ -39,6 +39,10 @@ message HttpConnectionManager {
// specified in this message.
ScopedRoutes scoped_routes = 31;
}
// Additional settings for HTTP requests handled by the connection manager. These will be
// applicable to both HTTP1 and HTTP2 requests.
config.core.v3.HttpProtocolOptions common_http_protocol_options = 35;
}
message Rds {

@ -0,0 +1,27 @@
// Copyright 2020 The 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.
// Local copy of Envoy xDS proto file, used for testing only.
syntax = "proto3";
package envoy.config.core.v3;
import "google/protobuf/duration.proto";
// [#next-free-field: 5]
message HttpProtocolOptions {
// The maximum duration of a connection.
google.protobuf.Duration max_stream_duration = 4;
}

@ -23,6 +23,7 @@ import "src/proto/grpc/testing/xds/v3/regex.proto";
import "src/proto/grpc/testing/xds/v3/percent.proto";
import "src/proto/grpc/testing/xds/v3/range.proto";
import "google/protobuf/duration.proto";
import "google/protobuf/wrappers.proto";
// [#protodoc-title: HTTP route components]
@ -180,7 +181,25 @@ message RouteMatch {
repeated QueryParameterMatcher query_parameters = 7;
}
// [#next-free-field: 34]
message MaxStreamDuration {
// Specifies the maximum duration allowed for streams on the route. If not specified, the value
// from the :ref:`max_stream_duration
// <envoy_api_field_config.core.v3.HttpProtocolOptions.max_stream_duration>` field in
// :ref:`HttpConnectionManager.common_http_protocol_options
// <envoy_api_field_extensions.filters.network.http_connection_manager.v3.HttpConnectionManager.common_http_protocol_options>`
// is used. If this field is set explicitly to zero, any
// HttpConnectionManager max_stream_duration timeout will be disabled for
// this route.
google.protobuf.Duration max_stream_duration = 1;
// If present, and the request contains a `grpc-timeout header
// <https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md>`_, use that value as the
// *max_stream_duration*, but limit the applied timeout to the maximum value specified here.
// If set to 0, the `grpc-timeout` header is used without modification.
google.protobuf.Duration grpc_timeout_header_max = 2;
}
// [#next-free-field: 37]
message RouteAction {
oneof cluster_specifier {
// Indicates the upstream cluster to which the request should be routed
@ -205,6 +224,8 @@ message RouteAction {
// for additional documentation.
WeightedCluster weighted_clusters = 3;
}
// Specifies the maximum stream duration for this route.
MaxStreamDuration max_stream_duration = 36;
}
// .. attention::

@ -1742,11 +1742,15 @@ class XdsEnd2endTest : public ::testing::TestWithParam<TestType> {
}
}
void CheckRpcSendFailure(const size_t times = 1,
const RpcOptions& rpc_options = RpcOptions()) {
void CheckRpcSendFailure(
const size_t times = 1, const RpcOptions& rpc_options = RpcOptions(),
const StatusCode expected_error_code = StatusCode::OK) {
for (size_t i = 0; i < times; ++i) {
const Status status = SendRpc(rpc_options);
EXPECT_FALSE(status.ok());
if (expected_error_code != StatusCode::OK) {
EXPECT_EQ(expected_error_code, status.error_code());
}
}
}
@ -4115,6 +4119,337 @@ TEST_P(LdsRdsTest, XdsRoutingClusterUpdateClustersWithPickingDelays) {
EXPECT_EQ(1, backends_[1]->backend_service()->request_count());
}
TEST_P(LdsRdsTest, XdsRoutingApplyXdsTimeout) {
const int64_t kTimeoutNano = 500000000;
const int64_t kTimeoutGrpcTimeoutHeaderMaxSecond = 1;
const int64_t kTimeoutMaxStreamDurationSecond = 2;
const int64_t kTimeoutHttpMaxStreamDurationSecond = 3;
const int64_t kTimeoutApplicationSecond = 4;
const char* kNewCluster1Name = "new_cluster_1";
const char* kNewEdsService1Name = "new_eds_service_name_1";
const char* kNewCluster2Name = "new_cluster_2";
const char* kNewEdsService2Name = "new_eds_service_name_2";
const char* kNewCluster3Name = "new_cluster_3";
const char* kNewEdsService3Name = "new_eds_service_name_3";
SetNextResolution({});
SetNextResolutionForLbChannelAllBalancers();
// Populate new EDS resources.
AdsServiceImpl::EdsResourceArgs args({
{"locality0", {g_port_saver->GetPort()}},
});
AdsServiceImpl::EdsResourceArgs args1({
{"locality0", {g_port_saver->GetPort()}},
});
AdsServiceImpl::EdsResourceArgs args2({
{"locality0", {g_port_saver->GetPort()}},
});
AdsServiceImpl::EdsResourceArgs args3({
{"locality0", {g_port_saver->GetPort()}},
});
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args));
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args1, kNewEdsService1Name));
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args2, kNewEdsService2Name));
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args3, kNewEdsService3Name));
// Populate new CDS resources.
Cluster new_cluster1 = balancers_[0]->ads_service()->default_cluster();
new_cluster1.set_name(kNewCluster1Name);
new_cluster1.mutable_eds_cluster_config()->set_service_name(
kNewEdsService1Name);
balancers_[0]->ads_service()->SetCdsResource(new_cluster1);
Cluster new_cluster2 = balancers_[0]->ads_service()->default_cluster();
new_cluster2.set_name(kNewCluster2Name);
new_cluster2.mutable_eds_cluster_config()->set_service_name(
kNewEdsService2Name);
balancers_[0]->ads_service()->SetCdsResource(new_cluster2);
Cluster new_cluster3 = balancers_[0]->ads_service()->default_cluster();
new_cluster3.set_name(kNewCluster3Name);
new_cluster3.mutable_eds_cluster_config()->set_service_name(
kNewEdsService3Name);
balancers_[0]->ads_service()->SetCdsResource(new_cluster3);
HttpConnectionManager http_connection_manager;
// Set up HTTP max_stream_duration of 3.5 seconds
auto* duration =
http_connection_manager.mutable_common_http_protocol_options()
->mutable_max_stream_duration();
duration->set_seconds(kTimeoutHttpMaxStreamDurationSecond);
duration->set_nanos(kTimeoutNano);
RouteConfiguration new_route_config =
balancers_[0]->ads_service()->default_route_config();
// route 1: Set max_stream_duration of 2.5 seconds, Set
// grpc_timeout_header_max of 1.5
auto* route1 = new_route_config.mutable_virtual_hosts(0)->mutable_routes(0);
route1->mutable_match()->set_path("/grpc.testing.EchoTest1Service/Echo1");
route1->mutable_route()->set_cluster(kNewCluster1Name);
auto* max_stream_duration =
route1->mutable_route()->mutable_max_stream_duration();
duration = max_stream_duration->mutable_max_stream_duration();
duration->set_seconds(kTimeoutMaxStreamDurationSecond);
duration->set_nanos(kTimeoutNano);
duration = max_stream_duration->mutable_grpc_timeout_header_max();
duration->set_seconds(kTimeoutGrpcTimeoutHeaderMaxSecond);
duration->set_nanos(kTimeoutNano);
// route 2: Set max_stream_duration of 2.5 seconds
auto* route2 = new_route_config.mutable_virtual_hosts(0)->add_routes();
route2->mutable_match()->set_path("/grpc.testing.EchoTest2Service/Echo2");
route2->mutable_route()->set_cluster(kNewCluster2Name);
max_stream_duration = route2->mutable_route()->mutable_max_stream_duration();
duration = max_stream_duration->mutable_max_stream_duration();
duration->set_seconds(kTimeoutMaxStreamDurationSecond);
duration->set_nanos(kTimeoutNano);
// route 3: No timeout values in route configuration
auto* route3 = new_route_config.mutable_virtual_hosts(0)->add_routes();
route3->mutable_match()->set_path("/grpc.testing.EchoTestService/Echo");
route3->mutable_route()->set_cluster(kNewCluster3Name);
if (GetParam().enable_rds_testing()) {
auto* rds = http_connection_manager.mutable_rds();
rds->set_route_config_name(kDefaultRouteConfigurationName);
rds->mutable_config_source()->mutable_ads();
auto listener = balancers_[0]->ads_service()->default_listener();
listener.mutable_api_listener()->mutable_api_listener()->PackFrom(
http_connection_manager);
balancers_[0]->ads_service()->SetLdsResource(listener);
SetRouteConfiguration(0, new_route_config);
} else {
*http_connection_manager.mutable_route_config() = new_route_config;
auto listener = balancers_[0]->ads_service()->default_listener();
listener.mutable_api_listener()->mutable_api_listener()->PackFrom(
http_connection_manager);
balancers_[0]->ads_service()->SetLdsResource(listener);
}
// Test grpc_timeout_header_max of 1.5 seconds applied
auto t0 = system_clock::now();
CheckRpcSendFailure(1,
RpcOptions()
.set_rpc_service(SERVICE_ECHO1)
.set_rpc_method(METHOD_ECHO1)
.set_wait_for_ready(true)
.set_timeout_ms(kTimeoutApplicationSecond * 1000),
StatusCode::DEADLINE_EXCEEDED);
auto ellapsed_nano_seconds =
std::chrono::duration_cast<std::chrono::nanoseconds>(system_clock::now() -
t0);
EXPECT_GT(ellapsed_nano_seconds.count(),
kTimeoutGrpcTimeoutHeaderMaxSecond * 1000000000 + kTimeoutNano);
EXPECT_LT(ellapsed_nano_seconds.count(),
kTimeoutMaxStreamDurationSecond * 1000000000);
// Test max_stream_duration of 2.5 seconds applied
t0 = system_clock::now();
CheckRpcSendFailure(1,
RpcOptions()
.set_rpc_service(SERVICE_ECHO2)
.set_rpc_method(METHOD_ECHO2)
.set_wait_for_ready(true)
.set_timeout_ms(kTimeoutApplicationSecond * 1000),
StatusCode::DEADLINE_EXCEEDED);
ellapsed_nano_seconds = std::chrono::duration_cast<std::chrono::nanoseconds>(
system_clock::now() - t0);
EXPECT_GT(ellapsed_nano_seconds.count(),
kTimeoutMaxStreamDurationSecond * 1000000000 + kTimeoutNano);
EXPECT_LT(ellapsed_nano_seconds.count(),
kTimeoutHttpMaxStreamDurationSecond * 1000000000);
// Test http_stream_duration of 3.5 seconds applied
t0 = system_clock::now();
CheckRpcSendFailure(1,
RpcOptions().set_wait_for_ready(true).set_timeout_ms(
kTimeoutApplicationSecond * 1000),
StatusCode::DEADLINE_EXCEEDED);
ellapsed_nano_seconds = std::chrono::duration_cast<std::chrono::nanoseconds>(
system_clock::now() - t0);
EXPECT_GT(ellapsed_nano_seconds.count(),
kTimeoutHttpMaxStreamDurationSecond * 1000000000 + kTimeoutNano);
EXPECT_LT(ellapsed_nano_seconds.count(),
kTimeoutApplicationSecond * 1000000000);
}
TEST_P(LdsRdsTest, XdsRoutingApplyApplicationTimeoutWhenXdsTimeoutExplicit0) {
const int64_t kTimeoutNano = 500000000;
const int64_t kTimeoutMaxStreamDurationSecond = 2;
const int64_t kTimeoutHttpMaxStreamDurationSecond = 3;
const int64_t kTimeoutApplicationSecond = 4;
const char* kNewCluster1Name = "new_cluster_1";
const char* kNewEdsService1Name = "new_eds_service_name_1";
const char* kNewCluster2Name = "new_cluster_2";
const char* kNewEdsService2Name = "new_eds_service_name_2";
SetNextResolution({});
SetNextResolutionForLbChannelAllBalancers();
// Populate new EDS resources.
AdsServiceImpl::EdsResourceArgs args({
{"locality0", {g_port_saver->GetPort()}},
});
AdsServiceImpl::EdsResourceArgs args1({
{"locality0", {g_port_saver->GetPort()}},
});
AdsServiceImpl::EdsResourceArgs args2({
{"locality0", {g_port_saver->GetPort()}},
});
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args));
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args1, kNewEdsService1Name));
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args2, kNewEdsService2Name));
// Populate new CDS resources.
Cluster new_cluster1 = balancers_[0]->ads_service()->default_cluster();
new_cluster1.set_name(kNewCluster1Name);
new_cluster1.mutable_eds_cluster_config()->set_service_name(
kNewEdsService1Name);
balancers_[0]->ads_service()->SetCdsResource(new_cluster1);
Cluster new_cluster2 = balancers_[0]->ads_service()->default_cluster();
new_cluster2.set_name(kNewCluster2Name);
new_cluster2.mutable_eds_cluster_config()->set_service_name(
kNewEdsService2Name);
balancers_[0]->ads_service()->SetCdsResource(new_cluster2);
HttpConnectionManager http_connection_manager;
// Set up HTTP max_stream_duration of 3.5 seconds
auto* duration =
http_connection_manager.mutable_common_http_protocol_options()
->mutable_max_stream_duration();
duration->set_seconds(kTimeoutHttpMaxStreamDurationSecond);
duration->set_nanos(kTimeoutNano);
RouteConfiguration new_route_config =
balancers_[0]->ads_service()->default_route_config();
// route 1: Set max_stream_duration of 2.5 seconds, Set
// grpc_timeout_header_max of 0
auto* route1 = new_route_config.mutable_virtual_hosts(0)->mutable_routes(0);
route1->mutable_match()->set_path("/grpc.testing.EchoTest1Service/Echo1");
route1->mutable_route()->set_cluster(kNewCluster1Name);
auto* max_stream_duration =
route1->mutable_route()->mutable_max_stream_duration();
duration = max_stream_duration->mutable_max_stream_duration();
duration->set_seconds(kTimeoutMaxStreamDurationSecond);
duration->set_nanos(kTimeoutNano);
duration = max_stream_duration->mutable_grpc_timeout_header_max();
duration->set_seconds(0);
duration->set_nanos(0);
// route 2: Set max_stream_duration to 0
auto* route2 = new_route_config.mutable_virtual_hosts(0)->add_routes();
route2->mutable_match()->set_path("/grpc.testing.EchoTest2Service/Echo2");
route2->mutable_route()->set_cluster(kNewCluster2Name);
max_stream_duration = route2->mutable_route()->mutable_max_stream_duration();
duration = max_stream_duration->mutable_max_stream_duration();
duration->set_seconds(0);
duration->set_nanos(0);
if (GetParam().enable_rds_testing()) {
auto* rds = http_connection_manager.mutable_rds();
rds->set_route_config_name(kDefaultRouteConfigurationName);
rds->mutable_config_source()->mutable_ads();
auto listener = balancers_[0]->ads_service()->default_listener();
listener.mutable_api_listener()->mutable_api_listener()->PackFrom(
http_connection_manager);
balancers_[0]->ads_service()->SetLdsResource(listener);
SetRouteConfiguration(0, new_route_config);
} else {
*http_connection_manager.mutable_route_config() = new_route_config;
auto listener = balancers_[0]->ads_service()->default_listener();
listener.mutable_api_listener()->mutable_api_listener()->PackFrom(
http_connection_manager);
balancers_[0]->ads_service()->SetLdsResource(listener);
}
// Test application timeout is applied for route 1
auto t0 = system_clock::now();
CheckRpcSendFailure(1,
RpcOptions()
.set_rpc_service(SERVICE_ECHO1)
.set_rpc_method(METHOD_ECHO1)
.set_wait_for_ready(true)
.set_timeout_ms(kTimeoutApplicationSecond * 1000),
StatusCode::DEADLINE_EXCEEDED);
auto ellapsed_nano_seconds =
std::chrono::duration_cast<std::chrono::nanoseconds>(system_clock::now() -
t0);
EXPECT_GT(ellapsed_nano_seconds.count(),
kTimeoutApplicationSecond * 1000000000);
// Test application timeout is applied for route 2
t0 = system_clock::now();
CheckRpcSendFailure(1,
RpcOptions()
.set_rpc_service(SERVICE_ECHO2)
.set_rpc_method(METHOD_ECHO2)
.set_wait_for_ready(true)
.set_timeout_ms(kTimeoutApplicationSecond * 1000),
StatusCode::DEADLINE_EXCEEDED);
ellapsed_nano_seconds = std::chrono::duration_cast<std::chrono::nanoseconds>(
system_clock::now() - t0);
EXPECT_GT(ellapsed_nano_seconds.count(),
kTimeoutApplicationSecond * 1000000000);
}
TEST_P(LdsRdsTest, XdsRoutingApplyApplicationTimeoutWhenHttpTimeoutExplicit0) {
const int64_t kTimeoutApplicationSecond = 4;
SetNextResolution({});
SetNextResolutionForLbChannelAllBalancers();
// Populate new EDS resources.
AdsServiceImpl::EdsResourceArgs args({
{"locality0", {g_port_saver->GetPort()}},
});
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args));
HttpConnectionManager http_connection_manager;
// Set up HTTP max_stream_duration to be explicit 0
auto* duration =
http_connection_manager.mutable_common_http_protocol_options()
->mutable_max_stream_duration();
duration->set_seconds(0);
duration->set_nanos(0);
RouteConfiguration new_route_config =
balancers_[0]->ads_service()->default_route_config();
if (GetParam().enable_rds_testing()) {
auto* rds = http_connection_manager.mutable_rds();
rds->set_route_config_name(kDefaultRouteConfigurationName);
rds->mutable_config_source()->mutable_ads();
auto listener = balancers_[0]->ads_service()->default_listener();
listener.mutable_api_listener()->mutable_api_listener()->PackFrom(
http_connection_manager);
balancers_[0]->ads_service()->SetLdsResource(listener);
SetRouteConfiguration(0, new_route_config);
} else {
*http_connection_manager.mutable_route_config() = new_route_config;
auto listener = balancers_[0]->ads_service()->default_listener();
listener.mutable_api_listener()->mutable_api_listener()->PackFrom(
http_connection_manager);
balancers_[0]->ads_service()->SetLdsResource(listener);
}
// Test application timeout is applied for route 1
auto t0 = system_clock::now();
CheckRpcSendFailure(1,
RpcOptions().set_wait_for_ready(true).set_timeout_ms(
kTimeoutApplicationSecond * 1000),
StatusCode::DEADLINE_EXCEEDED);
auto ellapsed_nano_seconds =
std::chrono::duration_cast<std::chrono::nanoseconds>(system_clock::now() -
t0);
EXPECT_GT(ellapsed_nano_seconds.count(),
kTimeoutApplicationSecond * 1000000000);
}
// Test to ensure application-specified deadline won't be affected when
// the xDS config does not specify a timeout.
TEST_P(LdsRdsTest, XdsRoutingWithOnlyApplicationTimeout) {
const int64_t kTimeoutApplicationSecond = 4;
SetNextResolution({});
SetNextResolutionForLbChannelAllBalancers();
// Populate new EDS resources.
AdsServiceImpl::EdsResourceArgs args({
{"locality0", {g_port_saver->GetPort()}},
});
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args));
auto t0 = system_clock::now();
CheckRpcSendFailure(1,
RpcOptions().set_wait_for_ready(true).set_timeout_ms(
kTimeoutApplicationSecond * 1000),
StatusCode::DEADLINE_EXCEEDED);
auto ellapsed_nano_seconds =
std::chrono::duration_cast<std::chrono::nanoseconds>(system_clock::now() -
t0);
EXPECT_GT(ellapsed_nano_seconds.count(),
kTimeoutApplicationSecond * 1000000000);
}
TEST_P(LdsRdsTest, XdsRoutingHeadersMatching) {
const char* kNewClusterName = "new_cluster";
const char* kNewEdsServiceName = "new_eds_service_name";

Loading…
Cancel
Save