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
BaseNode(EntityType type,std::string name)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
~BaseNode()67 BaseNode::~BaseNode() { ChannelzRegistry::Unregister(uuid_); }
68
RenderJsonString()69 std::string BaseNode::RenderJsonString() {
70 Json json = RenderJson();
71 return json.Dump();
72 }
73
74 //
75 // CallCountingHelper
76 //
77
CallCountingHelper()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
RecordCallStarted()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
RecordCallFailed()94 void CallCountingHelper::RecordCallFailed() {
95 per_cpu_counter_data_storage_[ExecCtx::Get()->starting_cpu()]
96 .calls_failed.FetchAdd(1, MemoryOrder::RELAXED);
97 }
98
RecordCallSucceeded()99 void CallCountingHelper::RecordCallSucceeded() {
100 per_cpu_counter_data_storage_[ExecCtx::Get()->starting_cpu()]
101 .calls_succeeded.FetchAdd(1, MemoryOrder::RELAXED);
102 }
103
CollectData(CounterData * out)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
PopulateCallCounts(Json::Object * json)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
ChannelNode(std::string target,size_t channel_tracer_max_nodes,bool is_internal_channel)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
GetChannelConnectivityStateChangeString(grpc_connectivity_state state)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
RenderJson()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
PopulateChildRefs(Json::Object * json)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
SetConnectivityState(grpc_connectivity_state state)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
AddChildChannel(intptr_t child_uuid)233 void ChannelNode::AddChildChannel(intptr_t child_uuid) {
234 MutexLock lock(&child_mu_);
235 child_channels_.insert(child_uuid);
236 }
237
RemoveChildChannel(intptr_t child_uuid)238 void ChannelNode::RemoveChildChannel(intptr_t child_uuid) {
239 MutexLock lock(&child_mu_);
240 child_channels_.erase(child_uuid);
241 }
242
AddChildSubchannel(intptr_t child_uuid)243 void ChannelNode::AddChildSubchannel(intptr_t child_uuid) {
244 MutexLock lock(&child_mu_);
245 child_subchannels_.insert(child_uuid);
246 }
247
RemoveChildSubchannel(intptr_t child_uuid)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
ServerNode(size_t channel_tracer_max_nodes)257 ServerNode::ServerNode(size_t channel_tracer_max_nodes)
258 : BaseNode(EntityType::kServer, ""), trace_(channel_tracer_max_nodes) {}
259
~ServerNode()260 ServerNode::~ServerNode() {}
261
AddChildSocket(RefCountedPtr<SocketNode> node)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
RemoveChildSocket(intptr_t child_uuid)267 void ServerNode::RemoveChildSocket(intptr_t child_uuid) {
268 MutexLock lock(&child_mu_);
269 child_sockets_.erase(child_uuid);
270 }
271
AddChildListenSocket(RefCountedPtr<ListenSocketNode> node)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
RemoveChildListenSocket(intptr_t child_uuid)277 void ServerNode::RemoveChildListenSocket(intptr_t child_uuid) {
278 MutexLock lock(&child_mu_);
279 child_listen_sockets_.erase(child_uuid);
280 }
281
RenderServerSockets(intptr_t start_socket_id,intptr_t max_results)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
RenderJson()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
RenderJson()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
RenderJson()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
SecurityArgCopy(void * p)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
SecurityArgDestroy(void * p)394 void SecurityArgDestroy(void* p) {
395 SocketNode::Security* xds_certificate_provider =
396 static_cast<SocketNode::Security*>(p);
397 xds_certificate_provider->Unref();
398 }
399
SecurityArgCmp(void * p,void * q)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
MakeChannelArg() const407 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
GetFromChannelArgs(const grpc_channel_args * args)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
PopulateSocketAddressJson(Json::Object * json,const char * name,const char * addr_str)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
SocketNode(std::string local,std::string remote,std::string name,RefCountedPtr<Security> security)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
RecordStreamStartedFromLocal()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
RecordStreamStartedFromRemote()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
RecordMessagesSent(uint32_t num_sent)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
RecordMessageReceived()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
RenderJson()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
ListenSocketNode(std::string local_addr,std::string name)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
RenderJson()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
593