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 "./impl/node_impl.h"
41 : : #include "./impl/types.h"
42 : : #include "./version.h"
43 : :
44 : : #if __has_include(<unistd.h>)
45 : : #include <unistd.h>
46 : : #endif
47 : :
48 : : #ifdef _WIN32
49 : : #include <Winsock2.h>
50 : : #include <ws2tcpip.h>
51 : : #else
52 : : #include <arpa/inet.h>
53 : : #include <netinet/in.h>
54 : : #include <sys/socket.h>
55 : : #endif
56 : :
57 : : #define VLINK_DISCOVERY_MULTICAST 1
58 : : #define VLINK_DISCOVERY_OFFLINE 0
59 : :
60 : : namespace vlink {
61 : :
62 : : #ifdef _WIN32
63 : : using SocketHandle = SOCKET;
64 : : static constexpr SocketHandle kInvalidSocket = INVALID_SOCKET;
65 : : #else
66 : : using SocketHandle = int;
67 : : static constexpr SocketHandle kInvalidSocket = -1;
68 : : #endif
69 : :
70 : : [[maybe_unused]] static constexpr int kReportFirstInterval = 100;
71 : : [[maybe_unused]] static constexpr int kReportInterval = 500;
72 : : [[maybe_unused]] static constexpr size_t kMaxTaskSize = 10000U;
73 : : [[maybe_unused]] static constexpr uint32_t kMaxElapsedTime = 1000;
74 : : [[maybe_unused]] static constexpr int kBroadcastSendPort = 51694;
75 : : [[maybe_unused]] static constexpr int kSendTTL = 3;
76 : : [[maybe_unused]] static constexpr int kMaxMtuSize = 1450;
77 : :
78 : : #if VLINK_DISCOVERY_MULTICAST
79 : : [[maybe_unused]] static constexpr const char* kBroadcastAddress = "239.255.0.100";
80 : : #else
81 : : [[maybe_unused]] static constexpr const char* kBroadcastAddress = "255.255.255.255";
82 : : #endif
83 : :
84 : : template <typename T>
85 : 978 : [[maybe_unused]] static std::string_view convert_type(const T& t) {
86 [ + + + + : 978 : switch (t) {
+ + - ]
87 : 42 : case kServer:
88 : 42 : return "Ser";
89 : 143 : case kClient:
90 : 143 : return "Cli";
91 : 61 : case kPublisher:
92 : 61 : return "Pub";
93 : 606 : case kSubscriber:
94 : 606 : return "Sub";
95 : 36 : case kSetter:
96 : 36 : return "Set";
97 : 90 : case kGetter:
98 : 90 : return "Get";
99 : : default: // LCOV_EXCL_LINE GCOVR_EXCL_LINE
100 : : return "Unk"; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
101 : : }
102 : : }
103 : :
104 : : // DiscoveryReporter::Impl
105 : : struct DiscoveryReporter::Impl final {
106 : : std::unordered_set<NodeImpl*> info_set;
107 : : std::mutex mtx;
108 : : std::vector<std::string> message_list;
109 : : Timer timer;
110 : : std::string runtime_version;
111 : : std::string local_message;
112 : : bool is_profiler_enabled{false};
113 : : int64_t seq{0};
114 : :
115 : : SocketHandle sock{kInvalidSocket};
116 : : sockaddr_in address;
117 : : bool enable_native_discovery{false};
118 : : #ifdef _WIN32
119 : : bool winsock_initialized{false};
120 : : #endif
121 : : };
122 : :
123 [ + - ]: 40 : DiscoveryReporter::DiscoveryReporter() : impl_(std::make_unique<Impl>()) {
124 [ + - + - ]: 40 : set_name("DiscoveryReporter");
125 : :
126 [ + - + - ]: 80 : const std::string native_discovery = Utils::get_env("VLINK_DISCOVER_NATIVE");
127 : :
128 [ - + ]: 40 : if (native_discovery == "1") {
129 : 0 : impl_->enable_native_discovery = true;
130 : : }
131 : :
132 : 40 : impl_->runtime_version = Version{VLINK_VERSION_MAJOR, VLINK_VERSION_MINOR, VLINK_VERSION_PATCH}.to_string();
133 : :
134 : 40 : impl_->local_message =
135 [ + - + - : 80 : Helpers::escape_field(get_host_name()) + ":" + Utils::get_pid_str() + ":" + Helpers::escape_field(get_app_name());
+ - + - +
- + - ]
136 : :
137 : 40 : impl_->is_profiler_enabled = CpuProfiler::is_global_enabled();
138 : :
139 : : #ifdef _WIN32
140 : : ::WSADATA wsa_data;
141 : :
142 : : if VUNLIKELY (::WSAStartup(MAKEWORD(2, 2), &wsa_data) != 0) {
143 : : VLOG_F("DiscoveryReporter: Failed to initialize winsock.");
144 : : return;
145 : : }
146 : :
147 : : impl_->winsock_initialized = true;
148 : : #endif
149 : 40 : impl_->sock = ::socket(AF_INET, SOCK_DGRAM, 0);
150 : :
151 [ - + ]: 40 : if VUNLIKELY (impl_->sock == kInvalidSocket) {
152 : : VLOG_F("DiscoveryReporter: Failed to create socket."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
153 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
154 : : }
155 : :
156 : : #ifdef IP_TTL
157 : :
158 [ - + ]: 40 : if VUNLIKELY (::setsockopt(impl_->sock, IPPROTO_IP, IP_TTL, reinterpret_cast<const char*>(&kSendTTL),
159 : : sizeof(kSendTTL)) < 0) {
160 : : VLOG_F("DiscoveryReporter: Failed to set TTL option."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
161 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
162 : : }
163 : : #endif
164 : :
165 : : #if !VLINK_DISCOVERY_MULTICAST
166 : : int enable_broadcast = 1;
167 : :
168 : : if VUNLIKELY (::setsockopt(impl_->sock, SOL_SOCKET, SO_BROADCAST, reinterpret_cast<const char*>(&enable_broadcast),
169 : : sizeof(enable_broadcast)) < 0) {
170 : : VLOG_F("DiscoveryReporter: Failed to enable broadcast.");
171 : : return;
172 : : }
173 : : #endif
174 : :
175 : 40 : std::memset(&impl_->address, 0, sizeof(impl_->address));
176 : :
177 : 40 : impl_->address.sin_family = AF_INET;
178 : 40 : impl_->address.sin_addr.s_addr = inet_addr(kBroadcastAddress);
179 : :
180 [ - + ]: 40 : if (impl_->enable_native_discovery) {
181 : : struct in_addr local_interface;
182 : 0 : local_interface.s_addr = inet_addr("127.0.0.1");
183 : :
184 [ # # ]: 0 : if VUNLIKELY (::setsockopt(impl_->sock, IPPROTO_IP, IP_MULTICAST_IF,
185 : : reinterpret_cast<const char*>(&local_interface), sizeof(local_interface)) < 0) {
186 : : VLOG_F("DiscoveryReporter: Failed to set multicast interface to loopback."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
187 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
188 : : }
189 : : }
190 : :
191 : 40 : impl_->address.sin_port = htons(kBroadcastSendPort);
192 : :
193 [ + - ]: 40 : impl_->timer.set_interval(kReportFirstInterval);
194 [ + - ]: 40 : impl_->timer.set_loop_count(Timer::kInfinite);
195 [ + - ]: 40 : impl_->timer.attach(this);
196 [ + - + - ]: 40 : impl_->timer.start([this]() {
197 : 123 : send_report();
198 : :
199 : 123 : impl_->timer.set_interval(kReportInterval);
200 : 123 : });
201 : :
202 [ + - + - ]: 79 : post_task([this]() { send_report(); });
203 [ + - ]: 40 : }
204 : :
205 : 79 : DiscoveryReporter::~DiscoveryReporter() {
206 : 40 : impl_->timer.stop();
207 : :
208 : 40 : quit(true);
209 : :
210 : 40 : wait_for_quit();
211 : :
212 : : #if VLINK_DISCOVERY_OFFLINE
213 : : send_offline();
214 : : #endif
215 : :
216 [ + - ]: 40 : if VLIKELY (impl_->sock != kInvalidSocket) {
217 : : #ifdef _WIN32
218 : : ::closesocket(impl_->sock);
219 : : #else
220 : 40 : ::close(impl_->sock);
221 : : #endif
222 : :
223 : 40 : impl_->sock = kInvalidSocket;
224 : : }
225 : :
226 : : #ifdef _WIN32
227 : : if VLIKELY (impl_->winsock_initialized) {
228 : : ::WSACleanup();
229 : : impl_->winsock_initialized = false;
230 : : }
231 : : #endif
232 : 79 : }
233 : :
234 : 454 : void DiscoveryReporter::add(NodeImpl* node) {
235 [ + - ]: 454 : std::lock_guard lock(impl_->mtx);
236 [ + - ]: 454 : impl_->info_set.emplace(node);
237 : :
238 [ + - ]: 454 : if (!impl_->is_profiler_enabled) {
239 [ + - ]: 454 : rebuild_message();
240 : : }
241 : 454 : }
242 : :
243 : 454 : void DiscoveryReporter::remove(NodeImpl* node) {
244 [ + - ]: 454 : std::lock_guard lock(impl_->mtx);
245 [ + - ]: 454 : impl_->info_set.erase(node);
246 : :
247 [ + - ]: 454 : if (!impl_->is_profiler_enabled) {
248 [ + - ]: 454 : rebuild_message();
249 : : }
250 : 454 : }
251 : :
252 : 163 : size_t DiscoveryReporter::get_max_task_count() const { return kMaxTaskSize; }
253 : :
254 : 487 : uint32_t DiscoveryReporter::get_max_elapsed_time() const { return kMaxElapsedTime; }
255 : :
256 : 39 : void DiscoveryReporter::on_begin() { MessageLoop::on_begin(); }
257 : :
258 : 39 : void DiscoveryReporter::on_end() { MessageLoop::on_end(); }
259 : :
260 : 908 : void DiscoveryReporter::rebuild_message() {
261 : 908 : impl_->message_list.clear();
262 : :
263 : 908 : std::string message_pack;
264 [ + - ]: 908 : message_pack.reserve(kMaxMtuSize);
265 : :
266 : 908 : std::map<std::tuple<int, std::string, std::string, SchemaType>, std::pair<bool, double>> profiler_value_map;
267 : :
268 [ - + ]: 908 : if (impl_->is_profiler_enabled) {
269 : : // LCOV_EXCL_START GCOVR_EXCL_START
270 : : for (auto* node : impl_->info_set) {
271 : : const std::string& trim_url = Helpers::trim_string(node->url);
272 : : const std::string& trim_ser_type = Helpers::trim_string(node->ser_type);
273 : :
274 : : auto& value =
275 : : profiler_value_map[std::make_tuple(node->impl_type, trim_url, trim_ser_type, node->schema_type)].second;
276 : :
277 : : if (node->profiler) {
278 : : value += node->profiler->restart();
279 : : }
280 : : }
281 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
282 : : }
283 : :
284 [ + + ]: 1886 : for (auto* node : impl_->info_set) {
285 : 978 : const std::string& trim_url = Helpers::trim_string(node->url);
286 : 978 : const std::string& trim_ser_type = Helpers::trim_string(node->ser_type);
287 : :
288 [ - + ]: 978 : if VUNLIKELY (trim_url.empty()) {
289 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
290 : : }
291 : :
292 [ - + ]: 978 : if VUNLIKELY (trim_url.size() > 300) {
293 : : VLOG_F("DiscoveryReporter: Url is too long [", trim_url, "]."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
294 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
295 : : }
296 : :
297 [ - + ]: 978 : if VUNLIKELY (trim_ser_type.size() > 300) {
298 : : VLOG_F("DiscoveryReporter: Ser type is too long [", trim_ser_type, "]."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
299 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
300 : : }
301 : :
302 : 978 : std::string message;
303 [ + - + - ]: 978 : message.append(convert_type(node->impl_type)).append(" ");
304 : :
305 : : auto& profiler_result =
306 [ + - + - ]: 978 : profiler_value_map[std::make_tuple(node->impl_type, trim_url, trim_ser_type, node->schema_type)];
307 : :
308 [ + + ]: 978 : if (profiler_result.first) {
309 : 216 : continue;
310 : : } else {
311 : 762 : profiler_result.first = true;
312 : :
313 : 762 : const std::string escaped_url = Helpers::escape_field(trim_url);
314 [ + + + - : 762 : const std::string escaped_ser_type = trim_ser_type.empty() ? "{}" : Helpers::escape_field(trim_ser_type);
+ + - - ]
315 : :
316 [ + - + - ]: 762 : message.append(escaped_url).append(" ");
317 [ + - + - ]: 762 : message.append(escaped_ser_type).append(" ");
318 [ + - + - : 762 : message.append(std::to_string(static_cast<uint32_t>(node->schema_type))).append(" ");
+ - ]
319 [ + - ]: 762 : message.append(impl_->local_message);
320 : :
321 [ - + ]: 762 : if (impl_->is_profiler_enabled) {
322 : : message.append(":" + Helpers::double_to_string(profiler_result.second, 4)); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
323 : : }
324 : :
325 [ + - ]: 762 : message.append("\n");
326 : 762 : }
327 : :
328 [ - + ]: 762 : if VUNLIKELY (message.size() > kMaxMtuSize) {
329 : : VLOG_F("DiscoveryReporter: Discovery message is too long [", trim_url, "]."); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
330 : : return; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
331 : : }
332 : :
333 [ - + ]: 762 : if VUNLIKELY (message_pack.size() + message.size() > kMaxMtuSize) {
334 : : if VLIKELY (!message_pack.empty()) { // LCOV_EXCL_LINE GCOVR_EXCL_LINE
335 : : impl_->message_list.emplace_back(message_pack); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
336 : : }
337 : :
338 : : message_pack.clear(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
339 : : }
340 : :
341 [ + - ]: 762 : message_pack.append(message);
342 [ + + - + : 1410 : }
+ - + +
- ]
343 : :
344 [ + + ]: 908 : if (!message_pack.empty()) {
345 [ + - ]: 529 : impl_->message_list.emplace_back(std::move(message_pack));
346 : : }
347 [ + - + - ]: 908 : }
348 : :
349 : 162 : void DiscoveryReporter::send_report() {
350 [ + - ]: 162 : std::lock_guard lock(impl_->mtx);
351 : :
352 [ - + ]: 162 : if (impl_->is_profiler_enabled) {
353 : : rebuild_message(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
354 : : }
355 : :
356 [ + + ]: 302 : for (const auto& message : impl_->message_list) {
357 [ + - - + ]: 140 : if VUNLIKELY (::sendto(impl_->sock, message.c_str(), message.length(), 0,
358 : : reinterpret_cast<struct sockaddr*>(&impl_->address), sizeof(impl_->address)) < 0) {
359 : : // VLOG_W("Failed to send broadcast message.");
360 : : Utils::yield_cpu(); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
361 : : continue; // LCOV_EXCL_LINE GCOVR_EXCL_LINE
362 : : }
363 : : }
364 : :
365 : 162 : ++impl_->seq;
366 : 162 : }
367 : :
368 : : // LCOV_EXCL_START GCOVR_EXCL_START
369 : : void DiscoveryReporter::send_offline() {
370 : : std::lock_guard lock(impl_->mtx);
371 : :
372 : : static std::string offline_message = "offline\n" + impl_->local_message;
373 : :
374 : : if VUNLIKELY (::sendto(impl_->sock, offline_message.c_str(), offline_message.length(), 0,
375 : : reinterpret_cast<struct sockaddr*>(&impl_->address), sizeof(impl_->address)) < 0) {
376 : : // VLOG_W("Failed to send broadcast message.");
377 : : Utils::yield_cpu();
378 : : }
379 : : }
380 : : // LCOV_EXCL_STOP GCOVR_EXCL_STOP
381 : :
382 : 40 : const std::string& DiscoveryReporter::get_host_name() {
383 : 40 : static std::string host_name = [] {
384 : 40 : std::string name = Utils::get_host_name();
385 : :
386 [ - + ]: 40 : if (name.size() > 50) {
387 : : name.resize(50); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
388 : : }
389 : :
390 : 40 : return name;
391 [ + - + - : 40 : }();
+ - - - ]
392 : :
393 : 40 : return host_name;
394 : : }
395 : :
396 : 40 : const std::string& DiscoveryReporter::get_app_name() {
397 : 40 : static std::string app_name = [] {
398 : 40 : std::string name = Utils::get_app_name();
399 : :
400 [ - + ]: 40 : if (name.size() > 50) {
401 : : name.resize(50); // LCOV_EXCL_LINE GCOVR_EXCL_LINE
402 : : }
403 : :
404 : 40 : return name;
405 [ + - + - : 40 : }();
+ - - - ]
406 : :
407 : 40 : return app_name;
408 : : }
409 : :
410 : : } // namespace vlink
|
