Refactor stream connection definition parsing (#21921)
* Refactor connection definition parsing and service determination logic - Extract and modularize `parse_connection_definition()` from `connect_to_this()` to improve readability and reusability. - Introduce `connect_to_definition_get_service()` for determining effective service or port. - Update `stream_parent_effective_port()` to utilize the new service extraction function. - Replace inline connection parsing with reusable logic, ensuring consistency and reducing redundancy. * Improve connection definition parsing and service extraction logic - Refactor `connect_to_definition_get_service()` to handle host and service parsing more effectively. - Add support for empty or null definitions, returning default service when applicable. - Simplify and streamline host length calculation and validation steps. * - Replace `stream_parent_effective_port()` with `stream_parent_effective_service()` for improved service extraction. - Update logging mechanisms to use service name instead of parsed port. - Improve clarity and consistency in service handling across connection logic.
Stelios Fragkakis committed
Mar 10, 2026 at 09:54 UTC
0d419462989d67f523587662f995df57b48a1b0b
3 files changed
+145
-52
src/libnetdata/socket/connect-to.c
+124
-50
@@ -217,6 +217,120 @@ int connect_to_this_ip46(
217
return fd;
218
}
219
220
+static int parse_connection_definition(
221
+ char *definition,
222
+ char **host,
223
+ char **service,
224
+ char **iface,
225
+ int *protocol,
226
+ int *socktype,
227
+ char **unix_path)
228
+{
229
+ if(!definition || !*definition)
230
+ return -1;
231
+
232
+ *host = definition;
233
+ *service = NULL;
234
+ *iface = "";
235
+ *protocol = IPPROTO_TCP;
236
+ *socktype = SOCK_STREAM;
237
+ *unix_path = NULL;
238
+
239
+ if(strncmp(*host, "tcp:", 4) == 0) {
240
+ *host += 4;
241
+ *protocol = IPPROTO_TCP;
242
+ *socktype = SOCK_STREAM;
243
+ }
244
+ else if(strncmp(*host, "udp:", 4) == 0) {
245
+ *host += 4;
246
+ *protocol = IPPROTO_UDP;
247
+ *socktype = SOCK_DGRAM;
248
+ }
249
+ else if(strncmp(*host, "unix:", 5) == 0) {
250
+ *unix_path = *host + 5;
251
+ return 1;
252
+ }
253
+ else if(**host == '/') {
254
+ *unix_path = *host;
255
+ return 1;
256
+ }
257
+
258
+ char *e = *host;
259
+ if(*e == '[') {
260
+ e = ++(*host);
261
+ while(*e && *e != ']') e++;
262
+ if(*e == ']') {
263
+ *e = '\0';
264
+ e++;
265
+ }
266
+ }
267
+ else {
268
+ while(*e && *e != ':' && *e != '%') e++;
269
+ }
270
+
271
+ if(*e == '%') {
272
+ *e = '\0';
273
+ e++;
274
+ *iface = e;
275
+ while(*e && *e != ':') e++;
276
+ }
277
+
278
+ if(*e == ':') {
279
+ *e = '\0';
280
+ e++;
281
+ *service = e;
282
+ }
283
+
284
+ if(!**host)
285
+ return -1;
286
+
287
+ return 0;
288
+}
289
+
290
+bool connect_to_definition_get_service(const char *definition, int default_port, char *service, size_t service_size) {
291
+ if(!service || !service_size)
292
+ return false;
293
+
294
+ snprintfz(service, service_size, "%d", default_port);
295
+
296
+ if(!definition || !*definition)
297
+ return true;
298
+
299
+ const char *s = definition;
300
+ if(strncmp(s, "tcp:", 4) == 0 || strncmp(s, "udp:", 4) == 0)
301
+ s += 4;
302
+ else if(strncmp(s, "unix:", 5) == 0 || *s == '/')
303
+ return true;
304
+
305
+ const char *e = s;
306
+ size_t host_len = 0;
307
+ if(*e == '[') {
308
+ const char *host_start = ++e;
309
+ while(*e && *e != ']') e++;
310
+ host_len = (size_t)(e - host_start);
311
+ if(*e == ']')
312
+ e++;
313
+ }
314
+ else {
315
+ const char *host_start = e;
316
+ while(*e && *e != ':' && *e != '%') e++;
317
+ host_len = (size_t)(e - host_start);
318
+ }
319
+
320
+ if(!host_len)
321
+ return false;
322
+
323
+ if(*e == '%') {
324
+ e++;
325
+ while(*e && *e != ':') e++;
326
+ }
327
+
328
+ if(*e == ':' && *(e + 1))
329
+ snprintfz(service, service_size, "%s", e + 1);
330
+
331
+ return true;
332
+}
333
+
334
// connect_to_this()
335
//
336
// definition format:
@@ -243,56 +357,16 @@ int connect_to_this(const char *definition, int default_port, struct timeval *ti
357
char default_service[10 + 1];
358
snprintfz(default_service, 10, "%d", default_port);
359
246
- char *host = buffer, *service = default_service, *iface = "";
247
- int protocol = IPPROTO_TCP, socktype = SOCK_STREAM;
360
+ char *host = NULL, *service = NULL, *iface = NULL, *unix_path = NULL;
361
+ int protocol = 0, socktype = 0;
362
uint32_t scope_id = 0;
363
250
- if(strncmp(host, "tcp:", 4) == 0) {
251
- host += 4;
252
- protocol = IPPROTO_TCP;
253
- socktype = SOCK_STREAM;
254
- }
255
- else if(strncmp(host, "udp:", 4) == 0) {
256
- host += 4;
257
- protocol = IPPROTO_UDP;
258
- socktype = SOCK_DGRAM;
259
- }
260
- else if(strncmp(host, "unix:", 5) == 0) {
261
- char *path = host + 5;
262
- return connect_to_unix(path, timeout);
263
- }
264
- else if(*host == '/') {
265
- char *path = host;
266
- return connect_to_unix(path, timeout);
267
- }
268
-
269
- char *e = host;
270
- if(*e == '[') {
271
- e = ++host;
272
- while(*e && *e != ']') e++;
273
- if(*e == ']') {
274
- *e = '\0';
275
- e++;
276
- }
277
- }
278
- else {
279
- while(*e && *e != ':' && *e != '%') e++;
280
- }
281
-
282
- if(*e == '%') {
283
- *e = '\0';
284
- e++;
285
- iface = e;
286
- while(*e && *e != ':') e++;
287
- }
288
-
289
- if(*e == ':') {
290
- *e = '\0';
291
- e++;
292
- service = e;
293
- }
364
+ int rc = parse_connection_definition(
365
+ buffer, &host, &service, &iface, &protocol, &socktype, &unix_path);
366
+ if(rc == 1)
367
+ return connect_to_unix(unix_path, timeout);
368
295
- if(!*host) {
369
+ if(rc == -1) {
370
nd_log(NDLS_DAEMON, NDLP_ERR,
371
"Definition '%s' does not specify a host.",
372
definition);
@@ -300,15 +374,15 @@ int connect_to_this(const char *definition, int default_port, struct timeval *ti
374
return -ND_SOCK_ERR_NO_HOST_IN_DEFINITION;
375
}
376
303
- if(*iface) {
377
+ if(iface && *iface) {
378
scope_id = if_nametoindex(iface);
379
if(!scope_id)
380
nd_log(NDLS_DAEMON, NDLP_ERR,
307
- "Cannot find a network interface named '%s'. Continuing with limiting the network interface",
381
+ "Cannot find a network interface named '%s'. Continuing without limiting the network interface",
382
iface);
383
}
384
311
- if(!*service)
385
+ if(!service || !*service)
386
service = default_service;
387
388
src/libnetdata/socket/connect-to.h
+4
@@ -12,6 +12,10 @@ int connect_to_this_ip46(
12
const char *service,
13
struct timeval *timeout,
14
bool *fallback_ipv4);
15
+// Fills `service` with the effective service from `definition`, or `default_port` when none is specified.
16
+// Returns false only when arguments are invalid or a non-empty definition is malformed (no host).
17
+// For NULL/empty definition, it keeps `service` set to `default_port` and returns true.
18
+bool connect_to_definition_get_service(const char *definition, int default_port, char *service, size_t service_size);
19
int connect_to_this(const char *definition, int default_port, struct timeval *timeout);
20
int connect_to_one_of(const char *destination, int default_port, struct timeval *timeout, size_t *reconnects_counter, char *connected_to, size_t connected_to_size);
21
int connect_to_one_of_urls(const char *destination, int default_port, struct timeval *timeout, size_t *reconnects_counter, char *connected_to, size_t connected_to_size);
src/streaming/stream-parents.c
+17
-2
@@ -45,6 +45,16 @@ struct stream_parent {
45
STREAM_PARENT *next;
46
};
47
48
+static const char *stream_parent_effective_service(const char *definition, int default_port, char *service, size_t service_size) {
49
+ if(!service || !service_size)
50
+ return "";
51
+
52
+ if(!connect_to_definition_get_service(definition, default_port, service, service_size))
53
+ snprintfz(service, service_size, "%d", default_port);
54
+
55
+ return service;
56
+}
57
+
58
// --------------------------------------------------------------------------------------------------------------------
59
// block unresponsive parents for some time, to allow speeding up the connection of the rest
60
@@ -374,9 +384,12 @@ static bool stream_info_json_parse_v1(struct json_object *jobj, const char *path
384
}
385
386
static bool stream_info_fetch(STREAM_PARENT *d, const char *uuid, int default_port, ND_SOCK *sender_sock, bool ssl, const char *hostname) {
387
+ char effective_service[NI_MAXSERV + 1];
388
+ stream_parent_effective_service(string2str(d->destination), default_port, effective_service, sizeof(effective_service));
389
+
390
ND_LOG_STACK lgs[] = {
391
ND_LOG_FIELD_STR(NDF_DST_IP, d->destination),
379
- ND_LOG_FIELD_I64(NDF_DST_PORT, default_port),
392
+ ND_LOG_FIELD_TXT(NDF_DST_PORT, effective_service),
393
ND_LOG_FIELD_TXT(NDF_REQUEST_METHOD, "GET"),
394
ND_LOG_FIELD_END(),
395
};
@@ -826,9 +839,11 @@ bool stream_parent_connect_to_one_unsafe(
839
rrdhost_hostname(host), string2str(d->destination), default_port,
840
i + 1, count);
841
842
+ char effective_service[NI_MAXSERV + 1];
843
+ stream_parent_effective_service(string2str(d->destination), default_port, effective_service, sizeof(effective_service));
844
ND_LOG_STACK lgs[] = {
845
ND_LOG_FIELD_STR(NDF_DST_IP, d->destination),
831
- ND_LOG_FIELD_I64(NDF_DST_PORT, default_port),
846
+ ND_LOG_FIELD_TXT(NDF_DST_PORT, effective_service),
847
ND_LOG_FIELD_END(),
848
};
849
ND_LOG_STACK_PUSH(lgs);