Merge pull request #16758 from ncteisen/channelz

Channelz Part 7: Server Listening Socket Support

src/core/ext/transport/chttp2/server/chttp2_server.cc

@@ -53,6 +53,8 @@ typedef struct {
   grpc_closure tcp_server_shutdown_complete;
   grpc_closure* server_destroy_listener_done;
   grpc_handshake_manager* pending_handshake_mgrs;
+  grpc_core::RefCountedPtr<grpc_core::channelz::ListenSocketNode>
+      channelz_listen_socket;
 } server_state;
 
 typedef struct {
@@ -224,6 +226,7 @@ static void tcp_server_shutdown_complete(void* arg, grpc_error* error) {
   GPR_ASSERT(state->shutdown);
   grpc_handshake_manager_pending_list_shutdown_all(
       state->pending_handshake_mgrs, GRPC_ERROR_REF(error));
+  state->channelz_listen_socket.reset();
   gpr_mu_unlock(&state->mu);
   // Flush queued work before destroying handshaker factory, since that
   // may do a synchronous unref.
@@ -263,6 +266,8 @@ grpc_error* grpc_chttp2_server_add_port(grpc_server* server, const char* addr,
   server_state* state = nullptr;
   grpc_error** errors = nullptr;
   size_t naddrs = 0;
+  const grpc_arg* arg = nullptr;
+  intptr_t socket_uuid = 0;
 
   *port_num = -1;
 
@@ -324,9 +329,16 @@ grpc_error* grpc_chttp2_server_add_port(grpc_server* server, const char* addr,
   }
   grpc_resolved_addresses_destroy(resolved);
 
+  arg = grpc_channel_args_find(args, GRPC_ARG_ENABLE_CHANNELZ);
+  if (grpc_channel_arg_get_bool(arg, false)) {
+    state->channelz_listen_socket =
+        grpc_core::MakeRefCounted<grpc_core::channelz::ListenSocketNode>();
+    socket_uuid = state->channelz_listen_socket->uuid();
+  }
+
   /* Register with the server only upon success */
   grpc_server_add_listener(server, state, server_start_listener,
-                           server_destroy_listener);
+                           server_destroy_listener, socket_uuid);
   goto done;
 
 /* Error path: cleanup and return */

src/core/lib/channel/channelz.cc

@@ -258,6 +258,20 @@ grpc_json* ServerNode::RenderJson() {
   }
   // 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;
 }
 
@@ -358,5 +372,22 @@ grpc_json* SocketNode::RenderJson() {
   return top_level_json;
 }
 
+ListenSocketNode::ListenSocketNode() : BaseNode(EntityType::kSocket) {}
+
+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());
+  return top_level_json;
+}
+
 }  // namespace channelz
 }  // namespace grpc_core

src/core/lib/channel/channelz.h

@@ -263,6 +263,15 @@ class SocketNode : public BaseNode {
   UniquePtr<char> peer_string_;
 };
 
+// Handles channelz bookkeeping for listen sockets
+class ListenSocketNode : public BaseNode {
+ public:
+  ListenSocketNode();
+  ~ListenSocketNode() override {}
+
+  grpc_json* RenderJson() override;
+};
+
 // Creation functions
 
 typedef RefCountedPtr<ChannelNode> (*ChannelNodeCreationFunc)(grpc_channel*,

src/core/lib/surface/server.cc

@@ -54,6 +54,7 @@ struct listener {
                 size_t pollset_count);
   void (*destroy)(grpc_server* server, void* arg, grpc_closure* closure);
   struct listener* next;
+  intptr_t socket_uuid;
   grpc_closure destroy_done;
 };
 
@@ -1230,6 +1231,15 @@ void grpc_server_populate_server_sockets(
   gpr_mu_unlock(&s->mu_global);
 }
 
+void grpc_server_populate_listen_sockets(
+    grpc_server* server, grpc_core::channelz::ChildRefsList* listen_sockets) {
+  gpr_mu_lock(&server->mu_global);
+  for (listener* l = server->listeners; l != nullptr; l = l->next) {
+    listen_sockets->push_back(l->socket_uuid);
+  }
+  gpr_mu_unlock(&server->mu_global);
+}
+
 void done_published_shutdown(void* done_arg, grpc_cq_completion* storage) {
   (void)done_arg;
   gpr_free(storage);
@@ -1363,11 +1373,13 @@ void grpc_server_add_listener(grpc_server* server, void* arg,
                                             grpc_pollset** pollsets,
                                             size_t pollset_count),
                               void (*destroy)(grpc_server* server, void* arg,
-                                              grpc_closure* on_done)) {
+                                              grpc_closure* on_done),
+                              intptr_t socket_uuid) {
   listener* l = static_cast<listener*>(gpr_malloc(sizeof(listener)));
   l->arg = arg;
   l->start = start;
   l->destroy = destroy;
+  l->socket_uuid = socket_uuid;
   l->next = server->listeners;
   server->listeners = l;
 }

src/core/lib/surface/server.h

@@ -39,7 +39,8 @@ void grpc_server_add_listener(grpc_server* server, void* listener,
                                             grpc_pollset** pollsets,
                                             size_t npollsets),
                               void (*destroy)(grpc_server* server, void* arg,
-                                              grpc_closure* on_done));
+                                              grpc_closure* on_done),
+                              intptr_t socket_uuid);
 
 /* Setup a transport - creates a channel stack, binds the transport to the
    server */
@@ -48,10 +49,15 @@ void grpc_server_setup_transport(grpc_server* server, grpc_transport* transport,
                                  const grpc_channel_args* args,
                                  intptr_t socket_uuid);
 
+/* fills in the uuids of all sockets used for connections on this server */
 void grpc_server_populate_server_sockets(
     grpc_server* server, grpc_core::channelz::ChildRefsList* server_sockets,
     intptr_t start_idx);
 
+/* fills in the uuids of all listen sockets on this server */
+void grpc_server_populate_listen_sockets(
+    grpc_server* server, grpc_core::channelz::ChildRefsList* listen_sockets);
+
 grpc_core::channelz::ServerNode* grpc_server_get_channelz_node(
     grpc_server* server);
 

src/cpp/server/channelz/channelz_service.cc

@@ -96,7 +96,6 @@ Status ChannelzService::GetSocket(ServerContext* unused,
                                   const channelz::v1::GetSocketRequest* request,
                                   channelz::v1::GetSocketResponse* response) {
   char* json_str = grpc_channelz_get_socket(request->socket_id());
-  gpr_log(GPR_ERROR, "%s", json_str);
   if (json_str == nullptr) {
     return Status(NOT_FOUND, "No object found for that SocketId");
   }