Full subchannel support

4 files changed, 36 insertions, 14 deletions

diff --git a/src/core/ext/filters/client_channel/client_channel.cc b/src/core/ext/filters/client_channel/client_channel.cc
index 024c9d737e..13a0b95511 100644
--- a/src/core/ext/filters/client_channel/client_channel.cc
+++ b/src/core/ext/filters/client_channel/client_channel.cc
@@ -892,6 +892,7 @@ typedef struct client_channel_call_data {
   grpc_millis deadline;
   gpr_arena* arena;
   grpc_call_stack* owning_call;
+  grpc_call* call;
   grpc_call_combiner* call_combiner;
 
   grpc_core::RefCountedPtr<ServerRetryThrottleData> retry_throttle_data;
@@ -2561,7 +2562,8 @@ static void create_subchannel_call(grpc_call_element* elem, grpc_error* error) {
       calld->arena,                         // arena
       calld->pick.subchannel_call_context,  // context
       calld->call_combiner,                 // call_combiner
-      parent_data_size                      // parent_data_size
+      parent_data_size,                     // parent_data_size
+      calld->call                           // call
   };
   grpc_error* new_error = calld->pick.connected_subchannel->CreateCall(
       call_args, &calld->subchannel_call);
@@ -3092,6 +3094,7 @@ static grpc_error* cc_init_call_elem(grpc_call_element* elem,
   calld->arena = args->arena;
   calld->owning_call = args->call_stack;
   calld->call_combiner = args->call_combiner;
+  calld->call = args->call;
   if (GPR_LIKELY(chand->deadline_checking_enabled)) {
     grpc_deadline_state_init(elem, args->call_stack, args->call_combiner,
                              calld->deadline);
diff --git a/src/core/ext/filters/client_channel/client_channel_channelz.cc b/src/core/ext/filters/client_channel/client_channel_channelz.cc
index 0db9799a65..a1ecbe75a2 100644
--- a/src/core/ext/filters/client_channel/client_channel_channelz.cc
+++ b/src/core/ext/filters/client_channel/client_channel_channelz.cc
@@ -126,6 +126,8 @@ grpc_json* ClientChannelNode::RenderJson() {
   // as CallCountingAndTracingNode to populate trace and call count data.
   PopulateTrace(json);
   PopulateCallData(json);
+  // reset to the top level
+  json = top_level_json;
   PopulateChildRefs(json);
   return top_level_json;
 }
diff --git a/src/core/ext/filters/client_channel/subchannel.cc b/src/core/ext/filters/client_channel/subchannel.cc
index 8dfbd33ffe..a15cfd43c3 100644
--- a/src/core/ext/filters/client_channel/subchannel.cc
+++ b/src/core/ext/filters/client_channel/subchannel.cc
@@ -46,6 +46,7 @@
 #include "src/core/lib/iomgr/timer.h"
 #include "src/core/lib/profiling/timers.h"
 #include "src/core/lib/slice/slice_internal.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/connectivity_state.h"
@@ -640,8 +641,8 @@ static bool publish_transport_locked(grpc_subchannel* c) {
   }
 
   /* publish */
-  c->connected_subchannel.reset(
-      grpc_core::New<grpc_core::ConnectedSubchannel>(stk));
+  c->connected_subchannel.reset(grpc_core::New<grpc_core::ConnectedSubchannel>(
+      stk, c->channelz_subchannel.get()));
   gpr_log(GPR_INFO, "New connected subchannel at %p for subchannel %p",
           c->connected_subchannel.get(), c);
 
@@ -796,9 +797,12 @@ grpc_arg grpc_create_subchannel_address_arg(const grpc_resolved_address* addr) {
 
 namespace grpc_core {
 
-ConnectedSubchannel::ConnectedSubchannel(grpc_channel_stack* channel_stack)
+ConnectedSubchannel::ConnectedSubchannel(
+    grpc_channel_stack* channel_stack,
+    channelz::SubchannelNode* channelz_subchannel)
     : RefCountedWithTracing<ConnectedSubchannel>(&grpc_trace_stream_refcount),
-      channel_stack_(channel_stack) {}
+      channel_stack_(channel_stack),
+      channelz_subchannel_(channelz_subchannel) {}
 
 ConnectedSubchannel::~ConnectedSubchannel() {
   GRPC_CHANNEL_STACK_UNREF(channel_stack_, "connected_subchannel_dtor");
@@ -845,14 +849,15 @@ grpc_error* ConnectedSubchannel::CreateCall(const CallArgs& args,
   connection.release();  // Ref is passed to the grpc_subchannel_call object.
   (*call)->connection = this;
   const grpc_call_element_args call_args = {
-      callstk,           /* call_stack */
-      nullptr,           /* server_transport_data */
-      args.context,      /* context */
-      args.path,         /* path */
-      args.start_time,   /* start_time */
-      args.deadline,     /* deadline */
-      args.arena,        /* arena */
-      args.call_combiner /* call_combiner */
+      callstk,            /* call_stack */
+      nullptr,            /* server_transport_data */
+      args.context,       /* context */
+      args.path,          /* path */
+      args.start_time,    /* start_time */
+      args.deadline,      /* deadline */
+      args.arena,         /* arena */
+      args.call_combiner, /* call_combiner */
+      args.call           /* call */
   };
   grpc_error* error = grpc_call_stack_init(
       channel_stack_, 1, subchannel_call_destroy, *call, &call_args);
@@ -861,6 +866,10 @@ grpc_error* ConnectedSubchannel::CreateCall(const CallArgs& args,
     gpr_log(GPR_ERROR, "error: %s", error_string);
     return error;
   }
+  if (channelz_subchannel_ != nullptr) {
+    channelz_subchannel_->RecordCallStarted();
+    grpc_call_set_channelz_subchannel(args.call, channelz_subchannel_);
+  }
   grpc_call_stack_set_pollset_or_pollset_set(callstk, args.pollent);
   return GRPC_ERROR_NONE;
 }
diff --git a/src/core/ext/filters/client_channel/subchannel.h b/src/core/ext/filters/client_channel/subchannel.h
index dd3a2d9621..d62348488e 100644
--- a/src/core/ext/filters/client_channel/subchannel.h
+++ b/src/core/ext/filters/client_channel/subchannel.h
@@ -83,9 +83,11 @@ class ConnectedSubchannel : public RefCountedWithTracing<ConnectedSubchannel> {
     grpc_call_context_element* context;
     grpc_call_combiner* call_combiner;
     size_t parent_data_size;
+    grpc_call* call;
   };
 
-  explicit ConnectedSubchannel(grpc_channel_stack* channel_stack);
+  explicit ConnectedSubchannel(grpc_channel_stack* channel_stack,
+                               channelz::SubchannelNode* channelz_subchannel);
   ~ConnectedSubchannel();
 
   grpc_channel_stack* channel_stack() { return channel_stack_; }
@@ -94,9 +96,15 @@ class ConnectedSubchannel : public RefCountedWithTracing<ConnectedSubchannel> {
                            grpc_closure* closure);
   void Ping(grpc_closure* on_initiate, grpc_closure* on_ack);
   grpc_error* CreateCall(const CallArgs& args, grpc_subchannel_call** call);
+  channelz::SubchannelNode* channelz_subchannel() {
+    return channelz_subchannel_;
+  }
 
  private:
   grpc_channel_stack* channel_stack_;
+  // backpointer to the channelz node in this connected subchannel's
+  // owning subchannel.
+  channelz::SubchannelNode* channelz_subchannel_;
 };
 
 }  // namespace grpc_core