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

Merge pull request #22648 from markdroth/xds_remove_old_flow

Browse Source 
xds: Remove fallback code and support for old xds LB configs.
reviewable/pr22280/r16^2
Mark D. Roth 5 years ago committed by GitHub
parent f198dc79b2 dc976d2a77
commit cbf18222af
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 71 additions and 586 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      include/grpc/impl/codegen/grpc_types.h
  2. 382
      src/core/ext/filters/client_channel/lb_policy/xds/eds.cc
  3. 271
      test/cpp/end2end/xds_end2end_test.cc

4
include/grpc/impl/codegen/grpc_types.h
Unescape View File

@ -344,10 +344,6 @@ typedef struct {
balancer before using fallback backend addresses from the resolver.
If 0, enter fallback mode immediately. Default value is 10000. */
#define GRPC_ARG_GRPCLB_FALLBACK_TIMEOUT_MS "grpc.grpclb_fallback_timeout_ms"
/* Timeout in milliseconds to wait for the serverlist from the xDS load
balancer before using fallback backend addresses from the resolver.
If 0, enter fallback mode immediately. Default value is 10000. */
#define GRPC_ARG_XDS_FALLBACK_TIMEOUT_MS "grpc.xds_fallback_timeout_ms"
/* Timeout in milliseconds to wait for the child of a specific priority to
complete its initial connection attempt before the priority LB policy fails
over to the next priority. Default value is 10 seconds. */

382
src/core/ext/filters/client_channel/lb_policy/xds/eds.cc
Unescape View File

@ -49,27 +49,22 @@ TraceFlag grpc_lb_eds_trace(false, "eds_lb");
namespace {
constexpr char kXds[] = "xds_experimental";
constexpr char kEds[] = "eds_experimental";
// Config for EDS LB policy.
class EdsLbConfig : public LoadBalancingPolicy::Config {
public:
EdsLbConfig(const char* name, std::string cluster_name,
std::string eds_service_name,
EdsLbConfig(std::string cluster_name, std::string eds_service_name,
absl::optional<std::string> lrs_load_reporting_server_name,
Json locality_picking_policy, Json endpoint_picking_policy,
RefCountedPtr<LoadBalancingPolicy::Config> fallback_policy)
: name_(name),
cluster_name_(std::move(cluster_name)),
Json locality_picking_policy, Json endpoint_picking_policy)
: cluster_name_(std::move(cluster_name)),
eds_service_name_(std::move(eds_service_name)),
lrs_load_reporting_server_name_(
std::move(lrs_load_reporting_server_name)),
locality_picking_policy_(std::move(locality_picking_policy)),
endpoint_picking_policy_(std::move(endpoint_picking_policy)),
fallback_policy_(std::move(fallback_policy)) {}
endpoint_picking_policy_(std::move(endpoint_picking_policy)) {}
const char* name() const override { return name_; }
const char* name() const override { return kEds; }
const std::string& cluster_name() const { return cluster_name_; }
const std::string& eds_service_name() const { return eds_service_name_; }
@ -82,26 +77,21 @@ class EdsLbConfig : public LoadBalancingPolicy::Config {
const Json& endpoint_picking_policy() const {
return endpoint_picking_policy_;
}
RefCountedPtr<LoadBalancingPolicy::Config> fallback_policy() const {
return fallback_policy_;
}
private:
const char* name_;
std::string cluster_name_;
std::string eds_service_name_;
absl::optional<std::string> lrs_load_reporting_server_name_;
Json locality_picking_policy_;
Json endpoint_picking_policy_;
RefCountedPtr<LoadBalancingPolicy::Config> fallback_policy_;
};
// EDS LB policy.
class EdsLb : public LoadBalancingPolicy {
public:
EdsLb(const char* name, Args args);
explicit EdsLb(Args args);
const char* name() const override { return name_; }
const char* name() const override { return kEds; }
void UpdateLocked(UpdateArgs args) override;
void ResetBackoffLocked() override;
@ -153,24 +143,6 @@ class EdsLb : public LoadBalancingPolicy {
RefCountedPtr<EdsLb> eds_policy_;
};
class FallbackHelper : public ChannelControlHelper {
public:
explicit FallbackHelper(RefCountedPtr<EdsLb> parent)
: parent_(std::move(parent)) {}
~FallbackHelper() { parent_.reset(DEBUG_LOCATION, "FallbackHelper"); }
RefCountedPtr<SubchannelInterface> CreateSubchannel(
const grpc_channel_args& args) override;
void UpdateState(grpc_connectivity_state state,
std::unique_ptr<SubchannelPicker> picker) override;
void RequestReresolution() override;
void AddTraceEvent(TraceSeverity severity, StringView message) override;
private:
RefCountedPtr<EdsLb> parent_;
};
~EdsLb();
void ShutdownLocked() override;
@ -185,15 +157,6 @@ class EdsLb : public LoadBalancingPolicy {
const grpc_channel_args* args_in);
void MaybeUpdateDropPickerLocked();
// Methods for dealing with fallback state.
void MaybeCancelFallbackAtStartupChecks();
static void OnFallbackTimer(void* arg, grpc_error* error);
static void OnFallbackTimerLocked(void* arg, grpc_error* error);
void UpdateFallbackPolicyLocked();
OrphanablePtr<LoadBalancingPolicy> CreateFallbackPolicyLocked(
const grpc_channel_args* args);
void MaybeExitFallbackMode();
// Caller must ensure that config_ is set before calling.
const StringView GetEdsResourceName() const {
if (xds_client_from_channel_ == nullptr) return server_name_;
@ -216,9 +179,6 @@ class EdsLb : public LoadBalancingPolicy {
: xds_client_.get();
}
// Policy name (kXds or kEds).
const char* name_;
// Server name from target URI.
std::string server_name_;
@ -251,26 +211,6 @@ class EdsLb : public LoadBalancingPolicy {
// The latest state and picker returned from the child policy.
grpc_connectivity_state child_state_;
RefCountedPtr<ChildPickerWrapper> child_picker_;
// Non-null iff we are in fallback mode.
OrphanablePtr<LoadBalancingPolicy> fallback_policy_;
// Whether the checks for fallback at startup are ALL pending. There are
// several cases where this can be reset:
// 1. The fallback timer fires, we enter fallback mode.
// 2. Before the fallback timer fires, the endpoint watcher reports an
// error, we enter fallback mode.
// 3. Before the fallback timer fires, if any child policy in the locality map
// becomes READY, we cancel the fallback timer.
bool fallback_at_startup_checks_pending_ = false;
// Timeout in milliseconds for before using fallback backend addresses.
// 0 means not using fallback.
const grpc_millis lb_fallback_timeout_ms_;
// The backend addresses from the resolver.
ServerAddressList fallback_backend_addresses_;
// Fallback timer.
grpc_timer lb_fallback_timer_;
grpc_closure lb_on_fallback_;
};
//
@ -331,15 +271,6 @@ void EdsLb::Helper::UpdateState(grpc_connectivity_state state,
eds_policy_->child_state_ = state;
eds_policy_->child_picker_ =
MakeRefCounted<ChildPickerWrapper>(std::move(picker));
// If the new state is READY, cancel the fallback-at-startup checks.
if (state == GRPC_CHANNEL_READY) {
eds_policy_->MaybeCancelFallbackAtStartupChecks();
eds_policy_->MaybeExitFallbackMode();
}
// TODO(roth): If the child reports TRANSIENT_FAILURE and the
// fallback-at-startup checks are pending, we should probably go into
// fallback mode immediately (cancelling the fallback-at-startup timer
// if needed).
// Wrap the picker in a DropPicker and pass it up.
eds_policy_->MaybeUpdateDropPickerLocked();
}
@ -349,33 +280,6 @@ void EdsLb::Helper::AddTraceEvent(TraceSeverity severity, StringView message) {
eds_policy_->channel_control_helper()->AddTraceEvent(severity, message);
}
//
// EdsLb::FallbackHelper
//
RefCountedPtr<SubchannelInterface> EdsLb::FallbackHelper::CreateSubchannel(
const grpc_channel_args& args) {
if (parent_->shutting_down_) return nullptr;
return parent_->channel_control_helper()->CreateSubchannel(args);
}
void EdsLb::FallbackHelper::UpdateState(
grpc_connectivity_state state, std::unique_ptr<SubchannelPicker> picker) {
if (parent_->shutting_down_) return;
parent_->channel_control_helper()->UpdateState(state, std::move(picker));
}
void EdsLb::FallbackHelper::RequestReresolution() {
if (parent_->shutting_down_) return;
parent_->channel_control_helper()->RequestReresolution();
}
void EdsLb::FallbackHelper::AddTraceEvent(TraceSeverity severity,
StringView message) {
if (parent_->shutting_down_) return;
parent_->channel_control_helper()->AddTraceEvent(severity, message);
}
//
// EdsLb::EndpointWatcher
//
@ -392,9 +296,6 @@ class EdsLb::EndpointWatcher : public XdsClient::EndpointWatcherInterface {
gpr_log(GPR_INFO, "[edslb %p] Received EDS update from xds client",
eds_policy_.get());
}
// If the balancer tells us to drop all the calls, we should exit fallback
// mode immediately.
if (update.drop_config->drop_all()) eds_policy_->MaybeExitFallbackMode();
// Update the drop config.
const bool drop_config_changed =
eds_policy_->drop_config_ == nullptr ||
@ -424,34 +325,18 @@ class EdsLb::EndpointWatcher : public XdsClient::EndpointWatcherInterface {
}
void OnError(grpc_error* error) override {
// If the fallback-at-startup checks are pending, go into fallback mode
// immediately. This short-circuits the timeout for the
// fallback-at-startup case.
if (eds_policy_->fallback_at_startup_checks_pending_) {
gpr_log(GPR_ERROR,
"[edslb %p] xds watcher reported error; entering fallback "
"mode: %s",
eds_policy_.get(), grpc_error_string(error));
eds_policy_->fallback_at_startup_checks_pending_ = false;
grpc_timer_cancel(&eds_policy_->lb_fallback_timer_);
eds_policy_->UpdateFallbackPolicyLocked();
// If the xds call failed, request re-resolution.
// TODO(roth): We check the error string contents here to
// differentiate between the xds call failing and the xds channel
// going into TRANSIENT_FAILURE. This is a pretty ugly hack,
// but it's okay for now, since we're not yet sure whether we will
// continue to support the current fallback functionality. If we
// decide to keep the fallback approach, then we should either
// find a cleaner way to expose the difference between these two
// cases or decide that we're okay re-resolving in both cases.
// Note that even if we do keep the current fallback functionality,
// this re-resolution will only be necessary if we are going to be
// using this LB policy with resolvers other than the xds resolver.
if (strstr(grpc_error_string(error), "xds call failed")) {
eds_policy_->channel_control_helper()->RequestReresolution();
}
gpr_log(GPR_ERROR, "[edslb %p] xds watcher reported error: %s",
eds_policy_.get(), grpc_error_string(error));
// Go into TRANSIENT_FAILURE if we have not yet created the child
// policy (i.e., we have not yet received data from xds). Otherwise,
// we keep running with the data we had previously.
if (eds_policy_->child_policy_ == nullptr) {
eds_policy_->channel_control_helper()->UpdateState(
GRPC_CHANNEL_TRANSIENT_FAILURE,
absl::make_unique<TransientFailurePicker>(error));
} else {
GRPC_ERROR_UNREF(error);
}
GRPC_ERROR_UNREF(error);
}
private:
@ -462,13 +347,9 @@ class EdsLb::EndpointWatcher : public XdsClient::EndpointWatcherInterface {
// EdsLb public methods
//
EdsLb::EdsLb(const char* name, Args args)
EdsLb::EdsLb(Args args)
: LoadBalancingPolicy(std::move(args)),
name_(name),
xds_client_from_channel_(XdsClient::GetFromChannelArgs(*args.args)),
lb_fallback_timeout_ms_(grpc_channel_args_find_integer(
args.args, GRPC_ARG_XDS_FALLBACK_TIMEOUT_MS,
{GRPC_EDS_DEFAULT_FALLBACK_TIMEOUT, 0, INT_MAX})) {
xds_client_from_channel_(XdsClient::GetFromChannelArgs(*args.args)) {
if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_eds_trace)) {
gpr_log(GPR_INFO, "[edslb %p] created -- xds client from channel: %p", this,
xds_client_from_channel_.get());
@ -499,7 +380,6 @@ void EdsLb::ShutdownLocked() {
gpr_log(GPR_INFO, "[edslb %p] shutting down", this);
}
shutting_down_ = true;
MaybeCancelFallbackAtStartupChecks();
// Drop our ref to the child's picker, in case it's holding a ref to
// the child.
child_picker_.reset();
@ -508,11 +388,6 @@ void EdsLb::ShutdownLocked() {
interested_parties());
child_policy_.reset();
}
if (fallback_policy_ != nullptr) {
grpc_pollset_set_del_pollset_set(fallback_policy_->interested_parties(),
interested_parties());
fallback_policy_.reset();
}
drop_stats_.reset();
// Cancel the endpoint watch here instead of in our dtor if we are using the
// xds resolver, because the watcher holds a ref to us and we might not be
@ -540,15 +415,10 @@ void EdsLb::UpdateLocked(UpdateArgs args) {
// Update config.
auto old_config = std::move(config_);
config_ = std::move(args.config);
// Update fallback address list.
fallback_backend_addresses_ = std::move(args.addresses);
// Update args.
grpc_channel_args_destroy(args_);
args_ = args.args;
args.args = nullptr;
// Update the existing fallback policy. The fallback policy config and/or the
// fallback addresses may be new.
if (fallback_policy_ != nullptr) UpdateFallbackPolicyLocked();
if (is_initial_update) {
// Initialize XdsClient.
if (xds_client_from_channel_ == nullptr) {
@ -556,7 +426,7 @@ void EdsLb::UpdateLocked(UpdateArgs args) {
xds_client_ = MakeOrphanable<XdsClient>(
combiner(), interested_parties(), GetEdsResourceName(),
nullptr /* service config watcher */, *args_, &error);
// TODO(roth): If we decide that we care about fallback mode, add
// TODO(roth): If we decide that we care about EDS-only mode, add
// proper error handling here.
GPR_ASSERT(error == GRPC_ERROR_NONE);
if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_eds_trace)) {
@ -564,13 +434,6 @@ void EdsLb::UpdateLocked(UpdateArgs args) {
xds_client_.get());
}
}
// Start fallback-at-startup checks.
grpc_millis deadline = ExecCtx::Get()->Now() + lb_fallback_timeout_ms_;
Ref(DEBUG_LOCATION, "on_fallback_timer").release(); // Held by closure
GRPC_CLOSURE_INIT(&lb_on_fallback_, &EdsLb::OnFallbackTimer, this,
grpc_schedule_on_exec_ctx);
fallback_at_startup_checks_pending_ = true;
grpc_timer_init(&lb_fallback_timer_, deadline, &lb_on_fallback_);
}
// Update drop stats for load reporting if needed.
if (is_initial_update || config_->lrs_load_reporting_server_name() !=
@ -609,9 +472,6 @@ void EdsLb::ResetBackoffLocked() {
if (child_policy_ != nullptr) {
child_policy_->ResetBackoffLocked();
}
if (fallback_policy_ != nullptr) {
fallback_policy_->ResetBackoffLocked();
}
}
//
@ -875,8 +735,6 @@ OrphanablePtr<LoadBalancingPolicy> EdsLb::CreateChildPolicyLocked(
}
void EdsLb::MaybeUpdateDropPickerLocked() {
// If we are in fallback mode, don't override the picker.
if (fallback_policy_ != nullptr) return;
// If we're dropping all calls, report READY, regardless of what (or
// whether) the child has reported.
if (drop_config_ != nullptr && drop_config_->drop_all()) {
@ -891,111 +749,24 @@ void EdsLb::MaybeUpdateDropPickerLocked() {
}
}
//
// fallback-related methods
//
void EdsLb::MaybeCancelFallbackAtStartupChecks() {
if (!fallback_at_startup_checks_pending_) return;
if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_eds_trace)) {
gpr_log(GPR_INFO, "[edslb %p] Cancelling fallback timer", this);
}
grpc_timer_cancel(&lb_fallback_timer_);
fallback_at_startup_checks_pending_ = false;
}
void EdsLb::OnFallbackTimer(void* arg, grpc_error* error) {
EdsLb* edslb_policy = static_cast<EdsLb*>(arg);
edslb_policy->combiner()->Run(
GRPC_CLOSURE_INIT(&edslb_policy->lb_on_fallback_,
&EdsLb::OnFallbackTimerLocked, edslb_policy, nullptr),
GRPC_ERROR_REF(error));
}
void EdsLb::OnFallbackTimerLocked(void* arg, grpc_error* error) {
EdsLb* edslb_policy = static_cast<EdsLb*>(arg);
// If some fallback-at-startup check is done after the timer fires but before
// this callback actually runs, don't fall back.
if (edslb_policy->fallback_at_startup_checks_pending_ &&
!edslb_policy->shutting_down_ && error == GRPC_ERROR_NONE) {
gpr_log(GPR_INFO,
"[edslb %p] Child policy not ready after fallback timeout; "
"entering fallback mode",
edslb_policy);
edslb_policy->fallback_at_startup_checks_pending_ = false;
edslb_policy->UpdateFallbackPolicyLocked();
}
edslb_policy->Unref(DEBUG_LOCATION, "on_fallback_timer");
}
void EdsLb::UpdateFallbackPolicyLocked() {
if (shutting_down_) return;
// Create policy if needed.
if (fallback_policy_ == nullptr) {
fallback_policy_ = CreateFallbackPolicyLocked(args_);
}
// Construct update args.
UpdateArgs update_args;
update_args.addresses = fallback_backend_addresses_;
update_args.config = config_->fallback_policy();
update_args.args = grpc_channel_args_copy(args_);
// Update the policy.
if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_eds_trace)) {
gpr_log(GPR_INFO, "[edslb %p] Updating fallback child policy handler %p",
this, fallback_policy_.get());
}
fallback_policy_->UpdateLocked(std::move(update_args));
}
OrphanablePtr<LoadBalancingPolicy> EdsLb::CreateFallbackPolicyLocked(
const grpc_channel_args* args) {
LoadBalancingPolicy::Args lb_policy_args;
lb_policy_args.combiner = combiner();
lb_policy_args.args = args;
lb_policy_args.channel_control_helper =
absl::make_unique<FallbackHelper>(Ref(DEBUG_LOCATION, "FallbackHelper"));
OrphanablePtr<LoadBalancingPolicy> lb_policy =
MakeOrphanable<ChildPolicyHandler>(std::move(lb_policy_args),
&grpc_lb_eds_trace);
if (GRPC_TRACE_FLAG_ENABLED(grpc_lb_eds_trace)) {
gpr_log(GPR_INFO, "[edslb %p] Created new fallback child policy handler %p",
this, lb_policy.get());
}
// Add our interested_parties pollset_set to that of the newly created
// child policy. This will make the child policy progress upon activity on
// this policy, which in turn is tied to the application's call.
grpc_pollset_set_add_pollset_set(lb_policy->interested_parties(),
interested_parties());
return lb_policy;
}
void EdsLb::MaybeExitFallbackMode() {
if (fallback_policy_ == nullptr) return;
gpr_log(GPR_INFO, "[edslb %p] Exiting fallback mode", this);
fallback_policy_.reset();
}
//
// factory
//
class EdsLbFactory : public LoadBalancingPolicyFactory {
public:
explicit EdsLbFactory(const char* name) : name_(name) {}
OrphanablePtr<LoadBalancingPolicy> CreateLoadBalancingPolicy(
LoadBalancingPolicy::Args args) const override {
return MakeOrphanable<EdsChildHandler>(std::move(args), &grpc_lb_eds_trace,
name_);
return MakeOrphanable<EdsChildHandler>(std::move(args), &grpc_lb_eds_trace);
}
const char* name() const override { return name_; }
const char* name() const override { return kEds; }
RefCountedPtr<LoadBalancingPolicy::Config> ParseLoadBalancingConfig(
const Json& json, grpc_error** error) const override {
GPR_DEBUG_ASSERT(error != nullptr && *error == GRPC_ERROR_NONE);
if (json.type() == Json::Type::JSON_NULL) {
// xds was mentioned as a policy in the deprecated loadBalancingPolicy
// eds was mentioned as a policy in the deprecated loadBalancingPolicy
// field or in the client API.
*error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"field:loadBalancingPolicy error:eds policy requires configuration. "
@ -1016,21 +787,15 @@ class EdsLbFactory : public LoadBalancingPolicyFactory {
}
// Cluster name.
std::string cluster_name;
if (name_ == kEds) {
it = json.object_value().find("clusterName");
if (it == json.object_value().end()) {
error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"field:clusterName error:required field missing"));
} else if (it->second.type() != Json::Type::STRING) {
error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"field:clusterName error:type should be string"));
} else {
cluster_name = it->second.string_value();
}
it = json.object_value().find("clusterName");
if (it == json.object_value().end()) {
error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"field:clusterName error:required field missing"));
} else if (it->second.type() != Json::Type::STRING) {
error_list.push_back(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"field:clusterName error:type should be string"));
} else {
// For xds policy, this field does not exist in the config, so it
// will always be set to the same value as edsServiceName.
cluster_name = eds_service_name;
cluster_name = it->second.string_value();
}
// LRS load reporting server name.
absl::optional<std::string> lrs_load_reporting_server_name;
@ -1043,20 +808,20 @@ class EdsLbFactory : public LoadBalancingPolicyFactory {
lrs_load_reporting_server_name.emplace(it->second.string_value());
}
}
// Locality-picking policy. Not supported for xds policy.
Json locality_picking_policy = Json::Array{
Json::Object{
{"weighted_target_experimental",
Json::Object{
{"targets", Json::Object()},
}},
},
};
if (name_ == kEds) {
it = json.object_value().find("localityPickingPolicy");
if (it != json.object_value().end()) {
locality_picking_policy = it->second;
}
// Locality-picking policy.
Json locality_picking_policy;
it = json.object_value().find("localityPickingPolicy");
if (it == json.object_value().end()) {
locality_picking_policy = Json::Array{
Json::Object{
{"weighted_target_experimental",
Json::Object{
{"targets", Json::Object()},
}},
},
};
} else {
locality_picking_policy = it->second;
}
grpc_error* parse_error = GRPC_ERROR_NONE;
if (LoadBalancingPolicyRegistry::ParseLoadBalancingConfig(
@ -1067,10 +832,8 @@ class EdsLbFactory : public LoadBalancingPolicyFactory {
GRPC_ERROR_UNREF(parse_error);
}
// Endpoint-picking policy. Called "childPolicy" for xds policy.
const char* field_name =
name_ == kEds ? "endpointPickingPolicy" : "childPolicy";
Json endpoint_picking_policy;
it = json.object_value().find(field_name);
it = json.object_value().find("endpointPickingPolicy");
if (it == json.object_value().end()) {
endpoint_picking_policy = Json::Array{
Json::Object{
@ -1085,36 +848,16 @@ class EdsLbFactory : public LoadBalancingPolicyFactory {
endpoint_picking_policy, &parse_error) == nullptr) {
GPR_DEBUG_ASSERT(parse_error != GRPC_ERROR_NONE);
error_list.push_back(GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
field_name, &parse_error, 1));
GRPC_ERROR_UNREF(parse_error);
}
// Fallback policy.
Json fallback_policy_config;
it = json.object_value().find("fallbackPolicy");
if (it == json.object_value().end()) {
fallback_policy_config = Json::Array{Json::Object{
{"round_robin", Json::Object()},
}};
} else {
fallback_policy_config = it->second;
}
parse_error = GRPC_ERROR_NONE;
RefCountedPtr<LoadBalancingPolicy::Config> fallback_policy =
LoadBalancingPolicyRegistry::ParseLoadBalancingConfig(
fallback_policy_config, &parse_error);
if (fallback_policy == nullptr) {
GPR_DEBUG_ASSERT(parse_error != GRPC_ERROR_NONE);
error_list.push_back(GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING(
"fallbackPolicy", &parse_error, 1));
"endpointPickingPolicy", &parse_error, 1));
GRPC_ERROR_UNREF(parse_error);
error_list.push_back(parse_error);
}
// Construct config.
if (error_list.empty()) {
return MakeRefCounted<EdsLbConfig>(
name_, std::move(cluster_name), std::move(eds_service_name),
std::move(cluster_name), std::move(eds_service_name),
std::move(lrs_load_reporting_server_name),
std::move(locality_picking_policy),
std::move(endpoint_picking_policy), std::move(fallback_policy));
std::move(endpoint_picking_policy));
} else {
*error = GRPC_ERROR_CREATE_FROM_VECTOR(
"eds_experimental LB policy config", &error_list);
@ -1125,14 +868,14 @@ class EdsLbFactory : public LoadBalancingPolicyFactory {
private:
class EdsChildHandler : public ChildPolicyHandler {
public:
EdsChildHandler(Args args, TraceFlag* tracer, const char* name)
: ChildPolicyHandler(std::move(args), tracer), name_(name) {}
EdsChildHandler(Args args, TraceFlag* tracer)
: ChildPolicyHandler(std::move(args), tracer) {}
bool ConfigChangeRequiresNewPolicyInstance(
LoadBalancingPolicy::Config* old_config,
LoadBalancingPolicy::Config* new_config) const override {
GPR_ASSERT(old_config->name() == name_);
GPR_ASSERT(new_config->name() == name_);
GPR_ASSERT(old_config->name() == kEds);
GPR_ASSERT(new_config->name() == kEds);
EdsLbConfig* old_eds_config = static_cast<EdsLbConfig*>(old_config);
EdsLbConfig* new_eds_config = static_cast<EdsLbConfig*>(new_config);
return old_eds_config->cluster_name() != new_eds_config->cluster_name() ||
@ -1142,14 +885,9 @@ class EdsLbFactory : public LoadBalancingPolicyFactory {
OrphanablePtr<LoadBalancingPolicy> CreateLoadBalancingPolicy(
const char* name, LoadBalancingPolicy::Args args) const override {
return MakeOrphanable<EdsLb>(name_, std::move(args));
return MakeOrphanable<EdsLb>(std::move(args));
}
private:
const char* name_;
};
const char* name_;
};
} // namespace
@ -1163,13 +901,7 @@ class EdsLbFactory : public LoadBalancingPolicyFactory {
void grpc_lb_policy_eds_init() {
grpc_core::LoadBalancingPolicyRegistry::Builder::
RegisterLoadBalancingPolicyFactory(
absl::make_unique<grpc_core::EdsLbFactory>(grpc_core::kEds));
// TODO(roth): This is here just for backward compatibility with some
// old tests we have internally. Remove this once they are upgraded
// to use the new policy name and config.
grpc_core::LoadBalancingPolicyRegistry::Builder::
RegisterLoadBalancingPolicyFactory(
absl::make_unique<grpc_core::EdsLbFactory>(grpc_core::kXds));
absl::make_unique<grpc_core::EdsLbFactory>());
}
void grpc_lb_policy_eds_shutdown() {}

271
test/cpp/end2end/xds_end2end_test.cc
Unescape View File

@ -1144,14 +1144,10 @@ class XdsEnd2endTest : public ::testing::TestWithParam<TestType> {
void ShutdownBackend(size_t index) { backends_[index]->Shutdown(); }
void ResetStub(int fallback_timeout = 0, int failover_timeout = 0,
void ResetStub(int failover_timeout = 0,
const grpc::string& expected_targets = "",
int xds_resource_does_not_exist_timeout = 0) {
ChannelArguments args;
// TODO(juanlishen): Add setter to ChannelArguments.
if (fallback_timeout > 0) {
args.SetInt(GRPC_ARG_XDS_FALLBACK_TIMEOUT_MS, fallback_timeout);
}
if (failover_timeout > 0) {
args.SetInt(GRPC_ARG_PRIORITY_FAILOVER_TIMEOUT_MS, failover_timeout);
}
@ -1285,8 +1281,8 @@ class XdsEnd2endTest : public ::testing::TestWithParam<TestType> {
: kDefaultServiceConfigWithoutLoadReporting_;
result.service_config =
grpc_core::ServiceConfig::Create(service_config_json, &error);
ASSERT_NE(result.service_config.get(), nullptr);
ASSERT_EQ(error, GRPC_ERROR_NONE) << grpc_error_string(error);
ASSERT_NE(result.service_config.get(), nullptr);
grpc_arg arg = grpc_core::FakeResolverResponseGenerator::MakeChannelArg(
lb_channel_response_generator == nullptr
? lb_channel_response_generator_.get()
@ -1519,7 +1515,8 @@ class XdsEnd2endTest : public ::testing::TestWithParam<TestType> {
"{\n"
" \"loadBalancingConfig\":[\n"
" { \"does_not_exist\":{} },\n"
" { \"xds_experimental\":{\n"
" { \"eds_experimental\":{\n"
" \"clusterName\": \"application_target_name\",\n"
" \"lrsLoadReportingServerName\": \"\"\n"
" } }\n"
" ]\n"
@ -1528,7 +1525,8 @@ class XdsEnd2endTest : public ::testing::TestWithParam<TestType> {
"{\n"
" \"loadBalancingConfig\":[\n"
" { \"does_not_exist\":{} },\n"
" { \"xds_experimental\":{\n"
" { \"eds_experimental\":{\n"
" \"clusterName\": \"application_target_name\"\n"
" } }\n"
" ]\n"
"}";
@ -1563,7 +1561,7 @@ TEST_P(BasicTest, Vanilla) {
}
// Check LB policy name for the channel.
EXPECT_EQ(
(GetParam().use_xds_resolver() ? "cds_experimental" : "xds_experimental"),
(GetParam().use_xds_resolver() ? "cds_experimental" : "eds_experimental"),
channel_->GetLoadBalancingPolicyName());
}
@ -1941,7 +1939,7 @@ using SecureNamingTest = BasicTest;
// Tests that secure naming check passes if target name is expected.
TEST_P(SecureNamingTest, TargetNameIsExpected) {
// TODO(juanlishen): Use separate fake creds for the balancer channel.
ResetStub(0, 0, kApplicationTargetName_ + ";lb");
ResetStub(0, kApplicationTargetName_ + ";lb");
SetNextResolution({});
SetNextResolutionForLbChannel({balancers_[0]->port()});
const size_t kNumRpcsPerAddress = 100;
@ -1971,7 +1969,7 @@ TEST_P(SecureNamingTest, TargetNameIsUnexpected) {
// the name from the balancer doesn't match expectations.
ASSERT_DEATH_IF_SUPPORTED(
{
ResetStub(0, 0, kApplicationTargetName_ + ";lb");
ResetStub(0, kApplicationTargetName_ + ";lb");
SetNextResolution({});
SetNextResolutionForLbChannel({balancers_[0]->port()});
channel_->WaitForConnected(grpc_timeout_seconds_to_deadline(1));
@ -2130,7 +2128,7 @@ TEST_P(LdsTest, RouteActionHasNoCluster) {
// Tests that LDS client times out when no response received.
TEST_P(LdsTest, Timeout) {
ResetStub(0, 0, "", 500);
ResetStub(0, "", 500);
balancers_[0]->ads_service()->SetResourceIgnore(kLdsTypeUrl);
SetNextResolution({});
SetNextResolutionForLbChannelAllBalancers();
@ -2265,7 +2263,7 @@ TEST_P(RdsTest, RouteActionHasNoCluster) {
// Tests that RDS client times out when no response received.
TEST_P(RdsTest, Timeout) {
ResetStub(0, 0, "", 500);
ResetStub(0, "", 500);
balancers_[0]->ads_service()->SetResourceIgnore(kRdsTypeUrl);
balancers_[0]->ads_service()->SetLdsToUseDynamicRds();
SetNextResolution({});
@ -2338,7 +2336,7 @@ TEST_P(CdsTest, WrongLrsServer) {
// Tests that CDS client times out when no response received.
TEST_P(CdsTest, Timeout) {
ResetStub(0, 0, "", 500);
ResetStub(0, "", 500);
balancers_[0]->ads_service()->SetResourceIgnore(kCdsTypeUrl);
SetNextResolution({});
SetNextResolutionForLbChannelAllBalancers();
@ -2348,7 +2346,7 @@ TEST_P(CdsTest, Timeout) {
using EdsTest = BasicTest;
TEST_P(EdsTest, Timeout) {
ResetStub(0, 0, "", 500);
ResetStub(0, "", 500);
balancers_[0]->ads_service()->SetResourceIgnore(kEdsTypeUrl);
SetNextResolution({});
SetNextResolutionForLbChannelAllBalancers();
@ -2608,7 +2606,7 @@ class FailoverTest : public BasicTest {
public:
void SetUp() override {
BasicTest::SetUp();
ResetStub(0, 100, "");
ResetStub(100, "");
}
};
@ -3045,241 +3043,6 @@ TEST_P(DropTest, DropAll) {
}
}
using FallbackTest = BasicTest;
// Tests that RPCs are handled by the fallback backends before the serverlist is
// received, but will be handled by the serverlist after it's received.
TEST_P(FallbackTest, Vanilla) {
const int kFallbackTimeoutMs = 200 * grpc_test_slowdown_factor();
const int kServerlistDelayMs = 500 * grpc_test_slowdown_factor();
const size_t kNumBackendsInResolution = backends_.size() / 2;
ResetStub(kFallbackTimeoutMs);
SetNextResolution(GetBackendPorts(0, kNumBackendsInResolution));
SetNextResolutionForLbChannelAllBalancers();
// Send non-empty serverlist only after kServerlistDelayMs.
AdsServiceImpl::EdsResourceArgs args({
{"locality0", GetBackendPorts(kNumBackendsInResolution)},
});
std::thread delayed_resource_setter(
std::bind(&BasicTest::SetEdsResourceWithDelay, this, 0,
AdsServiceImpl::BuildEdsResource(args), kServerlistDelayMs,
kDefaultResourceName));
// Wait until all the fallback backends are reachable.
WaitForAllBackends(0 /* start_index */,
kNumBackendsInResolution /* stop_index */);
gpr_log(GPR_INFO, "========= BEFORE FIRST BATCH ==========");
CheckRpcSendOk(kNumBackendsInResolution);
gpr_log(GPR_INFO, "========= DONE WITH FIRST BATCH ==========");
// Fallback is used: each backend returned by the resolver should have
// gotten one request.
for (size_t i = 0; i < kNumBackendsInResolution; ++i) {
EXPECT_EQ(1U, backends_[i]->backend_service()->request_count());
}
for (size_t i = kNumBackendsInResolution; i < backends_.size(); ++i) {
EXPECT_EQ(0U, backends_[i]->backend_service()->request_count());
}
// Wait until the serverlist reception has been processed and all backends
// in the serverlist are reachable.
WaitForAllBackends(kNumBackendsInResolution /* start_index */);
gpr_log(GPR_INFO, "========= BEFORE SECOND BATCH ==========");
CheckRpcSendOk(backends_.size() - kNumBackendsInResolution);
gpr_log(GPR_INFO, "========= DONE WITH SECOND BATCH ==========");
// Serverlist is used: each backend returned by the balancer should
// have gotten one request.
for (size_t i = 0; i < kNumBackendsInResolution; ++i) {
EXPECT_EQ(0U, backends_[i]->backend_service()->request_count());
}
for (size_t i = kNumBackendsInResolution; i < backends_.size(); ++i) {
EXPECT_EQ(1U, backends_[i]->backend_service()->request_count());
}
delayed_resource_setter.join();
}
// Tests that RPCs are handled by the updated fallback backends before
// serverlist is received,
TEST_P(FallbackTest, Update) {
const int kFallbackTimeoutMs = 200 * grpc_test_slowdown_factor();
const int kServerlistDelayMs = 500 * grpc_test_slowdown_factor();
const size_t kNumBackendsInResolution = backends_.size() / 3;
const size_t kNumBackendsInResolutionUpdate = backends_.size() / 3;
ResetStub(kFallbackTimeoutMs);
SetNextResolution(GetBackendPorts(0, kNumBackendsInResolution));
SetNextResolutionForLbChannelAllBalancers();
// Send non-empty serverlist only after kServerlistDelayMs.
AdsServiceImpl::EdsResourceArgs args({
{"locality0", GetBackendPorts(kNumBackendsInResolution +
kNumBackendsInResolutionUpdate)},
});
std::thread delayed_resource_setter(
std::bind(&BasicTest::SetEdsResourceWithDelay, this, 0,
AdsServiceImpl::BuildEdsResource(args), kServerlistDelayMs,
kDefaultResourceName));
// Wait until all the fallback backends are reachable.
WaitForAllBackends(0 /* start_index */,
kNumBackendsInResolution /* stop_index */);
gpr_log(GPR_INFO, "========= BEFORE FIRST BATCH ==========");
CheckRpcSendOk(kNumBackendsInResolution);
gpr_log(GPR_INFO, "========= DONE WITH FIRST BATCH ==========");
// Fallback is used: each backend returned by the resolver should have
// gotten one request.
for (size_t i = 0; i < kNumBackendsInResolution; ++i) {
EXPECT_EQ(1U, backends_[i]->backend_service()->request_count());
}
for (size_t i = kNumBackendsInResolution; i < backends_.size(); ++i) {
EXPECT_EQ(0U, backends_[i]->backend_service()->request_count());
}
SetNextResolution(GetBackendPorts(
kNumBackendsInResolution,
kNumBackendsInResolution + kNumBackendsInResolutionUpdate));
// Wait until the resolution update has been processed and all the new
// fallback backends are reachable.
WaitForAllBackends(kNumBackendsInResolution /* start_index */,
kNumBackendsInResolution +
kNumBackendsInResolutionUpdate /* stop_index */);
gpr_log(GPR_INFO, "========= BEFORE SECOND BATCH ==========");
CheckRpcSendOk(kNumBackendsInResolutionUpdate);
gpr_log(GPR_INFO, "========= DONE WITH SECOND BATCH ==========");
// The resolution update is used: each backend in the resolution update should
// have gotten one request.
for (size_t i = 0; i < kNumBackendsInResolution; ++i) {
EXPECT_EQ(0U, backends_[i]->backend_service()->request_count());
}
for (size_t i = kNumBackendsInResolution;
i < kNumBackendsInResolution + kNumBackendsInResolutionUpdate; ++i) {
EXPECT_EQ(1U, backends_[i]->backend_service()->request_count());
}
for (size_t i = kNumBackendsInResolution + kNumBackendsInResolutionUpdate;
i < backends_.size(); ++i) {
EXPECT_EQ(0U, backends_[i]->backend_service()->request_count());
}
// Wait until the serverlist reception has been processed and all backends
// in the serverlist are reachable.
WaitForAllBackends(kNumBackendsInResolution +
kNumBackendsInResolutionUpdate /* start_index */);
gpr_log(GPR_INFO, "========= BEFORE THIRD BATCH ==========");
CheckRpcSendOk(backends_.size() - kNumBackendsInResolution -
kNumBackendsInResolutionUpdate);
gpr_log(GPR_INFO, "========= DONE WITH THIRD BATCH ==========");
// Serverlist is used: each backend returned by the balancer should
// have gotten one request.
for (size_t i = 0;
i < kNumBackendsInResolution + kNumBackendsInResolutionUpdate; ++i) {
EXPECT_EQ(0U, backends_[i]->backend_service()->request_count());
}
for (size_t i = kNumBackendsInResolution + kNumBackendsInResolutionUpdate;
i < backends_.size(); ++i) {
EXPECT_EQ(1U, backends_[i]->backend_service()->request_count());
}
delayed_resource_setter.join();
}
// Tests that fallback will kick in immediately if the balancer channel fails.
TEST_P(FallbackTest, FallbackEarlyWhenBalancerChannelFails) {
const int kFallbackTimeoutMs = 10000 * grpc_test_slowdown_factor();
ResetStub(kFallbackTimeoutMs);
// Return an unreachable balancer and one fallback backend.
SetNextResolution({backends_[0]->port()});
SetNextResolutionForLbChannel({g_port_saver->GetPort()});
// Send RPC with deadline less than the fallback timeout and make sure it
// succeeds.
CheckRpcSendOk(/* times */ 1, /* timeout_ms */ 1000,
/* wait_for_ready */ false);
}
// Tests that fallback will kick in immediately if the balancer call fails.
TEST_P(FallbackTest, FallbackEarlyWhenBalancerCallFails) {
const int kFallbackTimeoutMs = 10000 * grpc_test_slowdown_factor();
ResetStub(kFallbackTimeoutMs);
// Return one balancer and one fallback backend.
SetNextResolution({backends_[0]->port()});
SetNextResolutionForLbChannelAllBalancers();
// Balancer drops call without sending a serverlist.
balancers_[0]->ads_service()->NotifyDoneWithAdsCall();
// Send RPC with deadline less than the fallback timeout and make sure it
// succeeds.
CheckRpcSendOk(/* times */ 1, /* timeout_ms */ 1000,
/* wait_for_ready */ false);
}
// Tests that fallback mode is entered if balancer response is received but the
// backends can't be reached.
TEST_P(FallbackTest, FallbackIfResponseReceivedButChildNotReady) {
const int kFallbackTimeoutMs = 500 * grpc_test_slowdown_factor();
ResetStub(kFallbackTimeoutMs);
SetNextResolution({backends_[0]->port()});
SetNextResolutionForLbChannelAllBalancers();
// Send a serverlist that only contains an unreachable backend before fallback
// timeout.
AdsServiceImpl::EdsResourceArgs args({
{"locality0", {g_port_saver->GetPort()}},
});
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args), kDefaultResourceName);
// Because no child policy is ready before fallback timeout, we enter fallback
// mode.
WaitForBackend(0);
}
// Tests that fallback mode is exited if the balancer tells the client to drop
// all the calls.
TEST_P(FallbackTest, FallbackModeIsExitedWhenBalancerSaysToDropAllCalls) {
// Return an unreachable balancer and one fallback backend.
SetNextResolution({backends_[0]->port()});
SetNextResolutionForLbChannel({g_port_saver->GetPort()});
// Enter fallback mode because the LB channel fails to connect.
WaitForBackend(0);
// Return a new balancer that sends a response to drop all calls.
AdsServiceImpl::EdsResourceArgs args({
{"locality0", GetBackendPorts()},
});
args.drop_categories = {{kLbDropType, 1000000}};
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args), kDefaultResourceName);
SetNextResolutionForLbChannelAllBalancers();
// Send RPCs until failure.
gpr_timespec deadline = gpr_time_add(
gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(5000, GPR_TIMESPAN));
do {
auto status = SendRpc();
if (!status.ok()) break;
} while (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), deadline) < 0);
CheckRpcSendFailure();
}
// Tests that fallback mode is exited if the child policy becomes ready.
TEST_P(FallbackTest, FallbackModeIsExitedAfterChildReady) {
// Return an unreachable balancer and one fallback backend.
SetNextResolution({backends_[0]->port()});
SetNextResolutionForLbChannel({g_port_saver->GetPort()});
// Enter fallback mode because the LB channel fails to connect.
WaitForBackend(0);
// Return a new balancer that sends a dead backend.
ShutdownBackend(1);
AdsServiceImpl::EdsResourceArgs args({
{"locality0", {backends_[1]->port()}},
});
balancers_[0]->ads_service()->SetEdsResource(
AdsServiceImpl::BuildEdsResource(args), kDefaultResourceName);
SetNextResolutionForLbChannelAllBalancers();
// The state (TRANSIENT_FAILURE) update from the child policy will be ignored
// because we are still in fallback mode.
gpr_timespec deadline = gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
gpr_time_from_millis(500, GPR_TIMESPAN));
// Send 0.5 second worth of RPCs.
do {
CheckRpcSendOk();
} while (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), deadline) < 0);
// After the backend is restarted, the child policy will eventually be READY,
// and we will exit fallback mode.
StartBackend(1);
WaitForBackend(1);
// We have exited fallback mode, so calls will go to the child policy
// exclusively.
CheckRpcSendOk(100);
EXPECT_EQ(0U, backends_[0]->backend_service()->request_count());
EXPECT_EQ(100U, backends_[1]->backend_service()->request_count());
}
class BalancerUpdateTest : public XdsEnd2endTest {
public:
BalancerUpdateTest() : XdsEnd2endTest(4, 3) {}
@ -3782,12 +3545,6 @@ INSTANTIATE_TEST_SUITE_P(XdsTest, DropTest,
TestType(true, true)),
&TestTypeName);
// Fallback does not work with xds resolver.
INSTANTIATE_TEST_SUITE_P(XdsTest, FallbackTest,
::testing::Values(TestType(false, true),
TestType(false, false)),
&TestTypeName);
INSTANTIATE_TEST_SUITE_P(XdsTest, BalancerUpdateTest,
::testing::Values(TestType(false, true),
TestType(false, false),

Write Preview
Loading…
Cancel
Save