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