|
|
|
|
@ -139,7 +139,7 @@ static grpc_error *initial_metadata_add_lb_token( |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void destroy_client_stats(void *arg) { |
|
|
|
|
grpc_grpclb_client_stats_unref(arg); |
|
|
|
|
grpc_grpclb_client_stats_unref((grpc_grpclb_client_stats *)arg); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
typedef struct wrapped_rr_closure_arg { |
|
|
|
|
@ -287,7 +287,7 @@ static void add_pending_ping(pending_ping **root, grpc_closure *notify) { |
|
|
|
|
* glb_lb_policy |
|
|
|
|
*/ |
|
|
|
|
typedef struct rr_connectivity_data rr_connectivity_data; |
|
|
|
|
static const grpc_lb_policy_vtable glb_lb_policy_vtable; |
|
|
|
|
|
|
|
|
|
typedef struct glb_lb_policy { |
|
|
|
|
/** base policy: must be first */ |
|
|
|
|
grpc_lb_policy base; |
|
|
|
|
@ -784,7 +784,7 @@ static void create_rr_locked(grpc_exec_ctx *exec_ctx, glb_lb_policy *glb_policy, |
|
|
|
|
/* Allocate the data for the tracking of the new RR policy's connectivity.
|
|
|
|
|
* It'll be deallocated in glb_rr_connectivity_changed() */ |
|
|
|
|
rr_connectivity_data *rr_connectivity = |
|
|
|
|
gpr_zalloc(sizeof(rr_connectivity_data)); |
|
|
|
|
(rr_connectivity_data *)gpr_zalloc(sizeof(rr_connectivity_data)); |
|
|
|
|
GRPC_CLOSURE_INIT(&rr_connectivity->on_change, |
|
|
|
|
glb_rr_connectivity_changed_locked, rr_connectivity, |
|
|
|
|
grpc_combiner_scheduler(glb_policy->base.combiner)); |
|
|
|
|
@ -924,7 +924,8 @@ static grpc_channel_args *build_lb_channel_args( |
|
|
|
|
grpc_lb_addresses *lb_addresses = |
|
|
|
|
grpc_lb_addresses_create(num_grpclb_addrs, NULL); |
|
|
|
|
grpc_slice_hash_table_entry *targets_info_entries = |
|
|
|
|
gpr_zalloc(sizeof(*targets_info_entries) * num_grpclb_addrs); |
|
|
|
|
(grpc_slice_hash_table_entry *)gpr_zalloc(sizeof(*targets_info_entries) * |
|
|
|
|
num_grpclb_addrs); |
|
|
|
|
|
|
|
|
|
size_t lb_addresses_idx = 0; |
|
|
|
|
for (size_t i = 0; i < addresses->num_addresses; ++i) { |
|
|
|
|
@ -966,97 +967,6 @@ static grpc_channel_args *build_lb_channel_args( |
|
|
|
|
return result; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void glb_lb_channel_on_connectivity_changed_cb(grpc_exec_ctx *exec_ctx, |
|
|
|
|
void *arg, |
|
|
|
|
grpc_error *error); |
|
|
|
|
static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_lb_policy_factory *factory, |
|
|
|
|
grpc_lb_policy_args *args) { |
|
|
|
|
/* Count the number of gRPC-LB addresses. There must be at least one. */ |
|
|
|
|
const grpc_arg *arg = |
|
|
|
|
grpc_channel_args_find(args->args, GRPC_ARG_LB_ADDRESSES); |
|
|
|
|
if (arg == NULL || arg->type != GRPC_ARG_POINTER) { |
|
|
|
|
return NULL; |
|
|
|
|
} |
|
|
|
|
grpc_lb_addresses *addresses = (grpc_lb_addresses *)arg->value.pointer.p; |
|
|
|
|
size_t num_grpclb_addrs = 0; |
|
|
|
|
for (size_t i = 0; i < addresses->num_addresses; ++i) { |
|
|
|
|
if (addresses->addresses[i].is_balancer) ++num_grpclb_addrs; |
|
|
|
|
} |
|
|
|
|
if (num_grpclb_addrs == 0) return NULL; |
|
|
|
|
|
|
|
|
|
glb_lb_policy *glb_policy = (glb_lb_policy *)gpr_zalloc(sizeof(*glb_policy)); |
|
|
|
|
|
|
|
|
|
/* Get server name. */ |
|
|
|
|
arg = grpc_channel_args_find(args->args, GRPC_ARG_SERVER_URI); |
|
|
|
|
GPR_ASSERT(arg != NULL); |
|
|
|
|
GPR_ASSERT(arg->type == GRPC_ARG_STRING); |
|
|
|
|
grpc_uri *uri = grpc_uri_parse(exec_ctx, arg->value.string, true); |
|
|
|
|
GPR_ASSERT(uri->path[0] != '\0'); |
|
|
|
|
glb_policy->server_name = |
|
|
|
|
gpr_strdup(uri->path[0] == '/' ? uri->path + 1 : uri->path); |
|
|
|
|
if (GRPC_TRACER_ON(grpc_lb_glb_trace)) { |
|
|
|
|
gpr_log(GPR_INFO, "Will use '%s' as the server name for LB request.", |
|
|
|
|
glb_policy->server_name); |
|
|
|
|
} |
|
|
|
|
grpc_uri_destroy(uri); |
|
|
|
|
|
|
|
|
|
glb_policy->cc_factory = args->client_channel_factory; |
|
|
|
|
GPR_ASSERT(glb_policy->cc_factory != NULL); |
|
|
|
|
|
|
|
|
|
arg = grpc_channel_args_find(args->args, GRPC_ARG_GRPCLB_CALL_TIMEOUT_MS); |
|
|
|
|
glb_policy->lb_call_timeout_ms = |
|
|
|
|
grpc_channel_arg_get_integer(arg, (grpc_integer_options){0, 0, INT_MAX}); |
|
|
|
|
|
|
|
|
|
arg = grpc_channel_args_find(args->args, GRPC_ARG_GRPCLB_FALLBACK_TIMEOUT_MS); |
|
|
|
|
glb_policy->lb_fallback_timeout_ms = grpc_channel_arg_get_integer( |
|
|
|
|
arg, (grpc_integer_options){GRPC_GRPCLB_DEFAULT_FALLBACK_TIMEOUT_MS, 0, |
|
|
|
|
INT_MAX}); |
|
|
|
|
|
|
|
|
|
// Make sure that GRPC_ARG_LB_POLICY_NAME is set in channel args,
|
|
|
|
|
// since we use this to trigger the client_load_reporting filter.
|
|
|
|
|
grpc_arg new_arg = |
|
|
|
|
grpc_channel_arg_string_create(GRPC_ARG_LB_POLICY_NAME, "grpclb"); |
|
|
|
|
static const char *args_to_remove[] = {GRPC_ARG_LB_POLICY_NAME}; |
|
|
|
|
glb_policy->args = grpc_channel_args_copy_and_add_and_remove( |
|
|
|
|
args->args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove), &new_arg, 1); |
|
|
|
|
|
|
|
|
|
/* Extract the backend addresses (may be empty) from the resolver for
|
|
|
|
|
* fallback. */ |
|
|
|
|
glb_policy->fallback_backend_addresses = |
|
|
|
|
extract_backend_addresses_locked(exec_ctx, addresses); |
|
|
|
|
|
|
|
|
|
/* Create a client channel over them to communicate with a LB service */ |
|
|
|
|
glb_policy->response_generator = |
|
|
|
|
grpc_fake_resolver_response_generator_create(); |
|
|
|
|
grpc_channel_args *lb_channel_args = build_lb_channel_args( |
|
|
|
|
exec_ctx, addresses, glb_policy->response_generator, args->args); |
|
|
|
|
char *uri_str; |
|
|
|
|
gpr_asprintf(&uri_str, "fake:///%s", glb_policy->server_name); |
|
|
|
|
glb_policy->lb_channel = grpc_lb_policy_grpclb_create_lb_channel( |
|
|
|
|
exec_ctx, uri_str, args->client_channel_factory, lb_channel_args); |
|
|
|
|
|
|
|
|
|
/* Propagate initial resolution */ |
|
|
|
|
grpc_fake_resolver_response_generator_set_response( |
|
|
|
|
exec_ctx, glb_policy->response_generator, lb_channel_args); |
|
|
|
|
grpc_channel_args_destroy(exec_ctx, lb_channel_args); |
|
|
|
|
gpr_free(uri_str); |
|
|
|
|
if (glb_policy->lb_channel == NULL) { |
|
|
|
|
gpr_free((void *)glb_policy->server_name); |
|
|
|
|
grpc_channel_args_destroy(exec_ctx, glb_policy->args); |
|
|
|
|
gpr_free(glb_policy); |
|
|
|
|
return NULL; |
|
|
|
|
} |
|
|
|
|
grpc_subchannel_index_ref(); |
|
|
|
|
GRPC_CLOSURE_INIT(&glb_policy->lb_channel_on_connectivity_changed, |
|
|
|
|
glb_lb_channel_on_connectivity_changed_cb, glb_policy, |
|
|
|
|
grpc_combiner_scheduler(args->combiner)); |
|
|
|
|
grpc_lb_policy_init(&glb_policy->base, &glb_lb_policy_vtable, args->combiner); |
|
|
|
|
grpc_connectivity_state_init(&glb_policy->state_tracker, GRPC_CHANNEL_IDLE, |
|
|
|
|
"grpclb"); |
|
|
|
|
return &glb_policy->base; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void glb_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { |
|
|
|
|
glb_lb_policy *glb_policy = (glb_lb_policy *)pol; |
|
|
|
|
GPR_ASSERT(glb_policy->pending_picks == NULL); |
|
|
|
|
@ -1385,7 +1295,8 @@ static void do_send_client_load_report_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
|
|
|
|
|
static bool load_report_counters_are_zero(grpc_grpclb_request *request) { |
|
|
|
|
grpc_grpclb_dropped_call_counts *drop_entries = |
|
|
|
|
request->client_stats.calls_finished_with_drop.arg; |
|
|
|
|
(grpc_grpclb_dropped_call_counts *) |
|
|
|
|
request->client_stats.calls_finished_with_drop.arg; |
|
|
|
|
return request->client_stats.num_calls_started == 0 && |
|
|
|
|
request->client_stats.num_calls_finished == 0 && |
|
|
|
|
request->client_stats.num_calls_finished_with_client_failed_to_send == |
|
|
|
|
@ -1757,7 +1668,7 @@ static void lb_call_on_retry_timer_locked(grpc_exec_ctx *exec_ctx, void *arg, |
|
|
|
|
|
|
|
|
|
static void lb_on_fallback_timer_locked(grpc_exec_ctx *exec_ctx, void *arg, |
|
|
|
|
grpc_error *error) { |
|
|
|
|
glb_lb_policy *glb_policy = arg; |
|
|
|
|
glb_lb_policy *glb_policy = (glb_lb_policy *)arg; |
|
|
|
|
/* If we receive a serverlist after the timer fires but before this callback
|
|
|
|
|
* actually runs, don't do anything. */ |
|
|
|
|
if (glb_policy->serverlist != NULL) return; |
|
|
|
|
@ -1985,6 +1896,94 @@ static const grpc_lb_policy_vtable glb_lb_policy_vtable = { |
|
|
|
|
glb_notify_on_state_change_locked, |
|
|
|
|
glb_update_locked}; |
|
|
|
|
|
|
|
|
|
static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_lb_policy_factory *factory, |
|
|
|
|
grpc_lb_policy_args *args) { |
|
|
|
|
/* Count the number of gRPC-LB addresses. There must be at least one. */ |
|
|
|
|
const grpc_arg *arg = |
|
|
|
|
grpc_channel_args_find(args->args, GRPC_ARG_LB_ADDRESSES); |
|
|
|
|
if (arg == NULL || arg->type != GRPC_ARG_POINTER) { |
|
|
|
|
return NULL; |
|
|
|
|
} |
|
|
|
|
grpc_lb_addresses *addresses = (grpc_lb_addresses *)arg->value.pointer.p; |
|
|
|
|
size_t num_grpclb_addrs = 0; |
|
|
|
|
for (size_t i = 0; i < addresses->num_addresses; ++i) { |
|
|
|
|
if (addresses->addresses[i].is_balancer) ++num_grpclb_addrs; |
|
|
|
|
} |
|
|
|
|
if (num_grpclb_addrs == 0) return NULL; |
|
|
|
|
|
|
|
|
|
glb_lb_policy *glb_policy = (glb_lb_policy *)gpr_zalloc(sizeof(*glb_policy)); |
|
|
|
|
|
|
|
|
|
/* Get server name. */ |
|
|
|
|
arg = grpc_channel_args_find(args->args, GRPC_ARG_SERVER_URI); |
|
|
|
|
GPR_ASSERT(arg != NULL); |
|
|
|
|
GPR_ASSERT(arg->type == GRPC_ARG_STRING); |
|
|
|
|
grpc_uri *uri = grpc_uri_parse(exec_ctx, arg->value.string, true); |
|
|
|
|
GPR_ASSERT(uri->path[0] != '\0'); |
|
|
|
|
glb_policy->server_name = |
|
|
|
|
gpr_strdup(uri->path[0] == '/' ? uri->path + 1 : uri->path); |
|
|
|
|
if (GRPC_TRACER_ON(grpc_lb_glb_trace)) { |
|
|
|
|
gpr_log(GPR_INFO, "Will use '%s' as the server name for LB request.", |
|
|
|
|
glb_policy->server_name); |
|
|
|
|
} |
|
|
|
|
grpc_uri_destroy(uri); |
|
|
|
|
|
|
|
|
|
glb_policy->cc_factory = args->client_channel_factory; |
|
|
|
|
GPR_ASSERT(glb_policy->cc_factory != NULL); |
|
|
|
|
|
|
|
|
|
arg = grpc_channel_args_find(args->args, GRPC_ARG_GRPCLB_CALL_TIMEOUT_MS); |
|
|
|
|
glb_policy->lb_call_timeout_ms = |
|
|
|
|
grpc_channel_arg_get_integer(arg, (grpc_integer_options){0, 0, INT_MAX}); |
|
|
|
|
|
|
|
|
|
arg = grpc_channel_args_find(args->args, GRPC_ARG_GRPCLB_FALLBACK_TIMEOUT_MS); |
|
|
|
|
glb_policy->lb_fallback_timeout_ms = grpc_channel_arg_get_integer( |
|
|
|
|
arg, (grpc_integer_options){GRPC_GRPCLB_DEFAULT_FALLBACK_TIMEOUT_MS, 0, |
|
|
|
|
INT_MAX}); |
|
|
|
|
|
|
|
|
|
// Make sure that GRPC_ARG_LB_POLICY_NAME is set in channel args,
|
|
|
|
|
// since we use this to trigger the client_load_reporting filter.
|
|
|
|
|
grpc_arg new_arg = |
|
|
|
|
grpc_channel_arg_string_create(GRPC_ARG_LB_POLICY_NAME, "grpclb"); |
|
|
|
|
static const char *args_to_remove[] = {GRPC_ARG_LB_POLICY_NAME}; |
|
|
|
|
glb_policy->args = grpc_channel_args_copy_and_add_and_remove( |
|
|
|
|
args->args, args_to_remove, GPR_ARRAY_SIZE(args_to_remove), &new_arg, 1); |
|
|
|
|
|
|
|
|
|
/* Extract the backend addresses (may be empty) from the resolver for
|
|
|
|
|
* fallback. */ |
|
|
|
|
glb_policy->fallback_backend_addresses = |
|
|
|
|
extract_backend_addresses_locked(exec_ctx, addresses); |
|
|
|
|
|
|
|
|
|
/* Create a client channel over them to communicate with a LB service */ |
|
|
|
|
glb_policy->response_generator = |
|
|
|
|
grpc_fake_resolver_response_generator_create(); |
|
|
|
|
grpc_channel_args *lb_channel_args = build_lb_channel_args( |
|
|
|
|
exec_ctx, addresses, glb_policy->response_generator, args->args); |
|
|
|
|
char *uri_str; |
|
|
|
|
gpr_asprintf(&uri_str, "fake:///%s", glb_policy->server_name); |
|
|
|
|
glb_policy->lb_channel = grpc_lb_policy_grpclb_create_lb_channel( |
|
|
|
|
exec_ctx, uri_str, args->client_channel_factory, lb_channel_args); |
|
|
|
|
|
|
|
|
|
/* Propagate initial resolution */ |
|
|
|
|
grpc_fake_resolver_response_generator_set_response( |
|
|
|
|
exec_ctx, glb_policy->response_generator, lb_channel_args); |
|
|
|
|
grpc_channel_args_destroy(exec_ctx, lb_channel_args); |
|
|
|
|
gpr_free(uri_str); |
|
|
|
|
if (glb_policy->lb_channel == NULL) { |
|
|
|
|
gpr_free((void *)glb_policy->server_name); |
|
|
|
|
grpc_channel_args_destroy(exec_ctx, glb_policy->args); |
|
|
|
|
gpr_free(glb_policy); |
|
|
|
|
return NULL; |
|
|
|
|
} |
|
|
|
|
grpc_subchannel_index_ref(); |
|
|
|
|
GRPC_CLOSURE_INIT(&glb_policy->lb_channel_on_connectivity_changed, |
|
|
|
|
glb_lb_channel_on_connectivity_changed_cb, glb_policy, |
|
|
|
|
grpc_combiner_scheduler(args->combiner)); |
|
|
|
|
grpc_lb_policy_init(&glb_policy->base, &glb_lb_policy_vtable, args->combiner); |
|
|
|
|
grpc_connectivity_state_init(&glb_policy->state_tracker, GRPC_CHANNEL_IDLE, |
|
|
|
|
"grpclb"); |
|
|
|
|
return &glb_policy->base; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void glb_factory_ref(grpc_lb_policy_factory *factory) {} |
|
|
|
|
|
|
|
|
|
static void glb_factory_unref(grpc_lb_policy_factory *factory) {} |
|
|
|
|
|
Yash Tibrewal
7 years ago
committed by
GitHub