|
|
|
|
@ -101,6 +101,7 @@ |
|
|
|
|
#include "src/core/ext/filters/client_channel/lb_policy_registry.h" |
|
|
|
|
#include "src/core/ext/filters/client_channel/parse_address.h" |
|
|
|
|
#include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h" |
|
|
|
|
#include "src/core/ext/filters/client_channel/subchannel_index.h" |
|
|
|
|
#include "src/core/lib/channel/channel_args.h" |
|
|
|
|
#include "src/core/lib/channel/channel_stack.h" |
|
|
|
|
#include "src/core/lib/iomgr/combiner.h" |
|
|
|
|
@ -137,7 +138,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 { |
|
|
|
|
@ -285,7 +286,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; |
|
|
|
|
@ -727,7 +728,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)); |
|
|
|
|
@ -869,7 +870,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) { |
|
|
|
|
@ -911,92 +913,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.
|
|
|
|
|
* TODO(roth): For now, we ignore non-balancer addresses, but in the |
|
|
|
|
* future, we may change the behavior such that we fall back to using |
|
|
|
|
* the non-balancer addresses if we cannot reach any balancers. In the |
|
|
|
|
* fallback case, we should use the LB policy indicated by |
|
|
|
|
* GRPC_ARG_LB_POLICY_NAME (although if that specifies grpclb or is |
|
|
|
|
* unset, we should default to pick_first). */ |
|
|
|
|
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}); |
|
|
|
|
|
|
|
|
|
// 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); |
|
|
|
|
|
|
|
|
|
/* 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_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); |
|
|
|
|
@ -1011,6 +927,7 @@ static void glb_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) { |
|
|
|
|
grpc_grpclb_destroy_serverlist(glb_policy->serverlist); |
|
|
|
|
} |
|
|
|
|
grpc_fake_resolver_response_generator_unref(glb_policy->response_generator); |
|
|
|
|
grpc_subchannel_index_unref(); |
|
|
|
|
if (glb_policy->pending_update_args != NULL) { |
|
|
|
|
grpc_channel_args_destroy(exec_ctx, glb_policy->pending_update_args->args); |
|
|
|
|
gpr_free(glb_policy->pending_update_args); |
|
|
|
|
@ -1303,7 +1220,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 == |
|
|
|
|
@ -1865,6 +1783,89 @@ 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.
|
|
|
|
|
* TODO(roth): For now, we ignore non-balancer addresses, but in the |
|
|
|
|
* future, we may change the behavior such that we fall back to using |
|
|
|
|
* the non-balancer addresses if we cannot reach any balancers. In the |
|
|
|
|
* fallback case, we should use the LB policy indicated by |
|
|
|
|
* GRPC_ARG_LB_POLICY_NAME (although if that specifies grpclb or is |
|
|
|
|
* unset, we should default to pick_first). */ |
|
|
|
|
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}); |
|
|
|
|
|
|
|
|
|
// 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); |
|
|
|
|
|
|
|
|
|
/* 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_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