1 files changed, 367 insertions, 61 deletions

diff --git a/src/core/lib/channel/channelz.cc b/src/core/lib/channel/channelz.cc
index 9d6002ed8a..8d589f5983 100644
--- a/src/core/lib/channel/channelz.cc
+++ b/src/core/lib/channel/channelz.cc
@@ -30,44 +30,122 @@
 
 #include "src/core/lib/channel/channelz_registry.h"
 #include "src/core/lib/channel/status_util.h"
+#include "src/core/lib/gpr/host_port.h"
 #include "src/core/lib/gpr/string.h"
 #include "src/core/lib/gpr/useful.h"
 #include "src/core/lib/gprpp/memory.h"
 #include "src/core/lib/iomgr/error.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/slice/b64.h"
 #include "src/core/lib/slice/slice_internal.h"
 #include "src/core/lib/surface/channel.h"
+#include "src/core/lib/surface/server.h"
 #include "src/core/lib/transport/error_utils.h"
+#include "src/core/lib/uri/uri_parser.h"
 
 namespace grpc_core {
 namespace channelz {
 
-ChannelNode::ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes,
-                         bool is_top_level_channel)
-    : channel_(channel),
-      target_(nullptr),
-      channel_uuid_(-1),
-      is_top_level_channel_(is_top_level_channel) {
-  trace_.Init(channel_tracer_max_nodes);
-  target_ = UniquePtr<char>(grpc_channel_get_target(channel_));
-  channel_uuid_ = ChannelzRegistry::RegisterChannelNode(this);
-  gpr_atm_no_barrier_store(&last_call_started_millis_,
-                           (gpr_atm)ExecCtx::Get()->Now());
+BaseNode::BaseNode(EntityType type) : type_(type), uuid_(-1) {
+  // The registry will set uuid_ under its lock.
+  ChannelzRegistry::Register(this);
 }
 
-ChannelNode::~ChannelNode() {
-  trace_.Destroy();
-  ChannelzRegistry::UnregisterChannelNode(channel_uuid_);
+BaseNode::~BaseNode() { ChannelzRegistry::Unregister(uuid_); }
+
+char* BaseNode::RenderJsonString() {
+  grpc_json* json = RenderJson();
+  GPR_ASSERT(json != nullptr);
+  char* json_str = grpc_json_dump_to_string(json, 0);
+  grpc_json_destroy(json);
+  return json_str;
 }
 
-void ChannelNode::RecordCallStarted() {
-  gpr_atm_no_barrier_fetch_add(&calls_started_, (gpr_atm)1);
-  gpr_atm_no_barrier_store(&last_call_started_millis_,
-                           (gpr_atm)ExecCtx::Get()->Now());
+CallCountingHelper::CallCountingHelper() {
+  num_cores_ = GPR_MAX(1, gpr_cpu_num_cores());
+  per_cpu_counter_data_storage_ = static_cast<AtomicCounterData*>(
+      gpr_zalloc(sizeof(AtomicCounterData) * num_cores_));
 }
 
-void ChannelNode::PopulateConnectivityState(grpc_json* json) {}
+CallCountingHelper::~CallCountingHelper() {
+  gpr_free(per_cpu_counter_data_storage_);
+}
 
-void ChannelNode::PopulateChildRefs(grpc_json* json) {}
+void CallCountingHelper::RecordCallStarted() {
+  gpr_atm_no_barrier_fetch_add(
+      &per_cpu_counter_data_storage_[grpc_core::ExecCtx::Get()->starting_cpu()]
+           .calls_started,
+      static_cast<gpr_atm>(1));
+  gpr_atm_no_barrier_store(
+      &per_cpu_counter_data_storage_[grpc_core::ExecCtx::Get()->starting_cpu()]
+           .last_call_started_millis,
+      (gpr_atm)ExecCtx::Get()->Now());
+}
+
+void CallCountingHelper::RecordCallFailed() {
+  gpr_atm_no_barrier_fetch_add(
+      &per_cpu_counter_data_storage_[grpc_core::ExecCtx::Get()->starting_cpu()]
+           .calls_failed,
+      static_cast<gpr_atm>(1));
+}
+
+void CallCountingHelper::RecordCallSucceeded() {
+  gpr_atm_no_barrier_fetch_add(
+      &per_cpu_counter_data_storage_[grpc_core::ExecCtx::Get()->starting_cpu()]
+           .calls_succeeded,
+      static_cast<gpr_atm>(1));
+}
+
+void CallCountingHelper::CollectData(CounterData* out) {
+  for (size_t core = 0; core < num_cores_; ++core) {
+    out->calls_started += gpr_atm_no_barrier_load(
+        &per_cpu_counter_data_storage_[core].calls_started);
+    out->calls_succeeded += gpr_atm_no_barrier_load(
+        &per_cpu_counter_data_storage_[core].calls_succeeded);
+    out->calls_failed += gpr_atm_no_barrier_load(
+        &per_cpu_counter_data_storage_[core].calls_failed);
+    gpr_atm last_call = gpr_atm_no_barrier_load(
+        &per_cpu_counter_data_storage_[core].last_call_started_millis);
+    if (last_call > out->last_call_started_millis) {
+      out->last_call_started_millis = last_call;
+    }
+  }
+}
+
+void CallCountingHelper::PopulateCallCounts(grpc_json* json) {
+  grpc_json* json_iterator = nullptr;
+  CounterData data;
+  CollectData(&data);
+  if (data.calls_started != 0) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "callsStarted", data.calls_started);
+  }
+  if (data.calls_succeeded != 0) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "callsSucceeded", data.calls_succeeded);
+  }
+  if (data.calls_failed) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "callsFailed", data.calls_failed);
+  }
+  if (data.calls_started != 0) {
+    gpr_timespec ts = grpc_millis_to_timespec(data.last_call_started_millis,
+                                              GPR_CLOCK_REALTIME);
+    json_iterator =
+        grpc_json_create_child(json_iterator, json, "lastCallStartedTimestamp",
+                               gpr_format_timespec(ts), GRPC_JSON_STRING, true);
+  }
+}
+
+ChannelNode::ChannelNode(grpc_channel* channel, size_t channel_tracer_max_nodes,
+                         bool is_top_level_channel)
+    : BaseNode(is_top_level_channel ? EntityType::kTopLevelChannel
+                                    : EntityType::kInternalChannel),
+      channel_(channel),
+      target_(UniquePtr<char>(grpc_channel_get_target(channel_))),
+      trace_(channel_tracer_max_nodes) {}
+
+ChannelNode::~ChannelNode() {}
 
 grpc_json* ChannelNode::RenderJson() {
   // We need to track these three json objects to build our object
@@ -80,7 +158,7 @@ grpc_json* ChannelNode::RenderJson() {
   json = json_iterator;
   json_iterator = nullptr;
   json_iterator = grpc_json_add_number_string_child(json, json_iterator,
-                                                    "channelId", channel_uuid_);
+                                                    "channelId", uuid());
   // reset json iterators to top level object
   json = top_level_json;
   json_iterator = nullptr;
@@ -89,51 +167,28 @@ grpc_json* ChannelNode::RenderJson() {
                                            GRPC_JSON_OBJECT, false);
   json = data;
   json_iterator = nullptr;
+  // template method. Child classes may override this to add their specific
+  // functionality.
   PopulateConnectivityState(json);
+  // populate the target.
   GPR_ASSERT(target_.get() != nullptr);
-  json_iterator = grpc_json_create_child(
-      json_iterator, json, "target", target_.get(), GRPC_JSON_STRING, false);
+  grpc_json_create_child(nullptr, json, "target", target_.get(),
+                         GRPC_JSON_STRING, false);
   // fill in the channel trace if applicable
-  grpc_json* trace = trace_->RenderJson();
-  if (trace != nullptr) {
-    // we manually link up and fill the child since it was created for us in
-    // ChannelTrace::RenderJson
-    trace->key = "trace";  // this object is named trace in channelz.proto
-    json_iterator = grpc_json_link_child(json, trace, json_iterator);
-  }
-  // reset the parent to be the data object.
-  json = data;
-  json_iterator = nullptr;
-  if (calls_started_ != 0) {
-    json_iterator = grpc_json_add_number_string_child(
-        json, json_iterator, "callsStarted", calls_started_);
-  }
-  if (calls_succeeded_ != 0) {
-    json_iterator = grpc_json_add_number_string_child(
-        json, json_iterator, "callsSucceeded", calls_succeeded_);
+  grpc_json* trace_json = trace_.RenderJson();
+  if (trace_json != nullptr) {
+    trace_json->key = "trace";  // this object is named trace in channelz.proto
+    grpc_json_link_child(json, trace_json, nullptr);
   }
-  if (calls_failed_) {
-    json_iterator = grpc_json_add_number_string_child(
-        json, json_iterator, "callsFailed", calls_failed_);
-  }
-  gpr_timespec ts =
-      grpc_millis_to_timespec(last_call_started_millis_, GPR_CLOCK_REALTIME);
-  json_iterator =
-      grpc_json_create_child(json_iterator, json, "lastCallStartedTimestamp",
-                             gpr_format_timespec(ts), GRPC_JSON_STRING, true);
+  // ask CallCountingHelper to populate trace and call count data.
+  call_counter_.PopulateCallCounts(json);
   json = top_level_json;
-  json_iterator = nullptr;
+  // template method. Child classes may override this to add their specific
+  // functionality.
   PopulateChildRefs(json);
   return top_level_json;
 }
 
-char* ChannelNode::RenderJsonString() {
-  grpc_json* json = RenderJson();
-  char* json_str = grpc_json_dump_to_string(json, 0);
-  grpc_json_destroy(json);
-  return json_str;
-}
-
 RefCountedPtr<ChannelNode> ChannelNode::MakeChannelNode(
     grpc_channel* channel, size_t channel_tracer_max_nodes,
     bool is_top_level_channel) {
@@ -141,12 +196,263 @@ RefCountedPtr<ChannelNode> ChannelNode::MakeChannelNode(
       channel, channel_tracer_max_nodes, is_top_level_channel);
 }
 
-SubchannelNode::SubchannelNode() {
-  subchannel_uuid_ = ChannelzRegistry::RegisterSubchannelNode(this);
+ServerNode::ServerNode(grpc_server* server, size_t channel_tracer_max_nodes)
+    : BaseNode(EntityType::kServer),
+      server_(server),
+      trace_(channel_tracer_max_nodes) {}
+
+ServerNode::~ServerNode() {}
+
+char* ServerNode::RenderServerSockets(intptr_t start_socket_id) {
+  grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT);
+  grpc_json* json = top_level_json;
+  grpc_json* json_iterator = nullptr;
+  ChildRefsList socket_refs;
+  // uuids index into entities one-off (idx 0 is really uuid 1, since 0 is
+  // reserved). However, we want to support requests coming in with
+  // start_server_id=0, which signifies "give me everything."
+  size_t start_idx = start_socket_id == 0 ? 0 : start_socket_id - 1;
+  grpc_server_populate_server_sockets(server_, &socket_refs, start_idx);
+  if (!socket_refs.empty()) {
+    // create list of socket refs
+    grpc_json* array_parent = grpc_json_create_child(
+        nullptr, json, "socketRef", nullptr, GRPC_JSON_ARRAY, false);
+    for (size_t i = 0; i < socket_refs.size(); ++i) {
+      json_iterator =
+          grpc_json_create_child(json_iterator, array_parent, nullptr, nullptr,
+                                 GRPC_JSON_OBJECT, false);
+      grpc_json_add_number_string_child(json_iterator, nullptr, "socketId",
+                                        socket_refs[i]);
+    }
+  }
+  // For now we do not have any pagination rules. In the future we could
+  // pick a constant for max_channels_sent for a GetServers request.
+  // Tracking: https://github.com/grpc/grpc/issues/16019.
+  json_iterator = grpc_json_create_child(nullptr, json, "end", nullptr,
+                                         GRPC_JSON_TRUE, false);
+  char* json_str = grpc_json_dump_to_string(top_level_json, 0);
+  grpc_json_destroy(top_level_json);
+  return json_str;
 }
 
-SubchannelNode::~SubchannelNode() {
-  ChannelzRegistry::UnregisterSubchannelNode(subchannel_uuid_);
+grpc_json* ServerNode::RenderJson() {
+  // We need to track these three json objects to build our object
+  grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT);
+  grpc_json* json = top_level_json;
+  grpc_json* json_iterator = nullptr;
+  // create and fill the ref child
+  json_iterator = grpc_json_create_child(json_iterator, json, "ref", nullptr,
+                                         GRPC_JSON_OBJECT, false);
+  json = json_iterator;
+  json_iterator = nullptr;
+  json_iterator = grpc_json_add_number_string_child(json, json_iterator,
+                                                    "serverId", uuid());
+  // reset json iterators to top level object
+  json = top_level_json;
+  json_iterator = nullptr;
+  // create and fill the data child.
+  grpc_json* data = grpc_json_create_child(json_iterator, json, "data", nullptr,
+                                           GRPC_JSON_OBJECT, false);
+  json = data;
+  json_iterator = nullptr;
+  // fill in the channel trace if applicable
+  grpc_json* trace_json = trace_.RenderJson();
+  if (trace_json != nullptr) {
+    trace_json->key = "trace";  // this object is named trace in channelz.proto
+    grpc_json_link_child(json, trace_json, nullptr);
+  }
+  // ask CallCountingHelper to populate trace and call count data.
+  call_counter_.PopulateCallCounts(json);
+  json = top_level_json;
+  ChildRefsList listen_sockets;
+  grpc_server_populate_listen_sockets(server_, &listen_sockets);
+  if (!listen_sockets.empty()) {
+    grpc_json* array_parent = grpc_json_create_child(
+        nullptr, json, "listenSocket", nullptr, GRPC_JSON_ARRAY, false);
+    for (size_t i = 0; i < listen_sockets.size(); ++i) {
+      json_iterator =
+          grpc_json_create_child(json_iterator, array_parent, nullptr, nullptr,
+                                 GRPC_JSON_OBJECT, false);
+      grpc_json_add_number_string_child(json_iterator, nullptr, "socketId",
+                                        listen_sockets[i]);
+    }
+  }
+  return top_level_json;
+}
+
+static void PopulateSocketAddressJson(grpc_json* json, const char* name,
+                                      const char* addr_str) {
+  if (addr_str == nullptr) return;
+  grpc_json* json_iterator = nullptr;
+  json_iterator = grpc_json_create_child(json_iterator, json, name, nullptr,
+                                         GRPC_JSON_OBJECT, false);
+  json = json_iterator;
+  json_iterator = nullptr;
+  grpc_uri* uri = grpc_uri_parse(addr_str, true);
+  if ((uri != nullptr) && ((strcmp(uri->scheme, "ipv4") == 0) ||
+                           (strcmp(uri->scheme, "ipv6") == 0))) {
+    const char* host_port = uri->path;
+    if (*host_port == '/') ++host_port;
+    char* host = nullptr;
+    char* port = nullptr;
+    GPR_ASSERT(gpr_split_host_port(host_port, &host, &port));
+    int port_num = -1;
+    if (port != nullptr) {
+      port_num = atoi(port);
+    }
+    char* b64_host = grpc_base64_encode(host, strlen(host), false, false);
+    json_iterator = grpc_json_create_child(json_iterator, json, "tcpip_address",
+                                           nullptr, GRPC_JSON_OBJECT, false);
+    json = json_iterator;
+    json_iterator = nullptr;
+    json_iterator = grpc_json_add_number_string_child(json, json_iterator,
+                                                      "port", port_num);
+    json_iterator = grpc_json_create_child(json_iterator, json, "ip_address",
+                                           b64_host, GRPC_JSON_STRING, true);
+    gpr_free(host);
+    gpr_free(port);
+
+  } else if (uri != nullptr && strcmp(uri->scheme, "unix") == 0) {
+    json_iterator = grpc_json_create_child(json_iterator, json, "uds_address",
+                                           nullptr, GRPC_JSON_OBJECT, false);
+    json = json_iterator;
+    json_iterator = nullptr;
+    json_iterator =
+        grpc_json_create_child(json_iterator, json, "filename",
+                               gpr_strdup(uri->path), GRPC_JSON_STRING, true);
+  } else {
+    json_iterator = grpc_json_create_child(json_iterator, json, "other_address",
+                                           nullptr, GRPC_JSON_OBJECT, false);
+    json = json_iterator;
+    json_iterator = nullptr;
+    json_iterator = grpc_json_create_child(json_iterator, json, "name",
+                                           addr_str, GRPC_JSON_STRING, false);
+  }
+  grpc_uri_destroy(uri);
+}
+
+SocketNode::SocketNode(UniquePtr<char> local, UniquePtr<char> remote)
+    : BaseNode(EntityType::kSocket),
+      local_(std::move(local)),
+      remote_(std::move(remote)) {}
+
+void SocketNode::RecordStreamStartedFromLocal() {
+  gpr_atm_no_barrier_fetch_add(&streams_started_, static_cast<gpr_atm>(1));
+  gpr_atm_no_barrier_store(&last_local_stream_created_millis_,
+                           (gpr_atm)ExecCtx::Get()->Now());
+}
+
+void SocketNode::RecordStreamStartedFromRemote() {
+  gpr_atm_no_barrier_fetch_add(&streams_started_, static_cast<gpr_atm>(1));
+  gpr_atm_no_barrier_store(&last_remote_stream_created_millis_,
+                           (gpr_atm)ExecCtx::Get()->Now());
+}
+
+void SocketNode::RecordMessagesSent(uint32_t num_sent) {
+  gpr_atm_no_barrier_fetch_add(&messages_sent_, static_cast<gpr_atm>(num_sent));
+  gpr_atm_no_barrier_store(&last_message_sent_millis_,
+                           (gpr_atm)ExecCtx::Get()->Now());
+}
+
+void SocketNode::RecordMessageReceived() {
+  gpr_atm_no_barrier_fetch_add(&messages_received_, static_cast<gpr_atm>(1));
+  gpr_atm_no_barrier_store(&last_message_received_millis_,
+                           (gpr_atm)ExecCtx::Get()->Now());
+}
+
+grpc_json* SocketNode::RenderJson() {
+  // We need to track these three json objects to build our object
+  grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT);
+  grpc_json* json = top_level_json;
+  grpc_json* json_iterator = nullptr;
+  // create and fill the ref child
+  json_iterator = grpc_json_create_child(json_iterator, json, "ref", nullptr,
+                                         GRPC_JSON_OBJECT, false);
+  json = json_iterator;
+  json_iterator = nullptr;
+  json_iterator = grpc_json_add_number_string_child(json, json_iterator,
+                                                    "socketId", uuid());
+  json = top_level_json;
+  PopulateSocketAddressJson(json, "remote", remote_.get());
+  PopulateSocketAddressJson(json, "local", local_.get());
+  // reset json iterators to top level object
+  json = top_level_json;
+  json_iterator = nullptr;
+  // create and fill the data child.
+  grpc_json* data = grpc_json_create_child(json_iterator, json, "data", nullptr,
+                                           GRPC_JSON_OBJECT, false);
+  json = data;
+  json_iterator = nullptr;
+  gpr_timespec ts;
+  if (streams_started_ != 0) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "streamsStarted", streams_started_);
+    if (last_local_stream_created_millis_ != 0) {
+      ts = grpc_millis_to_timespec(last_local_stream_created_millis_,
+                                   GPR_CLOCK_REALTIME);
+      json_iterator = grpc_json_create_child(
+          json_iterator, json, "lastLocalStreamCreatedTimestamp",
+          gpr_format_timespec(ts), GRPC_JSON_STRING, true);
+    }
+    if (last_remote_stream_created_millis_ != 0) {
+      ts = grpc_millis_to_timespec(last_remote_stream_created_millis_,
+                                   GPR_CLOCK_REALTIME);
+      json_iterator = grpc_json_create_child(
+          json_iterator, json, "lastRemoteStreamCreatedTimestamp",
+          gpr_format_timespec(ts), GRPC_JSON_STRING, true);
+    }
+  }
+  if (streams_succeeded_ != 0) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "streamsSucceeded", streams_succeeded_);
+  }
+  if (streams_failed_) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "streamsFailed", streams_failed_);
+  }
+  if (messages_sent_ != 0) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "messagesSent", messages_sent_);
+    ts = grpc_millis_to_timespec(last_message_sent_millis_, GPR_CLOCK_REALTIME);
+    json_iterator =
+        grpc_json_create_child(json_iterator, json, "lastMessageSentTimestamp",
+                               gpr_format_timespec(ts), GRPC_JSON_STRING, true);
+  }
+  if (messages_received_ != 0) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "messagesReceived", messages_received_);
+    ts = grpc_millis_to_timespec(last_message_received_millis_,
+                                 GPR_CLOCK_REALTIME);
+    json_iterator = grpc_json_create_child(
+        json_iterator, json, "lastMessageReceivedTimestamp",
+        gpr_format_timespec(ts), GRPC_JSON_STRING, true);
+  }
+  if (keepalives_sent_ != 0) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "keepAlivesSent", keepalives_sent_);
+  }
+  return top_level_json;
+}
+
+ListenSocketNode::ListenSocketNode(UniquePtr<char> local_addr)
+    : BaseNode(EntityType::kSocket), local_addr_(std::move(local_addr)) {}
+
+grpc_json* ListenSocketNode::RenderJson() {
+  // We need to track these three json objects to build our object
+  grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT);
+  grpc_json* json = top_level_json;
+  grpc_json* json_iterator = nullptr;
+  // create and fill the ref child
+  json_iterator = grpc_json_create_child(json_iterator, json, "ref", nullptr,
+                                         GRPC_JSON_OBJECT, false);
+  json = json_iterator;
+  json_iterator = nullptr;
+  json_iterator = grpc_json_add_number_string_child(json, json_iterator,
+                                                    "socketId", uuid());
+  json = top_level_json;
+  PopulateSocketAddressJson(json, "local", local_addr_.get());
+
+  return top_level_json;
 }
 
 }  // namespace channelz