|
|
|
|
@ -68,226 +68,185 @@ namespace grpc_core { |
|
|
|
|
|
|
|
|
|
TraceFlag grpc_xds_client_trace(false, "xds_client"); |
|
|
|
|
|
|
|
|
|
// Contains a channel to the xds server and all the data related to the
|
|
|
|
|
// channel. Holds a ref to the xds client object.
|
|
|
|
|
// TODO(roth): This is separate from the XdsClient object because it was
|
|
|
|
|
// originally designed to be able to swap itself out in case the
|
|
|
|
|
// balancer name changed. Now that the balancer name is going to be
|
|
|
|
|
// coming from the bootstrap file, we don't really need this level of
|
|
|
|
|
// indirection unless we decide to support watching the bootstrap file
|
|
|
|
|
// for changes. At some point, if we decide that we're never going to
|
|
|
|
|
// need to do that, then we can eliminate this class and move its
|
|
|
|
|
// contents directly into the XdsClient class.
|
|
|
|
|
class XdsClient::ChannelState : public InternallyRefCounted<ChannelState> { |
|
|
|
|
public: |
|
|
|
|
// An xds call wrapper that can restart a call upon failure. Holds a ref to
|
|
|
|
|
// the xds channel. The template parameter is the kind of wrapped xds call.
|
|
|
|
|
template <typename T> |
|
|
|
|
class RetryableCall : public InternallyRefCounted<RetryableCall<T>> { |
|
|
|
|
public: |
|
|
|
|
explicit RetryableCall(RefCountedPtr<ChannelState> chand); |
|
|
|
|
//
|
|
|
|
|
// Internal class declarations
|
|
|
|
|
//
|
|
|
|
|
|
|
|
|
|
void Orphan() override; |
|
|
|
|
// An xds call wrapper that can restart a call upon failure. Holds a ref to
|
|
|
|
|
// the xds channel. The template parameter is the kind of wrapped xds call.
|
|
|
|
|
template <typename T> |
|
|
|
|
class XdsClient::ChannelState::RetryableCall |
|
|
|
|
: public InternallyRefCounted<RetryableCall<T>> { |
|
|
|
|
public: |
|
|
|
|
explicit RetryableCall(RefCountedPtr<ChannelState> chand); |
|
|
|
|
|
|
|
|
|
void OnCallFinishedLocked(); |
|
|
|
|
void Orphan() override; |
|
|
|
|
|
|
|
|
|
T* calld() const { return calld_.get(); } |
|
|
|
|
ChannelState* chand() const { return chand_.get(); } |
|
|
|
|
void OnCallFinishedLocked(); |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
void StartNewCallLocked(); |
|
|
|
|
void StartRetryTimerLocked(); |
|
|
|
|
static void OnRetryTimer(void* arg, grpc_error* error); |
|
|
|
|
static void OnRetryTimerLocked(void* arg, grpc_error* error); |
|
|
|
|
|
|
|
|
|
// The wrapped call that talks to the xds server. It's instantiated
|
|
|
|
|
// every time we start a new call. It's null during call retry backoff.
|
|
|
|
|
OrphanablePtr<T> calld_; |
|
|
|
|
// The owning xds channel.
|
|
|
|
|
RefCountedPtr<ChannelState> chand_; |
|
|
|
|
|
|
|
|
|
// Retry state.
|
|
|
|
|
BackOff backoff_; |
|
|
|
|
grpc_timer retry_timer_; |
|
|
|
|
grpc_closure on_retry_timer_; |
|
|
|
|
bool retry_timer_callback_pending_ = false; |
|
|
|
|
|
|
|
|
|
bool shutting_down_ = false; |
|
|
|
|
}; |
|
|
|
|
T* calld() const { return calld_.get(); } |
|
|
|
|
ChannelState* chand() const { return chand_.get(); } |
|
|
|
|
|
|
|
|
|
// Contains an ADS call to the xds server.
|
|
|
|
|
class AdsCallState : public InternallyRefCounted<AdsCallState> { |
|
|
|
|
public: |
|
|
|
|
// The ctor and dtor should not be used directly.
|
|
|
|
|
explicit AdsCallState(RefCountedPtr<RetryableCall<AdsCallState>> parent); |
|
|
|
|
~AdsCallState() override; |
|
|
|
|
bool IsCurrentCallOnChannel() const; |
|
|
|
|
|
|
|
|
|
void Orphan() override; |
|
|
|
|
private: |
|
|
|
|
void StartNewCallLocked(); |
|
|
|
|
void StartRetryTimerLocked(); |
|
|
|
|
static void OnRetryTimer(void* arg, grpc_error* error); |
|
|
|
|
static void OnRetryTimerLocked(void* arg, grpc_error* error); |
|
|
|
|
|
|
|
|
|
// The wrapped xds call that talks to the xds server. It's instantiated
|
|
|
|
|
// every time we start a new call. It's null during call retry backoff.
|
|
|
|
|
OrphanablePtr<T> calld_; |
|
|
|
|
// The owning xds channel.
|
|
|
|
|
RefCountedPtr<ChannelState> chand_; |
|
|
|
|
|
|
|
|
|
// Retry state.
|
|
|
|
|
BackOff backoff_; |
|
|
|
|
grpc_timer retry_timer_; |
|
|
|
|
grpc_closure on_retry_timer_; |
|
|
|
|
bool retry_timer_callback_pending_ = false; |
|
|
|
|
|
|
|
|
|
RetryableCall<AdsCallState>* parent() const { return parent_.get(); } |
|
|
|
|
ChannelState* chand() const { return parent_->chand(); } |
|
|
|
|
XdsClient* xds_client() const { return chand()->xds_client(); } |
|
|
|
|
bool seen_response() const { return seen_response_; } |
|
|
|
|
bool shutting_down_ = false; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
static void OnResponseReceived(void* arg, grpc_error* error); |
|
|
|
|
static void OnStatusReceived(void* arg, grpc_error* error); |
|
|
|
|
static void OnResponseReceivedLocked(void* arg, grpc_error* error); |
|
|
|
|
static void OnStatusReceivedLocked(void* arg, grpc_error* error); |
|
|
|
|
// Contains an ADS call to the xds server.
|
|
|
|
|
class XdsClient::ChannelState::AdsCallState |
|
|
|
|
: public InternallyRefCounted<AdsCallState> { |
|
|
|
|
public: |
|
|
|
|
// The ctor and dtor should not be used directly.
|
|
|
|
|
explicit AdsCallState(RefCountedPtr<RetryableCall<AdsCallState>> parent); |
|
|
|
|
~AdsCallState() override; |
|
|
|
|
|
|
|
|
|
bool IsCurrentCallOnChannel() const; |
|
|
|
|
void Orphan() override; |
|
|
|
|
|
|
|
|
|
// The owning RetryableCall<>.
|
|
|
|
|
RefCountedPtr<RetryableCall<AdsCallState>> parent_; |
|
|
|
|
bool seen_response_ = false; |
|
|
|
|
RetryableCall<AdsCallState>* parent() const { return parent_.get(); } |
|
|
|
|
ChannelState* chand() const { return parent_->chand(); } |
|
|
|
|
XdsClient* xds_client() const { return chand()->xds_client(); } |
|
|
|
|
bool seen_response() const { return seen_response_; } |
|
|
|
|
|
|
|
|
|
// Always non-NULL.
|
|
|
|
|
grpc_call* call_; |
|
|
|
|
private: |
|
|
|
|
static void OnResponseReceived(void* arg, grpc_error* error); |
|
|
|
|
static void OnStatusReceived(void* arg, grpc_error* error); |
|
|
|
|
static void OnResponseReceivedLocked(void* arg, grpc_error* error); |
|
|
|
|
static void OnStatusReceivedLocked(void* arg, grpc_error* error); |
|
|
|
|
|
|
|
|
|
// recv_initial_metadata
|
|
|
|
|
grpc_metadata_array initial_metadata_recv_; |
|
|
|
|
bool IsCurrentCallOnChannel() const; |
|
|
|
|
|
|
|
|
|
// send_message
|
|
|
|
|
grpc_byte_buffer* send_message_payload_ = nullptr; |
|
|
|
|
// The owning RetryableCall<>.
|
|
|
|
|
RefCountedPtr<RetryableCall<AdsCallState>> parent_; |
|
|
|
|
bool seen_response_ = false; |
|
|
|
|
|
|
|
|
|
// recv_message
|
|
|
|
|
grpc_byte_buffer* recv_message_payload_ = nullptr; |
|
|
|
|
grpc_closure on_response_received_; |
|
|
|
|
// Always non-NULL.
|
|
|
|
|
grpc_call* call_; |
|
|
|
|
|
|
|
|
|
// recv_trailing_metadata
|
|
|
|
|
grpc_metadata_array trailing_metadata_recv_; |
|
|
|
|
grpc_status_code status_code_; |
|
|
|
|
grpc_slice status_details_; |
|
|
|
|
grpc_closure on_status_received_; |
|
|
|
|
}; |
|
|
|
|
// recv_initial_metadata
|
|
|
|
|
grpc_metadata_array initial_metadata_recv_; |
|
|
|
|
|
|
|
|
|
// Contains an LRS call to the xds server.
|
|
|
|
|
class LrsCallState : public InternallyRefCounted<LrsCallState> { |
|
|
|
|
public: |
|
|
|
|
// The ctor and dtor should not be used directly.
|
|
|
|
|
explicit LrsCallState(RefCountedPtr<RetryableCall<LrsCallState>> parent); |
|
|
|
|
~LrsCallState() override; |
|
|
|
|
// send_message
|
|
|
|
|
grpc_byte_buffer* send_message_payload_ = nullptr; |
|
|
|
|
|
|
|
|
|
void Orphan() override; |
|
|
|
|
// recv_message
|
|
|
|
|
grpc_byte_buffer* recv_message_payload_ = nullptr; |
|
|
|
|
grpc_closure on_response_received_; |
|
|
|
|
|
|
|
|
|
void MaybeStartReportingLocked(); |
|
|
|
|
// recv_trailing_metadata
|
|
|
|
|
grpc_metadata_array trailing_metadata_recv_; |
|
|
|
|
grpc_status_code status_code_; |
|
|
|
|
grpc_slice status_details_; |
|
|
|
|
grpc_closure on_status_received_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
RetryableCall<LrsCallState>* parent() { return parent_.get(); } |
|
|
|
|
ChannelState* chand() const { return parent_->chand(); } |
|
|
|
|
XdsClient* xds_client() const { return chand()->xds_client(); } |
|
|
|
|
bool seen_response() const { return seen_response_; } |
|
|
|
|
// Contains an LRS call to the xds server.
|
|
|
|
|
class XdsClient::ChannelState::LrsCallState |
|
|
|
|
: public InternallyRefCounted<LrsCallState> { |
|
|
|
|
public: |
|
|
|
|
// The ctor and dtor should not be used directly.
|
|
|
|
|
explicit LrsCallState(RefCountedPtr<RetryableCall<LrsCallState>> parent); |
|
|
|
|
~LrsCallState() override; |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
// Reports client-side load stats according to a fixed interval.
|
|
|
|
|
class Reporter : public InternallyRefCounted<Reporter> { |
|
|
|
|
public: |
|
|
|
|
Reporter(RefCountedPtr<LrsCallState> parent, grpc_millis report_interval) |
|
|
|
|
: parent_(std::move(parent)), report_interval_(report_interval) { |
|
|
|
|
ScheduleNextReportLocked(); |
|
|
|
|
} |
|
|
|
|
void Orphan() override; |
|
|
|
|
|
|
|
|
|
void Orphan() override; |
|
|
|
|
void MaybeStartReportingLocked(); |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
void ScheduleNextReportLocked(); |
|
|
|
|
static void OnNextReportTimer(void* arg, grpc_error* error); |
|
|
|
|
static void OnNextReportTimerLocked(void* arg, grpc_error* error); |
|
|
|
|
void SendReportLocked(); |
|
|
|
|
static void OnReportDone(void* arg, grpc_error* error); |
|
|
|
|
static void OnReportDoneLocked(void* arg, grpc_error* error); |
|
|
|
|
RetryableCall<LrsCallState>* parent() { return parent_.get(); } |
|
|
|
|
ChannelState* chand() const { return parent_->chand(); } |
|
|
|
|
XdsClient* xds_client() const { return chand()->xds_client(); } |
|
|
|
|
bool seen_response() const { return seen_response_; } |
|
|
|
|
|
|
|
|
|
bool IsCurrentReporterOnCall() const { |
|
|
|
|
return this == parent_->reporter_.get(); |
|
|
|
|
} |
|
|
|
|
XdsClient* xds_client() const { return parent_->xds_client(); } |
|
|
|
|
|
|
|
|
|
// The owning LRS call.
|
|
|
|
|
RefCountedPtr<LrsCallState> parent_; |
|
|
|
|
|
|
|
|
|
// The load reporting state.
|
|
|
|
|
const grpc_millis report_interval_; |
|
|
|
|
bool last_report_counters_were_zero_ = false; |
|
|
|
|
bool next_report_timer_callback_pending_ = false; |
|
|
|
|
grpc_timer next_report_timer_; |
|
|
|
|
grpc_closure on_next_report_timer_; |
|
|
|
|
grpc_closure on_report_done_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
static void OnInitialRequestSent(void* arg, grpc_error* error); |
|
|
|
|
static void OnResponseReceived(void* arg, grpc_error* error); |
|
|
|
|
static void OnStatusReceived(void* arg, grpc_error* error); |
|
|
|
|
static void OnInitialRequestSentLocked(void* arg, grpc_error* error); |
|
|
|
|
static void OnResponseReceivedLocked(void* arg, grpc_error* error); |
|
|
|
|
static void OnStatusReceivedLocked(void* arg, grpc_error* error); |
|
|
|
|
|
|
|
|
|
bool IsCurrentCallOnChannel() const; |
|
|
|
|
|
|
|
|
|
// The owning RetryableCall<>.
|
|
|
|
|
RefCountedPtr<RetryableCall<LrsCallState>> parent_; |
|
|
|
|
bool seen_response_ = false; |
|
|
|
|
|
|
|
|
|
// Always non-NULL.
|
|
|
|
|
grpc_call* call_; |
|
|
|
|
|
|
|
|
|
// recv_initial_metadata
|
|
|
|
|
grpc_metadata_array initial_metadata_recv_; |
|
|
|
|
|
|
|
|
|
// send_message
|
|
|
|
|
grpc_byte_buffer* send_message_payload_ = nullptr; |
|
|
|
|
grpc_closure on_initial_request_sent_; |
|
|
|
|
|
|
|
|
|
// recv_message
|
|
|
|
|
grpc_byte_buffer* recv_message_payload_ = nullptr; |
|
|
|
|
grpc_closure on_response_received_; |
|
|
|
|
|
|
|
|
|
// recv_trailing_metadata
|
|
|
|
|
grpc_metadata_array trailing_metadata_recv_; |
|
|
|
|
grpc_status_code status_code_; |
|
|
|
|
grpc_slice status_details_; |
|
|
|
|
grpc_closure on_status_received_; |
|
|
|
|
|
|
|
|
|
// Load reporting state.
|
|
|
|
|
UniquePtr<char> cluster_name_; |
|
|
|
|
grpc_millis load_reporting_interval_ = 0; |
|
|
|
|
OrphanablePtr<Reporter> reporter_; |
|
|
|
|
}; |
|
|
|
|
private: |
|
|
|
|
// Reports client-side load stats according to a fixed interval.
|
|
|
|
|
class Reporter : public InternallyRefCounted<Reporter> { |
|
|
|
|
public: |
|
|
|
|
Reporter(RefCountedPtr<LrsCallState> parent, grpc_millis report_interval) |
|
|
|
|
: parent_(std::move(parent)), report_interval_(report_interval) { |
|
|
|
|
ScheduleNextReportLocked(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ChannelState(RefCountedPtr<XdsClient> xds_client, |
|
|
|
|
const grpc_channel_args& args); |
|
|
|
|
~ChannelState(); |
|
|
|
|
void Orphan() override; |
|
|
|
|
|
|
|
|
|
void Orphan() override; |
|
|
|
|
private: |
|
|
|
|
void ScheduleNextReportLocked(); |
|
|
|
|
static void OnNextReportTimer(void* arg, grpc_error* error); |
|
|
|
|
static void OnNextReportTimerLocked(void* arg, grpc_error* error); |
|
|
|
|
void SendReportLocked(); |
|
|
|
|
static void OnReportDone(void* arg, grpc_error* error); |
|
|
|
|
static void OnReportDoneLocked(void* arg, grpc_error* error); |
|
|
|
|
|
|
|
|
|
bool IsCurrentReporterOnCall() const { |
|
|
|
|
return this == parent_->reporter_.get(); |
|
|
|
|
} |
|
|
|
|
XdsClient* xds_client() const { return parent_->xds_client(); } |
|
|
|
|
|
|
|
|
|
// The owning LRS call.
|
|
|
|
|
RefCountedPtr<LrsCallState> parent_; |
|
|
|
|
|
|
|
|
|
// The load reporting state.
|
|
|
|
|
const grpc_millis report_interval_; |
|
|
|
|
bool last_report_counters_were_zero_ = false; |
|
|
|
|
bool next_report_timer_callback_pending_ = false; |
|
|
|
|
grpc_timer next_report_timer_; |
|
|
|
|
grpc_closure on_next_report_timer_; |
|
|
|
|
grpc_closure on_report_done_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
grpc_channel* channel() const { return channel_; } |
|
|
|
|
XdsClient* xds_client() const { return xds_client_.get(); } |
|
|
|
|
AdsCallState* ads_calld() const { return ads_calld_->calld(); } |
|
|
|
|
LrsCallState* lrs_calld() const { return lrs_calld_->calld(); } |
|
|
|
|
static void OnInitialRequestSent(void* arg, grpc_error* error); |
|
|
|
|
static void OnResponseReceived(void* arg, grpc_error* error); |
|
|
|
|
static void OnStatusReceived(void* arg, grpc_error* error); |
|
|
|
|
static void OnInitialRequestSentLocked(void* arg, grpc_error* error); |
|
|
|
|
static void OnResponseReceivedLocked(void* arg, grpc_error* error); |
|
|
|
|
static void OnStatusReceivedLocked(void* arg, grpc_error* error); |
|
|
|
|
|
|
|
|
|
void MaybeStartAdsCall(); |
|
|
|
|
void StopAdsCall(); |
|
|
|
|
bool IsCurrentCallOnChannel() const; |
|
|
|
|
|
|
|
|
|
void MaybeStartLrsCall(); |
|
|
|
|
void StopLrsCall(); |
|
|
|
|
// The owning RetryableCall<>.
|
|
|
|
|
RefCountedPtr<RetryableCall<LrsCallState>> parent_; |
|
|
|
|
bool seen_response_ = false; |
|
|
|
|
|
|
|
|
|
bool HasActiveAdsCall() const { return ads_calld_->calld() != nullptr; } |
|
|
|
|
// Always non-NULL.
|
|
|
|
|
grpc_call* call_; |
|
|
|
|
|
|
|
|
|
void StartConnectivityWatchLocked(); |
|
|
|
|
void CancelConnectivityWatchLocked(); |
|
|
|
|
// recv_initial_metadata
|
|
|
|
|
grpc_metadata_array initial_metadata_recv_; |
|
|
|
|
|
|
|
|
|
private: |
|
|
|
|
class StateWatcher; |
|
|
|
|
// send_message
|
|
|
|
|
grpc_byte_buffer* send_message_payload_ = nullptr; |
|
|
|
|
grpc_closure on_initial_request_sent_; |
|
|
|
|
|
|
|
|
|
// The owning xds client.
|
|
|
|
|
RefCountedPtr<XdsClient> xds_client_; |
|
|
|
|
// recv_message
|
|
|
|
|
grpc_byte_buffer* recv_message_payload_ = nullptr; |
|
|
|
|
grpc_closure on_response_received_; |
|
|
|
|
|
|
|
|
|
// The channel and its status.
|
|
|
|
|
grpc_channel* channel_; |
|
|
|
|
bool shutting_down_ = false; |
|
|
|
|
StateWatcher* watcher_ = nullptr; |
|
|
|
|
// recv_trailing_metadata
|
|
|
|
|
grpc_metadata_array trailing_metadata_recv_; |
|
|
|
|
grpc_status_code status_code_; |
|
|
|
|
grpc_slice status_details_; |
|
|
|
|
grpc_closure on_status_received_; |
|
|
|
|
|
|
|
|
|
// The retryable XDS calls.
|
|
|
|
|
OrphanablePtr<RetryableCall<AdsCallState>> ads_calld_; |
|
|
|
|
OrphanablePtr<RetryableCall<LrsCallState>> lrs_calld_; |
|
|
|
|
// Load reporting state.
|
|
|
|
|
UniquePtr<char> cluster_name_; |
|
|
|
|
grpc_millis load_reporting_interval_ = 0; |
|
|
|
|
OrphanablePtr<Reporter> reporter_; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
//
|
|
|
|
|
@ -403,6 +362,20 @@ void XdsClient::ChannelState::Orphan() { |
|
|
|
|
Unref(DEBUG_LOCATION, "ChannelState+orphaned"); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
XdsClient::ChannelState::AdsCallState* XdsClient::ChannelState::ads_calld() |
|
|
|
|
const { |
|
|
|
|
return ads_calld_->calld(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
XdsClient::ChannelState::LrsCallState* XdsClient::ChannelState::lrs_calld() |
|
|
|
|
const { |
|
|
|
|
return lrs_calld_->calld(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
bool XdsClient::ChannelState::HasActiveAdsCall() const { |
|
|
|
|
return ads_calld_->calld() != nullptr; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsClient::ChannelState::MaybeStartAdsCall() { |
|
|
|
|
if (ads_calld_ != nullptr) return; |
|
|
|
|
ads_calld_.reset(New<RetryableCall<AdsCallState>>( |
|
|
|
|
@ -1296,7 +1269,13 @@ XdsClient::XdsClient(Combiner* combiner, grpc_pollset_set* interested_parties, |
|
|
|
|
} |
|
|
|
|
chand_ = MakeOrphanable<ChannelState>( |
|
|
|
|
Ref(DEBUG_LOCATION, "XdsClient+ChannelState"), channel_args); |
|
|
|
|
// TODO(roth): Start LDS call.
|
|
|
|
|
if (service_config_watcher_ != nullptr) { |
|
|
|
|
// TODO(juanlishen): Start LDS call and do not return service config
|
|
|
|
|
// until we get the first LDS response.
|
|
|
|
|
GRPC_CLOSURE_INIT(&service_config_notify_, NotifyOnServiceConfig, |
|
|
|
|
Ref().release(), nullptr); |
|
|
|
|
combiner_->Run(&service_config_notify_, GRPC_ERROR_NONE); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
XdsClient::~XdsClient() { GRPC_COMBINER_UNREF(combiner_, "xds_client"); } |
|
|
|
|
@ -1309,12 +1288,12 @@ void XdsClient::Orphan() { |
|
|
|
|
|
|
|
|
|
void XdsClient::WatchClusterData(StringView cluster, |
|
|
|
|
UniquePtr<ClusterWatcherInterface> watcher) { |
|
|
|
|
// TODO(roth): Implement.
|
|
|
|
|
// TODO(juanlishen): Implement.
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsClient::CancelClusterDataWatch(StringView cluster, |
|
|
|
|
ClusterWatcherInterface* watcher) { |
|
|
|
|
// TODO(roth): Implement.
|
|
|
|
|
// TODO(juanlishen): Implement.
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsClient::WatchEndpointData(StringView cluster, |
|
|
|
|
@ -1335,7 +1314,9 @@ void XdsClient::CancelEndpointDataWatch(StringView cluster, |
|
|
|
|
if (it != cluster_state_.endpoint_watchers.end()) { |
|
|
|
|
cluster_state_.endpoint_watchers.erase(it); |
|
|
|
|
} |
|
|
|
|
if (cluster_state_.endpoint_watchers.empty()) chand_->StopAdsCall(); |
|
|
|
|
if (chand_ != nullptr && cluster_state_.endpoint_watchers.empty()) { |
|
|
|
|
chand_->StopAdsCall(); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsClient::AddClientStats(StringView cluster, |
|
|
|
|
@ -1353,7 +1334,9 @@ void XdsClient::RemoveClientStats(StringView cluster, |
|
|
|
|
if (it != cluster_state_.client_stats.end()) { |
|
|
|
|
cluster_state_.client_stats.erase(it); |
|
|
|
|
} |
|
|
|
|
if (cluster_state_.client_stats.empty()) chand_->StopLrsCall(); |
|
|
|
|
if (chand_ != nullptr && cluster_state_.client_stats.empty()) { |
|
|
|
|
chand_->StopLrsCall(); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsClient::ResetBackoff() { |
|
|
|
|
@ -1363,9 +1346,6 @@ void XdsClient::ResetBackoff() { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsClient::NotifyOnError(grpc_error* error) { |
|
|
|
|
// TODO(roth): Once we implement the full LDS flow, it will not be
|
|
|
|
|
// necessary to check for the service config watcher being non-null,
|
|
|
|
|
// because that will always be true.
|
|
|
|
|
if (service_config_watcher_ != nullptr) { |
|
|
|
|
service_config_watcher_->OnError(GRPC_ERROR_REF(error)); |
|
|
|
|
} |
|
|
|
|
@ -1378,6 +1358,27 @@ void XdsClient::NotifyOnError(grpc_error* error) { |
|
|
|
|
GRPC_ERROR_UNREF(error); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void XdsClient::NotifyOnServiceConfig(void* arg, grpc_error* error) { |
|
|
|
|
XdsClient* self = static_cast<XdsClient*>(arg); |
|
|
|
|
// TODO(roth): When we add support for WeightedClusters, select the
|
|
|
|
|
// LB policy based on that functionality.
|
|
|
|
|
static const char* json = |
|
|
|
|
"{\n" |
|
|
|
|
" \"loadBalancingConfig\":[\n" |
|
|
|
|
" { \"xds_experimental\":{} }\n" |
|
|
|
|
" ]\n" |
|
|
|
|
"}"; |
|
|
|
|
RefCountedPtr<ServiceConfig> service_config = |
|
|
|
|
ServiceConfig::Create(json, &error); |
|
|
|
|
if (error != GRPC_ERROR_NONE) { |
|
|
|
|
self->service_config_watcher_->OnError(error); |
|
|
|
|
} else { |
|
|
|
|
self->service_config_watcher_->OnServiceConfigChanged( |
|
|
|
|
std::move(service_config)); |
|
|
|
|
} |
|
|
|
|
self->Unref(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void* XdsClient::ChannelArgCopy(void* p) { |
|
|
|
|
XdsClient* xds_client = static_cast<XdsClient*>(p); |
|
|
|
|
xds_client->Ref().release(); |
|
|
|
|
|
Yash Tibrewal
5 years ago