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