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 <stdio.h>
  24 #include <stdlib.h>
  25 #include <string.h>
  26
  27 #include "absl/strings/escaping.h"
  28 #include "absl/strings/strip.h"
  29
  30 #include <grpc/grpc.h>
  31 #include <grpc/support/alloc.h>
  32 #include <grpc/support/log.h>
  33 #include <grpc/support/string_util.h>
  34
  35 #include "src/core/lib/address_utils/sockaddr_utils.h"
  36 #include "src/core/lib/channel/channelz_registry.h"
  37 #include "src/core/lib/channel/status_util.h"
  38 #include "src/core/lib/gpr/string.h"
  39 #include "src/core/lib/gpr/useful.h"
  40 #include "src/core/lib/gprpp/atomic.h"
  41 #include "src/core/lib/gprpp/host_port.h"
  42 #include "src/core/lib/gprpp/memory.h"
  43 #include "src/core/lib/iomgr/error.h"
  44 #include "src/core/lib/iomgr/exec_ctx.h"
  45 #include "src/core/lib/iomgr/resolve_address.h"
  46 #include "src/core/lib/slice/b64.h"
  47 #include "src/core/lib/slice/slice_internal.h"
  48 #include "src/core/lib/surface/channel.h"
  49 #include "src/core/lib/surface/server.h"
  50 #include "src/core/lib/transport/connectivity_state.h"
  51 #include "src/core/lib/transport/error_utils.h"
  52 #include "src/core/lib/uri/uri_parser.h"
  53
  54 namespace grpc_core {
  55 namespace channelz {
  56
  57 //
  58 // BaseNode
  59 //
  60
  61 BaseNode::BaseNode(EntityType type, std::string name)
  62     : type_(type), uuid_(-1), name_(std::move(name)) {
  63   // The registry will set uuid_ under its lock.
  64   ChannelzRegistry::Register(this);
  65 }
  66
  67 BaseNode::~BaseNode() { ChannelzRegistry::Unregister(uuid_); }
  68
  69 std::string BaseNode::RenderJsonString() {
  70   Json json = RenderJson();
  71   return json.Dump();
  72 }
  73
  74 //
  75 // CallCountingHelper
  76 //
  77
  78 CallCountingHelper::CallCountingHelper() {
  79   num_cores_ = GPR_MAX(1, gpr_cpu_num_cores());
  80   per_cpu_counter_data_storage_.reserve(num_cores_);
  81   for (size_t i = 0; i < num_cores_; ++i) {
  82     per_cpu_counter_data_storage_.emplace_back();
  83   }
  84 }
  85
  86 void CallCountingHelper::RecordCallStarted() {
  87   AtomicCounterData& data =
  88       per_cpu_counter_data_storage_[ExecCtx::Get()->starting_cpu()];
  89   data.calls_started.FetchAdd(1, MemoryOrder::RELAXED);
  90   data.last_call_started_cycle.Store(gpr_get_cycle_counter(),
  91                                      MemoryOrder::RELAXED);
  92 }
  93
  94 void CallCountingHelper::RecordCallFailed() {
  95   per_cpu_counter_data_storage_[ExecCtx::Get()->starting_cpu()]
  96       .calls_failed.FetchAdd(1, MemoryOrder::RELAXED);
  97 }
  98
  99 void CallCountingHelper::RecordCallSucceeded() {
 100   per_cpu_counter_data_storage_[ExecCtx::Get()->starting_cpu()]
 101       .calls_succeeded.FetchAdd(1, MemoryOrder::RELAXED);
 102 }
 103
 104 void CallCountingHelper::CollectData(CounterData* out) {
 105   for (size_t core = 0; core < num_cores_; ++core) {
 106     AtomicCounterData& data = per_cpu_counter_data_storage_[core];
 107
 108     out->calls_started += data.calls_started.Load(MemoryOrder::RELAXED);
 109     out->calls_succeeded +=
 110         per_cpu_counter_data_storage_[core].calls_succeeded.Load(
 111             MemoryOrder::RELAXED);
 112     out->calls_failed += per_cpu_counter_data_storage_[core].calls_failed.Load(
 113         MemoryOrder::RELAXED);
 114     const gpr_cycle_counter last_call =
 115         per_cpu_counter_data_storage_[core].last_call_started_cycle.Load(
 116             MemoryOrder::RELAXED);
 117     if (last_call > out->last_call_started_cycle) {
 118       out->last_call_started_cycle = last_call;
 119     }
 120   }
 121 }
 122
 123 void CallCountingHelper::PopulateCallCounts(Json::Object* json) {
 124   CounterData data;
 125   CollectData(&data);
 126   if (data.calls_started != 0) {
 127     (*json)["callsStarted"] = std::to_string(data.calls_started);
 128     gpr_timespec ts = gpr_convert_clock_type(
 129         gpr_cycle_counter_to_time(data.last_call_started_cycle),
 130         GPR_CLOCK_REALTIME);
 131     (*json)["lastCallStartedTimestamp"] = gpr_format_timespec(ts);
 132   }
 133   if (data.calls_succeeded != 0) {
 134     (*json)["callsSucceeded"] = std::to_string(data.calls_succeeded);
 135   }
 136   if (data.calls_failed) {
 137     (*json)["callsFailed"] = std::to_string(data.calls_failed);
 138   }
 139 }
 140
 141 //
 142 // ChannelNode
 143 //
 144
 145 ChannelNode::ChannelNode(std::string target, size_t channel_tracer_max_nodes,
 146                          bool is_internal_channel)
 147     : BaseNode(is_internal_channel ? EntityType::kInternalChannel
 148                                    : EntityType::kTopLevelChannel,
 149                target),
 150       target_(std::move(target)),
 151       trace_(channel_tracer_max_nodes) {}
 152
 153 const char* ChannelNode::GetChannelConnectivityStateChangeString(
 154     grpc_connectivity_state state) {
 155   switch (state) {
 156     case GRPC_CHANNEL_IDLE:
 157       return "Channel state change to IDLE";
 158     case GRPC_CHANNEL_CONNECTING:
 159       return "Channel state change to CONNECTING";
 160     case GRPC_CHANNEL_READY:
 161       return "Channel state change to READY";
 162     case GRPC_CHANNEL_TRANSIENT_FAILURE:
 163       return "Channel state change to TRANSIENT_FAILURE";
 164     case GRPC_CHANNEL_SHUTDOWN:
 165       return "Channel state change to SHUTDOWN";
 166   }
 167   GPR_UNREACHABLE_CODE(return "UNKNOWN");
 168 }
 169
 170 Json ChannelNode::RenderJson() {
 171   Json::Object data = {
 172       {"target", target_},
 173   };
 174   // Connectivity state.
 175   // If low-order bit is on, then the field is set.
 176   int state_field = connectivity_state_.Load(MemoryOrder::RELAXED);
 177   if ((state_field & 1) != 0) {
 178     grpc_connectivity_state state =
 179         static_cast<grpc_connectivity_state>(state_field >> 1);
 180     data["state"] = Json::Object{
 181         {"state", ConnectivityStateName(state)},
 182     };
 183   }
 184   // Fill in the channel trace if applicable.
 185   Json trace_json = trace_.RenderJson();
 186   if (trace_json.type() != Json::Type::JSON_NULL) {
 187     data["trace"] = std::move(trace_json);
 188   }
 189   // Ask CallCountingHelper to populate call count data.
 190   call_counter_.PopulateCallCounts(&data);
 191   // Construct outer object.
 192   Json::Object json = {
 193       {"ref",
 194        Json::Object{
 195            {"channelId", std::to_string(uuid())},
 196        }},
 197       {"data", std::move(data)},
 198   };
 199   // Template method. Child classes may override this to add their specific
 200   // functionality.
 201   PopulateChildRefs(&json);
 202   return json;
 203 }
 204
 205 void ChannelNode::PopulateChildRefs(Json::Object* json) {
 206   MutexLock lock(&child_mu_);
 207   if (!child_subchannels_.empty()) {
 208     Json::Array array;
 209     for (intptr_t subchannel_uuid : child_subchannels_) {
 210       array.emplace_back(Json::Object{
 211           {"subchannelId", std::to_string(subchannel_uuid)},
 212       });
 213     }
 214     (*json)["subchannelRef"] = std::move(array);
 215   }
 216   if (!child_channels_.empty()) {
 217     Json::Array array;
 218     for (intptr_t channel_uuid : child_channels_) {
 219       array.emplace_back(Json::Object{
 220           {"channelId", std::to_string(channel_uuid)},
 221       });
 222     }
 223     (*json)["channelRef"] = std::move(array);
 224   }
 225 }
 226
 227 void ChannelNode::SetConnectivityState(grpc_connectivity_state state) {
 228   // Store with low-order bit set to indicate that the field is set.
 229   int state_field = (state << 1) + 1;
 230   connectivity_state_.Store(state_field, MemoryOrder::RELAXED);
 231 }
 232
 233 void ChannelNode::AddChildChannel(intptr_t child_uuid) {
 234   MutexLock lock(&child_mu_);
 235   child_channels_.insert(child_uuid);
 236 }
 237
 238 void ChannelNode::RemoveChildChannel(intptr_t child_uuid) {
 239   MutexLock lock(&child_mu_);
 240   child_channels_.erase(child_uuid);
 241 }
 242
 243 void ChannelNode::AddChildSubchannel(intptr_t child_uuid) {
 244   MutexLock lock(&child_mu_);
 245   child_subchannels_.insert(child_uuid);
 246 }
 247
 248 void ChannelNode::RemoveChildSubchannel(intptr_t child_uuid) {
 249   MutexLock lock(&child_mu_);
 250   child_subchannels_.erase(child_uuid);
 251 }
 252
 253 //
 254 // ServerNode
 255 //
 256
 257 ServerNode::ServerNode(size_t channel_tracer_max_nodes)
 258     : BaseNode(EntityType::kServer, ""), trace_(channel_tracer_max_nodes) {}
 259
 260 ServerNode::~ServerNode() {}
 261
 262 void ServerNode::AddChildSocket(RefCountedPtr<SocketNode> node) {
 263   MutexLock lock(&child_mu_);
 264   child_sockets_.insert(std::make_pair(node->uuid(), std::move(node)));
 265 }
 266
 267 void ServerNode::RemoveChildSocket(intptr_t child_uuid) {
 268   MutexLock lock(&child_mu_);
 269   child_sockets_.erase(child_uuid);
 270 }
 271
 272 void ServerNode::AddChildListenSocket(RefCountedPtr<ListenSocketNode> node) {
 273   MutexLock lock(&child_mu_);
 274   child_listen_sockets_.insert(std::make_pair(node->uuid(), std::move(node)));
 275 }
 276
 277 void ServerNode::RemoveChildListenSocket(intptr_t child_uuid) {
 278   MutexLock lock(&child_mu_);
 279   child_listen_sockets_.erase(child_uuid);
 280 }
 281
 282 std::string ServerNode::RenderServerSockets(intptr_t start_socket_id,
 283                                             intptr_t max_results) {
 284   GPR_ASSERT(start_socket_id >= 0);
 285   GPR_ASSERT(max_results >= 0);
 286   // If user does not set max_results, we choose 500.
 287   size_t pagination_limit = max_results == 0 ? 500 : max_results;
 288   Json::Object object;
 289   {
 290     MutexLock lock(&child_mu_);
 291     size_t sockets_rendered = 0;
 292     // Create list of socket refs.
 293     Json::Array array;
 294     auto it = child_sockets_.lower_bound(start_socket_id);
 295     for (; it != child_sockets_.end() && sockets_rendered < pagination_limit;
 296          ++it, ++sockets_rendered) {
 297       array.emplace_back(Json::Object{
 298           {"socketId", std::to_string(it->first)},
 299           {"name", it->second->name()},
 300       });
 301     }
 302     object["socketRef"] = std::move(array);
 303     if (it == child_sockets_.end()) object["end"] = true;
 304   }
 305   Json json = std::move(object);
 306   return json.Dump();
 307 }
 308
 309 Json ServerNode::RenderJson() {
 310   Json::Object data;
 311   // Fill in the channel trace if applicable.
 312   Json trace_json = trace_.RenderJson();
 313   if (trace_json.type() != Json::Type::JSON_NULL) {
 314     data["trace"] = std::move(trace_json);
 315   }
 316   // Ask CallCountingHelper to populate call count data.
 317   call_counter_.PopulateCallCounts(&data);
 318   // Construct top-level object.
 319   Json::Object object = {
 320       {"ref",
 321        Json::Object{
 322            {"serverId", std::to_string(uuid())},
 323        }},
 324       {"data", std::move(data)},
 325   };
 326   // Render listen sockets.
 327   {
 328     MutexLock lock(&child_mu_);
 329     if (!child_listen_sockets_.empty()) {
 330       Json::Array array;
 331       for (const auto& it : child_listen_sockets_) {
 332         array.emplace_back(Json::Object{
 333             {"socketId", std::to_string(it.first)},
 334             {"name", it.second->name()},
 335         });
 336       }
 337       object["listenSocket"] = std::move(array);
 338     }
 339   }
 340   return object;
 341 }
 342
 343 //
 344 // SocketNode::Security::Tls
 345 //
 346
 347 Json SocketNode::Security::Tls::RenderJson() {
 348   Json::Object data;
 349   if (type == NameType::kStandardName) {
 350     data["standard_name"] = name;
 351   } else if (type == NameType::kOtherName) {
 352     data["other_name"] = name;
 353   }
 354   if (!local_certificate.empty()) {
 355     data["local_certificate"] = absl::Base64Escape(local_certificate);
 356   }
 357   if (!remote_certificate.empty()) {
 358     data["remote_certificate"] = absl::Base64Escape(remote_certificate);
 359   }
 360   return data;
 361 }
 362
 363 //
 364 // SocketNode::Security
 365 //
 366
 367 Json SocketNode::Security::RenderJson() {
 368   Json::Object data;
 369   switch (type) {
 370     case ModelType::kUnset:
 371       break;
 372     case ModelType::kTls:
 373       if (tls) {
 374         data["tls"] = tls->RenderJson();
 375       }
 376       break;
 377     case ModelType::kOther:
 378       if (other) {
 379         data["other"] = *other;
 380       }
 381       break;
 382   }
 383   return data;
 384 }
 385
 386 namespace {
 387
 388 void* SecurityArgCopy(void* p) {
 389   SocketNode::Security* xds_certificate_provider =
 390       static_cast<SocketNode::Security*>(p);
 391   return xds_certificate_provider->Ref().release();
 392 }
 393
 394 void SecurityArgDestroy(void* p) {
 395   SocketNode::Security* xds_certificate_provider =
 396       static_cast<SocketNode::Security*>(p);
 397   xds_certificate_provider->Unref();
 398 }
 399
 400 int SecurityArgCmp(void* p, void* q) { return GPR_ICMP(p, q); }
 401
 402 const grpc_arg_pointer_vtable kChannelArgVtable = {
 403     SecurityArgCopy, SecurityArgDestroy, SecurityArgCmp};
 404
 405 }  // namespace
 406
 407 grpc_arg SocketNode::Security::MakeChannelArg() const {
 408   return grpc_channel_arg_pointer_create(
 409       const_cast<char*>(GRPC_ARG_CHANNELZ_SECURITY),
 410       const_cast<SocketNode::Security*>(this), &kChannelArgVtable);
 411 }
 412
 413 RefCountedPtr<SocketNode::Security> SocketNode::Security::GetFromChannelArgs(
 414     const grpc_channel_args* args) {
 415   Security* security = grpc_channel_args_find_pointer<Security>(
 416       args, GRPC_ARG_CHANNELZ_SECURITY);
 417   return security != nullptr ? security->Ref() : nullptr;
 418 }
 419
 420 //
 421 // SocketNode
 422 //
 423
 424 namespace {
 425
 426 void PopulateSocketAddressJson(Json::Object* json, const char* name,
 427                                const char* addr_str) {
 428   if (addr_str == nullptr) return;
 429   Json::Object data;
 430   absl::StatusOr<URI> uri = URI::Parse(addr_str);
 431   if (uri.ok() && (uri->scheme() == "ipv4" || uri->scheme() == "ipv6")) {
 432     std::string host;
 433     std::string port;
 434     GPR_ASSERT(
 435         SplitHostPort(absl::StripPrefix(uri->path(), "/"), &host, &port));
 436     int port_num = -1;
 437     if (!port.empty()) {
 438       port_num = atoi(port.data());
 439     }
 440     grpc_resolved_address resolved_host;
 441     grpc_error_handle error =
 442         grpc_string_to_sockaddr(&resolved_host, host.c_str(), port_num);
 443     if (error == GRPC_ERROR_NONE) {
 444       std::string packed_host = grpc_sockaddr_get_packed_host(&resolved_host);
 445       std::string b64_host = absl::Base64Escape(packed_host);
 446       data["tcpip_address"] = Json::Object{
 447           {"port", port_num},
 448           {"ip_address", b64_host},
 449       };
 450       (*json)[name] = std::move(data);
 451       return;
 452     }
 453     GRPC_ERROR_UNREF(error);
 454   }
 455   if (uri.ok() && uri->scheme() == "unix") {
 456     data["uds_address"] = Json::Object{
 457         {"filename", uri->path()},
 458     };
 459   } else {
 460     data["other_address"] = Json::Object{
 461         {"name", addr_str},
 462     };
 463   }
 464   (*json)[name] = std::move(data);
 465 }
 466
 467 }  // namespace
 468
 469 SocketNode::SocketNode(std::string local, std::string remote, std::string name,
 470                        RefCountedPtr<Security> security)
 471     : BaseNode(EntityType::kSocket, std::move(name)),
 472       local_(std::move(local)),
 473       remote_(std::move(remote)),
 474       security_(std::move(security)) {}
 475
 476 void SocketNode::RecordStreamStartedFromLocal() {
 477   streams_started_.FetchAdd(1, MemoryOrder::RELAXED);
 478   last_local_stream_created_cycle_.Store(gpr_get_cycle_counter(),
 479                                          MemoryOrder::RELAXED);
 480 }
 481
 482 void SocketNode::RecordStreamStartedFromRemote() {
 483   streams_started_.FetchAdd(1, MemoryOrder::RELAXED);
 484   last_remote_stream_created_cycle_.Store(gpr_get_cycle_counter(),
 485                                           MemoryOrder::RELAXED);
 486 }
 487
 488 void SocketNode::RecordMessagesSent(uint32_t num_sent) {
 489   messages_sent_.FetchAdd(num_sent, MemoryOrder::RELAXED);
 490   last_message_sent_cycle_.Store(gpr_get_cycle_counter(), MemoryOrder::RELAXED);
 491 }
 492
 493 void SocketNode::RecordMessageReceived() {
 494   messages_received_.FetchAdd(1, MemoryOrder::RELAXED);
 495   last_message_received_cycle_.Store(gpr_get_cycle_counter(),
 496                                      MemoryOrder::RELAXED);
 497 }
 498
 499 Json SocketNode::RenderJson() {
 500   // Create and fill the data child.
 501   Json::Object data;
 502   gpr_timespec ts;
 503   int64_t streams_started = streams_started_.Load(MemoryOrder::RELAXED);
 504   if (streams_started != 0) {
 505     data["streamsStarted"] = std::to_string(streams_started);
 506     gpr_cycle_counter last_local_stream_created_cycle =
 507         last_local_stream_created_cycle_.Load(MemoryOrder::RELAXED);
 508     if (last_local_stream_created_cycle != 0) {
 509       ts = gpr_convert_clock_type(
 510           gpr_cycle_counter_to_time(last_local_stream_created_cycle),
 511           GPR_CLOCK_REALTIME);
 512       data["lastLocalStreamCreatedTimestamp"] = gpr_format_timespec(ts);
 513     }
 514     gpr_cycle_counter last_remote_stream_created_cycle =
 515         last_remote_stream_created_cycle_.Load(MemoryOrder::RELAXED);
 516     if (last_remote_stream_created_cycle != 0) {
 517       ts = gpr_convert_clock_type(
 518           gpr_cycle_counter_to_time(last_remote_stream_created_cycle),
 519           GPR_CLOCK_REALTIME);
 520       data["lastRemoteStreamCreatedTimestamp"] = gpr_format_timespec(ts);
 521     }
 522   }
 523   int64_t streams_succeeded = streams_succeeded_.Load(MemoryOrder::RELAXED);
 524   if (streams_succeeded != 0) {
 525     data["streamsSucceeded"] = std::to_string(streams_succeeded);
 526   }
 527   int64_t streams_failed = streams_failed_.Load(MemoryOrder::RELAXED);
 528   if (streams_failed != 0) {
 529     data["streamsFailed"] = std::to_string(streams_failed);
 530   }
 531   int64_t messages_sent = messages_sent_.Load(MemoryOrder::RELAXED);
 532   if (messages_sent != 0) {
 533     data["messagesSent"] = std::to_string(messages_sent);
 534     ts = gpr_convert_clock_type(
 535         gpr_cycle_counter_to_time(
 536             last_message_sent_cycle_.Load(MemoryOrder::RELAXED)),
 537         GPR_CLOCK_REALTIME);
 538     data["lastMessageSentTimestamp"] = gpr_format_timespec(ts);
 539   }
 540   int64_t messages_received = messages_received_.Load(MemoryOrder::RELAXED);
 541   if (messages_received != 0) {
 542     data["messagesReceived"] = std::to_string(messages_received);
 543     ts = gpr_convert_clock_type(
 544         gpr_cycle_counter_to_time(
 545             last_message_received_cycle_.Load(MemoryOrder::RELAXED)),
 546         GPR_CLOCK_REALTIME);
 547     data["lastMessageReceivedTimestamp"] = gpr_format_timespec(ts);
 548   }
 549   int64_t keepalives_sent = keepalives_sent_.Load(MemoryOrder::RELAXED);
 550   if (keepalives_sent != 0) {
 551     data["keepAlivesSent"] = std::to_string(keepalives_sent);
 552   }
 553   // Create and fill the parent object.
 554   Json::Object object = {
 555       {"ref",
 556        Json::Object{
 557            {"socketId", std::to_string(uuid())},
 558            {"name", name()},
 559        }},
 560       {"data", std::move(data)},
 561   };
 562   if (security_ != nullptr &&
 563       security_->type != SocketNode::Security::ModelType::kUnset) {
 564     object["security"] = security_->RenderJson();
 565   }
 566   PopulateSocketAddressJson(&object, "remote", remote_.c_str());
 567   PopulateSocketAddressJson(&object, "local", local_.c_str());
 568   return object;
 569 }
 570
 571 //
 572 // ListenSocketNode
 573 //
 574
 575 ListenSocketNode::ListenSocketNode(std::string local_addr, std::string name)
 576     : BaseNode(EntityType::kSocket, std::move(name)),
 577       local_addr_(std::move(local_addr)) {}
 578
 579 Json ListenSocketNode::RenderJson() {
 580   Json::Object object = {
 581       {"ref",
 582        Json::Object{
 583            {"socketId", std::to_string(uuid())},
 584            {"name", name()},
 585        }},
 586   };
 587   PopulateSocketAddressJson(&object, "local", local_addr_.c_str());
 588   return object;
 589 }
 590
 591 }  // namespace channelz
 592 }  // namespace grpc_core