Merge pull request #5901 from dgquintas/lr_hook

Load Reporting hook

7 files changed, 407 insertions, 3 deletions

diff --git a/src/core/ext/census/grpc_filter.c b/src/core/ext/census/grpc_filter.c
index 5e278ef127..8c4c17ad9a 100644
--- a/src/core/ext/census/grpc_filter.c
+++ b/src/core/ext/census/grpc_filter.c
@@ -134,7 +134,9 @@ static void client_init_call_elem(grpc_exec_ctx *exec_ctx,
 }
 
 static void client_destroy_call_elem(grpc_exec_ctx *exec_ctx,
-                                     grpc_call_element *elem, void *ignored) {
+                                     grpc_call_element *elem,
+                                     const grpc_call_stats *stats,
+                                     void *ignored) {
   call_data *d = elem->call_data;
   GPR_ASSERT(d != NULL);
   /* TODO(hongyu): record rpc client stats and census_rpc_end_op here */
@@ -152,7 +154,9 @@ static void server_init_call_elem(grpc_exec_ctx *exec_ctx,
 }
 
 static void server_destroy_call_elem(grpc_exec_ctx *exec_ctx,
-                                     grpc_call_element *elem, void *ignored) {
+                                     grpc_call_element *elem,
+                                     const grpc_call_stats *stats,
+                                     void *ignored) {
   call_data *d = elem->call_data;
   GPR_ASSERT(d != NULL);
   /* TODO(hongyu): record rpc server stats and census_tracing_end_op here */
diff --git a/src/core/ext/client_config/client_channel.c b/src/core/ext/client_config/client_channel.c
index 9b5a078aec..642ae9ce59 100644
--- a/src/core/ext/client_config/client_channel.c
+++ b/src/core/ext/client_config/client_channel.c
@@ -416,6 +416,7 @@ static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
 
 /* Destructor for call_data */
 static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+                              const grpc_call_stats *stats,
                               void *and_free_memory) {
   grpc_subchannel_call_holder_destroy(exec_ctx, elem->call_data);
   gpr_free(and_free_memory);
diff --git a/src/core/ext/client_config/subchannel.c b/src/core/ext/client_config/subchannel.c
index cfd39e7cfb..f7d98ec4d9 100644
--- a/src/core/ext/client_config/subchannel.c
+++ b/src/core/ext/client_config/subchannel.c
@@ -645,7 +645,7 @@ static void subchannel_call_destroy(grpc_exec_ctx *exec_ctx, void *call,
   grpc_subchannel_call *c = call;
   GPR_TIMER_BEGIN("grpc_subchannel_call_unref.destroy", 0);
   grpc_connected_subchannel *connection = c->connection;
-  grpc_call_stack_destroy(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c), c);
+  grpc_call_stack_destroy(exec_ctx, SUBCHANNEL_CALL_TO_CALL_STACK(c), NULL, c);
   GRPC_CONNECTED_SUBCHANNEL_UNREF(exec_ctx, connection, "subchannel_call");
   GPR_TIMER_END("grpc_subchannel_call_unref.destroy", 0);
 }
diff --git a/src/core/ext/load_reporting/load_reporting.c b/src/core/ext/load_reporting/load_reporting.c
new file mode 100644
index 0000000000..60082dbaaa
--- /dev/null
+++ b/src/core/ext/load_reporting/load_reporting.c
@@ -0,0 +1,132 @@
+/*
+ *
+ * Copyright 2016, 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 <limits.h>
+#include <string.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/sync.h>
+
+#include "src/core/ext/load_reporting/load_reporting.h"
+#include "src/core/ext/load_reporting/load_reporting_filter.h"
+#include "src/core/lib/channel/channel_stack_builder.h"
+#include "src/core/lib/surface/channel_init.h"
+
+struct grpc_load_reporting_config {
+  grpc_load_reporting_fn fn;
+  void *user_data;
+};
+
+grpc_load_reporting_config *grpc_load_reporting_config_create(
+    grpc_load_reporting_fn fn, void *user_data) {
+  GPR_ASSERT(fn != NULL);
+  grpc_load_reporting_config *lrc =
+      gpr_malloc(sizeof(grpc_load_reporting_config));
+  lrc->fn = fn;
+  lrc->user_data = user_data;
+  return lrc;
+}
+
+grpc_load_reporting_config *grpc_load_reporting_config_copy(
+    grpc_load_reporting_config *src) {
+  return grpc_load_reporting_config_create(src->fn, src->user_data);
+}
+
+void grpc_load_reporting_config_destroy(grpc_load_reporting_config *lrc) {
+  gpr_free(lrc);
+}
+
+void grpc_load_reporting_config_call(
+    grpc_load_reporting_config *lrc,
+    const grpc_load_reporting_call_data *call_data) {
+  lrc->fn(call_data, lrc->user_data);
+}
+
+static bool is_load_reporting_enabled(const grpc_channel_args *a) {
+  if (a == NULL) return false;
+  for (size_t i = 0; i < a->num_args; i++) {
+    if (0 == strcmp(a->args[i].key, GRPC_ARG_ENABLE_LOAD_REPORTING)) {
+      return a->args[i].value.pointer.p != NULL;
+    }
+  }
+  return false;
+}
+
+static bool maybe_add_load_reporting_filter(grpc_channel_stack_builder *builder,
+                                            void *arg) {
+  const grpc_channel_args *args =
+      grpc_channel_stack_builder_get_channel_arguments(builder);
+  if (is_load_reporting_enabled(args)) {
+    return grpc_channel_stack_builder_prepend_filter(
+        builder, (const grpc_channel_filter *)arg, NULL, NULL);
+  }
+  return true;
+}
+
+static void lrd_arg_destroy(void *p) { grpc_load_reporting_config_destroy(p); }
+
+static void *lrd_arg_copy(void *p) {
+  return grpc_load_reporting_config_copy(p);
+}
+
+static int lrd_arg_cmp(void *a, void *b) {
+  grpc_load_reporting_config *lhs = a;
+  grpc_load_reporting_config *rhs = b;
+  return !(lhs->fn == rhs->fn && lhs->user_data == rhs->user_data);
+}
+
+static const grpc_arg_pointer_vtable lrd_ptr_vtable = {
+    lrd_arg_copy, lrd_arg_destroy, lrd_arg_cmp};
+
+grpc_arg grpc_load_reporting_config_create_arg(
+    grpc_load_reporting_config *lrc) {
+  grpc_arg arg;
+  arg.type = GRPC_ARG_POINTER;
+  arg.key = GRPC_ARG_ENABLE_LOAD_REPORTING;
+  arg.value.pointer.p = lrc;
+  arg.value.pointer.vtable = &lrd_ptr_vtable;
+  return arg;
+}
+
+/* Plugin registration */
+
+void grpc_load_reporting_plugin_init(void) {
+  grpc_channel_init_register_stage(GRPC_CLIENT_CHANNEL, INT_MAX,
+                                   maybe_add_load_reporting_filter,
+                                   (void *)&grpc_load_reporting_filter);
+  grpc_channel_init_register_stage(GRPC_SERVER_CHANNEL, INT_MAX,
+                                   maybe_add_load_reporting_filter,
+                                   (void *)&grpc_load_reporting_filter);
+}
+
+void grpc_load_reporting_plugin_shutdown() {}
diff --git a/src/core/ext/load_reporting/load_reporting.h b/src/core/ext/load_reporting/load_reporting.h
new file mode 100644
index 0000000000..316cd89bd7
--- /dev/null
+++ b/src/core/ext/load_reporting/load_reporting.h
@@ -0,0 +1,75 @@
+/*
+ *
+ * Copyright 2016, 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_LOAD_REPORTING_LOAD_REPORTING_H
+#define GRPC_CORE_EXT_LOAD_REPORTING_LOAD_REPORTING_H
+
+#include "src/core/lib/iomgr/closure.h"
+#include "src/core/lib/surface/call.h"
+
+typedef struct grpc_load_reporting_config grpc_load_reporting_config;
+
+/** Call information to be passed to the provided load reporting function upon
+ * completion of the call */
+typedef struct grpc_load_reporting_call_data {
+  const grpc_call_stats *stats;   /**< Stats for the call */
+  const char *trailing_md_string; /**< LR trailing metadata info */
+} grpc_load_reporting_call_data;
+
+/** Custom function to be called by the load reporting filter. */
+typedef void (*grpc_load_reporting_fn)(
+    const grpc_load_reporting_call_data *call_data, void *user_data);
+
+/** Register \a fn as the function to be invoked by the load reporting filter.
+ * \a fn will be invoked at the beginning and at the end of the call.
+ *
+ * For the first invocation, \a fn's first argument
+ * (grpc_load_reporting_call_data*) will be NULL. \a user_data is always passed
+ * as-is. */
+grpc_load_reporting_config *grpc_load_reporting_config_create(
+    grpc_load_reporting_fn fn, void *user_data);
+
+grpc_load_reporting_config *grpc_load_reporting_config_copy(
+    grpc_load_reporting_config *src);
+
+void grpc_load_reporting_config_destroy(grpc_load_reporting_config *lrc);
+
+/** Invoke the function registered by \a grpc_load_reporting_init. */
+void grpc_load_reporting_config_call(
+    grpc_load_reporting_config *lrc,
+    const grpc_load_reporting_call_data *call_data);
+
+/** Return a \a grpc_arg enabling load reporting */
+grpc_arg grpc_load_reporting_config_create_arg(grpc_load_reporting_config *lrc);
+
+#endif /* GRPC_CORE_EXT_LOAD_REPORTING_LOAD_REPORTING_H */
diff --git a/src/core/ext/load_reporting/load_reporting_filter.c b/src/core/ext/load_reporting/load_reporting_filter.c
new file mode 100644
index 0000000000..f49730fac3
--- /dev/null
+++ b/src/core/ext/load_reporting/load_reporting_filter.c
@@ -0,0 +1,151 @@
+/*
+ *
+ * Copyright 2016, 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 <grpc/support/log.h>
+#include <grpc/support/string_util.h>
+#include <grpc/support/sync.h>
+#include <string.h>
+
+#include "src/core/ext/load_reporting/load_reporting.h"
+#include "src/core/ext/load_reporting/load_reporting_filter.h"
+#include "src/core/lib/channel/channel_args.h"
+#include "src/core/lib/profiling/timers.h"
+#include "src/core/lib/transport/static_metadata.h"
+
+typedef struct call_data { const char *trailing_md_string; } call_data;
+typedef struct channel_data {
+  gpr_mu mu;
+  grpc_load_reporting_config *lrc;
+} channel_data;
+
+static void invoke_lr_fn_locked(grpc_load_reporting_config *lrc,
+                                grpc_load_reporting_call_data *lr_call_data) {
+  GPR_TIMER_BEGIN("load_reporting_config_fn", 0);
+  grpc_load_reporting_config_call(lrc, lr_call_data);
+  GPR_TIMER_END("load_reporting_config_fn", 0);
+}
+
+/* Constructor for call_data */
+static void init_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+                           grpc_call_element_args *args) {
+  call_data *calld = elem->call_data;
+  memset(calld, 0, sizeof(call_data));
+}
+
+/* Destructor for call_data */
+static void destroy_call_elem(grpc_exec_ctx *exec_ctx, grpc_call_element *elem,
+                              const grpc_call_stats *stats, void *ignored) {
+  channel_data *chand = elem->channel_data;
+  call_data *calld = elem->call_data;
+
+  grpc_load_reporting_call_data lr_call_data = {stats,
+                                                calld->trailing_md_string};
+
+  gpr_mu_lock(&chand->mu);
+  invoke_lr_fn_locked(chand->lrc, &lr_call_data);
+  gpr_mu_unlock(&chand->mu);
+}
+
+/* Constructor for channel_data */
+static void init_channel_elem(grpc_exec_ctx *exec_ctx,
+                              grpc_channel_element *elem,
+                              grpc_channel_element_args *args) {
+  GPR_ASSERT(!args->is_last);
+
+  channel_data *chand = elem->channel_data;
+  memset(chand, 0, sizeof(channel_data));
+
+  gpr_mu_init(&chand->mu);
+  for (size_t i = 0; i < args->channel_args->num_args; i++) {
+    if (0 == strcmp(args->channel_args->args[i].key,
+                    GRPC_ARG_ENABLE_LOAD_REPORTING)) {
+      grpc_load_reporting_config *arg_lrc =
+          args->channel_args->args[i].value.pointer.p;
+      chand->lrc = grpc_load_reporting_config_copy(arg_lrc);
+      GPR_ASSERT(chand->lrc != NULL);
+      break;
+    }
+  }
+  GPR_ASSERT(chand->lrc != NULL); /* arg actually found */
+
+  gpr_mu_lock(&chand->mu);
+  invoke_lr_fn_locked(chand->lrc, NULL);
+  gpr_mu_unlock(&chand->mu);
+}
+
+/* Destructor for channel data */
+static void destroy_channel_elem(grpc_exec_ctx *exec_ctx,
+                                 grpc_channel_element *elem) {
+  channel_data *chand = elem->channel_data;
+  gpr_mu_destroy(&chand->mu);
+  grpc_load_reporting_config_destroy(chand->lrc);
+}
+
+static grpc_mdelem *lr_trailing_md_filter(void *user_data, grpc_mdelem *md) {
+  grpc_call_element *elem = user_data;
+  call_data *calld = elem->call_data;
+
+  if (md->key == GRPC_MDSTR_LOAD_REPORTING) {
+    calld->trailing_md_string = gpr_strdup(grpc_mdstr_as_c_string(md->value));
+    return NULL;
+  }
+
+  return md;
+}
+
+static void lr_start_transport_stream_op(grpc_exec_ctx *exec_ctx,
+                                         grpc_call_element *elem,
+                                         grpc_transport_stream_op *op) {
+  GPR_TIMER_BEGIN("lr_start_transport_stream_op", 0);
+
+  if (op->send_trailing_metadata) {
+    grpc_metadata_batch_filter(op->send_trailing_metadata,
+                               lr_trailing_md_filter, elem);
+  }
+  grpc_call_next_op(exec_ctx, elem, op);
+
+  GPR_TIMER_END("lr_start_transport_stream_op", 0);
+}
+
+const grpc_channel_filter grpc_load_reporting_filter = {
+    lr_start_transport_stream_op,
+    grpc_channel_next_op,
+    sizeof(call_data),
+    init_call_elem,
+    grpc_call_stack_ignore_set_pollset,
+    destroy_call_elem,
+    sizeof(channel_data),
+    init_channel_elem,
+    destroy_channel_elem,
+    grpc_call_next_get_peer,
+    "load_reporting"};
diff --git a/src/core/ext/load_reporting/load_reporting_filter.h b/src/core/ext/load_reporting/load_reporting_filter.h
new file mode 100644
index 0000000000..f69cd6fdc6
--- /dev/null
+++ b/src/core/ext/load_reporting/load_reporting_filter.h
@@ -0,0 +1,41 @@
+/*
+ *
+ * Copyright 2016, 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_LOAD_REPORTING_LOAD_REPORTING_FILTER_H
+#define GRPC_CORE_EXT_LOAD_REPORTING_LOAD_REPORTING_FILTER_H
+
+#include "src/core/lib/channel/channel_stack.h"
+
+extern const grpc_channel_filter grpc_load_reporting_filter;
+
+#endif /* GRPC_CORE_EXT_LOAD_REPORTING_LOAD_REPORTING_FILTER_H */