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 // channel arg code
  53 //
  54
  55 namespace {
  56
  57 void* parent_uuid_copy(void* p) { return p; }
  58 void parent_uuid_destroy(void* /*p*/) {}
  59 int parent_uuid_cmp(void* p1, void* p2) { return GPR_ICMP(p1, p2); }
  60 const grpc_arg_pointer_vtable parent_uuid_vtable = {
  61     parent_uuid_copy, parent_uuid_destroy, parent_uuid_cmp};
  62
  63 }  // namespace
  64
  65 grpc_arg MakeParentUuidArg(intptr_t parent_uuid) {
  66   // We would ideally like to store the uuid in an integer argument.
  67   // Unfortunately, that won't work, because intptr_t (the type used for
  68   // uuids) doesn't fit in an int (the type used for integer args).
  69   // So instead, we use a hack to store it as a pointer, because
  70   // intptr_t should be the same size as void*.
  71   static_assert(sizeof(intptr_t) <= sizeof(void*),
  72                 "can't fit intptr_t inside of void*");
  73   return grpc_channel_arg_pointer_create(
  74       const_cast<char*>(GRPC_ARG_CHANNELZ_PARENT_UUID),
  75       reinterpret_cast<void*>(parent_uuid), &parent_uuid_vtable);
  76 }
  77
  78 intptr_t GetParentUuidFromArgs(const grpc_channel_args& args) {
  79   const grpc_arg* arg =
  80       grpc_channel_args_find(&args, GRPC_ARG_CHANNELZ_PARENT_UUID);
  81   if (arg == nullptr || arg->type != GRPC_ARG_POINTER) return 0;
  82   return reinterpret_cast<intptr_t>(arg->value.pointer.p);
  83 }
  84
  85 //
  86 // BaseNode
  87 //
  88
  89 BaseNode::BaseNode(EntityType type, std::string name)
  90     : type_(type), uuid_(-1), name_(std::move(name)) {
  91   // The registry will set uuid_ under its lock.
  92   ChannelzRegistry::Register(this);
  93 }
  94
  95 BaseNode::~BaseNode() { ChannelzRegistry::Unregister(uuid_); }
  96
  97 std::string BaseNode::RenderJsonString() {
  98   Json json = RenderJson();
  99   return json.Dump();
 100 }
 101
 102 //
 103 // CallCountingHelper
 104 //
 105
 106 CallCountingHelper::CallCountingHelper() {
 107   num_cores_ = GPR_MAX(1, gpr_cpu_num_cores());
 108   per_cpu_counter_data_storage_.reserve(num_cores_);
 109   for (size_t i = 0; i < num_cores_; ++i) {
 110     per_cpu_counter_data_storage_.emplace_back();
 111   }
 112 }
 113
 114 void CallCountingHelper::RecordCallStarted() {
 115   AtomicCounterData& data =
 116       per_cpu_counter_data_storage_[ExecCtx::Get()->starting_cpu()];
 117   data.calls_started.FetchAdd(1, MemoryOrder::RELAXED);
 118   data.last_call_started_cycle.Store(gpr_get_cycle_counter(),
 119                                      MemoryOrder::RELAXED);
 120 }
 121
 122 void CallCountingHelper::RecordCallFailed() {
 123   per_cpu_counter_data_storage_[ExecCtx::Get()->starting_cpu()]
 124       .calls_failed.FetchAdd(1, MemoryOrder::RELAXED);
 125 }
 126
 127 void CallCountingHelper::RecordCallSucceeded() {
 128   per_cpu_counter_data_storage_[ExecCtx::Get()->starting_cpu()]
 129       .calls_succeeded.FetchAdd(1, MemoryOrder::RELAXED);
 130 }
 131
 132 void CallCountingHelper::CollectData(CounterData* out) {
 133   for (size_t core = 0; core < num_cores_; ++core) {
 134     AtomicCounterData& data = per_cpu_counter_data_storage_[core];
 135
 136     out->calls_started += data.calls_started.Load(MemoryOrder::RELAXED);
 137     out->calls_succeeded +=
 138         per_cpu_counter_data_storage_[core].calls_succeeded.Load(
 139             MemoryOrder::RELAXED);
 140     out->calls_failed += per_cpu_counter_data_storage_[core].calls_failed.Load(
 141         MemoryOrder::RELAXED);
 142     const gpr_cycle_counter last_call =
 143         per_cpu_counter_data_storage_[core].last_call_started_cycle.Load(
 144             MemoryOrder::RELAXED);
 145     if (last_call > out->last_call_started_cycle) {
 146       out->last_call_started_cycle = last_call;
 147     }
 148   }
 149 }
 150
 151 void CallCountingHelper::PopulateCallCounts(Json::Object* object) {
 152   CounterData data;
 153   CollectData(&data);
 154   if (data.calls_started != 0) {
 155     (*object)["callsStarted"] = std::to_string(data.calls_started);
 156     gpr_timespec ts = gpr_convert_clock_type(
 157         gpr_cycle_counter_to_time(data.last_call_started_cycle),
 158         GPR_CLOCK_REALTIME);
 159     char* ts_str = gpr_format_timespec(ts);
 160     (*object)["lastCallStartedTimestamp"] = ts_str;
 161     gpr_free(ts_str);
 162   }
 163   if (data.calls_succeeded != 0) {
 164     (*object)["callsSucceeded"] = std::to_string(data.calls_succeeded);
 165   }
 166   if (data.calls_failed) {
 167     (*object)["callsFailed"] = std::to_string(data.calls_failed);
 168   }
 169 }
 170
 171 //
 172 // ChannelNode
 173 //
 174
 175 ChannelNode::ChannelNode(std::string target, size_t channel_tracer_max_nodes,
 176                          intptr_t parent_uuid)
 177     : BaseNode(parent_uuid == 0 ? EntityType::kTopLevelChannel
 178                                 : EntityType::kInternalChannel,
 179                target),
 180       target_(std::move(target)),
 181       trace_(channel_tracer_max_nodes),
 182       parent_uuid_(parent_uuid) {}
 183
 184 const char* ChannelNode::GetChannelConnectivityStateChangeString(
 185     grpc_connectivity_state state) {
 186   switch (state) {
 187     case GRPC_CHANNEL_IDLE:
 188       return "Channel state change to IDLE";
 189     case GRPC_CHANNEL_CONNECTING:
 190       return "Channel state change to CONNECTING";
 191     case GRPC_CHANNEL_READY:
 192       return "Channel state change to READY";
 193     case GRPC_CHANNEL_TRANSIENT_FAILURE:
 194       return "Channel state change to TRANSIENT_FAILURE";
 195     case GRPC_CHANNEL_SHUTDOWN:
 196       return "Channel state change to SHUTDOWN";
 197   }
 198   GPR_UNREACHABLE_CODE(return "UNKNOWN");
 199 }
 200
 201 Json ChannelNode::RenderJson() {
 202   Json::Object data = {
 203       {"target", target_},
 204   };
 205   // Connectivity state.
 206   // If low-order bit is on, then the field is set.
 207   int state_field = connectivity_state_.Load(MemoryOrder::RELAXED);
 208   if ((state_field & 1) != 0) {
 209     grpc_connectivity_state state =
 210         static_cast<grpc_connectivity_state>(state_field >> 1);
 211     data["state"] = Json::Object{
 212         {"state", ConnectivityStateName(state)},
 213     };
 214   }
 215   // Fill in the channel trace if applicable.
 216   Json trace_json = trace_.RenderJson();
 217   if (trace_json.type() != Json::Type::JSON_NULL) {
 218     data["trace"] = std::move(trace_json);
 219   }
 220   // Ask CallCountingHelper to populate call count data.
 221   call_counter_.PopulateCallCounts(&data);
 222   // Construct outer object.
 223   Json::Object json = {
 224       {"ref",
 225        Json::Object{
 226            {"channelId", std::to_string(uuid())},
 227        }},
 228       {"data", std::move(data)},
 229   };
 230   // Template method. Child classes may override this to add their specific
 231   // functionality.
 232   PopulateChildRefs(&json);
 233   return json;
 234 }
 235
 236 void ChannelNode::PopulateChildRefs(Json::Object* json) {
 237   MutexLock lock(&child_mu_);
 238   if (!child_subchannels_.empty()) {
 239     Json::Array array;
 240     for (const auto& p : child_subchannels_) {
 241       array.emplace_back(Json::Object{
 242           {"subchannelId", std::to_string(p.first)},
 243       });
 244     }
 245     (*json)["subchannelRef"] = std::move(array);
 246   }
 247   if (!child_channels_.empty()) {
 248     Json::Array array;
 249     for (const auto& p : child_channels_) {
 250       array.emplace_back(Json::Object{
 251           {"channelId", std::to_string(p.first)},
 252       });
 253     }
 254     (*json)["channelRef"] = std::move(array);
 255   }
 256 }
 257
 258 void ChannelNode::SetConnectivityState(grpc_connectivity_state state) {
 259   // Store with low-order bit set to indicate that the field is set.
 260   int state_field = (state << 1) + 1;
 261   connectivity_state_.Store(state_field, MemoryOrder::RELAXED);
 262 }
 263
 264 void ChannelNode::AddChildChannel(intptr_t child_uuid) {
 265   MutexLock lock(&child_mu_);
 266   child_channels_.insert(std::make_pair(child_uuid, true));
 267 }
 268
 269 void ChannelNode::RemoveChildChannel(intptr_t child_uuid) {
 270   MutexLock lock(&child_mu_);
 271   child_channels_.erase(child_uuid);
 272 }
 273
 274 void ChannelNode::AddChildSubchannel(intptr_t child_uuid) {
 275   MutexLock lock(&child_mu_);
 276   child_subchannels_.insert(std::make_pair(child_uuid, true));
 277 }
 278
 279 void ChannelNode::RemoveChildSubchannel(intptr_t child_uuid) {
 280   MutexLock lock(&child_mu_);
 281   child_subchannels_.erase(child_uuid);
 282 }
 283
 284 //
 285 // ServerNode
 286 //
 287
 288 ServerNode::ServerNode(grpc_server* /*server*/, size_t channel_tracer_max_nodes)
 289     : BaseNode(EntityType::kServer, ""), trace_(channel_tracer_max_nodes) {}
 290
 291 ServerNode::~ServerNode() {}
 292
 293 void ServerNode::AddChildSocket(RefCountedPtr<SocketNode> node) {
 294   MutexLock lock(&child_mu_);
 295   child_sockets_.insert(std::make_pair(node->uuid(), std::move(node)));
 296 }
 297
 298 void ServerNode::RemoveChildSocket(intptr_t child_uuid) {
 299   MutexLock lock(&child_mu_);
 300   child_sockets_.erase(child_uuid);
 301 }
 302
 303 void ServerNode::AddChildListenSocket(RefCountedPtr<ListenSocketNode> node) {
 304   MutexLock lock(&child_mu_);
 305   child_listen_sockets_.insert(std::make_pair(node->uuid(), std::move(node)));
 306 }
 307
 308 void ServerNode::RemoveChildListenSocket(intptr_t child_uuid) {
 309   MutexLock lock(&child_mu_);
 310   child_listen_sockets_.erase(child_uuid);
 311 }
 312
 313 std::string ServerNode::RenderServerSockets(intptr_t start_socket_id,
 314                                             intptr_t max_results) {
 315   // If user does not set max_results, we choose 500.
 316   size_t pagination_limit = max_results == 0 ? 500 : max_results;
 317   Json::Object object;
 318   {
 319     MutexLock lock(&child_mu_);
 320     size_t sockets_rendered = 0;
 321     if (!child_sockets_.empty()) {
 322       // Create list of socket refs.
 323       Json::Array array;
 324       const size_t limit = GPR_MIN(child_sockets_.size(), pagination_limit);
 325       for (auto it = child_sockets_.lower_bound(start_socket_id);
 326            it != child_sockets_.end() && sockets_rendered < limit;
 327            ++it, ++sockets_rendered) {
 328         array.emplace_back(Json::Object{
 329             {"socketId", std::to_string(it->first)},
 330             {"name", it->second->name()},
 331         });
 332       }
 333       object["socketRef"] = std::move(array);
 334     }
 335     if (sockets_rendered == child_sockets_.size()) object["end"] = true;
 336   }
 337   Json json = std::move(object);
 338   return json.Dump();
 339 }
 340
 341 Json ServerNode::RenderJson() {
 342   Json::Object data;
 343   // Fill in the channel trace if applicable.
 344   Json trace_json = trace_.RenderJson();
 345   if (trace_json.type() != Json::Type::JSON_NULL) {
 346     data["trace"] = std::move(trace_json);
 347   }
 348   // Ask CallCountingHelper to populate call count data.
 349   call_counter_.PopulateCallCounts(&data);
 350   // Construct top-level object.
 351   Json::Object object = {
 352       {"ref",
 353        Json::Object{
 354            {"serverId", std::to_string(uuid())},
 355        }},
 356       {"data", std::move(data)},
 357   };
 358   // Render listen sockets.
 359   {
 360     MutexLock lock(&child_mu_);
 361     if (!child_listen_sockets_.empty()) {
 362       Json::Array array;
 363       for (const auto& it : child_listen_sockets_) {
 364         array.emplace_back(Json::Object{
 365             {"socketId", std::to_string(it.first)},
 366             {"name", it.second->name()},
 367         });
 368       }
 369       object["listenSocket"] = std::move(array);
 370     }
 371   }
 372   return object;
 373 }
 374
 375 //
 376 // SocketNode
 377 //
 378
 379 namespace {
 380
 381 void PopulateSocketAddressJson(Json::Object* json, const char* name,
 382                                const char* addr_str) {
 383   if (addr_str == nullptr) return;
 384   Json::Object data;
 385   grpc_uri* uri = grpc_uri_parse(addr_str, true);
 386   if ((uri != nullptr) && ((strcmp(uri->scheme, "ipv4") == 0) ||
 387                            (strcmp(uri->scheme, "ipv6") == 0))) {
 388     const char* host_port = uri->path;
 389     if (*host_port == '/') ++host_port;
 390     grpc_core::UniquePtr<char> host;
 391     grpc_core::UniquePtr<char> port;
 392     GPR_ASSERT(SplitHostPort(host_port, &host, &port));
 393     int port_num = -1;
 394     if (port != nullptr) {
 395       port_num = atoi(port.get());
 396     }
 397     char* b64_host =
 398         grpc_base64_encode(host.get(), strlen(host.get()), false, false);
 399     data["tcpip_address"] = Json::Object{
 400         {"port", port_num},
 401         {"ip_address", b64_host},
 402     };
 403     gpr_free(b64_host);
 404   } else if (uri != nullptr && strcmp(uri->scheme, "unix") == 0) {
 405     data["uds_address"] = Json::Object{
 406         {"filename", uri->path},
 407     };
 408   } else {
 409     data["other_address"] = Json::Object{
 410         {"name", addr_str},
 411     };
 412   }
 413   grpc_uri_destroy(uri);
 414   (*json)[name] = std::move(data);
 415 }
 416
 417 }  // namespace
 418
 419 SocketNode::SocketNode(std::string local, std::string remote, std::string name)
 420     : BaseNode(EntityType::kSocket, std::move(name)),
 421       local_(std::move(local)),
 422       remote_(std::move(remote)) {}
 423
 424 void SocketNode::RecordStreamStartedFromLocal() {
 425   streams_started_.FetchAdd(1, MemoryOrder::RELAXED);
 426   last_local_stream_created_cycle_.Store(gpr_get_cycle_counter(),
 427                                          MemoryOrder::RELAXED);
 428 }
 429
 430 void SocketNode::RecordStreamStartedFromRemote() {
 431   streams_started_.FetchAdd(1, MemoryOrder::RELAXED);
 432   last_remote_stream_created_cycle_.Store(gpr_get_cycle_counter(),
 433                                           MemoryOrder::RELAXED);
 434 }
 435
 436 void SocketNode::RecordMessagesSent(uint32_t num_sent) {
 437   messages_sent_.FetchAdd(num_sent, MemoryOrder::RELAXED);
 438   last_message_sent_cycle_.Store(gpr_get_cycle_counter(), MemoryOrder::RELAXED);
 439 }
 440
 441 void SocketNode::RecordMessageReceived() {
 442   messages_received_.FetchAdd(1, MemoryOrder::RELAXED);
 443   last_message_received_cycle_.Store(gpr_get_cycle_counter(),
 444                                      MemoryOrder::RELAXED);
 445 }
 446
 447 Json SocketNode::RenderJson() {
 448   // Create and fill the data child.
 449   Json::Object data;
 450   gpr_timespec ts;
 451   int64_t streams_started = streams_started_.Load(MemoryOrder::RELAXED);
 452   if (streams_started != 0) {
 453     data["streamsStarted"] = std::to_string(streams_started);
 454     gpr_cycle_counter last_local_stream_created_cycle =
 455         last_local_stream_created_cycle_.Load(MemoryOrder::RELAXED);
 456     if (last_local_stream_created_cycle != 0) {
 457       ts = gpr_convert_clock_type(
 458           gpr_cycle_counter_to_time(last_local_stream_created_cycle),
 459           GPR_CLOCK_REALTIME);
 460       char* ts_str = gpr_format_timespec(ts);
 461       data["lastLocalStreamCreatedTimestamp"] = ts_str;
 462       gpr_free(ts_str);
 463     }
 464     gpr_cycle_counter last_remote_stream_created_cycle =
 465         last_remote_stream_created_cycle_.Load(MemoryOrder::RELAXED);
 466     if (last_remote_stream_created_cycle != 0) {
 467       ts = gpr_convert_clock_type(
 468           gpr_cycle_counter_to_time(last_remote_stream_created_cycle),
 469           GPR_CLOCK_REALTIME);
 470       char* ts_str = gpr_format_timespec(ts);
 471       data["lastRemoteStreamCreatedTimestamp"] = ts_str;
 472       gpr_free(ts_str);
 473     }
 474   }
 475   int64_t streams_succeeded = streams_succeeded_.Load(MemoryOrder::RELAXED);
 476   if (streams_succeeded != 0) {
 477     data["streamsSucceeded"] = std::to_string(streams_succeeded);
 478   }
 479   int64_t streams_failed = streams_failed_.Load(MemoryOrder::RELAXED);
 480   if (streams_failed != 0) {
 481     data["streamsFailed"] = std::to_string(streams_failed);
 482   }
 483   int64_t messages_sent = messages_sent_.Load(MemoryOrder::RELAXED);
 484   if (messages_sent != 0) {
 485     data["messagesSent"] = std::to_string(messages_sent);
 486     ts = gpr_convert_clock_type(
 487         gpr_cycle_counter_to_time(
 488             last_message_sent_cycle_.Load(MemoryOrder::RELAXED)),
 489         GPR_CLOCK_REALTIME);
 490     char* ts_str = gpr_format_timespec(ts);
 491     data["lastMessageSentTimestamp"] = ts_str;
 492     gpr_free(ts_str);
 493   }
 494   int64_t messages_received = messages_received_.Load(MemoryOrder::RELAXED);
 495   if (messages_received != 0) {
 496     data["messagesReceived"] = std::to_string(messages_received);
 497     ts = gpr_convert_clock_type(
 498         gpr_cycle_counter_to_time(
 499             last_message_received_cycle_.Load(MemoryOrder::RELAXED)),
 500         GPR_CLOCK_REALTIME);
 501     char* ts_str = gpr_format_timespec(ts);
 502     data["lastMessageReceivedTimestamp"] = ts_str;
 503     gpr_free(ts_str);
 504   }
 505   int64_t keepalives_sent = keepalives_sent_.Load(MemoryOrder::RELAXED);
 506   if (keepalives_sent != 0) {
 507     data["keepAlivesSent"] = std::to_string(keepalives_sent);
 508   }
 509   // Create and fill the parent object.
 510   Json::Object object = {
 511       {"ref",
 512        Json::Object{
 513            {"socketId", std::to_string(uuid())},
 514            {"name", name()},
 515        }},
 516       {"data", std::move(data)},
 517   };
 518   PopulateSocketAddressJson(&object, "remote", remote_.c_str());
 519   PopulateSocketAddressJson(&object, "local", local_.c_str());
 520   return object;
 521 }
 522
 523 //
 524 // ListenSocketNode
 525 //
 526
 527 ListenSocketNode::ListenSocketNode(std::string local_addr, std::string name)
 528     : BaseNode(EntityType::kSocket, std::move(name)),
 529       local_addr_(std::move(local_addr)) {}
 530
 531 Json ListenSocketNode::RenderJson() {
 532   Json::Object object = {
 533       {"ref",
 534        Json::Object{
 535            {"socketId", std::to_string(uuid())},
 536            {"name", name()},
 537        }},
 538   };
 539   PopulateSocketAddressJson(&object, "local", local_addr_.c_str());
 540   return object;
 541 }
 542
 543 }  // namespace channelz
 544 }  // namespace grpc_core