mirror of https://github.com/grpc/grpc.git
Merge pull request #10821 from markdroth/grpclb_client_side_load_reporting
Implement client-side load reporting for grpclb.pull/10845/head
Mark D. Roth
8 years ago
committed by
GitHub
commit
12056f1a0c
33 changed files with 1012 additions and 79 deletions
@ -0,0 +1,153 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2017, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#include "src/core/ext/filters/client_channel/lb_policy/grpclb/client_load_reporting_filter.h" |
||||
|
||||
#include <grpc/support/atm.h> |
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_client_stats.h" |
||||
#include "src/core/lib/iomgr/error.h" |
||||
#include "src/core/lib/profiling/timers.h" |
||||
|
||||
static grpc_error *init_channel_elem(grpc_exec_ctx *exec_ctx, |
||||
grpc_channel_element *elem, |
||||
grpc_channel_element_args *args) { |
||||
return GRPC_ERROR_NONE; |
||||
} |
||||
|
||||
static void destroy_channel_elem(grpc_exec_ctx *exec_ctx, |
||||
grpc_channel_element *elem) {} |
||||
|
||||
typedef struct { |
||||
// Stats object to update.
|
||||
grpc_grpclb_client_stats *client_stats; |
||||
// State for intercepting send_initial_metadata.
|
||||
grpc_closure on_complete_for_send; |
||||
grpc_closure *original_on_complete_for_send; |
||||
bool send_initial_metadata_succeeded; |
||||
// State for intercepting recv_initial_metadata.
|
||||
grpc_closure recv_initial_metadata_ready; |
||||
grpc_closure *original_recv_initial_metadata_ready; |
||||
bool recv_initial_metadata_succeeded; |
||||
} call_data; |
||||
|
||||
static void on_complete_for_send(grpc_exec_ctx *exec_ctx, void *arg, |
||||
grpc_error *error) { |
||||
call_data *calld = arg; |
||||
if (error == GRPC_ERROR_NONE) { |
||||
calld->send_initial_metadata_succeeded = true; |
||||
} |
||||
grpc_closure_run(exec_ctx, calld->original_on_complete_for_send, |
||||
GRPC_ERROR_REF(error)); |
||||
} |
||||
|
||||
static void recv_initial_metadata_ready(grpc_exec_ctx *exec_ctx, void *arg, |
||||
grpc_error *error) { |
||||
call_data *calld = arg; |
||||
if (error == GRPC_ERROR_NONE) { |
||||
calld->recv_initial_metadata_succeeded = true; |
||||
} |
||||
grpc_closure_run(exec_ctx, calld->original_recv_initial_metadata_ready, |
||||
GRPC_ERROR_REF(error)); |
||||
} |
||||
|
||||
static grpc_error *init_call_elem(grpc_exec_ctx *exec_ctx, |
||||
grpc_call_element *elem, |
||||
const grpc_call_element_args *args) { |
||||
call_data *calld = elem->call_data; |
||||
// Get stats object from context and take a ref.
|
||||
GPR_ASSERT(args->context != NULL); |
||||
GPR_ASSERT(args->context[GRPC_GRPCLB_CLIENT_STATS].value != NULL); |
||||
calld->client_stats = grpc_grpclb_client_stats_ref( |
||||
args->context[GRPC_GRPCLB_CLIENT_STATS].value); |
||||
// Record call started.
|
||||
grpc_grpclb_client_stats_add_call_started(calld->client_stats); |
||||
return GRPC_ERROR_NONE; |
||||
} |
||||
|
||||
static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem, |
||||
const grpc_call_final_info *final_info, |
||||
grpc_closure *ignored) { |
||||
call_data *calld = elem->call_data; |
||||
// Record call finished, optionally setting client_failed_to_send and
|
||||
// received.
|
||||
grpc_grpclb_client_stats_add_call_finished( |
||||
false /* drop_for_rate_limiting */, false /* drop_for_load_balancing */, |
||||
!calld->send_initial_metadata_succeeded /* client_failed_to_send */, |
||||
calld->recv_initial_metadata_succeeded /* known_received */, |
||||
calld->client_stats); |
||||
// All done, so unref the stats object.
|
||||
grpc_grpclb_client_stats_unref(calld->client_stats); |
||||
} |
||||
|
||||
static void start_transport_stream_op_batch( |
||||
grpc_exec_ctx *exec_ctx, grpc_call_element *elem, |
||||
grpc_transport_stream_op_batch *batch) { |
||||
call_data *calld = elem->call_data; |
||||
GPR_TIMER_BEGIN("clr_start_transport_stream_op_batch", 0); |
||||
// Intercept send_initial_metadata.
|
||||
if (batch->send_initial_metadata) { |
||||
calld->original_on_complete_for_send = batch->on_complete; |
||||
grpc_closure_init(&calld->on_complete_for_send, on_complete_for_send, calld, |
||||
grpc_schedule_on_exec_ctx); |
||||
batch->on_complete = &calld->on_complete_for_send; |
||||
} |
||||
// Intercept recv_initial_metadata.
|
||||
if (batch->recv_initial_metadata) { |
||||
calld->original_recv_initial_metadata_ready = |
||||
batch->payload->recv_initial_metadata.recv_initial_metadata_ready; |
||||
grpc_closure_init(&calld->recv_initial_metadata_ready, |
||||
recv_initial_metadata_ready, calld, |
||||
grpc_schedule_on_exec_ctx); |
||||
batch->payload->recv_initial_metadata.recv_initial_metadata_ready = |
||||
&calld->recv_initial_metadata_ready; |
||||
} |
||||
// Chain to next filter.
|
||||
grpc_call_next_op(exec_ctx, elem, batch); |
||||
GPR_TIMER_END("clr_start_transport_stream_op_batch", 0); |
||||
} |
||||
|
||||
const grpc_channel_filter grpc_client_load_reporting_filter = { |
||||
start_transport_stream_op_batch, |
||||
grpc_channel_next_op, |
||||
sizeof(call_data), |
||||
init_call_elem, |
||||
grpc_call_stack_ignore_set_pollset_or_pollset_set, |
||||
destroy_call_elem, |
||||
0, // sizeof(channel_data)
|
||||
init_channel_elem, |
||||
destroy_channel_elem, |
||||
grpc_call_next_get_peer, |
||||
grpc_channel_next_get_info, |
||||
"client_load_reporting"}; |
||||
@ -0,0 +1,42 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2017, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_LB_POLICY_GRPCLB_CLIENT_LOAD_REPORTING_FILTER_H |
||||
#define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_LB_POLICY_GRPCLB_CLIENT_LOAD_REPORTING_FILTER_H |
||||
|
||||
#include "src/core/lib/channel/channel_stack.h" |
||||
|
||||
extern const grpc_channel_filter grpc_client_load_reporting_filter; |
||||
|
||||
#endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_LB_POLICY_GRPCLB_CLIENT_LOAD_REPORTING_FILTER_H \ |
||||
*/ |
||||
@ -0,0 +1,133 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2017, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#include "src/core/ext/filters/client_channel/lb_policy/grpclb/grpclb_client_stats.h" |
||||
|
||||
#include <grpc/support/alloc.h> |
||||
#include <grpc/support/atm.h> |
||||
#include <grpc/support/sync.h> |
||||
#include <grpc/support/useful.h> |
||||
|
||||
#include "src/core/lib/channel/channel_args.h" |
||||
|
||||
#define GRPC_ARG_GRPCLB_CLIENT_STATS "grpc.grpclb_client_stats" |
||||
|
||||
struct grpc_grpclb_client_stats { |
||||
gpr_refcount refs; |
||||
gpr_atm num_calls_started; |
||||
gpr_atm num_calls_finished; |
||||
gpr_atm num_calls_finished_with_drop_for_rate_limiting; |
||||
gpr_atm num_calls_finished_with_drop_for_load_balancing; |
||||
gpr_atm num_calls_finished_with_client_failed_to_send; |
||||
gpr_atm num_calls_finished_known_received; |
||||
}; |
||||
|
||||
grpc_grpclb_client_stats* grpc_grpclb_client_stats_create() { |
||||
grpc_grpclb_client_stats* client_stats = gpr_zalloc(sizeof(*client_stats)); |
||||
gpr_ref_init(&client_stats->refs, 1); |
||||
return client_stats; |
||||
} |
||||
|
||||
grpc_grpclb_client_stats* grpc_grpclb_client_stats_ref( |
||||
grpc_grpclb_client_stats* client_stats) { |
||||
gpr_ref(&client_stats->refs); |
||||
return client_stats; |
||||
} |
||||
|
||||
void grpc_grpclb_client_stats_unref(grpc_grpclb_client_stats* client_stats) { |
||||
if (gpr_unref(&client_stats->refs)) { |
||||
gpr_free(client_stats); |
||||
} |
||||
} |
||||
|
||||
void grpc_grpclb_client_stats_add_call_started( |
||||
grpc_grpclb_client_stats* client_stats) { |
||||
gpr_atm_full_fetch_add(&client_stats->num_calls_started, (gpr_atm)1); |
||||
} |
||||
|
||||
void grpc_grpclb_client_stats_add_call_finished( |
||||
bool finished_with_drop_for_rate_limiting, |
||||
bool finished_with_drop_for_load_balancing, |
||||
bool finished_with_client_failed_to_send, bool finished_known_received, |
||||
grpc_grpclb_client_stats* client_stats) { |
||||
gpr_atm_full_fetch_add(&client_stats->num_calls_finished, (gpr_atm)1); |
||||
if (finished_with_drop_for_rate_limiting) { |
||||
gpr_atm_full_fetch_add( |
||||
&client_stats->num_calls_finished_with_drop_for_rate_limiting, |
||||
(gpr_atm)1); |
||||
} |
||||
if (finished_with_drop_for_load_balancing) { |
||||
gpr_atm_full_fetch_add( |
||||
&client_stats->num_calls_finished_with_drop_for_load_balancing, |
||||
(gpr_atm)1); |
||||
} |
||||
if (finished_with_client_failed_to_send) { |
||||
gpr_atm_full_fetch_add( |
||||
&client_stats->num_calls_finished_with_client_failed_to_send, |
||||
(gpr_atm)1); |
||||
} |
||||
if (finished_known_received) { |
||||
gpr_atm_full_fetch_add(&client_stats->num_calls_finished_known_received, |
||||
(gpr_atm)1); |
||||
} |
||||
} |
||||
|
||||
static void atomic_get_and_reset_counter(int64_t* value, gpr_atm* counter) { |
||||
*value = (int64_t)gpr_atm_acq_load(counter); |
||||
gpr_atm_full_fetch_add(counter, (gpr_atm)(-*value)); |
||||
} |
||||
|
||||
void grpc_grpclb_client_stats_get( |
||||
grpc_grpclb_client_stats* client_stats, int64_t* num_calls_started, |
||||
int64_t* num_calls_finished, |
||||
int64_t* num_calls_finished_with_drop_for_rate_limiting, |
||||
int64_t* num_calls_finished_with_drop_for_load_balancing, |
||||
int64_t* num_calls_finished_with_client_failed_to_send, |
||||
int64_t* num_calls_finished_known_received) { |
||||
atomic_get_and_reset_counter(num_calls_started, |
||||
&client_stats->num_calls_started); |
||||
atomic_get_and_reset_counter(num_calls_finished, |
||||
&client_stats->num_calls_finished); |
||||
atomic_get_and_reset_counter( |
||||
num_calls_finished_with_drop_for_rate_limiting, |
||||
&client_stats->num_calls_finished_with_drop_for_rate_limiting); |
||||
atomic_get_and_reset_counter( |
||||
num_calls_finished_with_drop_for_load_balancing, |
||||
&client_stats->num_calls_finished_with_drop_for_load_balancing); |
||||
atomic_get_and_reset_counter( |
||||
num_calls_finished_with_client_failed_to_send, |
||||
&client_stats->num_calls_finished_with_client_failed_to_send); |
||||
atomic_get_and_reset_counter( |
||||
num_calls_finished_known_received, |
||||
&client_stats->num_calls_finished_known_received); |
||||
} |
||||
@ -0,0 +1,65 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2017, Google Inc. |
||||
* All rights reserved. |
||||
* |
||||
* Redistribution and use in source and binary forms, with or without |
||||
* modification, are permitted provided that the following conditions are |
||||
* met: |
||||
* |
||||
* * Redistributions of source code must retain the above copyright |
||||
* notice, this list of conditions and the following disclaimer. |
||||
* * Redistributions in binary form must reproduce the above |
||||
* copyright notice, this list of conditions and the following disclaimer |
||||
* in the documentation and/or other materials provided with the |
||||
* distribution. |
||||
* * Neither the name of Google Inc. nor the names of its |
||||
* contributors may be used to endorse or promote products derived from |
||||
* this software without specific prior written permission. |
||||
* |
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_LB_POLICY_GRPCLB_GRPCLB_CLIENT_STATS_H |
||||
#define GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_LB_POLICY_GRPCLB_GRPCLB_CLIENT_STATS_H |
||||
|
||||
#include <stdbool.h> |
||||
|
||||
#include <grpc/impl/codegen/grpc_types.h> |
||||
|
||||
typedef struct grpc_grpclb_client_stats grpc_grpclb_client_stats; |
||||
|
||||
grpc_grpclb_client_stats* grpc_grpclb_client_stats_create(); |
||||
grpc_grpclb_client_stats* grpc_grpclb_client_stats_ref( |
||||
grpc_grpclb_client_stats* client_stats); |
||||
void grpc_grpclb_client_stats_unref(grpc_grpclb_client_stats* client_stats); |
||||
|
||||
void grpc_grpclb_client_stats_add_call_started( |
||||
grpc_grpclb_client_stats* client_stats); |
||||
void grpc_grpclb_client_stats_add_call_finished( |
||||
bool finished_with_drop_for_rate_limiting, |
||||
bool finished_with_drop_for_load_balancing, |
||||
bool finished_with_client_failed_to_send, bool finished_known_received, |
||||
grpc_grpclb_client_stats* client_stats); |
||||
|
||||
void grpc_grpclb_client_stats_get( |
||||
grpc_grpclb_client_stats* client_stats, int64_t* num_calls_started, |
||||
int64_t* num_calls_finished, |
||||
int64_t* num_calls_finished_with_drop_for_rate_limiting, |
||||
int64_t* num_calls_finished_with_drop_for_load_balancing, |
||||
int64_t* num_calls_finished_with_client_failed_to_send, |
||||
int64_t* num_calls_finished_known_received); |
||||
|
||||
#endif /* GRPC_CORE_EXT_FILTERS_CLIENT_CHANNEL_LB_POLICY_GRPCLB_GRPCLB_CLIENT_STATS_H \ |
||||
*/ |
||||
Loading…
Reference in new issue