Channelz socket support

include/grpc/grpc.h

@@ -510,6 +510,10 @@ GRPCAPI char* grpc_channelz_get_channel(intptr_t channel_id);
    is allocated and must be freed by the application. */
 GRPCAPI char* grpc_channelz_get_subchannel(intptr_t subchannel_id);
 
+/* Returns a single Socket, or else a NOT_FOUND code. The returned string
+   is allocated and must be freed by the application. */
+GRPCAPI char* grpc_channelz_get_socket(intptr_t socket_id);
+
 #ifdef __cplusplus
 }
 #endif

src/core/ext/transport/chttp2/transport/chttp2_transport.cc

@@ -335,6 +335,10 @@ static bool read_channel_args(grpc_chttp2_transport* t,
                 GRPC_ARG_OPTIMIZATION_TARGET,
                 channel_args->args[i].value.string);
       }
+    } else if (0 ==
+               strcmp(channel_args->args[i].key, GRPC_ARG_ENABLE_CHANNELZ)) {
+      t->channelz_socket =
+          grpc_core::MakeRefCounted<grpc_core::channelz::SocketNode>();
     } else {
       static const struct {
         const char* channel_arg_name;
@@ -720,6 +724,14 @@ static void destroy_stream_locked(void* sp, grpc_error* error) {
   grpc_chttp2_stream* s = static_cast<grpc_chttp2_stream*>(sp);
   grpc_chttp2_transport* t = s->t;
 
+  if (t->channelz_socket != nullptr) {
+    if ((t->is_client && s->eos_received) || (!t->is_client && s->eos_sent)) {
+      t->channelz_socket->RecordStreamSucceeded();
+    } else {
+      t->channelz_socket->RecordStreamFailed();
+    }
+  }
+
   GPR_ASSERT((s->write_closed && s->read_closed) || s->id == 0);
   if (s->id != 0) {
     GPR_ASSERT(grpc_chttp2_stream_map_find(&t->stream_map, s->id) == nullptr);
@@ -1407,6 +1419,9 @@ static void perform_stream_op_locked(void* stream_op,
   }
 
   if (op->send_initial_metadata) {
+    if (t->is_client && t->channelz_socket != nullptr) {
+      t->channelz_socket->RecordStreamStartedFromLocal();
+    }
     GRPC_STATS_INC_HTTP2_OP_SEND_INITIAL_METADATA();
     GPR_ASSERT(s->send_initial_metadata_finished == nullptr);
     on_complete->next_data.scratch |= CLOSURE_BARRIER_MAY_COVER_WRITE;
@@ -1492,6 +1507,9 @@ static void perform_stream_op_locked(void* stream_op,
 
   if (op->send_message) {
     GRPC_STATS_INC_HTTP2_OP_SEND_MESSAGE();
+    if (t->channelz_socket != nullptr) {
+      t->channelz_socket->RecordMessageSent();
+    }
     GRPC_STATS_INC_HTTP2_SEND_MESSAGE_SIZE(
         op->payload->send_message.send_message->length());
     on_complete->next_data.scratch |= CLOSURE_BARRIER_MAY_COVER_WRITE;
@@ -1609,6 +1627,9 @@ static void perform_stream_op_locked(void* stream_op,
 
   if (op->recv_message) {
     GRPC_STATS_INC_HTTP2_OP_RECV_MESSAGE();
+    if (t->channelz_socket != nullptr) {
+      t->channelz_socket->RecordMessageRecieved();
+    }
     size_t before = 0;
     GPR_ASSERT(s->recv_message_ready == nullptr);
     GPR_ASSERT(!s->pending_byte_stream);
@@ -2707,6 +2728,9 @@ static void start_keepalive_ping_locked(void* arg, grpc_error* error) {
   if (error != GRPC_ERROR_NONE) {
     return;
   }
+  if (t->channelz_socket != nullptr) {
+    t->channelz_socket->RecordKeepaliveSent();
+  }
   GRPC_CHTTP2_REF_TRANSPORT(t, "keepalive watchdog");
   grpc_timer_init(&t->keepalive_watchdog_timer,
                   grpc_core::ExecCtx::Get()->Now() + t->keepalive_timeout,

src/core/ext/transport/chttp2/transport/frame_data.cc

@@ -62,6 +62,7 @@ grpc_error* grpc_chttp2_data_parser_begin_frame(grpc_chttp2_data_parser* parser,
 
   if (flags & GRPC_CHTTP2_DATA_FLAG_END_STREAM) {
     s->received_last_frame = true;
+    s->eos_received = true;
   } else {
     s->received_last_frame = false;
   }

src/core/ext/transport/chttp2/transport/internal.h

@@ -36,6 +36,7 @@
 #include "src/core/ext/transport/chttp2/transport/hpack_parser.h"
 #include "src/core/ext/transport/chttp2/transport/incoming_metadata.h"
 #include "src/core/ext/transport/chttp2/transport/stream_map.h"
+#include "src/core/lib/channel/channelz.h"
 #include "src/core/lib/compression/stream_compression.h"
 #include "src/core/lib/gprpp/manual_constructor.h"
 #include "src/core/lib/iomgr/combiner.h"
@@ -471,6 +472,8 @@ struct grpc_chttp2_transport {
   bool keepalive_permit_without_calls;
   /** keep-alive state machine state */
   grpc_chttp2_keepalive_state keepalive_state;
+
+  grpc_core::RefCountedPtr<grpc_core::channelz::SocketNode> channelz_socket;
 };
 
 typedef enum {
@@ -534,6 +537,10 @@ struct grpc_chttp2_stream {
   /** Has trailing metadata been received. */
   bool received_trailing_metadata;
 
+  /* have we sent or received the EOS bit? */
+  bool eos_received;
+  bool eos_sent;
+
   /** the error that resulted in this stream being read-closed */
   grpc_error* read_closed_error;
   /** the error that resulted in this stream being write-closed */

src/core/ext/transport/chttp2/transport/parsing.cc

@@ -598,6 +598,9 @@ static grpc_error* init_header_frame_parser(grpc_chttp2_transport* t,
           gpr_log(GPR_ERROR, "grpc_chttp2_stream not accepted"));
       return init_skip_frame_parser(t, 1);
     }
+    if (t->channelz_socket != nullptr) {
+      t->channelz_socket->RecordStreamStartedFromRemote();
+    }
   } else {
     t->incoming_stream = s;
   }
@@ -611,6 +614,9 @@ static grpc_error* init_header_frame_parser(grpc_chttp2_transport* t,
   }
   t->parser = grpc_chttp2_header_parser_parse;
   t->parser_data = &t->hpack_parser;
+  if (t->header_eof) {
+    s->eos_received = true;
+  }
   switch (s->header_frames_received) {
     case 0:
       if (t->is_client && t->header_eof) {

src/core/ext/transport/chttp2/transport/writing.cc

@@ -569,6 +569,7 @@ class StreamWriteContext {
   void SentLastFrame() {
     s_->send_trailing_metadata = nullptr;
     s_->sent_trailing_metadata = true;
+    s_->eos_sent = true;
 
     if (!t_->is_client && !s_->read_closed) {
       grpc_slice_buffer_add(

src/core/lib/channel/channelz.cc

@@ -81,11 +81,13 @@ void CallCountingHelper::PopulateCallCounts(grpc_json* json) {
     json_iterator = grpc_json_add_number_string_child(
         json, json_iterator, "callsFailed", calls_failed_);
   }
-  gpr_timespec ts =
+  if (calls_started_ != 0) {
-      grpc_millis_to_timespec(last_call_started_millis_, GPR_CLOCK_REALTIME);
+    gpr_timespec ts =
-  json_iterator =
+        grpc_millis_to_timespec(last_call_started_millis_, GPR_CLOCK_REALTIME);
-      grpc_json_create_child(json_iterator, json, "lastCallStartedTimestamp",
+    json_iterator =
-                             gpr_format_timespec(ts), GRPC_JSON_STRING, true);
+        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,
@@ -184,5 +186,105 @@ grpc_json* ServerNode::RenderJson() {
   return top_level_json;
 }
 
+void SocketNode::RecordStreamStartedFromLocal() {
+  gpr_atm_no_barrier_fetch_add(&streams_started_, (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_, (gpr_atm)1);
+  gpr_atm_no_barrier_store(&last_remote_stream_created_millis_,
+                           (gpr_atm)ExecCtx::Get()->Now());
+}
+
+void SocketNode::RecordMessageSent() {
+  gpr_atm_no_barrier_fetch_add(&messages_sent_, (gpr_atm)1);
+  gpr_atm_no_barrier_store(&last_message_sent_millis_,
+                           (gpr_atm)ExecCtx::Get()->Now());
+}
+
+void SocketNode::RecordMessageRecieved() {
+  gpr_atm_no_barrier_fetch_add(&messages_recieved_, (gpr_atm)1);
+  gpr_atm_no_barrier_store(&last_message_recieved_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());
+  // 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;
+  if (streams_started_ != 0) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "streamsStarted", streams_started_);
+  }
+  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_);
+  }
+  if (messages_recieved_ != 0) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "messagesRecieved", messages_recieved_);
+  }
+  if (keepalives_sent_ != 0) {
+    json_iterator = grpc_json_add_number_string_child(
+        json, json_iterator, "keepAlivesSent", keepalives_sent_);
+  }
+  gpr_timespec ts;
+  if (streams_started_ != 0 && 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 (streams_started_ != 0 && 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 (messages_sent_ != 0) {
+    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_recieved_ != 0) {
+    ts = grpc_millis_to_timespec(last_message_recieved_millis_,
+                                 GPR_CLOCK_REALTIME);
+    json_iterator = grpc_json_create_child(
+        json_iterator, json, "lastMessageRecievedTimestamp",
+        gpr_format_timespec(ts), GRPC_JSON_STRING, true);
+  }
+  json = top_level_json;
+  return top_level_json;
+}
+
 }  // namespace channelz
 }  // namespace grpc_core

src/core/lib/channel/channelz.h

@@ -197,11 +197,38 @@ class ServerNode : public BaseNode {
 };
 
 // Handles channelz bookkeeping for sockets
-// TODO(ncteisen): implement in subsequent PR.
 class SocketNode : public BaseNode {
  public:
   SocketNode() : BaseNode(EntityType::kSocket) {}
   ~SocketNode() override {}
+
+  grpc_json* RenderJson() override;
+
+  void RecordStreamStartedFromLocal();
+  void RecordStreamStartedFromRemote();
+  void RecordStreamSucceeded() {
+    gpr_atm_no_barrier_fetch_add(&streams_succeeded_, (gpr_atm(1)));
+  }
+  void RecordStreamFailed() {
+    gpr_atm_no_barrier_fetch_add(&streams_failed_, (gpr_atm(1)));
+  }
+  void RecordMessageSent();
+  void RecordMessageRecieved();
+  void RecordKeepaliveSent() {
+    gpr_atm_no_barrier_fetch_add(&keepalives_sent_, (gpr_atm(1)));
+  }
+
+ private:
+  gpr_atm streams_started_ = 0;
+  gpr_atm streams_succeeded_ = 0;
+  gpr_atm streams_failed_ = 0;
+  gpr_atm messages_sent_ = 0;
+  gpr_atm messages_recieved_ = 0;
+  gpr_atm keepalives_sent_ = 0;
+  gpr_atm last_local_stream_created_millis_ = 0;
+  gpr_atm last_remote_stream_created_millis_ = 0;
+  gpr_atm last_message_sent_millis_ = 0;
+  gpr_atm last_message_recieved_millis_ = 0;
 };
 
 // Creation functions

src/core/lib/channel/channelz_registry.cc

@@ -197,3 +197,21 @@ char* grpc_channelz_get_subchannel(intptr_t subchannel_id) {
   grpc_json_destroy(top_level_json);
   return json_str;
 }
+
+char* grpc_channelz_get_socket(intptr_t socket_id) {
+  grpc_core::channelz::BaseNode* socket_node =
+      grpc_core::channelz::ChannelzRegistry::Get(socket_id);
+  if (socket_node == nullptr ||
+      socket_node->type() !=
+          grpc_core::channelz::BaseNode::EntityType::kSocket) {
+    return nullptr;
+  }
+  grpc_json* top_level_json = grpc_json_create(GRPC_JSON_OBJECT);
+  grpc_json* json = top_level_json;
+  grpc_json* socket_json = socket_node->RenderJson();
+  socket_json->key = "socket";
+  grpc_json_link_child(json, socket_json, nullptr);
+  char* json_str = grpc_json_dump_to_string(top_level_json, 0);
+  grpc_json_destroy(top_level_json);
+  return json_str;
+}

test/core/end2end/tests/channelz.cc

@@ -196,6 +196,178 @@ static void run_one_request(grpc_end2end_test_config config,
   cq_verifier_destroy(cqv);
 }
 
+/* Creates and returns a grpc_slice containing random alphanumeric characters.
+ */
+static grpc_slice generate_random_slice() {
+  size_t i;
+  static const char chars[] = "abcdefghijklmnopqrstuvwxyz1234567890";
+  char* output;
+  const size_t output_size = 1024 * 1024;
+  output = static_cast<char*>(gpr_malloc(output_size));
+  for (i = 0; i < output_size - 1; ++i) {
+    output[i] = chars[rand() % static_cast<int>(sizeof(chars) - 1)];
+  }
+  output[output_size - 1] = '\0';
+  grpc_slice out = grpc_slice_from_copied_string(output);
+  gpr_free(output);
+  return out;
+}
+
+static void run_one_request_with_payload(grpc_end2end_test_config config,
+                                         grpc_end2end_test_fixture f) {
+  /* Create large request and response bodies. These are big enough to require
+   * multiple round trips to deliver to the peer, and their exact contents of
+   * will be verified on completion. */
+  grpc_slice request_payload_slice = generate_random_slice();
+  grpc_slice response_payload_slice = generate_random_slice();
+
+  grpc_call* c;
+  grpc_call* s;
+  grpc_byte_buffer* request_payload =
+      grpc_raw_byte_buffer_create(&request_payload_slice, 1);
+  grpc_byte_buffer* response_payload =
+      grpc_raw_byte_buffer_create(&response_payload_slice, 1);
+  cq_verifier* cqv = cq_verifier_create(f.cq);
+  grpc_op ops[6];
+  grpc_op* op;
+  grpc_metadata_array initial_metadata_recv;
+  grpc_metadata_array trailing_metadata_recv;
+  grpc_metadata_array request_metadata_recv;
+  grpc_byte_buffer* request_payload_recv = nullptr;
+  grpc_byte_buffer* response_payload_recv = nullptr;
+  grpc_call_details call_details;
+  grpc_status_code status;
+  grpc_call_error error;
+  grpc_slice details;
+  int was_cancelled = 2;
+
+  gpr_timespec deadline = n_seconds_from_now(60);
+  c = grpc_channel_create_call(f.client, nullptr, GRPC_PROPAGATE_DEFAULTS, f.cq,
+                               grpc_slice_from_static_string("/foo"), nullptr,
+                               deadline, nullptr);
+  GPR_ASSERT(c);
+
+  grpc_metadata_array_init(&initial_metadata_recv);
+  grpc_metadata_array_init(&trailing_metadata_recv);
+  grpc_metadata_array_init(&request_metadata_recv);
+  grpc_call_details_init(&call_details);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = request_payload;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_SEND_CLOSE_FROM_CLIENT;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_RECV_INITIAL_METADATA;
+  op->data.recv_initial_metadata.recv_initial_metadata = &initial_metadata_recv;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &response_payload_recv;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_RECV_STATUS_ON_CLIENT;
+  op->data.recv_status_on_client.trailing_metadata = &trailing_metadata_recv;
+  op->data.recv_status_on_client.status = &status;
+  op->data.recv_status_on_client.status_details = &details;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  error = grpc_call_start_batch(c, ops, static_cast<size_t>(op - ops), tag(1),
+                                nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  error =
+      grpc_server_request_call(f.server, &s, &call_details,
+                               &request_metadata_recv, f.cq, f.cq, tag(101));
+  GPR_ASSERT(GRPC_CALL_OK == error);
+  CQ_EXPECT_COMPLETION(cqv, tag(101), 1);
+  cq_verify(cqv);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_SEND_INITIAL_METADATA;
+  op->data.send_initial_metadata.count = 0;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_RECV_MESSAGE;
+  op->data.recv_message.recv_message = &request_payload_recv;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(102),
+                                nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(102), 1);
+  cq_verify(cqv);
+
+  memset(ops, 0, sizeof(ops));
+  op = ops;
+  op->op = GRPC_OP_RECV_CLOSE_ON_SERVER;
+  op->data.recv_close_on_server.cancelled = &was_cancelled;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_SEND_MESSAGE;
+  op->data.send_message.send_message = response_payload;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  op->op = GRPC_OP_SEND_STATUS_FROM_SERVER;
+  op->data.send_status_from_server.trailing_metadata_count = 0;
+  op->data.send_status_from_server.status = GRPC_STATUS_OK;
+  grpc_slice status_details = grpc_slice_from_static_string("xyz");
+  op->data.send_status_from_server.status_details = &status_details;
+  op->flags = 0;
+  op->reserved = nullptr;
+  op++;
+  error = grpc_call_start_batch(s, ops, static_cast<size_t>(op - ops), tag(103),
+                                nullptr);
+  GPR_ASSERT(GRPC_CALL_OK == error);
+
+  CQ_EXPECT_COMPLETION(cqv, tag(103), 1);
+  CQ_EXPECT_COMPLETION(cqv, tag(1), 1);
+  cq_verify(cqv);
+
+  GPR_ASSERT(status == GRPC_STATUS_OK);
+  GPR_ASSERT(0 == grpc_slice_str_cmp(details, "xyz"));
+  GPR_ASSERT(0 == grpc_slice_str_cmp(call_details.method, "/foo"));
+  GPR_ASSERT(was_cancelled == 0);
+  GPR_ASSERT(byte_buffer_eq_slice(request_payload_recv, request_payload_slice));
+  GPR_ASSERT(
+      byte_buffer_eq_slice(response_payload_recv, response_payload_slice));
+
+  grpc_slice_unref(details);
+  grpc_metadata_array_destroy(&initial_metadata_recv);
+  grpc_metadata_array_destroy(&trailing_metadata_recv);
+  grpc_metadata_array_destroy(&request_metadata_recv);
+  grpc_call_details_destroy(&call_details);
+
+  grpc_call_unref(c);
+  grpc_call_unref(s);
+
+  cq_verifier_destroy(cqv);
+
+  grpc_byte_buffer_destroy(request_payload);
+  grpc_byte_buffer_destroy(response_payload);
+  grpc_byte_buffer_destroy(request_payload_recv);
+  grpc_byte_buffer_destroy(response_payload_recv);
+}
+
 static void test_channelz(grpc_end2end_test_config config) {
   grpc_end2end_test_fixture f;
 
@@ -258,6 +430,10 @@ static void test_channelz(grpc_end2end_test_config config) {
   GPR_ASSERT(nullptr == strstr(json, "\"severity\":\"CT_INFO\""));
   gpr_free(json);
 
+  // one successful request with payload to test socket data
+  // TODO(ncteisen): add some programatic spot checks on the socket json.
+  run_one_request_with_payload(config, f);
+
   end_test(&f);
   config.tear_down_data(&f);
 }