@ -125,8 +125,7 @@ class GrpcLb : public LoadBalancingPolicy {
const char * name ( ) const override { return kGrpclb ; } const char * name ( ) const override { return kGrpclb ; }
void UpdateLocked ( const grpc_channel_args & args , void UpdateLocked ( UpdateArgs args ) override ;
RefCountedPtr < Config > lb_config ) override ;
void ResetBackoffLocked ( ) override ; void ResetBackoffLocked ( ) override ;
void FillChildRefsForChannelz ( void FillChildRefsForChannelz (
channelz : : ChildRefsList * child_subchannels , channelz : : ChildRefsList * child_subchannels ,
@ -295,7 +294,8 @@ class GrpcLb : public LoadBalancingPolicy {
void ShutdownLocked ( ) override ; void ShutdownLocked ( ) override ;
// Helper functions used in UpdateLocked().
// Helper functions used in UpdateLocked().
void ProcessChannelArgsLocked ( const grpc_channel_args & args ) ; void ProcessAddressesAndChannelArgsLocked ( const ServerAddressList & addresses ,
const grpc_channel_args & args ) ;
void ParseLbConfig ( Config * grpclb_config ) ; void ParseLbConfig ( Config * grpclb_config ) ;
static void OnBalancerChannelConnectivityChangedLocked ( void * arg , static void OnBalancerChannelConnectivityChangedLocked ( void * arg ,
grpc_error * error ) ; grpc_error * error ) ;
@ -311,7 +311,8 @@ class GrpcLb : public LoadBalancingPolicy {
static void OnBalancerCallRetryTimerLocked ( void * arg , grpc_error * error ) ; static void OnBalancerCallRetryTimerLocked ( void * arg , grpc_error * error ) ;
// Methods for dealing with the child policy.
// Methods for dealing with the child policy.
grpc_channel_args * CreateChildPolicyArgsLocked ( ) ; grpc_channel_args * CreateChildPolicyArgsLocked (
bool is_backend_from_grpclb_load_balancer ) ;
OrphanablePtr < LoadBalancingPolicy > CreateChildPolicyLocked ( OrphanablePtr < LoadBalancingPolicy > CreateChildPolicyLocked (
const char * name , const grpc_channel_args * args ) ; const char * name , const grpc_channel_args * args ) ;
void CreateOrUpdateChildPolicyLocked ( ) ; void CreateOrUpdateChildPolicyLocked ( ) ;
@ -1204,7 +1205,6 @@ grpc_channel_args* BuildBalancerChannelArgs(
const ServerAddressList & addresses , const ServerAddressList & addresses ,
FakeResolverResponseGenerator * response_generator , FakeResolverResponseGenerator * response_generator ,
const grpc_channel_args * args ) { const grpc_channel_args * args ) {
ServerAddressList balancer_addresses = ExtractBalancerAddresses ( addresses ) ;
// Channel args to remove.
// Channel args to remove.
static const char * args_to_remove [ ] = { static const char * args_to_remove [ ] = {
// LB policy name, since we want to use the default (pick_first) in
// LB policy name, since we want to use the default (pick_first) in
@ -1217,15 +1217,6 @@ grpc_channel_args* BuildBalancerChannelArgs(
// the LB channel than for the parent channel. The client channel
// the LB channel than for the parent channel. The client channel
// factory will re-add this arg with the right value.
// factory will re-add this arg with the right value.
GRPC_ARG_SERVER_URI , GRPC_ARG_SERVER_URI ,
// The resolved addresses, which will be generated by the name resolver
// used in the LB channel. Note that the LB channel will use the fake
// resolver, so this won't actually generate a query to DNS (or some
// other name service). However, the addresses returned by the fake
// resolver will have is_balancer=false, whereas our own addresses have
// is_balancer=true. We need the LB channel to return addresses with
// is_balancer=false so that it does not wind up recursively using the
// grpclb LB policy.
GRPC_ARG_SERVER_ADDRESS_LIST ,
// The fake resolver response generator, because we are replacing it
// The fake resolver response generator, because we are replacing it
// with the one from the grpclb policy, used to propagate updates to
// with the one from the grpclb policy, used to propagate updates to
// the LB channel.
// the LB channel.
@ -1241,10 +1232,6 @@ grpc_channel_args* BuildBalancerChannelArgs(
} ; } ;
// Channel args to add.
// Channel args to add.
const grpc_arg args_to_add [ ] = { const grpc_arg args_to_add [ ] = {
// New address list.
// Note that we pass these in both when creating the LB channel
// and via the fake resolver. The latter is what actually gets used.
CreateServerAddressListChannelArg ( & balancer_addresses ) ,
// The fake resolver response generator, which we use to inject
// The fake resolver response generator, which we use to inject
// address updates into the LB channel.
// address updates into the LB channel.
grpc_core : : FakeResolverResponseGenerator : : MakeChannelArg ( grpc_core : : FakeResolverResponseGenerator : : MakeChannelArg (
@ -1262,7 +1249,7 @@ grpc_channel_args* BuildBalancerChannelArgs(
args , args_to_remove , GPR_ARRAY_SIZE ( args_to_remove ) , args_to_add , args , args_to_remove , GPR_ARRAY_SIZE ( args_to_remove ) , args_to_add ,
GPR_ARRAY_SIZE ( args_to_add ) ) ; GPR_ARRAY_SIZE ( args_to_add ) ) ;
// Make any necessary modifications for security.
// Make any necessary modifications for security.
return grpc_lb_policy_grpclb_modify_lb_channel_args ( new_args ) ; return grpc_lb_policy_grpclb_modify_lb_channel_args ( addresses , new_args ) ;
} }
//
//
@ -1388,11 +1375,10 @@ void GrpcLb::FillChildRefsForChannelz(
} }
} }
void GrpcLb : : UpdateLocked ( const grpc_channel_args & args , void GrpcLb : : UpdateLocked ( UpdateArgs args ) {
RefCountedPtr < Config > lb_config ) {
const bool is_initial_update = lb_channel_ = = nullptr ; const bool is_initial_update = lb_channel_ = = nullptr ;
ParseLbConfig ( lb_ config. get ( ) ) ; ParseLbConfig ( args . config . get ( ) ) ;
ProcessChannelArgsLocked ( args ) ; ProcessAddressesAnd ChannelArgsLocked ( args . addresses , * args . args ) ;
// Update the existing child policy.
// Update the existing child policy.
if ( child_policy_ ! = nullptr ) CreateOrUpdateChildPolicyLocked ( ) ; if ( child_policy_ ! = nullptr ) CreateOrUpdateChildPolicyLocked ( ) ;
// If this is the initial update, start the fallback-at-startup checks
// If this is the initial update, start the fallback-at-startup checks
@ -1442,18 +1428,10 @@ ServerAddressList ExtractBackendAddresses(const ServerAddressList& addresses) {
return backend_addresses ; return backend_addresses ;
} }
void GrpcLb : : ProcessChannelArgsLocked ( const grpc_channel_args & args ) { void GrpcLb : : ProcessAddressesAndChannelArgsLocked (
const ServerAddressList * addresses = FindServerAddressListChannelArg ( & args ) ; const ServerAddressList & addresses , const grpc_channel_args & args ) {
if ( addresses = = nullptr ) {
// Ignore this update.
gpr_log (
GPR_ERROR ,
" [grpclb %p] No valid LB addresses channel arg in update, ignoring. " ,
this ) ;
return ;
}
// Update fallback address list.
// Update fallback address list.
fallback_backend_addresses_ = ExtractBackendAddresses ( * addresses ) ; fallback_backend_addresses_ = ExtractBackendAddresses ( addresses ) ;
// Make sure that GRPC_ARG_LB_POLICY_NAME is set in channel args,
// Make sure that GRPC_ARG_LB_POLICY_NAME is set in channel args,
// since we use this to trigger the client_load_reporting filter.
// since we use this to trigger the client_load_reporting filter.
static const char * args_to_remove [ ] = { GRPC_ARG_LB_POLICY_NAME } ; static const char * args_to_remove [ ] = { GRPC_ARG_LB_POLICY_NAME } ;
@ -1463,8 +1441,9 @@ void GrpcLb::ProcessChannelArgsLocked(const grpc_channel_args& args) {
args_ = grpc_channel_args_copy_and_add_and_remove ( args_ = grpc_channel_args_copy_and_add_and_remove (
& args , args_to_remove , GPR_ARRAY_SIZE ( args_to_remove ) , & new_arg , 1 ) ; & args , args_to_remove , GPR_ARRAY_SIZE ( args_to_remove ) , & new_arg , 1 ) ;
// Construct args for balancer channel.
// Construct args for balancer channel.
grpc_channel_args * lb_channel_args = ServerAddressList balancer_addresses = ExtractBalancerAddresses ( addresses ) ;
BuildBalancerChannelArgs ( * addresses , response_generator_ . get ( ) , & args ) ; grpc_channel_args * lb_channel_args = BuildBalancerChannelArgs (
balancer_addresses , response_generator_ . get ( ) , & args ) ;
// Create balancer channel if needed.
// Create balancer channel if needed.
if ( lb_channel_ = = nullptr ) { if ( lb_channel_ = = nullptr ) {
char * uri_str ; char * uri_str ;
@ -1481,8 +1460,10 @@ void GrpcLb::ProcessChannelArgsLocked(const grpc_channel_args& args) {
} }
// Propagate updates to the LB channel (pick_first) through the fake
// Propagate updates to the LB channel (pick_first) through the fake
// resolver.
// resolver.
response_generator_ - > SetResponse ( lb_channel_args ) ; Resolver : : Result result ;
grpc_channel_args_destroy ( lb_channel_args ) ; result . addresses = std : : move ( balancer_addresses ) ;
result . args = lb_channel_args ;
response_generator_ - > SetResponse ( std : : move ( result ) ) ;
} }
void GrpcLb : : ParseLbConfig ( Config * grpclb_config ) { void GrpcLb : : ParseLbConfig ( Config * grpclb_config ) {
@ -1649,25 +1630,9 @@ void GrpcLb::OnFallbackTimerLocked(void* arg, grpc_error* error) {
// code for interacting with the child policy
// code for interacting with the child policy
//
//
grpc_channel_args * GrpcLb : : CreateChildPolicyArgsLocked ( ) { grpc_channel_args * GrpcLb : : CreateChildPolicyArgsLocked (
ServerAddressList tmp_addresses ; bool is_backend_from_grpclb_load_balancer ) {
ServerAddressList * addresses = & tmp_addresses ; grpc_arg args_to_add [ 2 ] = {
bool is_backend_from_grpclb_load_balancer = false ;
if ( fallback_mode_ ) {
// Note: If fallback backend address list is empty, the child policy
// will go into state TRANSIENT_FAILURE.
addresses = & fallback_backend_addresses_ ;
} else {
tmp_addresses = serverlist_ - > GetServerAddressList (
lb_calld_ = = nullptr ? nullptr : lb_calld_ - > client_stats ( ) ) ;
is_backend_from_grpclb_load_balancer = true ;
}
GPR_ASSERT ( addresses ! = nullptr ) ;
// Replace the server address list in the channel args that we pass down to
// the subchannel.
static const char * keys_to_remove [ ] = { GRPC_ARG_SERVER_ADDRESS_LIST } ;
grpc_arg args_to_add [ 3 ] = {
CreateServerAddressListChannelArg ( addresses ) ,
// A channel arg indicating if the target is a backend inferred from a
// A channel arg indicating if the target is a backend inferred from a
// grpclb load balancer.
// grpclb load balancer.
grpc_channel_arg_integer_create ( grpc_channel_arg_integer_create (
@ -1675,15 +1640,12 @@ grpc_channel_args* GrpcLb::CreateChildPolicyArgsLocked() {
GRPC_ARG_ADDRESS_IS_BACKEND_FROM_GRPCLB_LOAD_BALANCER ) , GRPC_ARG_ADDRESS_IS_BACKEND_FROM_GRPCLB_LOAD_BALANCER ) ,
is_backend_from_grpclb_load_balancer ) , is_backend_from_grpclb_load_balancer ) ,
} ; } ;
size_t num_args_to_add = 2 ; size_t num_args_to_add = 1 ;
if ( is_backend_from_grpclb_load_balancer ) { if ( is_backend_from_grpclb_load_balancer ) {
args_to_add [ 2 ] = grpc_channel_arg_integer_create ( args_to_add [ num_args_to_add + + ] = grpc_channel_arg_integer_create (
const_cast < char * > ( GRPC_ARG_INHIBIT_HEALTH_CHECKING ) , 1 ) ; const_cast < char * > ( GRPC_ARG_INHIBIT_HEALTH_CHECKING ) , 1 ) ;
+ + num_args_to_add ;
} }
return grpc_channel_args_copy_and_add_and_remove ( return grpc_channel_args_copy_and_add ( args_ , args_to_add , num_args_to_add ) ;
args_ , keys_to_remove , GPR_ARRAY_SIZE ( keys_to_remove ) , args_to_add ,
num_args_to_add ) ;
} }
OrphanablePtr < LoadBalancingPolicy > GrpcLb : : CreateChildPolicyLocked ( OrphanablePtr < LoadBalancingPolicy > GrpcLb : : CreateChildPolicyLocked (
@ -1717,8 +1679,25 @@ OrphanablePtr<LoadBalancingPolicy> GrpcLb::CreateChildPolicyLocked(
void GrpcLb : : CreateOrUpdateChildPolicyLocked ( ) { void GrpcLb : : CreateOrUpdateChildPolicyLocked ( ) {
if ( shutting_down_ ) return ; if ( shutting_down_ ) return ;
grpc_channel_args * args = CreateChildPolicyArgsLocked ( ) ; // Construct update args.
GPR_ASSERT ( args ! = nullptr ) ; UpdateArgs update_args ;
bool is_backend_from_grpclb_load_balancer = false ;
if ( fallback_mode_ ) {
// If CreateOrUpdateChildPolicyLocked() is invoked when we haven't
// received any serverlist from the balancer, we use the fallback backends
// returned by the resolver. Note that the fallback backend list may be
// empty, in which case the new round_robin policy will keep the requested
// picks pending.
update_args . addresses = fallback_backend_addresses_ ;
} else {
update_args . addresses = serverlist_ - > GetServerAddressList (
lb_calld_ = = nullptr ? nullptr : lb_calld_ - > client_stats ( ) ) ;
is_backend_from_grpclb_load_balancer = true ;
}
update_args . args =
CreateChildPolicyArgsLocked ( is_backend_from_grpclb_load_balancer ) ;
GPR_ASSERT ( update_args . args ! = nullptr ) ;
update_args . config = child_policy_config_ ;
// If the child policy name changes, we need to create a new child
// If the child policy name changes, we need to create a new child
// policy. When this happens, we leave child_policy_ as-is and store
// policy. When this happens, we leave child_policy_ as-is and store
// the new child policy in pending_child_policy_. Once the new child
// the new child policy in pending_child_policy_. Once the new child
@ -1789,7 +1768,8 @@ void GrpcLb::CreateOrUpdateChildPolicyLocked() {
gpr_log ( GPR_INFO , " [grpclb %p] Creating new %schild policy %s " , this , gpr_log ( GPR_INFO , " [grpclb %p] Creating new %schild policy %s " , this ,
child_policy_ = = nullptr ? " " : " pending " , child_policy_name ) ; child_policy_ = = nullptr ? " " : " pending " , child_policy_name ) ;
} }
auto new_policy = CreateChildPolicyLocked ( child_policy_name , args ) ; auto new_policy =
CreateChildPolicyLocked ( child_policy_name , update_args . args ) ;
// Swap the policy into place.
// Swap the policy into place.
auto & lb_policy = auto & lb_policy =
child_policy_ = = nullptr ? child_policy_ : pending_child_policy_ ; child_policy_ = = nullptr ? child_policy_ : pending_child_policy_ ;
@ -1813,9 +1793,7 @@ void GrpcLb::CreateOrUpdateChildPolicyLocked() {
policy_to_update = = pending_child_policy_ . get ( ) ? " pending " : " " , policy_to_update = = pending_child_policy_ . get ( ) ? " pending " : " " ,
policy_to_update ) ; policy_to_update ) ;
} }
policy_to_update - > UpdateLocked ( * args , child_policy_config_ ) ; policy_to_update - > UpdateLocked ( std : : move ( update_args ) ) ;
// Clean up.
grpc_channel_args_destroy ( args ) ;
} }
//
//
Karthik Ravi Shankar
6 years ago