|
|
@@ -108,13 +108,16 @@
|
|
|
|
|
|
#include "src/core/ext/filters/client_channel/client_channel.h"
|
|
|
#include "src/core/ext/filters/client_channel/client_channel_factory.h"
|
|
|
+#include "src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.h"
|
|
|
#include "src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb.h"
|
|
|
#include "src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_channel.h"
|
|
|
+#include "src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_client_stats.h"
|
|
|
#include "src/core/ext/filters/client_channel/lb_policy/grpclb/load_balancer_api.h"
|
|
|
#include "src/core/ext/filters/client_channel/lb_policy_factory.h"
|
|
|
#include "src/core/ext/filters/client_channel/lb_policy_registry.h"
|
|
|
#include "src/core/ext/filters/client_channel/parse_address.h"
|
|
|
#include "src/core/lib/channel/channel_args.h"
|
|
|
+#include "src/core/lib/channel/channel_stack.h"
|
|
|
#include "src/core/lib/iomgr/combiner.h"
|
|
|
#include "src/core/lib/iomgr/sockaddr.h"
|
|
|
#include "src/core/lib/iomgr/sockaddr_utils.h"
|
|
|
@@ -126,6 +129,7 @@
|
|
|
#include "src/core/lib/support/string.h"
|
|
|
#include "src/core/lib/surface/call.h"
|
|
|
#include "src/core/lib/surface/channel.h"
|
|
|
+#include "src/core/lib/surface/channel_init.h"
|
|
|
#include "src/core/lib/transport/static_metadata.h"
|
|
|
|
|
|
#define GRPC_GRPCLB_MIN_CONNECT_TIMEOUT_SECONDS 20
|
|
|
@@ -147,6 +151,10 @@ static grpc_error *initial_metadata_add_lb_token(
|
|
|
lb_token_mdelem_storage, lb_token);
|
|
|
}
|
|
|
|
|
|
+static void destroy_client_stats(void *arg) {
|
|
|
+ grpc_grpclb_client_stats_unref(arg);
|
|
|
+}
|
|
|
+
|
|
|
typedef struct wrapped_rr_closure_arg {
|
|
|
/* the closure instance using this struct as argument */
|
|
|
grpc_closure wrapper_closure;
|
|
|
@@ -163,6 +171,13 @@ typedef struct wrapped_rr_closure_arg {
|
|
|
* initial metadata */
|
|
|
grpc_connected_subchannel **target;
|
|
|
|
|
|
+ /* the context to be populated for the subchannel call */
|
|
|
+ grpc_call_context_element *context;
|
|
|
+
|
|
|
+ /* Stats for client-side load reporting. Note that this holds a
|
|
|
+ * reference, which must be either passed on via context or unreffed. */
|
|
|
+ grpc_grpclb_client_stats *client_stats;
|
|
|
+
|
|
|
/* the LB token associated with the pick */
|
|
|
grpc_mdelem lb_token;
|
|
|
|
|
|
@@ -202,6 +217,12 @@ static void wrapped_rr_closure(grpc_exec_ctx *exec_ctx, void *arg,
|
|
|
(void *)*wc_arg->target, (void *)wc_arg->rr_policy);
|
|
|
abort();
|
|
|
}
|
|
|
+ // Pass on client stats via context. Passes ownership of the reference.
|
|
|
+ GPR_ASSERT(wc_arg->client_stats != NULL);
|
|
|
+ wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].value = wc_arg->client_stats;
|
|
|
+ wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].destroy = destroy_client_stats;
|
|
|
+ } else {
|
|
|
+ grpc_grpclb_client_stats_unref(wc_arg->client_stats);
|
|
|
}
|
|
|
if (grpc_lb_glb_trace) {
|
|
|
gpr_log(GPR_INFO, "Unreffing RR %p", (void *)wc_arg->rr_policy);
|
|
|
@@ -237,6 +258,7 @@ typedef struct pending_pick {
|
|
|
static void add_pending_pick(pending_pick **root,
|
|
|
const grpc_lb_policy_pick_args *pick_args,
|
|
|
grpc_connected_subchannel **target,
|
|
|
+ grpc_call_context_element *context,
|
|
|
grpc_closure *on_complete) {
|
|
|
pending_pick *pp = gpr_zalloc(sizeof(*pp));
|
|
|
pp->next = *root;
|
|
|
@@ -244,6 +266,7 @@ static void add_pending_pick(pending_pick **root,
|
|
|
pp->target = target;
|
|
|
pp->wrapped_on_complete_arg.wrapped_closure = on_complete;
|
|
|
pp->wrapped_on_complete_arg.target = target;
|
|
|
+ pp->wrapped_on_complete_arg.context = context;
|
|
|
pp->wrapped_on_complete_arg.initial_metadata = pick_args->initial_metadata;
|
|
|
pp->wrapped_on_complete_arg.lb_token_mdelem_storage =
|
|
|
pick_args->lb_token_mdelem_storage;
|
|
|
@@ -316,6 +339,10 @@ typedef struct glb_lb_policy {
|
|
|
/************************************************************/
|
|
|
/* client data associated with the LB server communication */
|
|
|
/************************************************************/
|
|
|
+
|
|
|
+ /* Finished sending initial request. */
|
|
|
+ grpc_closure lb_on_sent_initial_request;
|
|
|
+
|
|
|
/* Status from the LB server has been received. This signals the end of the LB
|
|
|
* call. */
|
|
|
grpc_closure lb_on_server_status_received;
|
|
|
@@ -348,6 +375,23 @@ typedef struct glb_lb_policy {
|
|
|
|
|
|
/** LB call retry timer */
|
|
|
grpc_timer lb_call_retry_timer;
|
|
|
+
|
|
|
+ bool initial_request_sent;
|
|
|
+ bool seen_initial_response;
|
|
|
+
|
|
|
+ /* Stats for client-side load reporting. Should be unreffed and
|
|
|
+ * recreated whenever lb_call is replaced. */
|
|
|
+ grpc_grpclb_client_stats *client_stats;
|
|
|
+ /* Interval and timer for next client load report. */
|
|
|
+ gpr_timespec client_stats_report_interval;
|
|
|
+ grpc_timer client_load_report_timer;
|
|
|
+ bool client_load_report_timer_pending;
|
|
|
+ bool last_client_load_report_counters_were_zero;
|
|
|
+ /* Closure used for either the load report timer or the callback for
|
|
|
+ * completion of sending the load report. */
|
|
|
+ grpc_closure client_load_report_closure;
|
|
|
+ /* Client load report message payload. */
|
|
|
+ grpc_byte_buffer *client_load_report_payload;
|
|
|
} glb_lb_policy;
|
|
|
|
|
|
/* Keeps track and reacts to changes in connectivity of the RR instance */
|
|
|
@@ -552,8 +596,8 @@ static bool pick_from_internal_rr_locked(
|
|
|
grpc_connected_subchannel **target, wrapped_rr_closure_arg *wc_arg) {
|
|
|
GPR_ASSERT(rr_policy != NULL);
|
|
|
const bool pick_done = grpc_lb_policy_pick_locked(
|
|
|
- exec_ctx, rr_policy, pick_args, target, (void **)&wc_arg->lb_token,
|
|
|
- &wc_arg->wrapper_closure);
|
|
|
+ exec_ctx, rr_policy, pick_args, target, wc_arg->context,
|
|
|
+ (void **)&wc_arg->lb_token, &wc_arg->wrapper_closure);
|
|
|
if (pick_done) {
|
|
|
/* synchronous grpc_lb_policy_pick call. Unref the RR policy. */
|
|
|
if (grpc_lb_glb_trace) {
|
|
|
@@ -567,7 +611,12 @@ static bool pick_from_internal_rr_locked(
|
|
|
pick_args->lb_token_mdelem_storage,
|
|
|
GRPC_MDELEM_REF(wc_arg->lb_token));
|
|
|
|
|
|
- gpr_free(wc_arg);
|
|
|
+ // Pass on client stats via context. Passes ownership of the reference.
|
|
|
+ GPR_ASSERT(wc_arg->client_stats != NULL);
|
|
|
+ wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].value = wc_arg->client_stats;
|
|
|
+ wc_arg->context[GRPC_GRPCLB_CLIENT_STATS].destroy = destroy_client_stats;
|
|
|
+
|
|
|
+ gpr_free(wc_arg->free_when_done);
|
|
|
}
|
|
|
/* else, the pending pick will be registered and taken care of by the
|
|
|
* pending pick list inside the RR policy (glb_policy->rr_policy).
|
|
|
@@ -690,6 +739,8 @@ static void rr_handover_locked(grpc_exec_ctx *exec_ctx,
|
|
|
glb_policy->pending_picks = pp->next;
|
|
|
GRPC_LB_POLICY_REF(glb_policy->rr_policy, "rr_handover_pending_pick");
|
|
|
pp->wrapped_on_complete_arg.rr_policy = glb_policy->rr_policy;
|
|
|
+ pp->wrapped_on_complete_arg.client_stats =
|
|
|
+ grpc_grpclb_client_stats_ref(glb_policy->client_stats);
|
|
|
if (grpc_lb_glb_trace) {
|
|
|
gpr_log(GPR_INFO, "Pending pick about to PICK from 0x%" PRIxPTR "",
|
|
|
(intptr_t)glb_policy->rr_policy);
|
|
|
@@ -864,9 +915,18 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
|
|
|
grpc_uri_destroy(uri);
|
|
|
|
|
|
glb_policy->cc_factory = args->client_channel_factory;
|
|
|
- glb_policy->args = grpc_channel_args_copy(args->args);
|
|
|
GPR_ASSERT(glb_policy->cc_factory != NULL);
|
|
|
|
|
|
+ // 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;
|
|
|
+ new_arg.key = GRPC_ARG_LB_POLICY_NAME;
|
|
|
+ new_arg.type = GRPC_ARG_STRING;
|
|
|
+ new_arg.value.string = "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);
|
|
|
+
|
|
|
grpc_slice_hash_table *targets_info = NULL;
|
|
|
/* Create a client channel over them to communicate with a LB service */
|
|
|
char *lb_service_target_addresses =
|
|
|
@@ -880,6 +940,8 @@ static grpc_lb_policy *glb_create(grpc_exec_ctx *exec_ctx,
|
|
|
grpc_channel_args_destroy(exec_ctx, lb_channel_args);
|
|
|
gpr_free(lb_service_target_addresses);
|
|
|
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;
|
|
|
}
|
|
|
@@ -895,6 +957,9 @@ static void glb_destroy(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
|
|
|
GPR_ASSERT(glb_policy->pending_pings == NULL);
|
|
|
gpr_free((void *)glb_policy->server_name);
|
|
|
grpc_channel_args_destroy(exec_ctx, glb_policy->args);
|
|
|
+ if (glb_policy->client_stats != NULL) {
|
|
|
+ grpc_grpclb_client_stats_unref(glb_policy->client_stats);
|
|
|
+ }
|
|
|
grpc_channel_destroy(glb_policy->lb_channel);
|
|
|
glb_policy->lb_channel = NULL;
|
|
|
grpc_connectivity_state_destroy(exec_ctx, &glb_policy->state_tracker);
|
|
|
@@ -1011,7 +1076,8 @@ static void glb_exit_idle_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol) {
|
|
|
|
|
|
static int glb_pick_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
|
|
|
const grpc_lb_policy_pick_args *pick_args,
|
|
|
- grpc_connected_subchannel **target, void **user_data,
|
|
|
+ grpc_connected_subchannel **target,
|
|
|
+ grpc_call_context_element *context, void **user_data,
|
|
|
grpc_closure *on_complete) {
|
|
|
if (pick_args->lb_token_mdelem_storage == NULL) {
|
|
|
*target = NULL;
|
|
|
@@ -1039,6 +1105,10 @@ static int glb_pick_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
|
|
|
grpc_schedule_on_exec_ctx);
|
|
|
wc_arg->rr_policy = glb_policy->rr_policy;
|
|
|
wc_arg->target = target;
|
|
|
+ wc_arg->context = context;
|
|
|
+ GPR_ASSERT(glb_policy->client_stats != NULL);
|
|
|
+ wc_arg->client_stats =
|
|
|
+ grpc_grpclb_client_stats_ref(glb_policy->client_stats);
|
|
|
wc_arg->wrapped_closure = on_complete;
|
|
|
wc_arg->lb_token_mdelem_storage = pick_args->lb_token_mdelem_storage;
|
|
|
wc_arg->initial_metadata = pick_args->initial_metadata;
|
|
|
@@ -1052,7 +1122,7 @@ static int glb_pick_locked(grpc_exec_ctx *exec_ctx, grpc_lb_policy *pol,
|
|
|
"picks",
|
|
|
(void *)(glb_policy));
|
|
|
}
|
|
|
- add_pending_pick(&glb_policy->pending_picks, pick_args, target,
|
|
|
+ add_pending_pick(&glb_policy->pending_picks, pick_args, target, context,
|
|
|
on_complete);
|
|
|
|
|
|
if (!glb_policy->started_picking) {
|
|
|
@@ -1093,6 +1163,104 @@ static void glb_notify_on_state_change_locked(grpc_exec_ctx *exec_ctx,
|
|
|
exec_ctx, &glb_policy->state_tracker, current, notify);
|
|
|
}
|
|
|
|
|
|
+static void send_client_load_report_locked(grpc_exec_ctx *exec_ctx, void *arg,
|
|
|
+ grpc_error *error);
|
|
|
+
|
|
|
+static void schedule_next_client_load_report(grpc_exec_ctx *exec_ctx,
|
|
|
+ glb_lb_policy *glb_policy) {
|
|
|
+ const gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
|
|
|
+ const gpr_timespec next_client_load_report_time =
|
|
|
+ gpr_time_add(now, glb_policy->client_stats_report_interval);
|
|
|
+ grpc_closure_init(&glb_policy->client_load_report_closure,
|
|
|
+ send_client_load_report_locked, glb_policy,
|
|
|
+ grpc_combiner_scheduler(glb_policy->base.combiner, false));
|
|
|
+ grpc_timer_init(exec_ctx, &glb_policy->client_load_report_timer,
|
|
|
+ next_client_load_report_time,
|
|
|
+ &glb_policy->client_load_report_closure, now);
|
|
|
+}
|
|
|
+
|
|
|
+static void client_load_report_done_locked(grpc_exec_ctx *exec_ctx, void *arg,
|
|
|
+ grpc_error *error) {
|
|
|
+ glb_lb_policy *glb_policy = arg;
|
|
|
+ grpc_byte_buffer_destroy(glb_policy->client_load_report_payload);
|
|
|
+ glb_policy->client_load_report_payload = NULL;
|
|
|
+ if (error != GRPC_ERROR_NONE || glb_policy->lb_call == NULL) {
|
|
|
+ glb_policy->client_load_report_timer_pending = false;
|
|
|
+ GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &glb_policy->base,
|
|
|
+ "client_load_report");
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ schedule_next_client_load_report(exec_ctx, glb_policy);
|
|
|
+}
|
|
|
+
|
|
|
+static void do_send_client_load_report_locked(grpc_exec_ctx *exec_ctx,
|
|
|
+ glb_lb_policy *glb_policy) {
|
|
|
+ grpc_op op;
|
|
|
+ memset(&op, 0, sizeof(op));
|
|
|
+ op.op = GRPC_OP_SEND_MESSAGE;
|
|
|
+ op.data.send_message.send_message = glb_policy->client_load_report_payload;
|
|
|
+ grpc_closure_init(&glb_policy->client_load_report_closure,
|
|
|
+ client_load_report_done_locked, glb_policy,
|
|
|
+ grpc_combiner_scheduler(glb_policy->base.combiner, false));
|
|
|
+ grpc_call_error call_error = grpc_call_start_batch_and_execute(
|
|
|
+ exec_ctx, glb_policy->lb_call, &op, 1,
|
|
|
+ &glb_policy->client_load_report_closure);
|
|
|
+ GPR_ASSERT(GRPC_CALL_OK == call_error);
|
|
|
+}
|
|
|
+
|
|
|
+static bool load_report_counters_are_zero(grpc_grpclb_request *request) {
|
|
|
+ return request->client_stats.num_calls_started == 0 &&
|
|
|
+ request->client_stats.num_calls_finished == 0 &&
|
|
|
+ request->client_stats.num_calls_finished_with_drop_for_rate_limiting ==
|
|
|
+ 0 &&
|
|
|
+ request->client_stats
|
|
|
+ .num_calls_finished_with_drop_for_load_balancing == 0 &&
|
|
|
+ request->client_stats.num_calls_finished_with_client_failed_to_send ==
|
|
|
+ 0 &&
|
|
|
+ request->client_stats.num_calls_finished_known_received == 0;
|
|
|
+}
|
|
|
+
|
|
|
+static void send_client_load_report_locked(grpc_exec_ctx *exec_ctx, void *arg,
|
|
|
+ grpc_error *error) {
|
|
|
+ glb_lb_policy *glb_policy = arg;
|
|
|
+ if (error == GRPC_ERROR_CANCELLED || glb_policy->lb_call == NULL) {
|
|
|
+ glb_policy->client_load_report_timer_pending = false;
|
|
|
+ GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &glb_policy->base,
|
|
|
+ "client_load_report");
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ // Construct message payload.
|
|
|
+ GPR_ASSERT(glb_policy->client_load_report_payload == NULL);
|
|
|
+ grpc_grpclb_request *request =
|
|
|
+ grpc_grpclb_load_report_request_create(glb_policy->client_stats);
|
|
|
+ // Skip client load report if the counters were all zero in the last
|
|
|
+ // report and they are still zero in this one.
|
|
|
+ if (load_report_counters_are_zero(request)) {
|
|
|
+ if (glb_policy->last_client_load_report_counters_were_zero) {
|
|
|
+ grpc_grpclb_request_destroy(request);
|
|
|
+ schedule_next_client_load_report(exec_ctx, glb_policy);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ glb_policy->last_client_load_report_counters_were_zero = true;
|
|
|
+ } else {
|
|
|
+ glb_policy->last_client_load_report_counters_were_zero = false;
|
|
|
+ }
|
|
|
+ grpc_slice request_payload_slice = grpc_grpclb_request_encode(request);
|
|
|
+ glb_policy->client_load_report_payload =
|
|
|
+ grpc_raw_byte_buffer_create(&request_payload_slice, 1);
|
|
|
+ grpc_slice_unref_internal(exec_ctx, request_payload_slice);
|
|
|
+ grpc_grpclb_request_destroy(request);
|
|
|
+ // If we've already sent the initial request, then we can go ahead and
|
|
|
+ // sent the load report. Otherwise, we need to wait until the initial
|
|
|
+ // request has been sent to send this
|
|
|
+ // (see lb_on_sent_initial_request_locked() below).
|
|
|
+ if (glb_policy->initial_request_sent) {
|
|
|
+ do_send_client_load_report_locked(exec_ctx, glb_policy);
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+static void lb_on_sent_initial_request_locked(grpc_exec_ctx *exec_ctx,
|
|
|
+ void *arg, grpc_error *error);
|
|
|
static void lb_on_server_status_received_locked(grpc_exec_ctx *exec_ctx,
|
|
|
void *arg, grpc_error *error);
|
|
|
static void lb_on_response_received_locked(grpc_exec_ctx *exec_ctx, void *arg,
|
|
|
@@ -1114,6 +1282,11 @@ static void lb_call_init_locked(grpc_exec_ctx *exec_ctx,
|
|
|
&host, glb_policy->deadline, NULL);
|
|
|
grpc_slice_unref_internal(exec_ctx, host);
|
|
|
|
|
|
+ if (glb_policy->client_stats != NULL) {
|
|
|
+ grpc_grpclb_client_stats_unref(glb_policy->client_stats);
|
|
|
+ }
|
|
|
+ glb_policy->client_stats = grpc_grpclb_client_stats_create();
|
|
|
+
|
|
|
grpc_metadata_array_init(&glb_policy->lb_initial_metadata_recv);
|
|
|
grpc_metadata_array_init(&glb_policy->lb_trailing_metadata_recv);
|
|
|
|
|
|
@@ -1125,6 +1298,9 @@ static void lb_call_init_locked(grpc_exec_ctx *exec_ctx,
|
|
|
grpc_slice_unref_internal(exec_ctx, request_payload_slice);
|
|
|
grpc_grpclb_request_destroy(request);
|
|
|
|
|
|
+ grpc_closure_init(&glb_policy->lb_on_sent_initial_request,
|
|
|
+ lb_on_sent_initial_request_locked, glb_policy,
|
|
|
+ grpc_combiner_scheduler(glb_policy->base.combiner, false));
|
|
|
grpc_closure_init(&glb_policy->lb_on_server_status_received,
|
|
|
lb_on_server_status_received_locked, glb_policy,
|
|
|
grpc_combiner_scheduler(glb_policy->base.combiner, false));
|
|
|
@@ -1138,6 +1314,10 @@ static void lb_call_init_locked(grpc_exec_ctx *exec_ctx,
|
|
|
GRPC_GRPCLB_RECONNECT_JITTER,
|
|
|
GRPC_GRPCLB_MIN_CONNECT_TIMEOUT_SECONDS * 1000,
|
|
|
GRPC_GRPCLB_RECONNECT_MAX_BACKOFF_SECONDS * 1000);
|
|
|
+
|
|
|
+ glb_policy->initial_request_sent = false;
|
|
|
+ glb_policy->seen_initial_response = false;
|
|
|
+ glb_policy->last_client_load_report_counters_were_zero = false;
|
|
|
}
|
|
|
|
|
|
static void lb_call_destroy_locked(grpc_exec_ctx *exec_ctx,
|
|
|
@@ -1151,6 +1331,10 @@ static void lb_call_destroy_locked(grpc_exec_ctx *exec_ctx,
|
|
|
|
|
|
grpc_byte_buffer_destroy(glb_policy->lb_request_payload);
|
|
|
grpc_slice_unref_internal(exec_ctx, glb_policy->lb_call_status_details);
|
|
|
+
|
|
|
+ if (!glb_policy->client_load_report_timer_pending) {
|
|
|
+ grpc_timer_cancel(exec_ctx, &glb_policy->client_load_report_timer);
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
/*
|
|
|
@@ -1179,21 +1363,27 @@ static void query_for_backends_locked(grpc_exec_ctx *exec_ctx,
|
|
|
op->flags = 0;
|
|
|
op->reserved = NULL;
|
|
|
op++;
|
|
|
-
|
|
|
op->op = GRPC_OP_RECV_INITIAL_METADATA;
|
|
|
op->data.recv_initial_metadata.recv_initial_metadata =
|
|
|
&glb_policy->lb_initial_metadata_recv;
|
|
|
op->flags = 0;
|
|
|
op->reserved = NULL;
|
|
|
op++;
|
|
|
-
|
|
|
GPR_ASSERT(glb_policy->lb_request_payload != NULL);
|
|
|
op->op = GRPC_OP_SEND_MESSAGE;
|
|
|
op->data.send_message.send_message = glb_policy->lb_request_payload;
|
|
|
op->flags = 0;
|
|
|
op->reserved = NULL;
|
|
|
op++;
|
|
|
+ /* take a weak ref (won't prevent calling of \a glb_shutdown if the strong ref
|
|
|
+ * count goes to zero) to be unref'd in lb_on_sent_initial_request_locked() */
|
|
|
+ GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "lb_on_server_status_received");
|
|
|
+ call_error = grpc_call_start_batch_and_execute(
|
|
|
+ exec_ctx, glb_policy->lb_call, ops, (size_t)(op - ops),
|
|
|
+ &glb_policy->lb_on_sent_initial_request);
|
|
|
+ GPR_ASSERT(GRPC_CALL_OK == call_error);
|
|
|
|
|
|
+ op = ops;
|
|
|
op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
|
|
|
op->data.recv_status_on_client.trailing_metadata =
|
|
|
&glb_policy->lb_trailing_metadata_recv;
|
|
|
@@ -1225,6 +1415,19 @@ static void query_for_backends_locked(grpc_exec_ctx *exec_ctx,
|
|
|
GPR_ASSERT(GRPC_CALL_OK == call_error);
|
|
|
}
|
|
|
|
|
|
+static void lb_on_sent_initial_request_locked(grpc_exec_ctx *exec_ctx,
|
|
|
+ void *arg, grpc_error *error) {
|
|
|
+ glb_lb_policy *glb_policy = arg;
|
|
|
+ glb_policy->initial_request_sent = true;
|
|
|
+ // If we attempted to send a client load report before the initial
|
|
|
+ // request was sent, send the load report now.
|
|
|
+ if (glb_policy->client_load_report_payload != NULL) {
|
|
|
+ do_send_client_load_report_locked(exec_ctx, glb_policy);
|
|
|
+ }
|
|
|
+ GRPC_LB_POLICY_WEAK_UNREF(exec_ctx, &glb_policy->base,
|
|
|
+ "lb_on_response_received_locked");
|
|
|
+}
|
|
|
+
|
|
|
static void lb_on_response_received_locked(grpc_exec_ctx *exec_ctx, void *arg,
|
|
|
grpc_error *error) {
|
|
|
glb_lb_policy *glb_policy = arg;
|
|
|
@@ -1240,58 +1443,91 @@ static void lb_on_response_received_locked(grpc_exec_ctx *exec_ctx, void *arg,
|
|
|
grpc_byte_buffer_reader_init(&bbr, glb_policy->lb_response_payload);
|
|
|
grpc_slice response_slice = grpc_byte_buffer_reader_readall(&bbr);
|
|
|
grpc_byte_buffer_destroy(glb_policy->lb_response_payload);
|
|
|
- grpc_grpclb_serverlist *serverlist =
|
|
|
- grpc_grpclb_response_parse_serverlist(response_slice);
|
|
|
- if (serverlist != NULL) {
|
|
|
- GPR_ASSERT(glb_policy->lb_call != NULL);
|
|
|
- grpc_slice_unref_internal(exec_ctx, response_slice);
|
|
|
- if (grpc_lb_glb_trace) {
|
|
|
- gpr_log(GPR_INFO, "Serverlist with %lu servers received",
|
|
|
- (unsigned long)serverlist->num_servers);
|
|
|
- for (size_t i = 0; i < serverlist->num_servers; ++i) {
|
|
|
- grpc_resolved_address addr;
|
|
|
- parse_server(serverlist->servers[i], &addr);
|
|
|
- char *ipport;
|
|
|
- grpc_sockaddr_to_string(&ipport, &addr, false);
|
|
|
- gpr_log(GPR_INFO, "Serverlist[%lu]: %s", (unsigned long)i, ipport);
|
|
|
- gpr_free(ipport);
|
|
|
+
|
|
|
+ grpc_grpclb_initial_response *response = NULL;
|
|
|
+ if (!glb_policy->seen_initial_response &&
|
|
|
+ (response = grpc_grpclb_initial_response_parse(response_slice)) !=
|
|
|
+ NULL) {
|
|
|
+ if (response->has_client_stats_report_interval) {
|
|
|
+ glb_policy->client_stats_report_interval =
|
|
|
+ gpr_time_max(gpr_time_from_seconds(1, GPR_TIMESPAN),
|
|
|
+ grpc_grpclb_duration_to_timespec(
|
|
|
+ &response->client_stats_report_interval));
|
|
|
+ if (grpc_lb_glb_trace) {
|
|
|
+ gpr_log(GPR_INFO,
|
|
|
+ "received initial LB response message; "
|
|
|
+ "client load reporting interval = %" PRId64 ".%09d sec",
|
|
|
+ glb_policy->client_stats_report_interval.tv_sec,
|
|
|
+ glb_policy->client_stats_report_interval.tv_nsec);
|
|
|
}
|
|
|
+ /* take a weak ref (won't prevent calling of \a glb_shutdown() if the
|
|
|
+ * strong ref count goes to zero) to be unref'd in
|
|
|
+ * send_client_load_report() */
|
|
|
+ glb_policy->client_load_report_timer_pending = true;
|
|
|
+ GRPC_LB_POLICY_WEAK_REF(&glb_policy->base, "client_load_report");
|
|
|
+ schedule_next_client_load_report(exec_ctx, glb_policy);
|
|
|
+ } else if (grpc_lb_glb_trace) {
|
|
|
+ gpr_log(GPR_INFO,
|
|
|
+ "received initial LB response message; "
|
|
|
+ "client load reporting NOT enabled");
|
|
|
}
|
|
|
+ grpc_grpclb_initial_response_destroy(response);
|
|
|
+ glb_policy->seen_initial_response = true;
|
|
|
+ } else {
|
|
|
+ grpc_grpclb_serverlist *serverlist =
|
|
|
+ grpc_grpclb_response_parse_serverlist(response_slice);
|
|
|
+ if (serverlist != NULL) {
|
|
|
+ GPR_ASSERT(glb_policy->lb_call != NULL);
|
|
|
+ if (grpc_lb_glb_trace) {
|
|
|
+ gpr_log(GPR_INFO, "Serverlist with %lu servers received",
|
|
|
+ (unsigned long)serverlist->num_servers);
|
|
|
+ for (size_t i = 0; i < serverlist->num_servers; ++i) {
|
|
|
+ grpc_resolved_address addr;
|
|
|
+ parse_server(serverlist->servers[i], &addr);
|
|
|
+ char *ipport;
|
|
|
+ grpc_sockaddr_to_string(&ipport, &addr, false);
|
|
|
+ gpr_log(GPR_INFO, "Serverlist[%lu]: %s", (unsigned long)i, ipport);
|
|
|
+ gpr_free(ipport);
|
|
|
+ }
|
|
|
+ }
|
|
|
|
|
|
- /* update serverlist */
|
|
|
- if (serverlist->num_servers > 0) {
|
|
|
- if (grpc_grpclb_serverlist_equals(glb_policy->serverlist, serverlist)) {
|
|
|
+ /* update serverlist */
|
|
|
+ if (serverlist->num_servers > 0) {
|
|
|
+ if (grpc_grpclb_serverlist_equals(glb_policy->serverlist,
|
|
|
+ serverlist)) {
|
|
|
+ if (grpc_lb_glb_trace) {
|
|
|
+ gpr_log(GPR_INFO,
|
|
|
+ "Incoming server list identical to current, ignoring.");
|
|
|
+ }
|
|
|
+ grpc_grpclb_destroy_serverlist(serverlist);
|
|
|
+ } else { /* new serverlist */
|
|
|
+ if (glb_policy->serverlist != NULL) {
|
|
|
+ /* dispose of the old serverlist */
|
|
|
+ grpc_grpclb_destroy_serverlist(glb_policy->serverlist);
|
|
|
+ }
|
|
|
+ /* and update the copy in the glb_lb_policy instance. This
|
|
|
+ * serverlist instance will be destroyed either upon the next
|
|
|
+ * update or in glb_destroy() */
|
|
|
+ glb_policy->serverlist = serverlist;
|
|
|
+
|
|
|
+ rr_handover_locked(exec_ctx, glb_policy);
|
|
|
+ }
|
|
|
+ } else {
|
|
|
if (grpc_lb_glb_trace) {
|
|
|
gpr_log(GPR_INFO,
|
|
|
- "Incoming server list identical to current, ignoring.");
|
|
|
+ "Received empty server list. Picks will stay pending until "
|
|
|
+ "a response with > 0 servers is received");
|
|
|
}
|
|
|
grpc_grpclb_destroy_serverlist(serverlist);
|
|
|
- } else { /* new serverlist */
|
|
|
- if (glb_policy->serverlist != NULL) {
|
|
|
- /* dispose of the old serverlist */
|
|
|
- grpc_grpclb_destroy_serverlist(glb_policy->serverlist);
|
|
|
- }
|
|
|
- /* and update the copy in the glb_lb_policy instance. This serverlist
|
|
|
- * instance will be destroyed either upon the next update or in
|
|
|
- * glb_destroy() */
|
|
|
- glb_policy->serverlist = serverlist;
|
|
|
-
|
|
|
- rr_handover_locked(exec_ctx, glb_policy);
|
|
|
}
|
|
|
- } else {
|
|
|
- if (grpc_lb_glb_trace) {
|
|
|
- gpr_log(GPR_INFO,
|
|
|
- "Received empty server list. Picks will stay pending until a "
|
|
|
- "response with > 0 servers is received");
|
|
|
- }
|
|
|
- grpc_grpclb_destroy_serverlist(serverlist);
|
|
|
+ } else { /* serverlist == NULL */
|
|
|
+ gpr_log(GPR_ERROR, "Invalid LB response received: '%s'. Ignoring.",
|
|
|
+ grpc_dump_slice(response_slice, GPR_DUMP_ASCII | GPR_DUMP_HEX));
|
|
|
}
|
|
|
- } else { /* serverlist == NULL */
|
|
|
- gpr_log(GPR_ERROR, "Invalid LB response received: '%s'. Ignoring.",
|
|
|
- grpc_dump_slice(response_slice, GPR_DUMP_ASCII | GPR_DUMP_HEX));
|
|
|
- grpc_slice_unref_internal(exec_ctx, response_slice);
|
|
|
}
|
|
|
|
|
|
+ grpc_slice_unref_internal(exec_ctx, response_slice);
|
|
|
+
|
|
|
if (!glb_policy->shutting_down) {
|
|
|
/* keep listening for serverlist updates */
|
|
|
op->op = GRPC_OP_RECV_MESSAGE;
|
|
|
@@ -1403,9 +1639,29 @@ grpc_lb_policy_factory *grpc_glb_lb_factory_create() {
|
|
|
}
|
|
|
|
|
|
/* Plugin registration */
|
|
|
+
|
|
|
+// Only add client_load_reporting filter if the grpclb LB policy is used.
|
|
|
+static bool maybe_add_client_load_reporting_filter(
|
|
|
+ grpc_exec_ctx *exec_ctx, grpc_channel_stack_builder *builder, void *arg) {
|
|
|
+ const grpc_channel_args *args =
|
|
|
+ grpc_channel_stack_builder_get_channel_arguments(builder);
|
|
|
+ const grpc_arg *channel_arg =
|
|
|
+ grpc_channel_args_find(args, GRPC_ARG_LB_POLICY_NAME);
|
|
|
+ if (channel_arg != NULL && channel_arg->type == GRPC_ARG_STRING &&
|
|
|
+ strcmp(channel_arg->value.string, "grpclb") == 0) {
|
|
|
+ return grpc_channel_stack_builder_append_filter(
|
|
|
+ builder, (const grpc_channel_filter *)arg, NULL, NULL);
|
|
|
+ }
|
|
|
+ return true;
|
|
|
+}
|
|
|
+
|
|
|
void grpc_lb_policy_grpclb_init() {
|
|
|
grpc_register_lb_policy(grpc_glb_lb_factory_create());
|
|
|
grpc_register_tracer("glb", &grpc_lb_glb_trace);
|
|
|
+ grpc_channel_init_register_stage(GRPC_CLIENT_SUBCHANNEL,
|
|
|
+ GRPC_CHANNEL_INIT_BUILTIN_PRIORITY,
|
|
|
+ maybe_add_client_load_reporting_filter,
|
|
|
+ (void *)&grpc_client_load_reporting_filter);
|
|
|
}
|
|
|
|
|
|
void grpc_lb_policy_grpclb_shutdown() {}
|
Mark D. Roth