Branch data Line data Source code
1 : : /*
2 : : * Copyright (C) 2026 by Thun Lu. All rights reserved.
3 : : * Author: Thun Lu <thun.lu@zohomail.cn>
4 : : * Repo: https://github.com/thun-res/vlink
5 : : * _ __ __ _ __
6 : : * | | / / / / (_) ____ / /__
7 : : * | | / / / / / / / __ \ / //_/
8 : : * | |/ / / /___ / / / / / / / ,<
9 : : * |___/ /_____/ /_/ /_/ /_/ /_/|_|
10 : : *
11 : : * Licensed under the Apache License, Version 2.0 (the "License");
12 : : * you may not use this file except in compliance with the License.
13 : : * You may obtain a copy of the License at
14 : : *
15 : : * http://www.apache.org/licenses/LICENSE-2.0
16 : : *
17 : : * Unless required by applicable law or agreed to in writing, software
18 : : * distributed under the License is distributed on an "AS IS" BASIS,
19 : : * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
20 : : * See the License for the specific language governing permissions and
21 : : * limitations under the License.
22 : : */
23 : :
24 : : #include "./extension/discovery_reporter.h"
25 : :
26 : : #include <cstring>
27 : : #include <map>
28 : : #include <memory>
29 : : #include <mutex>
30 : : #include <string>
31 : : #include <tuple>
32 : : #include <unordered_set>
33 : : #include <utility>
34 : : #include <vector>
35 : :
36 : : #include "./base/cpu_profiler.h"
37 : : #include "./base/helpers.h"
38 : : #include "./base/logger.h"
39 : : #include "./base/utils.h"
40 : : #include "./extension/discovery_viewer.h"
41 : : #include "./impl/node_impl.h"
42 : : #include "./impl/types.h"
43 : : #include "./version.h"
44 : :
45 : : #if __has_include(<unistd.h>)
46 : : #include <unistd.h>
47 : : #endif
48 : :
49 : : #ifdef _WIN32
50 : : #include <Winsock2.h>
51 : : #include <ws2tcpip.h>
52 : : #else
53 : : #include <arpa/inet.h>
54 : : #include <netinet/in.h>
55 : : #include <sys/socket.h>
56 : : #endif
57 : :
58 : : #define VLINK_DISCOVERY_MULTICAST 1
59 : : #define VLINK_DISCOVERY_OFFLINE 0
60 : :
61 : : namespace vlink {
62 : :
63 : : #ifdef _WIN32
64 : : using SocketHandle = SOCKET;
65 : : static constexpr SocketHandle kInvalidSocket = INVALID_SOCKET;
66 : : #else
67 : : using SocketHandle = int;
68 : : static constexpr SocketHandle kInvalidSocket = -1;
69 : : #endif
70 : :
71 : : [[maybe_unused]] static constexpr int kReportFirstInterval = 100;
72 : : [[maybe_unused]] static constexpr int kReportInterval = 500;
73 : : [[maybe_unused]] static constexpr size_t kMaxTaskSize = 10000U;
74 : : [[maybe_unused]] static constexpr uint32_t kMaxElapsedTime = 1000;
75 : : [[maybe_unused]] static constexpr int kInterfaceWarnInterval = 60000;
76 : : [[maybe_unused]] static constexpr int kSendTTL = 3;
77 : : [[maybe_unused]] static constexpr int kMaxMtuSize = 1450;
78 : :
79 : : #if VLINK_DISCOVERY_MULTICAST
80 : : [[maybe_unused]] static constexpr const char* kBroadcastAddress = "239.255.0.100";
81 : : #else
82 : : [[maybe_unused]] static constexpr const char* kBroadcastAddress = "255.255.255.255";
83 : : #endif
84 : :
85 : : template <typename T>
86 : 1119 : [[maybe_unused]] static std::string_view convert_type(const T& t) {
87 [ + + + + : 1119 : switch (t) {
+ + - ]
88 : 54 : case kServer:
89 : 54 : return "Ser";
90 : 147 : case kClient:
91 : 147 : return "Cli";
92 : 102 : case kPublisher:
93 : 102 : return "Pub";
94 : 628 : case kSubscriber:
95 : 628 : return "Sub";
96 : 60 : case kSetter:
97 : 60 : return "Set";
98 : 128 : case kGetter:
99 : 128 : return "Get";
100 : : default: // LCOV_EXCL_LINE GCOVR_EXCL_LINE
101 : : return "Unk"; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
102 : : }
103 : : }
104 : :
105 : 165 : static void send_message_list(SocketHandle sock, const sockaddr_in& address, const std::vector<std::string>& list) {
106 [ + + ]: 312 : for (const auto& message : list) {
107 [ + - - + ]: 147 : if VUNLIKELY (::sendto(sock, message.c_str(), message.length(), 0,
108 : : reinterpret_cast<const struct sockaddr*>(&address), sizeof(address)) < 0) {
109 : : Utils::yield_cpu(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
110 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
111 : : }
112 : : }
113 : 165 : }
114 : :
115 : 0 : [[maybe_unused]] static void send_multicast_list(SocketHandle sock, const sockaddr_in& address,
116 : : const std::vector<std::string>& ip_list,
117 : : const std::vector<std::string>& message_list) {
118 [ # # ]: 0 : for (const auto& ip : ip_list) {
119 : : in_addr interface_addr;
120 : 0 : interface_addr.s_addr = inet_addr(ip.c_str());
121 : :
122 [ # # ]: 0 : if VUNLIKELY (::setsockopt(sock, IPPROTO_IP, IP_MULTICAST_IF, reinterpret_cast<const char*>(&interface_addr),
123 : : sizeof(interface_addr)) < 0) {
124 [ # # # # : 0 : VLOG_W_EVERY_MS(kInterfaceWarnInterval, "DiscoveryReporter: Failed to set multicast interface [", ip, "].");
# # ]
125 : 0 : continue;
126 : 0 : }
127 : :
128 [ # # ]: 0 : send_message_list(sock, address, message_list);
129 : : }
130 : 0 : }
131 : :
132 : : // DiscoveryReporter::Impl
133 : : struct DiscoveryReporter::Impl final {
134 : : std::unordered_set<NodeImpl*> info_set;
135 : : std::mutex mtx;
136 : : std::vector<std::string> message_list;
137 : : Timer timer;
138 : : std::string runtime_version;
139 : : std::string local_message;
140 : : bool is_profiler_enabled{false};
141 : : int64_t seq{0};
142 : :
143 : : SocketHandle sock{kInvalidSocket};
144 : : sockaddr_in address;
145 : : std::vector<std::string> ip_list;
146 : : #ifdef _WIN32
147 : : bool winsock_initialized{false};
148 : : #endif
149 : : };
150 : :
151 [ + - ]: 40 : DiscoveryReporter::DiscoveryReporter() : impl_(std::make_unique<Impl>()) {
152 [ + - + - ]: 40 : set_name("DiscoveryReporter");
153 : :
154 [ + - + - ]: 80 : const std::string native_discovery = Utils::get_env("VLINK_DISCOVER_NATIVE");
155 : :
156 [ - + ]: 40 : if (native_discovery == "1") {
157 [ # # ]: 0 : impl_->ip_list.emplace_back("127.0.0.1");
158 : : } else {
159 [ + - + - ]: 40 : impl_->ip_list = Helpers::split_any(Utils::get_env("VLINK_DISCOVER_IP"));
160 : : }
161 : :
162 : 40 : impl_->runtime_version = Version{VLINK_VERSION_MAJOR, VLINK_VERSION_MINOR, VLINK_VERSION_PATCH}.to_string();
163 : :
164 : 40 : impl_->local_message =
165 [ + - + - : 80 : Helpers::escape_field(get_host_name()) + ":" + Utils::get_pid_str() + ":" + Helpers::escape_field(get_app_name());
+ - + - +
- + - ]
166 : :
167 : 40 : impl_->is_profiler_enabled = CpuProfiler::is_global_enabled();
168 : :
169 : : #ifdef _WIN32
170 : : ::WSADATA wsa_data;
171 : :
172 : : if VUNLIKELY (::WSAStartup(MAKEWORD(2, 2), &wsa_data) != 0) {
173 : : VLOG_F("DiscoveryReporter: Failed to initialize winsock.");
174 : : return;
175 : : }
176 : :
177 : : impl_->winsock_initialized = true;
178 : : #endif
179 : 40 : impl_->sock = ::socket(AF_INET, SOCK_DGRAM, 0);
180 : :
181 [ - + ]: 40 : if VUNLIKELY (impl_->sock == kInvalidSocket) {
182 : : VLOG_F("DiscoveryReporter: Failed to create socket."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
183 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
184 : : }
185 : :
186 : : #ifdef IP_TTL
187 : :
188 [ - + ]: 40 : if VUNLIKELY (::setsockopt(impl_->sock, IPPROTO_IP, IP_TTL, reinterpret_cast<const char*>(&kSendTTL),
189 : : sizeof(kSendTTL)) < 0) {
190 : : VLOG_F("DiscoveryReporter: Failed to set TTL option."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
191 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
192 : : }
193 : : #endif
194 : :
195 : : #if !VLINK_DISCOVERY_MULTICAST
196 : : int enable_broadcast = 1;
197 : :
198 : : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_BROADCAST, reinterpret_cast<const char*>(&enable_broadcast),
199 : : sizeof(enable_broadcast)) < 0) {
200 : : VLOG_F("DiscoveryReporter: Failed to enable broadcast.");
201 : : return;
202 : : }
203 : : #endif
204 : :
205 [ + - ]: 40 : DiscoveryViewer::warn_listen_domain();
206 : :
207 : 40 : std::memset(&impl_->address, 0, sizeof(impl_->address));
208 : :
209 : 40 : impl_->address.sin_family = AF_INET;
210 : 40 : impl_->address.sin_addr.s_addr = inet_addr(kBroadcastAddress);
211 [ + - ]: 40 : impl_->address.sin_port = htons(DiscoveryViewer::get_listen_port());
212 : :
213 [ + - ]: 40 : impl_->timer.set_interval(kReportFirstInterval);
214 [ + - ]: 40 : impl_->timer.set_loop_count(Timer::kInfinite);
215 [ + - ]: 40 : impl_->timer.attach(this);
216 [ + - + - ]: 40 : impl_->timer.start([this]() {
217 : 126 : send_report();
218 : :
219 : 126 : impl_->timer.set_interval(kReportInterval);
220 : 126 : });
221 : :
222 [ + - + - ]: 79 : post_task([this]() { send_report(); });
223 [ + - ]: 40 : }
224 : :
225 : 79 : DiscoveryReporter::~DiscoveryReporter() {
226 : : #ifdef _WIN32
227 : : if (Utils::is_terminating()) {
228 : : (void)impl_.release();
229 : : return;
230 : : }
231 : : #endif
232 : :
233 : 40 : impl_->timer.stop();
234 : :
235 : 40 : quit(true);
236 : :
237 : 40 : wait_for_quit();
238 : :
239 : : #if VLINK_DISCOVERY_OFFLINE
240 : : send_offline();
241 : : #endif
242 : :
243 [ + - ]: 40 : if VLIKELY (impl_->sock != kInvalidSocket) {
244 : : #ifdef _WIN32
245 : : ::closesocket(impl_->sock);
246 : : #else
247 : 40 : ::close(impl_->sock);
248 : : #endif
249 : :
250 : 40 : impl_->sock = kInvalidSocket;
251 : : }
252 : :
253 : : #ifdef _WIN32
254 : : if VLIKELY (impl_->winsock_initialized) {
255 : : ::WSACleanup();
256 : : impl_->winsock_initialized = false;
257 : : }
258 : : #endif
259 : 79 : }
260 : :
261 : 507 : void DiscoveryReporter::add(NodeImpl* node) {
262 [ + - ]: 507 : std::lock_guard lock(impl_->mtx);
263 [ + - ]: 507 : impl_->info_set.emplace(node);
264 : :
265 [ + - ]: 507 : if (!impl_->is_profiler_enabled) {
266 [ + - ]: 507 : rebuild_message();
267 : : }
268 : 507 : }
269 : :
270 : 507 : void DiscoveryReporter::remove(NodeImpl* node) {
271 [ + - ]: 507 : std::lock_guard lock(impl_->mtx);
272 [ + - ]: 507 : impl_->info_set.erase(node);
273 : :
274 [ + - ]: 507 : if (!impl_->is_profiler_enabled) {
275 [ + - ]: 507 : rebuild_message();
276 : : }
277 : 507 : }
278 : :
279 : 166 : size_t DiscoveryReporter::get_max_task_count() const { return kMaxTaskSize; }
280 : :
281 : 496 : uint32_t DiscoveryReporter::get_max_elapsed_time() const { return kMaxElapsedTime; }
282 : :
283 : 39 : void DiscoveryReporter::on_begin() { MessageLoop::on_begin(); }
284 : :
285 : 39 : void DiscoveryReporter::on_end() { MessageLoop::on_end(); }
286 : :
287 : 1014 : void DiscoveryReporter::rebuild_message() {
288 : 1014 : impl_->message_list.clear();
289 : :
290 : 1014 : std::string message_pack;
291 [ + - ]: 1014 : message_pack.reserve(kMaxMtuSize);
292 : :
293 : 1014 : std::map<std::tuple<int, std::string, std::string, SchemaType, bool>, std::pair<bool, double>> profiler_value_map;
294 : :
295 [ - + ]: 1014 : if (impl_->is_profiler_enabled) {
296 : : // LCOV_EXCL_START GCOVR_EXCL_START
297 : : for (auto* node : impl_->info_set) {
298 : : const std::string& trim_url = Helpers::trim_string(node->url);
299 : : const std::string& trim_ser_type = Helpers::trim_string(node->ser_type);
300 : :
301 : : auto& value = profiler_value_map[std::make_tuple(
302 : : node->impl_type, trim_url, trim_ser_type, node->schema_type,
303 : : node->impl_type == kPublisher && node->get_property("proxy.bridge") == "1")]
304 : : .second;
305 : :
306 : : if (node->profiler) {
307 : : value += node->profiler->restart();
308 : : }
309 : : }
310 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
311 : : }
312 : :
313 [ + + ]: 2133 : for (auto* node : impl_->info_set) {
314 : 1119 : const std::string& trim_url = Helpers::trim_string(node->url);
315 : 1119 : const std::string& trim_ser_type = Helpers::trim_string(node->ser_type);
316 : :
317 [ - + ]: 1119 : if VUNLIKELY (trim_url.empty()) {
318 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
319 : : }
320 : :
321 [ - + ]: 1119 : if VUNLIKELY (trim_url.size() > 300) {
322 : : VLOG_F("DiscoveryReporter: Url is too long [", trim_url, "]."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
323 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
324 : : }
325 : :
326 [ - + ]: 1119 : if VUNLIKELY (trim_ser_type.size() > 300) {
327 : : VLOG_F("DiscoveryReporter: Ser type is too long [", trim_ser_type, "]."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
328 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
329 : : }
330 : :
331 : 1119 : std::string message;
332 [ + - + - ]: 1119 : message.append(convert_type(node->impl_type)).append(" ");
333 : :
334 [ + + + - : 1119 : const bool bridge = node->impl_type == kPublisher && node->get_property("proxy.bridge") == "1";
+ - - + +
+ + + + +
- - - - -
- ]
335 : : auto& profiler_result =
336 [ + - + - ]: 1119 : profiler_value_map[std::make_tuple(node->impl_type, trim_url, trim_ser_type, node->schema_type, bridge)];
337 : :
338 [ + + ]: 1119 : if (profiler_result.first) {
339 : 228 : continue;
340 : : } else {
341 : 891 : profiler_result.first = true;
342 : :
343 : 891 : const std::string escaped_url = Helpers::escape_field(trim_url);
344 [ + + + - : 891 : const std::string escaped_ser_type = trim_ser_type.empty() ? "{}" : Helpers::escape_field(trim_ser_type);
+ + - - ]
345 : :
346 [ + - + - ]: 891 : message.append(escaped_url).append(" ");
347 [ + - + - ]: 891 : message.append(escaped_ser_type).append(" ");
348 [ + - + - : 891 : message.append(std::to_string(static_cast<uint32_t>(node->schema_type))).append(" ");
+ - ]
349 [ + - ]: 891 : message.append(impl_->local_message);
350 : :
351 [ - + ]: 891 : if (impl_->is_profiler_enabled) {
352 : : message.append(":" + Helpers::double_to_string(profiler_result.second, 4)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
353 : : }
354 : :
355 [ - + ]: 891 : if (bridge) {
356 [ # # ]: 0 : message.append(" bridge=1");
357 : : }
358 : :
359 [ + - ]: 891 : message.append("\n");
360 : 891 : }
361 : :
362 [ - + ]: 891 : if VUNLIKELY (message.size() > kMaxMtuSize) {
363 : : VLOG_F("DiscoveryReporter: Discovery message is too long [", trim_url, "]."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
364 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
365 : : }
366 : :
367 [ - + ]: 891 : if VUNLIKELY (message_pack.size() + message.size() > kMaxMtuSize) {
368 : : if VLIKELY (!message_pack.empty()) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
369 : : impl_->message_list.emplace_back(message_pack); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
370 : : }
371 : :
372 : : message_pack.clear(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
373 : : }
374 : :
375 [ + - ]: 891 : message_pack.append(message);
376 [ + + - + : 1575 : }
+ - + +
- ]
377 : :
378 [ + + ]: 1014 : if (!message_pack.empty()) {
379 [ + - ]: 617 : impl_->message_list.emplace_back(std::move(message_pack));
380 : : }
381 [ + - + - ]: 1014 : }
382 : :
383 : 165 : void DiscoveryReporter::send_report() {
384 [ + - ]: 165 : std::lock_guard lock(impl_->mtx);
385 : :
386 [ - + ]: 165 : if (impl_->is_profiler_enabled) {
387 : : rebuild_message(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
388 : : }
389 : :
390 : : #if VLINK_DISCOVERY_MULTICAST
391 [ + - ]: 165 : if (impl_->ip_list.empty()) {
392 [ + - ]: 165 : send_message_list(impl_->sock, impl_->address, impl_->message_list);
393 : : } else {
394 [ # # ]: 0 : send_multicast_list(impl_->sock, impl_->address, impl_->ip_list, impl_->message_list);
395 : : }
396 : : #else
397 : : send_message_list(impl_->sock, impl_->address, impl_->message_list);
398 : : #endif
399 : :
400 : 165 : ++impl_->seq;
401 : 165 : }
402 : :
403 : : // LCOV_EXCL_START GCOVR_EXCL_START
404 : : void DiscoveryReporter::send_offline() {
405 : : std::lock_guard lock(impl_->mtx);
406 : :
407 : : const std::vector<std::string> offline_list{"offline\n" + impl_->local_message};
408 : :
409 : : #if VLINK_DISCOVERY_MULTICAST
410 : : if (impl_->ip_list.empty()) {
411 : : send_message_list(impl_->sock, impl_->address, offline_list);
412 : : } else {
413 : : send_multicast_list(impl_->sock, impl_->address, impl_->ip_list, offline_list);
414 : : }
415 : : #else
416 : : send_message_list(impl_->sock, impl_->address, offline_list);
417 : : #endif
418 : : }
419 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
420 : :
421 : 40 : const std::string& DiscoveryReporter::get_host_name() {
422 : 40 : static std::string host_name = [] {
423 : 40 : std::string name = Utils::get_host_name();
424 : :
425 [ - + ]: 40 : if (name.size() > 50) {
426 : : name.resize(50); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
427 : : }
428 : :
429 : 40 : return name;
430 [ + - + - : 40 : }();
+ - - - ]
431 : :
432 : 40 : return host_name;
433 : : }
434 : :
435 : 40 : const std::string& DiscoveryReporter::get_app_name() {
436 : 40 : static std::string app_name = [] {
437 : 40 : std::string name = Utils::get_app_name();
438 : :
439 [ - + ]: 40 : if (name.size() > 50) {
440 : : name.resize(50); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
441 : : }
442 : :
443 : 40 : return name;
444 [ + - + - : 40 : }();
+ - - - ]
445 : :
446 : 40 : return app_name;
447 : : }
448 : :
449 : : } // namespace vlink
|
