@ -74,12 +74,12 @@ namespace {
constexpr char kXds [ ] = " xds_experimental " ;
class Parsed XdsConfig: public LoadBalancingPolicy : : Config {
class XdsConfig : public LoadBalancingPolicy : : Config {
public :
Parsed XdsConfig( RefCountedPtr < LoadBalancingPolicy : : Config > child_policy ,
XdsConfig ( RefCountedPtr < LoadBalancingPolicy : : Config > child_policy ,
RefCountedPtr < LoadBalancingPolicy : : Config > fallback_policy ,
grpc_core : : UniquePtr < char > eds_service_name ,
grpc_core : : UniquePtr < char > lrs_load_reporting_server_name )
std : : string eds_service_name ,
Optional < std : : string > lrs_load_reporting_server_name )
: child_policy_ ( std : : move ( child_policy ) ) ,
fallback_policy_ ( std : : move ( fallback_policy ) ) ,
eds_service_name_ ( std : : move ( eds_service_name ) ) ,
@ -96,17 +96,19 @@ class ParsedXdsConfig : public LoadBalancingPolicy::Config {
return fallback_policy_ ;
}
const char * eds_service_name ( ) const { return eds_service_name_ . get ( ) ; } ;
const char * eds_service_name ( ) const {
return eds_service_name_ . empty ( ) ? nullptr : eds_service_name_ . c_str ( ) ;
} ;
const char * lrs_load_reporting_server_name ( ) const {
return lrs_load_reporting_server_name_ . get ( ) ;
const Optional < std : : string > & lrs_load_reporting_server_name ( ) const {
return lrs_load_reporting_server_name_ ;
} ;
private :
RefCountedPtr < LoadBalancingPolicy : : Config > child_policy_ ;
RefCountedPtr < LoadBalancingPolicy : : Config > fallback_policy_ ;
grpc_core : : UniquePtr < char > eds_service_name_ ;
grpc_core : : UniquePtr < char > lrs_load_reporting_server_name_ ;
std : : string eds_service_name_ ;
Optional < std : : string > lrs_load_reporting_server_name_ ;
} ;
class XdsLb : public LoadBalancingPolicy {
@ -160,6 +162,8 @@ class XdsLb : public LoadBalancingPolicy {
pickers_ ( std : : move ( pickers ) ) ,
drop_config_ ( xds_policy_ - > drop_config_ ) { }
~ LocalityPicker ( ) { xds_policy_ . reset ( DEBUG_LOCATION , " LocalityPicker " ) ; }
PickResult Pick ( PickArgs args ) override ;
private :
@ -195,10 +199,6 @@ class XdsLb : public LoadBalancingPolicy {
LoadBalancingPolicy * child_ = nullptr ;
} ;
// There is only one PriorityList instance, which has the same lifetime with
// the XdsLb instance.
class PriorityList {
public :
// Each LocalityMap holds a ref to the XdsLb.
class LocalityMap : public InternallyRefCounted < LocalityMap > {
public :
@ -209,8 +209,7 @@ class XdsLb : public LoadBalancingPolicy {
RefCountedPtr < XdsLocalityName > name ) ;
~ Locality ( ) ;
void UpdateLocked ( uint32_t locality_weight ,
ServerAddressList serverlist ) ;
void UpdateLocked ( uint32_t locality_weight , ServerAddressList serverlist ) ;
void ShutdownLocked ( ) ;
void ResetBackoffLocked ( ) ;
void DeactivateLocked ( ) ;
@ -243,8 +242,7 @@ class XdsLb : public LoadBalancingPolicy {
// This is a no-op, because we get the addresses from the xds
// client, which is a watch-based API.
void RequestReresolution ( ) override { }
void AddTraceEvent ( TraceSeverity severity ,
StringView message ) override ;
void AddTraceEvent ( TraceSeverity severity , StringView message ) override ;
void set_child ( LoadBalancingPolicy * child ) { child_ = child ; }
private :
@ -285,6 +283,8 @@ class XdsLb : public LoadBalancingPolicy {
LocalityMap ( RefCountedPtr < XdsLb > xds_policy , uint32_t priority ) ;
~ LocalityMap ( ) { xds_policy_ . reset ( DEBUG_LOCATION , " LocalityMap " ) ; }
void UpdateLocked (
const XdsPriorityListUpdate : : LocalityMap & locality_map_update ) ;
void ResetBackoffLocked ( ) ;
@ -316,9 +316,6 @@ class XdsLb : public LoadBalancingPolicy {
static void OnDelayedRemovalTimerLocked ( void * arg , grpc_error * error ) ;
static void OnFailoverTimerLocked ( void * arg , grpc_error * error ) ;
PriorityList * priority_list ( ) const {
return & xds_policy_ - > priority_list_ ;
}
const XdsPriorityListUpdate & priority_list_update ( ) const {
return xds_policy_ - > priority_list_update_ ;
}
@ -345,19 +342,24 @@ class XdsLb : public LoadBalancingPolicy {
bool failover_timer_callback_pending_ = false ;
} ;
explicit PriorityList ( XdsLb * xds_policy ) : xds_policy_ ( xds_policy ) { }
~ XdsLb ( ) ;
void UpdateLocked ( ) ;
void ResetBackoffLocked ( ) ;
void ShutdownLocked ( ) ;
void UpdateXdsPickerLocked ( ) ;
void ShutdownLocked ( ) override ;
const XdsPriorityListUpdate & priority_list_update ( ) const {
return xds_policy_ - > priority_list_update_ ;
const char * eds_service_name ( ) const {
if ( config_ ! = nullptr & & config_ - > eds_service_name ( ) ! = nullptr ) {
return config_ - > eds_service_name ( ) ;
}
return server_name_ . c_str ( ) ;
}
uint32_t current_priority ( ) const { return current_priority_ ; }
private :
XdsClient * xds_client ( ) const {
return xds_client_from_channel_ ! = nullptr ? xds_client_from_channel_ . get ( )
: xds_client_ . get ( ) ;
}
void UpdatePrioritiesLocked ( ) ;
void UpdateXdsPickerLocked ( ) ;
void MaybeCreateLocalityMapLocked ( uint32_t priority ) ;
void FailoverOnConnectionFailureLocked ( ) ;
void FailoverOnDisconnectionLocked ( uint32_t failed_priority ) ;
@ -371,19 +373,6 @@ class XdsLb : public LoadBalancingPolicy {
}
bool Contains ( uint32_t priority ) { return priority < priorities_ . size ( ) ; }
XdsLb * xds_policy_ ;
// The list of locality maps, indexed by priority. P0 is the highest
// priority.
InlinedVector < OrphanablePtr < LocalityMap > , 2 > priorities_ ;
// The priority that is being used.
uint32_t current_priority_ = UINT32_MAX ;
} ;
~ XdsLb ( ) ;
void ShutdownLocked ( ) override ;
// Methods for dealing with fallback state.
void MaybeCancelFallbackAtStartupChecks ( ) ;
static void OnFallbackTimer ( void * arg , grpc_error * error ) ;
@ -393,24 +382,12 @@ class XdsLb : public LoadBalancingPolicy {
const char * name , const grpc_channel_args * args ) ;
void MaybeExitFallbackMode ( ) ;
const char * eds_service_name ( ) const {
if ( config_ ! = nullptr & & config_ - > eds_service_name ( ) ! = nullptr ) {
return config_ - > eds_service_name ( ) ;
}
return server_name_ . get ( ) ;
}
XdsClient * xds_client ( ) const {
return xds_client_from_channel_ ! = nullptr ? xds_client_from_channel_ . get ( )
: xds_client_ . get ( ) ;
}
// Server name from target URI.
grpc_core : : UniquePtr < char > server_name_ ;
std : : string server_name_ ;
// Current channel args and config from the resolver.
const grpc_channel_args * args_ = nullptr ;
RefCountedPtr < Parsed XdsConfig> config_ ;
RefCountedPtr < XdsConfig > config_ ;
// Internal state.
bool shutting_down_ = false ;
@ -448,8 +425,11 @@ class XdsLb : public LoadBalancingPolicy {
const grpc_millis locality_retention_interval_ms_ ;
const grpc_millis locality_map_failover_timeout_ms_ ;
// A list of locality maps indexed by priority.
PriorityList priority_list_ ;
// The list of locality maps, indexed by priority. P0 is the highest
// priority.
InlinedVector < OrphanablePtr < LocalityMap > , 2 > priorities_ ;
// The priority that is being used.
uint32_t current_priority_ = UINT32_MAX ;
// The update for priority_list_.
XdsPriorityListUpdate priority_list_update_ ;
@ -495,7 +475,7 @@ LoadBalancingPolicy::PickResult XdsLb::EndpointPickerWrapper::Pick(
XdsLb : : PickResult XdsLb : : LocalityPicker : : Pick ( PickArgs args ) {
// Handle drop.
const grpc_core : : UniquePtr < char > * drop_category ;
const std : : string * drop_category ;
if ( drop_config_ - > ShouldDrop ( & drop_category ) ) {
xds_policy_ - > client_stats_ . AddCallDropped ( * drop_category ) ;
PickResult result ;
@ -612,6 +592,8 @@ class XdsLb::EndpointWatcher : public XdsClient::EndpointWatcherInterface {
explicit EndpointWatcher ( RefCountedPtr < XdsLb > xds_policy )
: xds_policy_ ( std : : move ( xds_policy ) ) { }
~ EndpointWatcher ( ) { xds_policy_ . reset ( DEBUG_LOCATION , " EndpointWatcher " ) ; }
void OnEndpointChanged ( EdsUpdate update ) override {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_xds_trace ) ) {
gpr_log ( GPR_INFO , " [xdslb %p] Received EDS update from xds client " ,
@ -634,13 +616,13 @@ class XdsLb::EndpointWatcher : public XdsClient::EndpointWatcherInterface {
xds_policy_ . get ( ) , drop_config_changed ) ;
}
if ( drop_config_changed ) {
xds_policy_ - > priority_list_ . UpdateXdsPickerLocked ( ) ;
xds_policy_ - > UpdateXdsPickerLocked ( ) ;
}
return ;
}
// Update the priority list.
xds_policy_ - > priority_list_update_ = std : : move ( update . priority_list_update ) ;
xds_policy_ - > priority_list_ . Update Locked( ) ;
xds_policy_ - > UpdatePriorities Locked( ) ;
}
void OnError ( grpc_error * error ) override {
@ -693,8 +675,7 @@ XdsLb::XdsLb(Args args)
{ GRPC_XDS_DEFAULT_LOCALITY_RETENTION_INTERVAL_MS , 0 , INT_MAX } ) ) ,
locality_map_failover_timeout_ms_ ( grpc_channel_args_find_integer (
args . args , GRPC_ARG_XDS_FAILOVER_TIMEOUT_MS ,
{ GRPC_XDS_DEFAULT_FAILOVER_TIMEOUT_MS , 0 , INT_MAX } ) ) ,
priority_list_ ( this ) {
{ GRPC_XDS_DEFAULT_FAILOVER_TIMEOUT_MS , 0 , INT_MAX } ) ) {
if ( xds_client_from_channel_ ! = nullptr & &
GRPC_TRACE_FLAG_ENABLED ( grpc_lb_xds_trace ) ) {
gpr_log ( GPR_INFO , " [xdslb %p] Using xds client %p from channel " , this ,
@ -706,11 +687,10 @@ XdsLb::XdsLb(Args args)
GPR_ASSERT ( server_uri ! = nullptr ) ;
grpc_uri * uri = grpc_uri_parse ( server_uri , true ) ;
GPR_ASSERT ( uri - > path [ 0 ] ! = ' \0 ' ) ;
server_name_ . reset (
gpr_strdup ( uri - > path [ 0 ] = = ' / ' ? uri - > path + 1 : uri - > path ) ) ;
server_name_ = uri - > path [ 0 ] = = ' / ' ? uri - > path + 1 : uri - > path ;
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_xds_trace ) ) {
gpr_log ( GPR_INFO , " [xdslb %p] server name from channel: %s " , this ,
server_name_ . get ( ) ) ;
server_name_ . c_str ( ) ) ;
}
grpc_uri_destroy ( uri ) ;
}
@ -728,7 +708,7 @@ void XdsLb::ShutdownLocked() {
}
shutting_down_ = true ;
MaybeCancelFallbackAtStartupChecks ( ) ;
priority_list_ . ShutdownLocked ( ) ;
priorities_ . clear ( ) ;
if ( fallback_policy_ ! = nullptr ) {
grpc_pollset_set_del_pollset_set ( fallback_policy_ - > interested_parties ( ) ,
interested_parties ( ) ) ;
@ -739,16 +719,23 @@ void XdsLb::ShutdownLocked() {
}
fallback_policy_ . reset ( ) ;
pending_fallback_policy_ . reset ( ) ;
// Cancel the endpoint watch here instead of in our dtor, because the
// watcher holds a ref to us.
// Cancel the endpoint watch here instead of in our dtor if we are using the
// XdsResolver, because the watcher holds a ref to us and we might not be
// destroying the Xds client leading to a situation where the Xds lb policy is
// never destroyed.
if ( xds_client_from_channel_ ! = nullptr ) {
xds_client ( ) - > CancelEndpointDataWatch ( StringView ( eds_service_name ( ) ) ,
endpoint_watcher_ ) ;
if ( config_ - > lrs_load_reporting_server_name ( ) ! = nullptr ) {
if ( config_ - > lrs_load_reporting_server_name ( ) . has_value ( ) ) {
// TODO(roth): We should pass the cluster name (in addition to the
// eds_service_name) when adding the client stats. To do so, we need to
// first find a way to plumb the cluster name down into this LB policy.
xds_client ( ) - > RemoveClientStats (
StringView ( config_ - > lrs_load_reporting_server_name ( ) ) ,
StringView ( config_ - > lrs_load_reporting_server_name ( ) . value ( ) . c_str ( ) ) ,
StringView ( eds_service_name ( ) ) , & client_stats_ ) ;
}
xds_client_from_channel_ . reset ( ) ;
}
xds_client_ . reset ( ) ;
}
@ -761,7 +748,9 @@ void XdsLb::ResetBackoffLocked() {
// LB policy, this is done via the resolver, so we don't need to do it
// for xds_client_from_channel_ here.
if ( xds_client_ ! = nullptr ) xds_client_ - > ResetBackoff ( ) ;
priority_list_ . ResetBackoffLocked ( ) ;
for ( size_t i = 0 ; i < priorities_ . size ( ) ; + + i ) {
priorities_ [ i ] - > ResetBackoffLocked ( ) ;
}
if ( fallback_policy_ ! = nullptr ) {
fallback_policy_ - > ResetBackoffLocked ( ) ;
}
@ -786,7 +775,7 @@ void XdsLb::UpdateLocked(UpdateArgs args) {
args_ = args . args ;
args . args = nullptr ;
// Update priority list.
priority_list_ . Update Locked( ) ;
UpdatePriorities Locked( ) ;
// Update the existing fallback policy. The fallback policy config and/or the
// fallback addresses may be new.
if ( fallback_policy_ ! = nullptr ) UpdateFallbackPolicyLocked ( ) ;
@ -820,7 +809,8 @@ void XdsLb::UpdateLocked(UpdateArgs args) {
xds_client ( ) - > CancelEndpointDataWatch ( StringView ( old_eds_service_name ) ,
endpoint_watcher_ ) ;
}
auto watcher = MakeUnique < EndpointWatcher > ( Ref ( ) ) ;
auto watcher = grpc_core : : MakeUnique < EndpointWatcher > (
Ref ( DEBUG_LOCATION , " EndpointWatcher " ) ) ;
endpoint_watcher_ = watcher . get ( ) ;
xds_client ( ) - > WatchEndpointData ( StringView ( eds_service_name ( ) ) ,
std : : move ( watcher ) ) ;
@ -831,21 +821,25 @@ void XdsLb::UpdateLocked(UpdateArgs args) {
// all of the pickers whenever load reporting is enabled or disabled
// here.
if ( is_initial_update | |
( config_ - > lrs_load_reporting_server_name ( ) = = nullptr ) ! =
( old_config - > lrs_load_reporting_server_name ( ) = = nullptr ) | |
( config_ - > lrs_load_reporting_server_name ( ) ! = nullptr & &
old_config - > lrs_load_reporting_server_name ( ) ! = nullptr & &
strcmp ( config_ - > lrs_load_reporting_server_name ( ) ,
old_config - > lrs_load_reporting_server_name ( ) ) ! = 0 ) ) {
( config_ - > lrs_load_reporting_server_name ( ) . has_value ( ) ) ! =
( old_config - > lrs_load_reporting_server_name ( ) . has_value ( ) ) | |
( config_ - > lrs_load_reporting_server_name ( ) . has_value ( ) & &
old_config - > lrs_load_reporting_server_name ( ) . has_value ( ) & &
config_ - > lrs_load_reporting_server_name ( ) . value ( ) ! =
old_config - > lrs_load_reporting_server_name ( ) . value ( ) ) ) {
if ( old_config ! = nullptr & &
old_config - > lrs_load_reporting_server_name ( ) ! = nullptr ) {
old_config - > lrs_load_reporting_server_name ( ) . has_value ( ) ) {
xds_client ( ) - > RemoveClientStats (
StringView ( old_config - > lrs_load_reporting_server_name ( ) ) ,
StringView (
old_config - > lrs_load_reporting_server_name ( ) . value ( ) . c_str ( ) ) ,
StringView ( old_eds_service_name ) , & client_stats_ ) ;
}
if ( config_ - > lrs_load_reporting_server_name ( ) ! = nullptr ) {
if ( config_ - > lrs_load_reporting_server_name ( ) . has_value ( ) ) {
// TODO(roth): We should pass the cluster name (in addition to the
// eds_service_name) when adding the client stats. To do so, we need to
// first find a way to plumb the cluster name down into this LB policy.
xds_client ( ) - > AddClientStats (
StringView ( config_ - > lrs_load_reporting_server_name ( ) ) ,
StringView ( config_ - > lrs_load_reporting_server_name ( ) . value ( ) . c_str ( ) ) ,
StringView ( eds_service_name ( ) ) , & client_stats_ ) ;
}
}
@ -1028,17 +1022,16 @@ void XdsLb::MaybeExitFallbackMode() {
}
//
// XdsLb::PriorityList
// priority list-related methods
//
void XdsLb : : PriorityList : : UpdateLocked ( ) {
const auto & priority_list_update = xds_policy_ - > priority_list_update_ ;
void XdsLb : : UpdatePrioritiesLocked ( ) {
// 1. Remove from the priority list the priorities that are not in the update.
DeactivatePrioritiesLowerThan ( priority_list_update . LowestPriority ( ) ) ;
DeactivatePrioritiesLowerThan ( priority_list_update_ . LowestPriority ( ) ) ;
// 2. Update all the existing priorities.
for ( uint32_t priority = 0 ; priority < priorities_ . size ( ) ; + + priority ) {
LocalityMap * locality_map = priorities_ [ priority ] . get ( ) ;
const auto * locality_map_update = priority_list_update . Find ( priority ) ;
const auto * locality_map_update = priority_list_update_ . Find ( priority ) ;
// Propagate locality_map_update.
// TODO(juanlishen): Find a clean way to skip duplicate update for a
// priority.
@ -1056,52 +1049,43 @@ void XdsLb::PriorityList::UpdateLocked() {
}
}
void XdsLb : : PriorityList : : ResetBackoffLocked ( ) {
for ( size_t i = 0 ; i < priorities_ . size ( ) ; + + i ) {
priorities_ [ i ] - > ResetBackoffLocked ( ) ;
}
}
void XdsLb : : PriorityList : : ShutdownLocked ( ) { priorities_ . clear ( ) ; }
void XdsLb : : PriorityList : : UpdateXdsPickerLocked ( ) {
void XdsLb : : UpdateXdsPickerLocked ( ) {
// If we are in fallback mode, don't generate an xds picker from localities.
if ( xds_policy_ - > fallback_policy_ ! = nullptr ) return ;
if ( current_priority ( ) = = UINT32_MAX ) {
if ( fallback_policy_ ! = nullptr ) return ;
if ( current_priority_ = = UINT32_MAX ) {
grpc_error * error = grpc_error_set_int (
GRPC_ERROR_CREATE_FROM_STATIC_STRING ( " no ready locality map " ) ,
GRPC_ERROR_INT_GRPC_STATUS , GRPC_STATUS_UNAVAILABLE ) ;
xds_policy_ - > channel_control_helper ( ) - > UpdateState (
channel_control_helper ( ) - > UpdateState (
GRPC_CHANNEL_TRANSIENT_FAILURE ,
MakeUnique < TransientFailurePicker > ( error ) ) ;
grpc_core : : MakeUnique < TransientFailurePicker > ( error ) ) ;
return ;
}
priorities_ [ current_priority_ ] - > UpdateXdsPickerLocked ( ) ;
}
void XdsLb : : PriorityList : : MaybeCreateLocalityMapLocked ( uint32_t priority ) {
void XdsLb : : MaybeCreateLocalityMapLocked ( uint32_t priority ) {
// Exhausted priorities in the update.
if ( ! priority_list_update ( ) . Contains ( priority ) ) return ;
auto new_locality_map = new LocalityMap (
xds_policy_ - > Ref ( DEBUG_LOCATION , " XdsLb+ LocalityMap" ) , priority ) ;
if ( ! priority_list_update_ . Contains ( priority ) ) return ;
auto new_locality_map =
new LocalityMap ( Ref ( DEBUG_LOCATION , " LocalityMap " ) , priority ) ;
priorities_ . emplace_back ( OrphanablePtr < LocalityMap > ( new_locality_map ) ) ;
new_locality_map - > UpdateLocked ( * priority_list_update ( ) . Find ( priority ) ) ;
new_locality_map - > UpdateLocked ( * priority_list_update_ . Find ( priority ) ) ;
}
void XdsLb : : PriorityList : : FailoverOnConnectionFailureLocked ( ) {
void XdsLb : : FailoverOnConnectionFailureLocked ( ) {
const uint32_t failed_priority = LowestPriority ( ) ;
// If we're failing over from the lowest priority, report TRANSIENT_FAILURE.
if ( failed_priority = = priority_list_update ( ) . LowestPriority ( ) ) {
if ( failed_priority = = priority_list_update_ . LowestPriority ( ) ) {
UpdateXdsPickerLocked ( ) ;
}
MaybeCreateLocalityMapLocked ( failed_priority + 1 ) ;
}
void XdsLb : : PriorityList : : FailoverOnDisconnectionLocked (
uint32_t failed_priority ) {
void XdsLb : : FailoverOnDisconnectionLocked ( uint32_t failed_priority ) {
current_priority_ = UINT32_MAX ;
for ( uint32_t next_priority = failed_priority + 1 ;
next_priority < = priority_list_update ( ) . LowestPriority ( ) ;
next_priority < = priority_list_update_ . LowestPriority ( ) ;
+ + next_priority ) {
if ( ! Contains ( next_priority ) ) {
MaybeCreateLocalityMapLocked ( next_priority ) ;
@ -1111,17 +1095,17 @@ void XdsLb::PriorityList::FailoverOnDisconnectionLocked(
}
}
void XdsLb : : PriorityList : : SwitchToHigherPriorityLocked ( uint32_t priority ) {
void XdsLb : : SwitchToHigherPriorityLocked ( uint32_t priority ) {
current_priority_ = priority ;
DeactivatePrioritiesLowerThan ( current_priority_ ) ;
UpdateXdsPickerLocked ( ) ;
}
void XdsLb : : PriorityList : : DeactivatePrioritiesLowerThan ( uint32_t priority ) {
void XdsLb : : DeactivatePrioritiesLowerThan ( uint32_t priority ) {
if ( priorities_ . empty ( ) ) return ;
// Deactivate the locality maps from the lowest priority.
for ( uint32_t p = LowestPriority ( ) ; p > priority ; - - p ) {
if ( xds_policy_ - > locality_retention_interval_ms_ = = 0 ) {
if ( locality_retention_interval_ms_ = = 0 ) {
priorities_ . pop_back ( ) ;
} else {
priorities_ [ p ] - > DeactivateLocked ( ) ;
@ -1129,8 +1113,7 @@ void XdsLb::PriorityList::DeactivatePrioritiesLowerThan(uint32_t priority) {
}
}
OrphanablePtr < XdsLb : : PriorityList : : LocalityMap : : Locality >
XdsLb : : PriorityList : : ExtractLocalityLocked (
OrphanablePtr < XdsLb : : LocalityMap : : Locality > XdsLb : : ExtractLocalityLocked (
const RefCountedPtr < XdsLocalityName > & name , uint32_t exclude_priority ) {
for ( uint32_t priority = 0 ; priority < priorities_ . size ( ) ; + + priority ) {
if ( priority = = exclude_priority ) continue ;
@ -1142,17 +1125,16 @@ XdsLb::PriorityList::ExtractLocalityLocked(
}
//
// XdsLb::PriorityList:: LocalityMap
// XdsLb::LocalityMap
//
XdsLb : : PriorityList : : LocalityMap : : LocalityMap ( RefCountedPtr < XdsLb > xds_policy ,
XdsLb : : LocalityMap : : LocalityMap ( RefCountedPtr < XdsLb > xds_policy ,
uint32_t priority )
: xds_policy_ ( std : : move ( xds_policy ) ) , priority_ ( priority ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_xds_trace ) ) {
gpr_log ( GPR_INFO , " [xdslb %p] Creating priority % " PRIu32 ,
xds_policy_ . get ( ) , priority_ ) ;
}
GRPC_CLOSURE_INIT ( & on_failover_timer_ , OnFailoverTimer , this ,
grpc_schedule_on_exec_ctx ) ;
// Start the failover timer.
@ -1165,12 +1147,13 @@ XdsLb::PriorityList::LocalityMap::LocalityMap(RefCountedPtr<XdsLb> xds_policy,
// This is the first locality map ever created, report CONNECTING.
if ( priority_ = = 0 ) {
xds_policy_ - > channel_control_helper ( ) - > UpdateState (
GRPC_CHANNEL_CONNECTING , MakeUnique < QueuePicker > ( xds_policy_ - > Ref (
DEBUG_LOCATION , " QueuePicker " ) ) ) ;
GRPC_CHANNEL_CONNECTING ,
grpc_core : : MakeUnique < QueuePicker > (
xds_policy_ - > Ref ( DEBUG_LOCATION , " QueuePicker " ) ) ) ;
}
}
void XdsLb : : PriorityList : : LocalityMap : : UpdateLocked (
void XdsLb : : LocalityMap : : UpdateLocked (
const XdsPriorityListUpdate : : LocalityMap & locality_map_update ) {
if ( xds_policy_ - > shutting_down_ ) return ;
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_xds_trace ) ) {
@ -1202,7 +1185,7 @@ void XdsLb::PriorityList::LocalityMap::UpdateLocked(
OrphanablePtr < Locality > & locality = localities_ [ name ] ;
if ( locality = = nullptr ) {
// Move from another locality map if possible.
locality = priority_list ( ) - > ExtractLocalityLocked ( name , priority_ ) ;
locality = xds_policy_ - > ExtractLocalityLocked ( name , priority_ ) ;
if ( locality ! = nullptr ) {
locality - > set_locality_map (
Ref ( DEBUG_LOCATION , " LocalityMap+Locality_move " ) ) ;
@ -1218,11 +1201,11 @@ void XdsLb::PriorityList::LocalityMap::UpdateLocked(
}
}
void XdsLb : : PriorityList : : LocalityMap : : ResetBackoffLocked ( ) {
void XdsLb : : LocalityMap : : ResetBackoffLocked ( ) {
for ( auto & p : localities_ ) p . second - > ResetBackoffLocked ( ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : UpdateXdsPickerLocked ( ) {
void XdsLb : : LocalityMap : : UpdateXdsPickerLocked ( ) {
// Construct a new xds picker which maintains a map of all locality pickers
// that are ready. Each locality is represented by a portion of the range
// proportional to its weight, such that the total range is the sum of the
@ -1239,13 +1222,14 @@ void XdsLb::PriorityList::LocalityMap::UpdateXdsPickerLocked() {
picker_list . push_back ( std : : make_pair ( end , locality - > picker_wrapper ( ) ) ) ;
}
xds_policy ( ) - > channel_control_helper ( ) - > UpdateState (
GRPC_CHANNEL_READY , grpc_core : : MakeUnique < LocalityPicker > (
xds_policy_ - > Ref ( DEBUG_LOCATION , " XdsLb+Picker " ) ,
GRPC_CHANNEL_READY ,
grpc_core : : MakeUnique < LocalityPicker > (
xds_policy_ - > Ref ( DEBUG_LOCATION , " LocalityPicker " ) ,
std : : move ( picker_list ) ) ) ;
}
OrphanablePtr < XdsLb : : PriorityList : : LocalityMap : : Locality >
XdsLb : : PriorityList : : LocalityMap : : ExtractLocalityLocked (
OrphanablePtr < XdsLb : : LocalityMap : : Locality >
XdsLb : : LocalityMap : : ExtractLocalityLocked (
const RefCountedPtr < XdsLocalityName > & name ) {
for ( auto iter = localities_ . begin ( ) ; iter ! = localities_ . end ( ) ; + + iter ) {
const auto & name_in_map = iter - > first ;
@ -1258,7 +1242,7 @@ XdsLb::PriorityList::LocalityMap::ExtractLocalityLocked(
return nullptr ;
}
void XdsLb : : PriorityList : : LocalityMap : : DeactivateLocked ( ) {
void XdsLb : : LocalityMap : : DeactivateLocked ( ) {
// If already deactivated, don't do it again.
if ( delayed_removal_timer_callback_pending_ ) return ;
MaybeCancelFailoverTimerLocked ( ) ;
@ -1279,24 +1263,24 @@ void XdsLb::PriorityList::LocalityMap::DeactivateLocked() {
delayed_removal_timer_callback_pending_ = true ;
}
bool XdsLb : : PriorityList : : LocalityMap : : MaybeReactivateLocked ( ) {
bool XdsLb : : LocalityMap : : MaybeReactivateLocked ( ) {
// Don't reactivate a priority that is not higher than the current one.
if ( priority_ > = priority_list ( ) - > current_priority ( ) ) return false ;
if ( priority_ > = xds_policy_ - > current_priority_ ) return false ;
// Reactivate this priority by cancelling deletion timer.
if ( delayed_removal_timer_callback_pending_ ) {
grpc_timer_cancel ( & delayed_removal_timer_ ) ;
}
// Switch to this higher priority if it's READY.
if ( connectivity_state_ ! = GRPC_CHANNEL_READY ) return false ;
priority_list ( ) - > SwitchToHigherPriorityLocked ( priority_ ) ;
xds_policy_ - > SwitchToHigherPriorityLocked ( priority_ ) ;
return true ;
}
void XdsLb : : PriorityList : : LocalityMap : : MaybeCancelFailoverTimerLocked ( ) {
void XdsLb : : LocalityMap : : MaybeCancelFailoverTimerLocked ( ) {
if ( failover_timer_callback_pending_ ) grpc_timer_cancel ( & failover_timer_ ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : Orphan ( ) {
void XdsLb : : LocalityMap : : Orphan ( ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_xds_trace ) ) {
gpr_log ( GPR_INFO , " [xdslb %p] Priority % " PRIu32 " orphaned. " , xds_policy ( ) ,
priority_ ) ;
@ -1309,11 +1293,11 @@ void XdsLb::PriorityList::LocalityMap::Orphan() {
Unref ( DEBUG_LOCATION , " LocalityMap+Orphan " ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : OnLocalityStateUpdateLocked ( ) {
void XdsLb : : LocalityMap : : OnLocalityStateUpdateLocked ( ) {
UpdateConnectivityStateLocked ( ) ;
// Ignore priorities not in priority_list_update.
if ( ! priority_list_update ( ) . Contains ( priority_ ) ) return ;
const uint32_t current_priority = priority_list ( ) - > current_priority ( ) ;
const uint32_t current_priority = xds_policy_ - > current_priority_ ;
// Ignore lower-than-current priorities.
if ( priority_ > current_priority ) return ;
// Maybe update fallback state.
@ -1327,13 +1311,13 @@ void XdsLb::PriorityList::LocalityMap::OnLocalityStateUpdateLocked() {
if ( connectivity_state_ = = GRPC_CHANNEL_READY ) {
MaybeCancelFailoverTimerLocked ( ) ;
// If a higher-than-current priority becomes READY, switch to use it.
priority_list ( ) - > SwitchToHigherPriorityLocked ( priority_ ) ;
xds_policy_ - > SwitchToHigherPriorityLocked ( priority_ ) ;
} else if ( connectivity_state_ = = GRPC_CHANNEL_TRANSIENT_FAILURE ) {
// If a higher-than-current priority becomes TRANSIENT_FAILURE, only
// handle it if it's the priority that is still in failover timeout.
if ( failover_timer_callback_pending_ ) {
MaybeCancelFailoverTimerLocked ( ) ;
priority_list ( ) - > FailoverOnConnectionFailureLocked ( ) ;
xds_policy_ - > FailoverOnConnectionFailureLocked ( ) ;
}
}
return ;
@ -1341,7 +1325,7 @@ void XdsLb::PriorityList::LocalityMap::OnLocalityStateUpdateLocked() {
// Update is for current priority.
if ( connectivity_state_ ! = GRPC_CHANNEL_READY ) {
// Fail over if it's no longer READY.
priority_list ( ) - > FailoverOnDisconnectionLocked ( priority_ ) ;
xds_policy_ - > FailoverOnDisconnectionLocked ( priority_ ) ;
}
// At this point, one of the following things has happened to the current
// priority.
@ -1349,10 +1333,10 @@ void XdsLb::PriorityList::LocalityMap::OnLocalityStateUpdateLocked() {
// 2. It changed to a lower priority due to failover.
// 3. It became invalid because failover didn't yield a READY priority.
// In any case, update the xds picker.
priority_list ( ) - > UpdateXdsPickerLocked ( ) ;
xds_policy_ - > UpdateXdsPickerLocked ( ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : UpdateConnectivityStateLocked ( ) {
void XdsLb : : LocalityMap : : UpdateConnectivityStateLocked ( ) {
size_t num_ready = 0 ;
size_t num_connecting = 0 ;
size_t num_idle = 0 ;
@ -1400,8 +1384,7 @@ void XdsLb::PriorityList::LocalityMap::UpdateConnectivityStateLocked() {
}
}
void XdsLb : : PriorityList : : LocalityMap : : OnDelayedRemovalTimer (
void * arg , grpc_error * error ) {
void XdsLb : : LocalityMap : : OnDelayedRemovalTimer ( void * arg , grpc_error * error ) {
LocalityMap * self = static_cast < LocalityMap * > ( arg ) ;
self - > xds_policy_ - > combiner ( ) - > Run (
GRPC_CLOSURE_INIT ( & self - > on_delayed_removal_timer_ ,
@ -1409,14 +1392,13 @@ void XdsLb::PriorityList::LocalityMap::OnDelayedRemovalTimer(
GRPC_ERROR_REF ( error ) ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : OnDelayedRemovalTimerLocked (
void * arg , grpc_error * error ) {
void XdsLb : : LocalityMap : : OnDelayedRemovalTimerLocked ( void * arg ,
grpc_error * error ) {
LocalityMap * self = static_cast < LocalityMap * > ( arg ) ;
self - > delayed_removal_timer_callback_pending_ = false ;
if ( error = = GRPC_ERROR_NONE & & ! self - > xds_policy_ - > shutting_down_ ) {
auto * priority_list = self - > priority_list ( ) ;
const bool keep = self - > priority_list_update ( ) . Contains ( self - > priority_ ) & &
self - > priority_ < = priority_list - > current_priority ( ) ;
self - > priority_ < = self - > xds_policy_ - > current_priority_ ;
if ( ! keep ) {
// This check is to make sure we always delete the locality maps from
// the lowest priority even if the closures of the back-to-back timers
@ -1425,8 +1407,8 @@ void XdsLb::PriorityList::LocalityMap::OnDelayedRemovalTimerLocked(
// deactivated locality maps when out-of-order closures are run.
// TODO(juanlishen): Check the timer implementation to see if this
// defense is necessary.
if ( self - > priority_ = = priority_list - > LowestPriority ( ) ) {
priority_list - > priorities_ . pop_back ( ) ;
if ( self - > priority_ = = self - > xds_policy_ - > LowestPriority ( ) ) {
self - > xds_policy_ - > priorities_ . pop_back ( ) ;
} else {
gpr_log ( GPR_ERROR ,
" [xdslb %p] Priority % " PRIu32
@ -1439,8 +1421,7 @@ void XdsLb::PriorityList::LocalityMap::OnDelayedRemovalTimerLocked(
self - > Unref ( DEBUG_LOCATION , " LocalityMap+timer " ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : OnFailoverTimer ( void * arg ,
grpc_error * error ) {
void XdsLb : : LocalityMap : : OnFailoverTimer ( void * arg , grpc_error * error ) {
LocalityMap * self = static_cast < LocalityMap * > ( arg ) ;
self - > xds_policy_ - > combiner ( ) - > Run (
GRPC_CLOSURE_INIT ( & self - > on_failover_timer_ , OnFailoverTimerLocked , self ,
@ -1448,22 +1429,20 @@ void XdsLb::PriorityList::LocalityMap::OnFailoverTimer(void* arg,
GRPC_ERROR_REF ( error ) ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : OnFailoverTimerLocked (
void * arg , grpc_error * error ) {
void XdsLb : : LocalityMap : : OnFailoverTimerLocked ( void * arg , grpc_error * error ) {
LocalityMap * self = static_cast < LocalityMap * > ( arg ) ;
self - > failover_timer_callback_pending_ = false ;
if ( error = = GRPC_ERROR_NONE & & ! self - > xds_policy_ - > shutting_down_ ) {
self - > priority_list ( ) - > FailoverOnConnectionFailureLocked ( ) ;
self - > xds_policy_ - > FailoverOnConnectionFailureLocked ( ) ;
}
self - > Unref ( DEBUG_LOCATION , " LocalityMap+OnFailoverTimerLocked " ) ;
}
//
// XdsLb::PriorityList:: LocalityMap::Locality
// XdsLb::LocalityMap::Locality
//
XdsLb : : PriorityList : : LocalityMap : : Locality : : Locality (
RefCountedPtr < LocalityMap > locality_map ,
XdsLb : : LocalityMap : : Locality : : Locality ( RefCountedPtr < LocalityMap > locality_map ,
RefCountedPtr < XdsLocalityName > name )
: locality_map_ ( std : : move ( locality_map ) ) , name_ ( std : : move ( name ) ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_xds_trace ) ) {
@ -1472,7 +1451,7 @@ XdsLb::PriorityList::LocalityMap::Locality::Locality(
}
}
XdsLb : : PriorityList : : LocalityMap : : Locality : : ~ Locality ( ) {
XdsLb : : LocalityMap : : Locality : : ~ Locality ( ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_xds_trace ) ) {
gpr_log ( GPR_INFO , " [xdslb %p] Locality %p %s: destroying locality " ,
xds_policy ( ) , this , name_ - > AsHumanReadableString ( ) ) ;
@ -1480,8 +1459,7 @@ XdsLb::PriorityList::LocalityMap::Locality::~Locality() {
locality_map_ . reset ( DEBUG_LOCATION , " Locality " ) ;
}
grpc_channel_args *
XdsLb : : PriorityList : : LocalityMap : : Locality : : CreateChildPolicyArgsLocked (
grpc_channel_args * XdsLb : : LocalityMap : : Locality : : CreateChildPolicyArgsLocked (
const grpc_channel_args * args_in ) {
const grpc_arg args_to_add [ ] = {
// A channel arg indicating if the target is a backend inferred from a
@ -1499,7 +1477,7 @@ XdsLb::PriorityList::LocalityMap::Locality::CreateChildPolicyArgsLocked(
}
OrphanablePtr < LoadBalancingPolicy >
XdsLb : : PriorityList : : LocalityMap : : Locality : : CreateChildPolicyLocked (
XdsLb : : LocalityMap : : Locality : : CreateChildPolicyLocked (
const char * name , const grpc_channel_args * args ) {
Helper * helper = new Helper ( this - > Ref ( DEBUG_LOCATION , " Helper " ) ) ;
LoadBalancingPolicy : : Args lb_policy_args ;
@ -1531,8 +1509,8 @@ XdsLb::PriorityList::LocalityMap::Locality::CreateChildPolicyLocked(
return lb_policy ;
}
void XdsLb : : PriorityList : : LocalityMap : : Locality : : UpdateLocked (
uint32_t locality_weight , ServerAddressList serverlist ) {
void XdsLb : : LocalityMap : : Locality : : UpdateLocked ( uint32_t locality_weight ,
ServerAddressList serverlist ) {
if ( xds_policy ( ) - > shutting_down_ ) return ;
// Update locality weight.
weight_ = locality_weight ;
@ -1641,7 +1619,7 @@ void XdsLb::PriorityList::LocalityMap::Locality::UpdateLocked(
policy_to_update - > UpdateLocked ( std : : move ( update_args ) ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : Locality : : ShutdownLocked ( ) {
void XdsLb : : LocalityMap : : Locality : : ShutdownLocked ( ) {
if ( GRPC_TRACE_FLAG_ENABLED ( grpc_lb_xds_trace ) ) {
gpr_log ( GPR_INFO , " [xdslb %p] Locality %p %s: shutting down locality " ,
xds_policy ( ) , this , name_ - > AsHumanReadableString ( ) ) ;
@ -1666,19 +1644,19 @@ void XdsLb::PriorityList::LocalityMap::Locality::ShutdownLocked() {
shutdown_ = true ;
}
void XdsLb : : PriorityList : : LocalityMap : : Locality : : ResetBackoffLocked ( ) {
void XdsLb : : LocalityMap : : Locality : : ResetBackoffLocked ( ) {
child_policy_ - > ResetBackoffLocked ( ) ;
if ( pending_child_policy_ ! = nullptr ) {
pending_child_policy_ - > ResetBackoffLocked ( ) ;
}
}
void XdsLb : : PriorityList : : LocalityMap : : Locality : : Orphan ( ) {
void XdsLb : : LocalityMap : : Locality : : Orphan ( ) {
ShutdownLocked ( ) ;
Unref ( ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : Locality : : DeactivateLocked ( ) {
void XdsLb : : LocalityMap : : Locality : : DeactivateLocked ( ) {
// If already deactivated, don't do that again.
if ( weight_ = = 0 ) return ;
// Set the locality weight to 0 so that future xds picker won't contain this
@ -1695,8 +1673,8 @@ void XdsLb::PriorityList::LocalityMap::Locality::DeactivateLocked() {
delayed_removal_timer_callback_pending_ = true ;
}
void XdsLb : : PriorityList : : LocalityMap : : Locality : : OnDelayedRemovalTimer (
void * arg , grpc_error * error ) {
void XdsLb : : LocalityMap : : Locality : : OnDelayedRemovalTimer ( void * arg ,
grpc_error * error ) {
Locality * self = static_cast < Locality * > ( arg ) ;
self - > xds_policy ( ) - > combiner ( ) - > Run (
GRPC_CLOSURE_INIT ( & self - > on_delayed_removal_timer_ ,
@ -1704,7 +1682,7 @@ void XdsLb::PriorityList::LocalityMap::Locality::OnDelayedRemovalTimer(
GRPC_ERROR_REF ( error ) ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : Locality : : OnDelayedRemovalTimerLocked (
void XdsLb : : LocalityMap : : Locality : : OnDelayedRemovalTimerLocked (
void * arg , grpc_error * error ) {
Locality * self = static_cast < Locality * > ( arg ) ;
self - > delayed_removal_timer_callback_pending_ = false ;
@ -1718,20 +1696,18 @@ void XdsLb::PriorityList::LocalityMap::Locality::OnDelayedRemovalTimerLocked(
// XdsLb::Locality::Helper
//
bool XdsLb : : PriorityList : : LocalityMap : : Locality : : Helper : : CalledByPendingChild ( )
const {
bool XdsLb : : LocalityMap : : Locality : : Helper : : CalledByPendingChild ( ) const {
GPR_ASSERT ( child_ ! = nullptr ) ;
return child_ = = locality_ - > pending_child_policy_ . get ( ) ;
}
bool XdsLb : : PriorityList : : LocalityMap : : Locality : : Helper : : CalledByCurrentChild ( )
const {
bool XdsLb : : LocalityMap : : Locality : : Helper : : CalledByCurrentChild ( ) const {
GPR_ASSERT ( child_ ! = nullptr ) ;
return child_ = = locality_ - > child_policy_ . get ( ) ;
}
RefCountedPtr < SubchannelInterface >
XdsLb : : PriorityList : : LocalityMap : : Locality : : Helper : : CreateSubchannel (
XdsLb : : LocalityMap : : Locality : : Helper : : CreateSubchannel (
const grpc_channel_args & args ) {
if ( locality_ - > xds_policy ( ) - > shutting_down_ | |
( ! CalledByPendingChild ( ) & & ! CalledByCurrentChild ( ) ) ) {
@ -1741,7 +1717,7 @@ XdsLb::PriorityList::LocalityMap::Locality::Helper::CreateSubchannel(
args ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : Locality : : Helper : : UpdateState (
void XdsLb : : LocalityMap : : Locality : : Helper : : UpdateState (
grpc_connectivity_state state , std : : unique_ptr < SubchannelPicker > picker ) {
if ( locality_ - > xds_policy ( ) - > shutting_down_ ) return ;
// If this request is from the pending child policy, ignore it until
@ -1777,8 +1753,8 @@ void XdsLb::PriorityList::LocalityMap::Locality::Helper::UpdateState(
locality_ - > locality_map_ - > OnLocalityStateUpdateLocked ( ) ;
}
void XdsLb : : PriorityList : : LocalityMap : : Locality : : Helper : : AddTraceEvent (
TraceSeverity severity , StringView message ) {
void XdsLb : : LocalityMap : : Locality : : Helper : : AddTraceEvent ( TraceSeverity severity ,
StringView message ) {
if ( locality_ - > xds_policy ( ) - > shutting_down_ | |
( ! CalledByPendingChild ( ) & & ! CalledByCurrentChild ( ) ) ) {
return ;
@ -1801,9 +1777,9 @@ class XdsFactory : public LoadBalancingPolicyFactory {
const char * name ( ) const override { return kXds ; }
RefCountedPtr < LoadBalancingPolicy : : Config > ParseLoadBalancingConfig (
const grpc_json * json , grpc_error * * error ) const override {
const Json & json , grpc_error * * error ) const override {
GPR_DEBUG_ASSERT ( error ! = nullptr & & * error = = GRPC_ERROR_NONE ) ;
if ( json = = nullptr ) {
if ( json . type ( ) = = Json : : Type : : JSON_NULL ) {
// xds was mentioned as a policy in the deprecated loadBalancingPolicy
// field or in the client API.
* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING (
@ -1811,69 +1787,69 @@ class XdsFactory : public LoadBalancingPolicyFactory {
" Please use loadBalancingConfig field of service config instead. " ) ;
return nullptr ;
}
GPR_DEBUG_ASSERT ( strcmp ( json - > key , name ( ) ) = = 0 ) ;
InlinedVector < grpc_error * , 3 > error_list ;
std : : vector < grpc_error * > error_list ;
// Child policy.
RefCountedPtr < LoadBalancingPolicy : : Config > child_policy ;
RefCountedPtr < LoadBalancingPolicy : : Config > fallback_policy ;
const char * eds_service_name = nullptr ;
const char * lrs_load_reporting_server_name = nullptr ;
for ( const grpc_json * field = json - > child ; field ! = nullptr ;
field = field - > next ) {
if ( field - > key = = nullptr ) continue ;
if ( strcmp ( field - > key , " childPolicy " ) = = 0 ) {
if ( child_policy ! = nullptr ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:childPolicy error:Duplicate entry " ) ) ;
}
auto it = json . object_value ( ) . find ( " childPolicy " ) ;
if ( it ! = json . object_value ( ) . end ( ) ) {
grpc_error * parse_error = GRPC_ERROR_NONE ;
child_policy = LoadBalancingPolicyRegistry : : ParseLoadBalancingConfig (
fiel d , & parse_error ) ;
it - > second , & parse_error ) ;
if ( child_policy = = nullptr ) {
GPR_DEBUG_ASSERT ( parse_error ! = GRPC_ERROR_NONE ) ;
error_list . push_back ( parse_error ) ;
std : : vector < grpc_error * > child_errors ;
child_errors . push_back ( parse_error ) ;
error_list . push_back (
GRPC_ERROR_CREATE_FROM_VECTOR ( " field:childPolicy " , & child_errors ) ) ;
}
} else if ( strcmp ( field - > key , " fallbackPolicy " ) = = 0 ) {
if ( fallback_policy ! = nullptr ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:fallbackPolicy error:Duplicate entry " ) ) ;
}
// Fallback policy.
RefCountedPtr < LoadBalancingPolicy : : Config > fallback_policy ;
it = json . object_value ( ) . find ( " fallbackPolicy " ) ;
if ( it ! = json . object_value ( ) . end ( ) ) {
grpc_error * parse_error = GRPC_ERROR_NONE ;
fallback_policy = LoadBalancingPolicyRegistry : : ParseLoadBalancingConfig (
field , & parse_error ) ;
it - > secon d, & parse_error ) ;
if ( fallback_policy = = nullptr ) {
GPR_DEBUG_ASSERT ( parse_error ! = GRPC_ERROR_NONE ) ;
error_list . push_back ( parse_error ) ;
std : : vector < grpc_error * > child_errors ;
child_errors . push_back ( parse_error ) ;
error_list . push_back ( GRPC_ERROR_CREATE_FROM_VECTOR (
" field:fallbackPolicy " , & child_errors ) ) ;
}
} else if ( strcmp ( field - > key , " edsServiceName " ) = = 0 ) {
if ( eds_service_name ! = nullptr ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:edsServiceName error:Duplicate entry " ) ) ;
}
if ( field - > type ! = GRPC_JSON_STRING ) {
// EDS service name.
const char * eds_service_name = nullptr ;
it = json . object_value ( ) . find ( " edsServiceName " ) ;
if ( it ! = json . object_value ( ) . end ( ) ) {
if ( it - > second . type ( ) ! = Json : : Type : : STRING ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:edsServiceName error:type should be string " ) ) ;
continue ;
} else {
eds_service_name = it - > second . string_value ( ) . c_str ( ) ;
}
eds_service_name = field - > value ;
} else if ( strcmp ( field - > key , " lrsLoadReportingServerName " ) = = 0 ) {
if ( lrs_load_reporting_server_name ! = nullptr ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:lrsLoadReportingServerName error:Duplicate entry " ) ) ;
}
if ( field - > type ! = GRPC_JSON_STRING ) {
// LRS load reporting server name.
const char * lrs_load_reporting_server_name = nullptr ;
it = json . object_value ( ) . find ( " lrsLoadReportingServerName " ) ;
if ( it ! = json . object_value ( ) . end ( ) ) {
if ( it - > second . type ( ) ! = Json : : Type : : STRING ) {
error_list . push_back ( GRPC_ERROR_CREATE_FROM_STATIC_STRING (
" field:lrsLoadReportingServerName error:type should be string " ) ) ;
continue ;
}
lrs_load_reporting_server_name = field - > value ;
} else {
lrs_load_reporting_server_name = it - > second . string_value ( ) . c_str ( ) ;
}
}
if ( error_list . empty ( ) ) {
return MakeRefCounted < ParsedXdsConfig > (
Optional < std : : string > optional_lrs_load_reporting_server_name ;
if ( lrs_load_reporting_server_name ! = nullptr ) {
optional_lrs_load_reporting_server_name . emplace (
std : : string ( lrs_load_reporting_server_name ) ) ;
}
return MakeRefCounted < XdsConfig > (
std : : move ( child_policy ) , std : : move ( fallback_policy ) ,
grpc_core : : UniquePtr < char > ( gpr_strdup ( eds_service_name ) ) ,
grpc_core : : UniquePtr < char > (
gpr_strdup ( lrs_load_reporting_server_name ) ) ) ;
eds_service_name = = nullptr ? " " : eds_service_name ,
std : : move ( optional_lrs_load_reporting_server_name ) ) ;
} else {
* error = GRPC_ERROR_CREATE_FROM_VECTOR ( " Xds Parser " , & error_list ) ;
return nullptr ;
Zhanghui Mao
5 years ago
committed by
GitHub