src/core/lib/channel/channelz.cc

   1 /*
   2  *
   3  * Copyright 2017 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #include <grpc/impl/codegen/port_platform.h>
  20
  21 #include "src/core/lib/channel/channelz.h"
  22
  23 #include <grpc/grpc.h>
  24 #include <grpc/support/alloc.h>
  25 #include <grpc/support/log.h>
  26 #include <grpc/support/string_util.h>
  27 #include <stdio.h>
  28 #include <stdlib.h>
  29 #include <string.h>
  30
  31 #include "src/core/lib/channel/channelz_registry.h"
  32 #include "src/core/lib/channel/status_util.h"
  33 #include "src/core/lib/gpr/string.h"
  34 #include "src/core/lib/gpr/useful.h"
  35 #include "src/core/lib/gprpp/atomic.h"
  36 #include "src/core/lib/gprpp/host_port.h"
  37 #include "src/core/lib/gprpp/memory.h"
  38 #include "src/core/lib/iomgr/error.h"
  39 #include "src/core/lib/iomgr/exec_ctx.h"
  40 #include "src/core/lib/slice/b64.h"
  41 #include "src/core/lib/slice/slice_internal.h"
  42 #include "src/core/lib/surface/channel.h"
  43 #include "src/core/lib/surface/server.h"
  44 #include "src/core/lib/transport/connectivity_state.h"
  45 #include "src/core/lib/transport/error_utils.h"
  46 #include "src/core/lib/uri/uri_parser.h"
  47
  48 namespace grpc_core {
  49 namespace channelz {
  50
  51 //
  52 // BaseNode
  53 //
  54
  55 BaseNode::BaseNode(EntityType type, std::string name)
  56     : type_(type), uuid_(-1), name_(std::move(name)) {
  57   // The registry will set uuid_ under its lock.
  58   ChannelzRegistry::Register(this);
  59 }
  60
  61 BaseNode::~BaseNode() { ChannelzRegistry::Unregister(uuid_); }
  62
  63 std::string BaseNode::RenderJsonString() {
  64   Json json = RenderJson();
  65   return json.Dump();
  66 }
  67
  68 //
  69 // CallCountingHelper
  70 //
  71
  72 CallCountingHelper::CallCountingHelper() {
  73   num_cores_ = GPR_MAX(1, gpr_cpu_num_cores());
  74   per_cpu_counter_data_storage_.reserve(num_cores_);
  75   for (size_t i = 0; i < num_cores_; ++i) {
  76     per_cpu_counter_data_storage_.emplace_back();
  77   }
  78 }
  79
  80 void CallCountingHelper::RecordCallStarted() {
  81   AtomicCounterData& data =
  82       per_cpu_counter_data_storage_[ExecCtx::Get()->starting_cpu()];
  83   data.calls_started.FetchAdd(1, MemoryOrder::RELAXED);
  84   data.last_call_started_cycle.Store(gpr_get_cycle_counter(),
  85                                      MemoryOrder::RELAXED);
  86 }
  87
  88 void CallCountingHelper::RecordCallFailed() {
  89   per_cpu_counter_data_storage_[ExecCtx::Get()->starting_cpu()]
  90       .calls_failed.FetchAdd(1, MemoryOrder::RELAXED);
  91 }
  92
  93 void CallCountingHelper::RecordCallSucceeded() {
  94   per_cpu_counter_data_storage_[ExecCtx::Get()->starting_cpu()]
  95       .calls_succeeded.FetchAdd(1, MemoryOrder::RELAXED);
  96 }
  97
  98 void CallCountingHelper::CollectData(CounterData* out) {
  99   for (size_t core = 0; core < num_cores_; ++core) {
 100     AtomicCounterData& data = per_cpu_counter_data_storage_[core];
 101
 102     out->calls_started += data.calls_started.Load(MemoryOrder::RELAXED);
 103     out->calls_succeeded +=
 104         per_cpu_counter_data_storage_[core].calls_succeeded.Load(
 105             MemoryOrder::RELAXED);
 106     out->calls_failed += per_cpu_counter_data_storage_[core].calls_failed.Load(
 107         MemoryOrder::RELAXED);
 108     const gpr_cycle_counter last_call =
 109         per_cpu_counter_data_storage_[core].last_call_started_cycle.Load(
 110             MemoryOrder::RELAXED);
 111     if (last_call > out->last_call_started_cycle) {
 112       out->last_call_started_cycle = last_call;
 113     }
 114   }
 115 }
 116
 117 void CallCountingHelper::PopulateCallCounts(Json::Object* object) {
 118   CounterData data;
 119   CollectData(&data);
 120   if (data.calls_started != 0) {
 121     (*object)["callsStarted"] = std::to_string(data.calls_started);
 122     gpr_timespec ts = gpr_convert_clock_type(
 123         gpr_cycle_counter_to_time(data.last_call_started_cycle),
 124         GPR_CLOCK_REALTIME);
 125     (*object)["lastCallStartedTimestamp"] = gpr_format_timespec(ts);
 126   }
 127   if (data.calls_succeeded != 0) {
 128     (*object)["callsSucceeded"] = std::to_string(data.calls_succeeded);
 129   }
 130   if (data.calls_failed) {
 131     (*object)["callsFailed"] = std::to_string(data.calls_failed);
 132   }
 133 }
 134
 135 //
 136 // ChannelNode
 137 //
 138
 139 ChannelNode::ChannelNode(std::string target, size_t channel_tracer_max_nodes,
 140                          bool is_internal_channel)
 141     : BaseNode(is_internal_channel ? EntityType::kInternalChannel
 142                                    : EntityType::kTopLevelChannel,
 143                target),
 144       target_(std::move(target)),
 145       trace_(channel_tracer_max_nodes) {}
 146
 147 const char* ChannelNode::GetChannelConnectivityStateChangeString(
 148     grpc_connectivity_state state) {
 149   switch (state) {
 150     case GRPC_CHANNEL_IDLE:
 151       return "Channel state change to IDLE";
 152     case GRPC_CHANNEL_CONNECTING:
 153       return "Channel state change to CONNECTING";
 154     case GRPC_CHANNEL_READY:
 155       return "Channel state change to READY";
 156     case GRPC_CHANNEL_TRANSIENT_FAILURE:
 157       return "Channel state change to TRANSIENT_FAILURE";
 158     case GRPC_CHANNEL_SHUTDOWN:
 159       return "Channel state change to SHUTDOWN";
 160   }
 161   GPR_UNREACHABLE_CODE(return "UNKNOWN");
 162 }
 163
 164 Json ChannelNode::RenderJson() {
 165   Json::Object data = {
 166       {"target", target_},
 167   };
 168   // Connectivity state.
 169   // If low-order bit is on, then the field is set.
 170   int state_field = connectivity_state_.Load(MemoryOrder::RELAXED);
 171   if ((state_field & 1) != 0) {
 172     grpc_connectivity_state state =
 173         static_cast<grpc_connectivity_state>(state_field >> 1);
 174     data["state"] = Json::Object{
 175         {"state", ConnectivityStateName(state)},
 176     };
 177   }
 178   // Fill in the channel trace if applicable.
 179   Json trace_json = trace_.RenderJson();
 180   if (trace_json.type() != Json::Type::JSON_NULL) {
 181     data["trace"] = std::move(trace_json);
 182   }
 183   // Ask CallCountingHelper to populate call count data.
 184   call_counter_.PopulateCallCounts(&data);
 185   // Construct outer object.
 186   Json::Object json = {
 187       {"ref",
 188        Json::Object{
 189            {"channelId", std::to_string(uuid())},
 190        }},
 191       {"data", std::move(data)},
 192   };
 193   // Template method. Child classes may override this to add their specific
 194   // functionality.
 195   PopulateChildRefs(&json);
 196   return json;
 197 }
 198
 199 void ChannelNode::PopulateChildRefs(Json::Object* json) {
 200   MutexLock lock(&child_mu_);
 201   if (!child_subchannels_.empty()) {
 202     Json::Array array;
 203     for (intptr_t subchannel_uuid : child_subchannels_) {
 204       array.emplace_back(Json::Object{
 205           {"subchannelId", std::to_string(subchannel_uuid)},
 206       });
 207     }
 208     (*json)["subchannelRef"] = std::move(array);
 209   }
 210   if (!child_channels_.empty()) {
 211     Json::Array array;
 212     for (intptr_t channel_uuid : child_channels_) {
 213       array.emplace_back(Json::Object{
 214           {"channelId", std::to_string(channel_uuid)},
 215       });
 216     }
 217     (*json)["channelRef"] = std::move(array);
 218   }
 219 }
 220
 221 void ChannelNode::SetConnectivityState(grpc_connectivity_state state) {
 222   // Store with low-order bit set to indicate that the field is set.
 223   int state_field = (state << 1) + 1;
 224   connectivity_state_.Store(state_field, MemoryOrder::RELAXED);
 225 }
 226
 227 void ChannelNode::AddChildChannel(intptr_t child_uuid) {
 228   MutexLock lock(&child_mu_);
 229   child_channels_.insert(child_uuid);
 230 }
 231
 232 void ChannelNode::RemoveChildChannel(intptr_t child_uuid) {
 233   MutexLock lock(&child_mu_);
 234   child_channels_.erase(child_uuid);
 235 }
 236
 237 void ChannelNode::AddChildSubchannel(intptr_t child_uuid) {
 238   MutexLock lock(&child_mu_);
 239   child_subchannels_.insert(child_uuid);
 240 }
 241
 242 void ChannelNode::RemoveChildSubchannel(intptr_t child_uuid) {
 243   MutexLock lock(&child_mu_);
 244   child_subchannels_.erase(child_uuid);
 245 }
 246
 247 //
 248 // ServerNode
 249 //
 250
 251 ServerNode::ServerNode(size_t channel_tracer_max_nodes)
 252     : BaseNode(EntityType::kServer, ""), trace_(channel_tracer_max_nodes) {}
 253
 254 ServerNode::~ServerNode() {}
 255
 256 void ServerNode::AddChildSocket(RefCountedPtr<SocketNode> node) {
 257   MutexLock lock(&child_mu_);
 258   child_sockets_.insert(std::make_pair(node->uuid(), std::move(node)));
 259 }
 260
 261 void ServerNode::RemoveChildSocket(intptr_t child_uuid) {
 262   MutexLock lock(&child_mu_);
 263   child_sockets_.erase(child_uuid);
 264 }
 265
 266 void ServerNode::AddChildListenSocket(RefCountedPtr<ListenSocketNode> node) {
 267   MutexLock lock(&child_mu_);
 268   child_listen_sockets_.insert(std::make_pair(node->uuid(), std::move(node)));
 269 }
 270
 271 void ServerNode::RemoveChildListenSocket(intptr_t child_uuid) {
 272   MutexLock lock(&child_mu_);
 273   child_listen_sockets_.erase(child_uuid);
 274 }
 275
 276 std::string ServerNode::RenderServerSockets(intptr_t start_socket_id,
 277                                             intptr_t max_results) {
 278   GPR_ASSERT(start_socket_id >= 0);
 279   GPR_ASSERT(max_results >= 0);
 280   // If user does not set max_results, we choose 500.
 281   size_t pagination_limit = max_results == 0 ? 500 : max_results;
 282   Json::Object object;
 283   {
 284     MutexLock lock(&child_mu_);
 285     size_t sockets_rendered = 0;
 286     // Create list of socket refs.
 287     Json::Array array;
 288     auto it = child_sockets_.lower_bound(start_socket_id);
 289     for (; it != child_sockets_.end() && sockets_rendered < pagination_limit;
 290          ++it, ++sockets_rendered) {
 291       array.emplace_back(Json::Object{
 292           {"socketId", std::to_string(it->first)},
 293           {"name", it->second->name()},
 294       });
 295     }
 296     object["socketRef"] = std::move(array);
 297     if (it == child_sockets_.end()) object["end"] = true;
 298   }
 299   Json json = std::move(object);
 300   return json.Dump();
 301 }
 302
 303 Json ServerNode::RenderJson() {
 304   Json::Object data;
 305   // Fill in the channel trace if applicable.
 306   Json trace_json = trace_.RenderJson();
 307   if (trace_json.type() != Json::Type::JSON_NULL) {
 308     data["trace"] = std::move(trace_json);
 309   }
 310   // Ask CallCountingHelper to populate call count data.
 311   call_counter_.PopulateCallCounts(&data);
 312   // Construct top-level object.
 313   Json::Object object = {
 314       {"ref",
 315        Json::Object{
 316            {"serverId", std::to_string(uuid())},
 317        }},
 318       {"data", std::move(data)},
 319   };
 320   // Render listen sockets.
 321   {
 322     MutexLock lock(&child_mu_);
 323     if (!child_listen_sockets_.empty()) {
 324       Json::Array array;
 325       for (const auto& it : child_listen_sockets_) {
 326         array.emplace_back(Json::Object{
 327             {"socketId", std::to_string(it.first)},
 328             {"name", it.second->name()},
 329         });
 330       }
 331       object["listenSocket"] = std::move(array);
 332     }
 333   }
 334   return object;
 335 }
 336
 337 //
 338 // SocketNode
 339 //
 340
 341 namespace {
 342
 343 void PopulateSocketAddressJson(Json::Object* json, const char* name,
 344                                const char* addr_str) {
 345   if (addr_str == nullptr) return;
 346   Json::Object data;
 347   grpc_uri* uri = grpc_uri_parse(addr_str, true);
 348   if ((uri != nullptr) && ((strcmp(uri->scheme, "ipv4") == 0) ||
 349                            (strcmp(uri->scheme, "ipv6") == 0))) {
 350     const char* host_port = uri->path;
 351     if (*host_port == '/') ++host_port;
 352     std::string host;
 353     std::string port;
 354     GPR_ASSERT(SplitHostPort(host_port, &host, &port));
 355     int port_num = -1;
 356     if (!port.empty()) {
 357       port_num = atoi(port.data());
 358     }
 359     char* b64_host = grpc_base64_encode(host.data(), host.size(), false, false);
 360     data["tcpip_address"] = Json::Object{
 361         {"port", port_num},
 362         {"ip_address", b64_host},
 363     };
 364     gpr_free(b64_host);
 365   } else if (uri != nullptr && strcmp(uri->scheme, "unix") == 0) {
 366     data["uds_address"] = Json::Object{
 367         {"filename", uri->path},
 368     };
 369   } else {
 370     data["other_address"] = Json::Object{
 371         {"name", addr_str},
 372     };
 373   }
 374   grpc_uri_destroy(uri);
 375   (*json)[name] = std::move(data);
 376 }
 377
 378 }  // namespace
 379
 380 SocketNode::SocketNode(std::string local, std::string remote, std::string name)
 381     : BaseNode(EntityType::kSocket, std::move(name)),
 382       local_(std::move(local)),
 383       remote_(std::move(remote)) {}
 384
 385 void SocketNode::RecordStreamStartedFromLocal() {
 386   streams_started_.FetchAdd(1, MemoryOrder::RELAXED);
 387   last_local_stream_created_cycle_.Store(gpr_get_cycle_counter(),
 388                                          MemoryOrder::RELAXED);
 389 }
 390
 391 void SocketNode::RecordStreamStartedFromRemote() {
 392   streams_started_.FetchAdd(1, MemoryOrder::RELAXED);
 393   last_remote_stream_created_cycle_.Store(gpr_get_cycle_counter(),
 394                                           MemoryOrder::RELAXED);
 395 }
 396
 397 void SocketNode::RecordMessagesSent(uint32_t num_sent) {
 398   messages_sent_.FetchAdd(num_sent, MemoryOrder::RELAXED);
 399   last_message_sent_cycle_.Store(gpr_get_cycle_counter(), MemoryOrder::RELAXED);
 400 }
 401
 402 void SocketNode::RecordMessageReceived() {
 403   messages_received_.FetchAdd(1, MemoryOrder::RELAXED);
 404   last_message_received_cycle_.Store(gpr_get_cycle_counter(),
 405                                      MemoryOrder::RELAXED);
 406 }
 407
 408 Json SocketNode::RenderJson() {
 409   // Create and fill the data child.
 410   Json::Object data;
 411   gpr_timespec ts;
 412   int64_t streams_started = streams_started_.Load(MemoryOrder::RELAXED);
 413   if (streams_started != 0) {
 414     data["streamsStarted"] = std::to_string(streams_started);
 415     gpr_cycle_counter last_local_stream_created_cycle =
 416         last_local_stream_created_cycle_.Load(MemoryOrder::RELAXED);
 417     if (last_local_stream_created_cycle != 0) {
 418       ts = gpr_convert_clock_type(
 419           gpr_cycle_counter_to_time(last_local_stream_created_cycle),
 420           GPR_CLOCK_REALTIME);
 421       data["lastLocalStreamCreatedTimestamp"] = gpr_format_timespec(ts);
 422     }
 423     gpr_cycle_counter last_remote_stream_created_cycle =
 424         last_remote_stream_created_cycle_.Load(MemoryOrder::RELAXED);
 425     if (last_remote_stream_created_cycle != 0) {
 426       ts = gpr_convert_clock_type(
 427           gpr_cycle_counter_to_time(last_remote_stream_created_cycle),
 428           GPR_CLOCK_REALTIME);
 429       data["lastRemoteStreamCreatedTimestamp"] = gpr_format_timespec(ts);
 430     }
 431   }
 432   int64_t streams_succeeded = streams_succeeded_.Load(MemoryOrder::RELAXED);
 433   if (streams_succeeded != 0) {
 434     data["streamsSucceeded"] = std::to_string(streams_succeeded);
 435   }
 436   int64_t streams_failed = streams_failed_.Load(MemoryOrder::RELAXED);
 437   if (streams_failed != 0) {
 438     data["streamsFailed"] = std::to_string(streams_failed);
 439   }
 440   int64_t messages_sent = messages_sent_.Load(MemoryOrder::RELAXED);
 441   if (messages_sent != 0) {
 442     data["messagesSent"] = std::to_string(messages_sent);
 443     ts = gpr_convert_clock_type(
 444         gpr_cycle_counter_to_time(
 445             last_message_sent_cycle_.Load(MemoryOrder::RELAXED)),
 446         GPR_CLOCK_REALTIME);
 447     data["lastMessageSentTimestamp"] = gpr_format_timespec(ts);
 448   }
 449   int64_t messages_received = messages_received_.Load(MemoryOrder::RELAXED);
 450   if (messages_received != 0) {
 451     data["messagesReceived"] = std::to_string(messages_received);
 452     ts = gpr_convert_clock_type(
 453         gpr_cycle_counter_to_time(
 454             last_message_received_cycle_.Load(MemoryOrder::RELAXED)),
 455         GPR_CLOCK_REALTIME);
 456     data["lastMessageReceivedTimestamp"] = gpr_format_timespec(ts);
 457   }
 458   int64_t keepalives_sent = keepalives_sent_.Load(MemoryOrder::RELAXED);
 459   if (keepalives_sent != 0) {
 460     data["keepAlivesSent"] = std::to_string(keepalives_sent);
 461   }
 462   // Create and fill the parent object.
 463   Json::Object object = {
 464       {"ref",
 465        Json::Object{
 466            {"socketId", std::to_string(uuid())},
 467            {"name", name()},
 468        }},
 469       {"data", std::move(data)},
 470   };
 471   PopulateSocketAddressJson(&object, "remote", remote_.c_str());
 472   PopulateSocketAddressJson(&object, "local", local_.c_str());
 473   return object;
 474 }
 475
 476 //
 477 // ListenSocketNode
 478 //
 479
 480 ListenSocketNode::ListenSocketNode(std::string local_addr, std::string name)
 481     : BaseNode(EntityType::kSocket, std::move(name)),
 482       local_addr_(std::move(local_addr)) {}
 483
 484 Json ListenSocketNode::RenderJson() {
 485   Json::Object object = {
 486       {"ref",
 487        Json::Object{
 488            {"socketId", std::to_string(uuid())},
 489            {"name", name()},
 490        }},
 491   };
 492   PopulateSocketAddressJson(&object, "local", local_addr_.c_str());
 493   return object;
 494 }
 495
 496 }  // namespace channelz
 497 }  // namespace grpc_core