Streaming alerts (#19586)
* remove undocumented "is ephemeral node" from stream.conf; ensure there always is an "_is_ephemeral" host label in nodes * added netdatacli command mark-stale-nodes-ephemeral; default value for [db].cleanup ephemeral hosts after = off; service thread now also cleans up hosts when they dont have retention; the minimum time for cleaning up ephemeral nodes is the orphan to archived time * do not print a tone of newlines * unify the logic about ephemerality for virtual nodes and streaming; now external plugins can control the ephemerality of virtual nodes by setting the label _is_ephemeral to true or false * netdata.streaming_inbound now has 2 instances, one of permanent nodes and another for ephemeral nodes, allowing users to have alert on any kind of state * log PCLAIMID when processing parent claimid, in an attempt to find the cases parent claim id remains empty when indirectly claimed * added alerts about missing permanent nodes * dequeue from hub before deleting contexts * validate node id and claim id * some more inlining * add the ephemerality status to streaming function * the alert now uses netdata.uptime.uptime * turn some PCLAIMID logs into debug logs * always have InReason in streaming function * add rowOptions and in/out connections counters to streaming function * fixing streaming sender statuses - incomplete * sender reason in function streaming * fix stale disconnected and pulse counters * added documentation * updated documentation
Costa Tsaousis committed
Feb 10, 2025 at 09:06 UTC
5907efae73f92a01ce27d70a3a38c1b8aeab8f8d
35 files changed
+767
-366
docs/nodes-ephemerality.md
new
+89
@@ -0,0 +1,89 @@
1
+
2
+# Nodes Ephemerality in Netdata
3
+
4
+## Overview
5
+
6
+In distributed monitoring environments, maintaining a reliable and consistent observability system is crucial. Netdata v2.23 introduces significant improvements to how ephemeral nodes are managed, ensuring a better balance between alerting consistency and flexibility for transient infrastructure.
7
+
8
+Previously, ephemeral nodes were defined as "nodes that are forgotten a day after they last disconnect." This approach sometimes led to unexpected inconsistencies in monitoring, particularly for users operating highly dynamic environments. With v2.23, ephemeral nodes are now defined as "nodes that are expected to disconnect without alerts being raised."
9
+
10
+This change serves three key objectives:
11
+
12
+1. **Stronger Monitoring Consistency for Permanent Nodes**: By ensuring that only permanent nodes trigger disconnection alerts, users can focus on real operational issues without being overwhelmed by alert noise.
13
+
14
+2. **Enhanced Flexibility for Transient Environments**: Users managing auto-scaling cloud instances, containers, and other volatile infrastructure can now configure nodes as ephemeral, preventing unnecessary alerts and making monitoring more effective.
15
+
16
+3. **Automated Cleanup for Ephemeral Nodes**: Netdata provides an automated way for the monitoring system to clean up itself by "forgetting" ephemeral nodes after a defined period. By default, the retention period is determined by the parent nodes' data retention settings. However, given that Netdata's tiered storage may provide retention for months or years, users may configure a shorter expiration time for ephemeral nodes.
17
+
18
+By introducing these changes, Netdata significantly enhances its ability to monitor itself as a mesh-like distributed observability system, ensuring that alerts reflect actual system health rather than expected, routine disconnections. Additionally, the automatic cleanup feature prevents stale ephemeral nodes from accumulating in the monitoring system, keeping dashboards clean and up-to-date.
19
+
20
+## Understanding Ephemeral Nodes
21
+
22
+When it comes to ephemerality, Netdata supports 2 types of nodes:
23
+
24
+- **Ephemeral Nodes**: nodes that are expected to disconnect and/or reconnect frequently, or nodes that are expected to shut down or vanish at any point in time. Such nodes may be:
25
+
26
+ - Auto-scaling cloud instances.
27
+ - Containers and VMs that are created and destroyed dynamically.
28
+ - IoT devices with intermittent connectivity.
29
+ - Development/test environments where nodes frequently restart.
30
+- **Permanent Nodes**: nodes that are expected to always be online, and disconnections are a strong indication of some kind of failure that operations teams should be aware of.
31
+
32
+
33
+## Configuring Ephemeral Nodes
34
+
35
+By default, all nodes in Netdata are **permanent**. Users can mark nodes as ephemeral like this:
36
+
37
+At the `netdata.conf` of the ephemeral node, set:
38
+
39
+```ini
40
+[global]
41
+ is ephemeral node = yes
42
+
43
+```
44
+
45
+And restart the node. This ephemerality flag is propagated to Netdata Parents and Netdata Cloud via the `_is_ephemeral` host label (boolean: true/false).
46
+
47
+## Netdata Parents Alerts
48
+
49
+Netdata v2.23 introduces two alerts for **permanent** nodes:
50
+
51
+- `streaming_never_connected`: Counts the number of **permanent** nodes never connected to a Netdata Parent (since its last restart) and transitions to WARNING when this number is non-zero.
52
+- `streaming_disconnected`: Counts the number of **permanent** nodes that have been connected but are now disconnected from the Netdata Parent, and transitions to WARNING when this number is non-zero.
53
+
54
+To identify the exact nodes that trigger these alarms, use the `Netdata-streaming` function under the `Top` tab of the dashboard. This Netdata Function presents a list (a table) of all nodes known to a Netdata Parent and provides detailed state information for the lifecycle of the node, including database status, ingestion status, streaming status, health and alerts status, and more.
55
+
56
+In this table, red lines indicate a problem during ingestion, and yellow lines indicate a problem during (re)streaming. Colored lines are only related to **permanent** nodes. Filter the table using `Ephemerality` by selecting `permanent` and use the table columns `InStatus`, `InReason`, and `InAge` to understand the ingestion issue at hand. Similarly, for (re)streaming to another Netdata Parent, use `OutStatus`, `OutReason`, and `OutAge`.
57
+
58
+### How to Mark Archived Nodes as Ephemeral
59
+
60
+In case there are **permanent** nodes that are no longer available, in order to clear the alerts, the following command must be run on each of the Netdata Parents having these alerts raised:
61
+
62
+```sh
63
+netdatacli mark-stale-nodes-ephemeral ALL_NODES
64
+
65
+```
66
+
67
+This command instructs Netdata to mark as **ephemeral** all the nodes not currently online.
68
+
69
+Keep in mind that nodes will be marked again as **permanent** if they reconnect and they have not been configured in their `netdata.conf` to be **ephemeral**. So, marking them at the parents is only useful for nodes that are not expected to connect again.
70
+
71
+## Netdata Cloud Alerts
72
+
73
+Before Netdata v2.23, Netdata Cloud was sending node unreachable notifications for all nodes, independently of their ephemerality.
74
+
75
+Since Netdata v2.23, Netdata Cloud is sending node unreachable notifications only for **permanent** nodes.
76
+
77
+## Automatically "Forgetting" Ephemeral Nodes
78
+
79
+Netdata versions prior to v2.23 were automatically "forgetting" ephemeral nodes if they disconnected for more than 1 day. In Netdata v2.23+, this feature is now **disabled** by default.
80
+
81
+To enable it again, set this in `netdata.conf` of the Netdata Parents that are expected to "forget" the ephemeral nodes:
82
+
83
+```ini
84
+[db]
85
+ cleanup ephemeral hosts after = 1d
86
+
87
+```
88
+
89
+The above instructs the Netdata Parent to automatically "forget" ephemeral nodes 1 day after they disconnect. When a node is "forgotten," its data is no longer available for queries, and when all parents reporting the node to Netdata Cloud "forget" it, Netdata Cloud automatically deletes the node.
\ No newline at end of file
src/collectors/proc.plugin/sys_devices_pci_aer.c
+2
-2
@@ -195,8 +195,8 @@ int do_proc_sys_devices_pci_aer(int update_every, usec_t dt __maybe_unused) {
195
int do_root_ports = CONFIG_BOOLEAN_AUTO;
196
int do_pci_slots = CONFIG_BOOLEAN_NO;
197
198
- char buffer[100 + 1] = "";
199
- rrdlabels_get_value_strcpyz(localhost->rrdlabels, buffer, 100, "_virtualization");
198
+ char buffer[128];
199
+ rrdlabels_get_value_strcpyz(localhost->rrdlabels, buffer, sizeof(buffer), "_virtualization");
200
if(strcmp(buffer, "none") != 0) {
201
// no need to run on virtualized environments
202
do_root_ports = CONFIG_BOOLEAN_NO;
src/daemon/commands.c
+79
-59
@@ -47,23 +47,27 @@ static cmd_status_t cmd_ping_execute(char *args, char **message);
47
static cmd_status_t cmd_aclk_state(char *args, char **message);
48
static cmd_status_t cmd_version(char *args, char **message);
49
static cmd_status_t cmd_dumpconfig(char *args, char **message);
50
-static cmd_status_t cmd_remove_node(char *args, char **message);
50
+static cmd_status_t cmd_remove_stale_node(char *args, char **message);
51
+static cmd_status_t cmd_mark_stale_nodes_ephemeral(char *args, char **message);
52
53
static command_info_t command_info_array[] = {
53
- {"help", cmd_help_execute, CMD_TYPE_HIGH_PRIORITY}, // show help menu
54
- {"reload-health", cmd_reload_health_execute, CMD_TYPE_ORTHOGONAL}, // reload health configuration
55
- {"reopen-logs", cmd_reopen_logs_execute, CMD_TYPE_ORTHOGONAL}, // Close and reopen log files
56
- {"shutdown-agent", cmd_exit_execute, CMD_TYPE_EXCLUSIVE}, // exit cleanly
57
- {"fatal-agent", cmd_fatal_execute, CMD_TYPE_HIGH_PRIORITY}, // exit with fatal error
58
- {"reload-claiming-state", cmd_reload_claiming_state_execute, CMD_TYPE_ORTHOGONAL}, // reload claiming state
59
- {"reload-labels", cmd_reload_labels_execute, CMD_TYPE_ORTHOGONAL}, // reload the labels
60
- {"read-config", cmd_read_config_execute, CMD_TYPE_CONCURRENT},
61
- {"write-config", cmd_write_config_execute, CMD_TYPE_ORTHOGONAL},
62
- {"ping", cmd_ping_execute, CMD_TYPE_ORTHOGONAL},
63
- {"aclk-state", cmd_aclk_state, CMD_TYPE_ORTHOGONAL},
64
- {"version", cmd_version, CMD_TYPE_ORTHOGONAL},
65
- {"dumpconfig", cmd_dumpconfig, CMD_TYPE_ORTHOGONAL},
66
- {"remove-stale-node", cmd_remove_node, CMD_TYPE_ORTHOGONAL}
54
+ {"help", "", "Show this help menu.", cmd_help_execute, CMD_TYPE_HIGH_PRIORITY}, // show help menu
55
+ {"reload-health", "", "Reload health configuration.", cmd_reload_health_execute, CMD_TYPE_ORTHOGONAL}, // reload health configuration
56
+ {"reopen-logs", "", "Close and reopen log files.", cmd_reopen_logs_execute, CMD_TYPE_ORTHOGONAL}, // Close and reopen log files
57
+ {"shutdown-agent", "", "Cleanup and exit the netdata agent.", cmd_exit_execute, CMD_TYPE_EXCLUSIVE}, // exit cleanly
58
+ {"fatal-agent", "", "Log the state and halt the netdata agent.", cmd_fatal_execute, CMD_TYPE_HIGH_PRIORITY}, // exit with fatal error
59
+ {"reload-claiming-state", "", "Reload agent claiming state from disk.", cmd_reload_claiming_state_execute, CMD_TYPE_ORTHOGONAL}, // reload claiming state
60
+ {"reload-labels", "", "Reload all localhost labels.", cmd_reload_labels_execute, CMD_TYPE_ORTHOGONAL}, // reload the labels
61
+ {"read-config", "", "", cmd_read_config_execute, CMD_TYPE_CONCURRENT},
62
+ {"write-config", "", "", cmd_write_config_execute, CMD_TYPE_ORTHOGONAL},
63
+ {"ping", "", "Return with 'pong' if agent is alive.", cmd_ping_execute, CMD_TYPE_ORTHOGONAL},
64
+ {"aclk-state", "[json]", "Returns current state of ACLK and Netdata Cloud connection. (optionally in json).", cmd_aclk_state, CMD_TYPE_ORTHOGONAL},
65
+ {"version", "", "Returns the netdata version.", cmd_version, CMD_TYPE_ORTHOGONAL},
66
+ {"dumpconfig", "", "Returns the current netdata.conf on stdout.", cmd_dumpconfig, CMD_TYPE_ORTHOGONAL},
67
+ {"remove-stale-node", "<node_id | machine_guid | hostname | ALL_NODES>",
68
+ "Marks one or all disconnected nodes as ephemeral, and removes them\n so that they are no longer available for queries, from both this\n Netdata Agent dashboard and Netdata Cloud.", cmd_remove_stale_node, CMD_TYPE_ORTHOGONAL},
69
+ {"mark-stale-nodes-ephemeral", "<node_id | machine_guid | hostname | ALL_NODES>",
70
+ "Marks one or all disconnected nodes as ephemeral, while keeping their retention\n available for queries on both this Netdata Agent dashboard and Netdata Cloud", cmd_mark_stale_nodes_ephemeral, CMD_TYPE_ORTHOGONAL},
71
};
72
73
/* Mutexes for commands of type CMD_TYPE_ORTHOGONAL */
@@ -105,35 +109,26 @@ static command_lock_t *cmd_unlock_by_type[] = {
109
static cmd_status_t cmd_help_execute(char *args, char **message)
110
{
111
(void)args;
112
+ CLEAN_BUFFER *wb = buffer_create(0, NULL);
113
+
114
+ buffer_strcat(wb, "The commands are:\n\n");
115
+ for(size_t i = 0; i < _countof(command_info_array); i++) {
116
+ const command_info_t *t = &command_info_array[i];
117
+ if(!t->help || !t->help[0]) continue;
118
+
119
+ buffer_strcat(wb, " ");
120
+ buffer_strcat(wb, t->cmd_str);
121
+ if(t->params && t->params[0]) {
122
+ buffer_putc(wb, ' ');
123
+ buffer_strcat(wb, t->params);
124
+ }
125
+ buffer_putc(wb, '\n');
126
+ buffer_strcat(wb, " ");
127
+ buffer_strcat(wb, t->help);
128
+ buffer_strcat(wb, "\n\n");
129
+ }
130
109
- *message = mallocz(MAX_COMMAND_LENGTH);
110
- strncpyz(*message,
111
- "\nThe commands are:\n\n"
112
- "help\n"
113
- " Show this help menu.\n\n"
114
- "reload-health\n"
115
- " Reload health configuration.\n\n"
116
- "reload-labels\n"
117
- " Reload all labels.\n\n"
118
- "reopen-logs\n"
119
- " Close and reopen log files.\n\n"
120
- "shutdown-agent\n"
121
- " Cleanup and exit the netdata agent.\n\n"
122
- "fatal-agent\n"
123
- " Log the state and halt the netdata agent.\n\n"
124
- "reload-claiming-state\n"
125
- " Reload agent claiming state from disk.\n\n"
126
- "ping\n"
127
- " Return with 'pong' if agent is alive.\n\n"
128
- "aclk-state [json]\n"
129
- " Returns current state of ACLK and Cloud connection. (optionally in json).\n\n"
130
- "dumpconfig\n"
131
- " Returns the current netdata.conf on stdout.\n\n"
132
- "remove-stale-node <node_id | machine_guid | hostname | ALL_NODES>\n"
133
- " Unregisters and removes a node from the cloud.\n\n"
134
- "version\n"
135
- " Returns the netdata version.\n",
136
- MAX_COMMAND_LENGTH - 1);
131
+ *message = strdupz(buffer_tostring(wb));
132
return CMD_STATUS_SUCCESS;
133
}
134
@@ -351,40 +346,53 @@ static cmd_status_t cmd_dumpconfig(char *args, char **message)
346
return CMD_STATUS_SUCCESS;
347
}
348
354
-static int remove_ephemeral_host(BUFFER *wb, RRDHOST *host, bool report_error)
349
+static int remove_ephemeral_host(BUFFER *wb, RRDHOST *host, bool report_error, bool unregister)
350
{
351
if (host == localhost) {
352
if (report_error)
358
- buffer_sprintf(wb, "You cannot unregister the parent node (%s)", rrdhost_hostname(host));
353
+ buffer_sprintf(wb, "Node '%s' (machine guid: %s) is our localhost - not changing it",
354
+ rrdhost_hostname(host), host->machine_guid);
355
return 0;
356
}
357
358
if (rrdhost_is_online(host)) {
359
if (report_error)
364
- buffer_sprintf(wb, "Cannot unregister a live node (%s)", rrdhost_hostname(host));
360
+ buffer_sprintf(wb, "Node '%s' (machine guid: %s) is online - not changing it",
361
+ rrdhost_hostname(host), host->machine_guid);
362
return 0;
363
}
364
365
if (!rrdhost_option_check(host, RRDHOST_OPTION_EPHEMERAL_HOST)) {
366
rrdhost_option_set(host, RRDHOST_OPTION_EPHEMERAL_HOST);
367
sql_set_host_label(&host->host_id.uuid, "_is_ephemeral", "true");
371
- aclk_host_state_update(host, 0, 0);
372
- unregister_node(host->machine_guid);
373
- host->node_id = UUID_ZERO;
374
- buffer_sprintf(wb, "Unregistering node with machine guid %s, hostname = %s", host->machine_guid, rrdhost_hostname(host));
375
- rrd_wrlock();
376
- rrdhost_free___while_having_rrd_wrlock(host, true);
377
- rrd_wrunlock();
368
+
369
+ if(unregister) {
370
+ aclk_host_state_update(host, 0, 0);
371
+ unregister_node(host->machine_guid);
372
+ host->node_id = UUID_ZERO;
373
+ buffer_sprintf(wb, "Node '%s' (machine guid: %s) has been unregistered",
374
+ rrdhost_hostname(host), host->machine_guid);
375
+ rrd_wrlock();
376
+ rrdhost_free___while_having_rrd_wrlock(host);
377
+ rrd_wrunlock();
378
+ }
379
+ else {
380
+ pulse_host_status(host, 0, 0);
381
+ buffer_sprintf(wb, "Node '%s' (machine guid: %s) has been marked ephemeral",
382
+ rrdhost_hostname(host), host->machine_guid);
383
+ }
384
+
385
return 1;
386
}
387
if (report_error)
381
- buffer_sprintf(wb, "Node with machine guid %s, hostname = %s is already unregistered", host->machine_guid, rrdhost_hostname(host));
388
+ buffer_sprintf(wb, "Node '%s' (machine guid: %s) is already ephemeral - not changing it",
389
+ rrdhost_hostname(host), host->machine_guid);
390
return 0;
391
}
392
393
#define SQL_HOSTNAME_TO_REMOVE "SELECT host_id FROM host WHERE (hostname = @hostname OR @hostname = 'ALL_NODES')"
394
387
-static cmd_status_t cmd_remove_node(char *args, char **message)
395
+static cmd_status_t cmd_remove_stale_node_internal(char *args, char **message, bool unregister)
396
{
397
(void)args;
398
@@ -402,7 +410,7 @@ static cmd_status_t cmd_remove_node(char *args, char **message)
410
if (!host) {
411
sqlite3_stmt *res = NULL;
412
405
- bool report_error = strcmp(args, "ALL_NODES");
413
+ bool report_error = strcmp(args, "ALL_NODES") != 0;
414
415
if (!PREPARE_STATEMENT(db_meta, SQL_HOSTNAME_TO_REMOVE, &res)) {
416
buffer_sprintf(wb, "Failed to prepare database statement to check for stale nodes");
@@ -419,9 +427,11 @@ static cmd_status_t cmd_remove_node(char *args, char **message)
427
uuid_unparse_lower(*(nd_uuid_t *)sqlite3_column_blob(res, 0), guid);
428
host = rrdhost_find_by_guid(guid);
429
if (host) {
422
- if (cnt)
430
+ int rc = remove_ephemeral_host(wb, host, report_error, unregister);
431
+ if(rc) {
432
+ cnt += rc;
433
buffer_fast_strcat(wb, "\n", 1);
424
- cnt += remove_ephemeral_host(wb, host, report_error);
434
+ }
435
}
436
}
437
if (!cnt && buffer_strlen(wb) == 0) {
@@ -435,7 +445,7 @@ static cmd_status_t cmd_remove_node(char *args, char **message)
445
SQLITE_FINALIZE(res);
446
}
447
else
438
- (void) remove_ephemeral_host(wb, host, true);
448
+ (void) remove_ephemeral_host(wb, host, true, unregister);
449
450
done:
451
*message = strdupz(buffer_tostring(wb));
@@ -443,6 +453,16 @@ done:
453
return CMD_STATUS_SUCCESS;
454
}
455
456
+static cmd_status_t cmd_remove_stale_node(char *args, char **message)
457
+{
458
+ return cmd_remove_stale_node_internal(args, message, true);
459
+}
460
+
461
+static cmd_status_t cmd_mark_stale_nodes_ephemeral(char *args, char **message)
462
+{
463
+ return cmd_remove_stale_node_internal(args, message, false);
464
+}
465
+
466
static void cmd_lock_exclusive(unsigned index)
467
{
468
(void)index;
src/daemon/commands.h
+3
@@ -21,6 +21,7 @@ typedef enum cmd {
21
CMD_VERSION,
22
CMD_DUMPCONFIG,
23
CMD_REMOVE_NODE,
24
+ CMD_MARK_NODE,
25
CMD_TOTAL_COMMANDS
26
} cmd_t;
27
@@ -65,6 +66,8 @@ typedef cmd_status_t (command_action_t) (char *args, char **message);
66
67
typedef struct command_info {
68
char *cmd_str; // the command string
69
+ char *params;
70
+ char *help;
71
command_action_t *func; // the function that executes the command
72
cmd_type_t type; // Concurrency control information for the command
73
} command_info_t;
src/daemon/config/netdata-conf-db.c
+17
-11
@@ -12,8 +12,8 @@ static size_t storage_tiers_grouping_iterations[RRD_STORAGE_TIERS] = {1, 60, 60,
12
static time_t storage_tiers_retention_time_s[RRD_STORAGE_TIERS] = {14 * DAYS, 90 * DAYS, 2 * 365 * DAYS, 2 * 365 * DAYS, 2 * 365 * DAYS};
13
14
time_t rrdset_free_obsolete_time_s = 3600;
15
-time_t rrdhost_free_orphan_time_s = 3600;
16
-time_t rrdhost_free_ephemeral_time_s = 86400;
15
+time_t rrdhost_cleanup_orphan_to_archive_time_s = 3600;
16
+time_t rrdhost_free_ephemeral_time_s = 0;
17
18
size_t get_tier_grouping(size_t tier) {
19
if(unlikely(tier >= nd_profile.storage_tiers)) tier = nd_profile.storage_tiers - 1;
@@ -344,11 +344,6 @@ void netdata_conf_section_db(void) {
344
if(run) return;
345
run = true;
346
347
- // ------------------------------------------------------------------------
348
-
349
- rrdhost_free_orphan_time_s =
350
- inicfg_get_duration_seconds(&netdata_config, CONFIG_SECTION_DB, "cleanup orphan hosts after", rrdhost_free_orphan_time_s);
351
-
347
// ------------------------------------------------------------------------
348
// get default database update frequency
349
@@ -404,16 +399,27 @@ void netdata_conf_section_db(void) {
399
400
// --------------------------------------------------------------------
401
402
+ rrdhost_cleanup_orphan_to_archive_time_s =
403
+ inicfg_get_duration_seconds(&netdata_config, CONFIG_SECTION_DB, "cleanup orphan hosts after", rrdhost_cleanup_orphan_to_archive_time_s);
404
+ if(rrdhost_cleanup_orphan_to_archive_time_s < 10) {
405
+ rrdhost_cleanup_orphan_to_archive_time_s = 10;
406
+ inicfg_set_duration_seconds(&netdata_config, CONFIG_SECTION_DB, "cleanup orphan hosts after", rrdhost_cleanup_orphan_to_archive_time_s);
407
+ }
408
+
409
rrdhost_free_ephemeral_time_s =
410
inicfg_get_duration_seconds(&netdata_config, CONFIG_SECTION_DB, "cleanup ephemeral hosts after", rrdhost_free_ephemeral_time_s);
411
+ if(rrdhost_free_ephemeral_time_s && rrdhost_free_ephemeral_time_s < rrdhost_cleanup_orphan_to_archive_time_s) {
412
+ // the free ephemeral time cannot be less than the cleanup orphan time
413
+ rrdhost_free_ephemeral_time_s = rrdhost_cleanup_orphan_to_archive_time_s;
414
+ inicfg_set_duration_seconds(&netdata_config, CONFIG_SECTION_DB, "cleanup ephemeral hosts after", rrdhost_free_ephemeral_time_s);
415
+ }
416
417
rrdset_free_obsolete_time_s =
418
inicfg_get_duration_seconds(&netdata_config, CONFIG_SECTION_DB, "cleanup obsolete charts after", rrdset_free_obsolete_time_s);
412
-
413
- // Current chart locking and invalidation scheme doesn't prevent Netdata from segmentation faults if a short
414
- // cleanup delay is set. Extensive stress tests showed that 10 seconds is quite a safe delay. Look at
415
- // https://github.com/netdata/netdata/pull/11222#issuecomment-868367920 for more information.
419
if (rrdset_free_obsolete_time_s < 10) {
420
+ // Current chart locking and invalidation scheme doesn't prevent Netdata from segmentation faults if a short
421
+ // cleanup delay is set. Extensive stress tests showed that 10 seconds is quite a safe delay. Look at
422
+ // https://github.com/netdata/netdata/pull/11222#issuecomment-868367920 for more information.
423
rrdset_free_obsolete_time_s = 10;
424
netdata_log_info("The \"cleanup obsolete charts after\" option was set to 10 seconds.");
425
inicfg_set_duration_seconds(&netdata_config, CONFIG_SECTION_DB, "cleanup obsolete charts after", rrdset_free_obsolete_time_s);
src/daemon/pulse/pulse-parents.c
+121
-70
@@ -28,15 +28,28 @@ struct {
28
struct by_reason disconnects_by_reason;
29
30
// gauges
31
- ssize_t nodes_local;
32
- ssize_t nodes_virtual;
33
- ssize_t nodes_loading;
34
- ssize_t nodes_archived;
35
- ssize_t nodes_offline;
36
- ssize_t nodes_waiting;
37
- ssize_t nodes_replicating;
38
- ssize_t nodes_replication_waiting;
39
- ssize_t nodes_running;
31
+ struct {
32
+ RRDSET *st_nodes;
33
+ RRDDIM *rd_loading;
34
+ RRDDIM *rd_local;
35
+ RRDDIM *rd_virtual;
36
+ RRDDIM *rd_archived;
37
+ RRDDIM *rd_offline;
38
+ RRDDIM *rd_waiting;
39
+ RRDDIM *rd_replication_waiting;
40
+ RRDDIM *rd_replicating;
41
+ RRDDIM *rd_running;
42
+
43
+ ssize_t nodes_local;
44
+ ssize_t nodes_virtual;
45
+ ssize_t nodes_loading;
46
+ ssize_t nodes_archived;
47
+ ssize_t nodes_offline;
48
+ ssize_t nodes_waiting;
49
+ ssize_t nodes_replicating;
50
+ ssize_t nodes_replication_waiting;
51
+ ssize_t nodes_running;
52
+ } type[2];
53
} parent;
54
55
struct {
@@ -53,6 +66,7 @@ struct {
66
ssize_t nodes_replicating;
67
ssize_t nodes_running;
68
ssize_t nodes_no_dst;
69
+ ssize_t nodes_no_dst_failed;
70
} sender;
71
72
} p = { 0 };
@@ -101,6 +115,9 @@ static void update_reason(struct by_reason *b, STREAM_HANDSHAKE reason) {
115
}
116
117
static void pulse_host_add_sub_status(PULSE_HOST_STATUS status, ssize_t val, STREAM_HANDSHAKE reason) {
118
+ size_t idx = status & PULSE_HOST_STATUS_EPHEMERAL ? 1 : 0;
119
+ status &= ~(PULSE_HOST_STATUS_EPHEMERAL | PULSE_HOST_STATUS_PERMANENT);
120
+
121
while(status) {
122
PULSE_HOST_STATUS s = 1 << (__builtin_ffs(status) - 1);
123
status &= ~s;
@@ -112,42 +129,42 @@ static void pulse_host_add_sub_status(PULSE_HOST_STATUS status, ssize_t val, STR
129
break;
130
131
case PULSE_HOST_STATUS_LOCAL:
115
- __atomic_add_fetch(&p.parent.nodes_local, val, __ATOMIC_RELAXED);
132
+ __atomic_add_fetch(&p.parent.type[idx].nodes_local, val, __ATOMIC_RELAXED);
133
break;
134
135
case PULSE_HOST_STATUS_VIRTUAL:
119
- __atomic_add_fetch(&p.parent.nodes_virtual, val, __ATOMIC_RELAXED);
136
+ __atomic_add_fetch(&p.parent.type[idx].nodes_virtual, val, __ATOMIC_RELAXED);
137
break;
138
139
case PULSE_HOST_STATUS_LOADING:
123
- __atomic_add_fetch(&p.parent.nodes_loading, val, __ATOMIC_RELAXED);
140
+ __atomic_add_fetch(&p.parent.type[idx].nodes_loading, val, __ATOMIC_RELAXED);
141
break;
142
143
case PULSE_HOST_STATUS_ARCHIVED:
127
- __atomic_add_fetch(&p.parent.nodes_archived, val, __ATOMIC_RELAXED);
144
+ __atomic_add_fetch(&p.parent.type[idx].nodes_archived, val, __ATOMIC_RELAXED);
145
break;
146
147
case PULSE_HOST_STATUS_RCV_OFFLINE:
131
- __atomic_add_fetch(&p.parent.nodes_offline, val, __ATOMIC_RELAXED);
148
+ __atomic_add_fetch(&p.parent.type[idx].nodes_offline, val, __ATOMIC_RELAXED);
149
do_parent_reason = true;
150
break;
151
152
case PULSE_HOST_STATUS_RCV_WAITING:
136
- __atomic_add_fetch(&p.parent.nodes_waiting, val, __ATOMIC_RELAXED);
153
+ __atomic_add_fetch(&p.parent.type[idx].nodes_waiting, val, __ATOMIC_RELAXED);
154
do_parent_reason = true;
155
reason = 0;
156
break;
157
158
case PULSE_HOST_STATUS_RCV_REPLICATION_WAIT:
142
- __atomic_add_fetch(&p.parent.nodes_replication_waiting, val, __ATOMIC_RELAXED);
159
+ __atomic_add_fetch(&p.parent.type[idx].nodes_replication_waiting, val, __ATOMIC_RELAXED);
160
break;
161
162
case PULSE_HOST_STATUS_RCV_REPLICATING:
146
- __atomic_add_fetch(&p.parent.nodes_replicating, val, __ATOMIC_RELAXED);
163
+ __atomic_add_fetch(&p.parent.type[idx].nodes_replicating, val, __ATOMIC_RELAXED);
164
break;
165
166
case PULSE_HOST_STATUS_RCV_RUNNING:
150
- __atomic_add_fetch(&p.parent.nodes_running, val, __ATOMIC_RELAXED);
167
+ __atomic_add_fetch(&p.parent.type[idx].nodes_running, val, __ATOMIC_RELAXED);
168
break;
169
170
case PULSE_HOST_STATUS_SND_OFFLINE:
@@ -179,6 +196,10 @@ static void pulse_host_add_sub_status(PULSE_HOST_STATUS status, ssize_t val, STR
196
case PULSE_HOST_STATUS_SND_NO_DST:
197
__atomic_add_fetch(&p.sender.nodes_no_dst, val, __ATOMIC_RELAXED);
198
break;
199
+
200
+ case PULSE_HOST_STATUS_SND_NO_DST_FAILED:
201
+ __atomic_add_fetch(&p.sender.nodes_no_dst_failed, val, __ATOMIC_RELAXED);
202
+ break;
203
}
204
205
if(do_parent_reason && val > 0)
@@ -195,30 +216,47 @@ void pulse_host_status(RRDHOST *host, PULSE_HOST_STATUS status, STREAM_HANDSHAKE
216
if(!status)
217
status = pulse_host_detect_receiver_status(host);
218
198
- PULSE_HOST_STATUS basic = PULSE_HOST_STATUS_LOCAL|PULSE_HOST_STATUS_VIRTUAL| PULSE_HOST_STATUS_LOADING |PULSE_HOST_STATUS_ARCHIVED|PULSE_HOST_STATUS_DELETED;
199
- PULSE_HOST_STATUS rcv = PULSE_HOST_STATUS_RCV_OFFLINE|PULSE_HOST_STATUS_RCV_WAITING|PULSE_HOST_STATUS_RCV_REPLICATING|PULSE_HOST_STATUS_RCV_REPLICATION_WAIT|PULSE_HOST_STATUS_RCV_RUNNING;
200
- PULSE_HOST_STATUS snd = PULSE_HOST_STATUS_SND_OFFLINE|PULSE_HOST_STATUS_SND_PENDING|PULSE_HOST_STATUS_SND_CONNECTING|PULSE_HOST_STATUS_SND_WAITING|PULSE_HOST_STATUS_SND_REPLICATING|PULSE_HOST_STATUS_SND_RUNNING|PULSE_HOST_STATUS_SND_NO_DST;
219
+ static const PULSE_HOST_STATUS ephemerality =
220
+ PULSE_HOST_STATUS_EPHEMERAL | PULSE_HOST_STATUS_PERMANENT;
221
+
222
+ static const PULSE_HOST_STATUS basic =
223
+ PULSE_HOST_STATUS_LOCAL | PULSE_HOST_STATUS_VIRTUAL | PULSE_HOST_STATUS_LOADING |
224
+ PULSE_HOST_STATUS_ARCHIVED | PULSE_HOST_STATUS_DELETED;
225
+
226
+ static const PULSE_HOST_STATUS receiver =
227
+ PULSE_HOST_STATUS_RCV_OFFLINE | PULSE_HOST_STATUS_RCV_WAITING | PULSE_HOST_STATUS_RCV_REPLICATING |
228
+ PULSE_HOST_STATUS_RCV_REPLICATION_WAIT | PULSE_HOST_STATUS_RCV_RUNNING;
229
+
230
+ static const PULSE_HOST_STATUS sender =
231
+ PULSE_HOST_STATUS_SND_OFFLINE | PULSE_HOST_STATUS_SND_PENDING | PULSE_HOST_STATUS_SND_CONNECTING |
232
+ PULSE_HOST_STATUS_SND_WAITING | PULSE_HOST_STATUS_SND_REPLICATING | PULSE_HOST_STATUS_SND_RUNNING |
233
+ PULSE_HOST_STATUS_SND_NO_DST | PULSE_HOST_STATUS_SND_NO_DST_FAILED;
234
+
235
+ if((status & (basic | receiver)) && !(status & ephemerality))
236
+ status |= rrdhost_option_check(host, RRDHOST_OPTION_EPHEMERAL_HOST) ?
237
+ PULSE_HOST_STATUS_EPHEMERAL : PULSE_HOST_STATUS_PERMANENT;
238
239
if(status & basic)
203
- remove = ~0;
204
- else if(status & rcv)
205
- remove = basic | rcv;
206
- else if(status & snd)
207
- remove = snd;
240
+ remove = basic | receiver | ephemerality | sender;
241
+ else if(status & receiver)
242
+ remove = basic | receiver | ephemerality;
243
+ else if(status & sender)
244
+ remove = sender;
245
246
spinlock_lock(&p.spinlock);
247
PULSE_HOST_STATUS old = PHOST_GET(&p.index, (uintptr_t)host);
211
- if(status == PULSE_HOST_STATUS_DELETED)
248
+ if(status == PULSE_HOST_STATUS_DELETED) {
249
PHOST_DEL(&p.index, (uintptr_t)host);
250
+ status = 0; // do not add anything, just remove the old flags
251
+ }
252
else
253
PHOST_SET(&p.index, (uintptr_t)host, (old & ~remove) | status);
254
spinlock_unlock(&p.spinlock);
255
256
remove &= old;
218
- pulse_host_add_sub_status(remove, -1, 0);
257
220
- if(status != PULSE_HOST_STATUS_DELETED)
221
- pulse_host_add_sub_status(status, 1, reason);
258
+ pulse_host_add_sub_status(remove, -1, 0);
259
+ pulse_host_add_sub_status(status, 1, reason);
260
}
261
262
void pulse_parent_stream_info_received_request(void) {
@@ -250,7 +288,7 @@ void pulse_sender_connection_failed(const char *destination __maybe_unused, STRE
288
289
// --------------------------------------------------------------------------------------------------------------------
290
253
-static void chart_by_reason(struct by_reason *b, const char *id, const char *context, const char *title, int priority) {
291
+static void chart_by_reason(struct by_reason *b, const char *id, const char *context, const char *title, const char *label, int priority) {
292
if(!b->st) {
293
b->st = rrdset_create_localhost(
294
"netdata"
@@ -278,6 +316,9 @@ static void chart_by_reason(struct by_reason *b, const char *id, const char *con
316
317
b->rd[i] = rrddim_add(b->st, buf, NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
318
}
319
+
320
+ rrdlabels_add(b->st->rrdlabels, "type", label, RRDLABEL_SRC_AUTO);
321
+
322
b->rd[STREAM_HANDSHAKE_STREAM_INFO] = rrddim_add(b->st, "info", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
323
b->rd[STREAM_HANDSHAKE_CONNECT] = rrddim_add(b->st, "connect", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
324
b->rd[STREAM_HANDSHAKE_OTHER] = rrddim_add(b->st, "other", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
@@ -291,22 +332,22 @@ static void chart_by_reason(struct by_reason *b, const char *id, const char *con
332
333
void pulse_parents_do(bool extended) {
334
if(netdata_conf_is_parent()) {
294
- {
295
- static RRDSET *st_nodes = NULL;
296
- static RRDDIM *rd_loading = NULL;
297
- static RRDDIM *rd_local = NULL;
298
- static RRDDIM *rd_virtual = NULL;
299
- static RRDDIM *rd_archived = NULL;
300
- static RRDDIM *rd_offline = NULL;
301
- static RRDDIM *rd_waiting = NULL;
302
- static RRDDIM *rd_replication_waiting = NULL;
303
- static RRDDIM *rd_replicating = NULL;
304
- static RRDDIM *rd_running = NULL;
305
-
306
- if (unlikely(!st_nodes)) {
307
- st_nodes = rrdset_create_localhost(
335
+ for(size_t idx = 0; idx < _countof(p.parent.type) ; idx++) {
336
+ if (unlikely(!p.parent.type[idx].st_nodes)) {
337
+ const char *type;
338
+ const char *id;
339
+ if(idx == 0) {
340
+ type = "permanent";
341
+ id = "netdata.streaming_inbound_permanent";
342
+ }
343
+ else {
344
+ type = "ephemeral";
345
+ id = "netdata.streaming_inbound_ephemeral";
346
+ }
347
+
348
+ p.parent.type[idx].st_nodes = rrdset_create_localhost(
349
"netdata"
309
- , "streaming_inbound"
350
+ , id
351
, NULL
352
, "Streaming"
353
, "netdata.streaming_inbound"
@@ -319,28 +360,30 @@ void pulse_parents_do(bool extended) {
360
, RRDSET_TYPE_LINE
361
);
362
322
- rd_local = rrddim_add(st_nodes, "local", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
323
- rd_virtual = rrddim_add(st_nodes, "virtual", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
324
- rd_loading = rrddim_add(st_nodes, "loading", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
325
- rd_archived = rrddim_add(st_nodes, "stale archived", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
326
- rd_offline = rrddim_add(st_nodes, "stale disconnected", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
327
- rd_waiting = rrddim_add(st_nodes, "waiting", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
328
- rd_replication_waiting = rrddim_add(st_nodes, "waiting replication", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
329
- rd_replicating = rrddim_add(st_nodes, "replicating", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
330
- rd_running = rrddim_add(st_nodes, "running", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
363
+ rrdlabels_add(p.parent.type[idx].st_nodes->rrdlabels, "type", type, RRDLABEL_SRC_AUTO);
364
+
365
+ p.parent.type[idx].rd_local = rrddim_add(p.parent.type[idx].st_nodes, "local", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
366
+ p.parent.type[idx].rd_virtual = rrddim_add(p.parent.type[idx].st_nodes, "virtual", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
367
+ p.parent.type[idx].rd_loading = rrddim_add(p.parent.type[idx].st_nodes, "loading", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
368
+ p.parent.type[idx].rd_archived = rrddim_add(p.parent.type[idx].st_nodes, "stale archived", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
369
+ p.parent.type[idx].rd_offline = rrddim_add(p.parent.type[idx].st_nodes, "stale disconnected", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
370
+ p.parent.type[idx].rd_waiting = rrddim_add(p.parent.type[idx].st_nodes, "waiting", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
371
+ p.parent.type[idx].rd_replication_waiting = rrddim_add(p.parent.type[idx].st_nodes, "waiting replication", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
372
+ p.parent.type[idx].rd_replicating = rrddim_add(p.parent.type[idx].st_nodes, "replicating", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
373
+ p.parent.type[idx].rd_running = rrddim_add(p.parent.type[idx].st_nodes, "running", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
374
}
375
333
- rrddim_set_by_pointer(st_nodes, rd_local, (collected_number)__atomic_load_n(&p.parent.nodes_local, __ATOMIC_RELAXED));
334
- rrddim_set_by_pointer(st_nodes, rd_virtual, (collected_number)__atomic_load_n(&p.parent.nodes_virtual, __ATOMIC_RELAXED));
335
- rrddim_set_by_pointer(st_nodes,rd_loading, (collected_number)__atomic_load_n(&p.parent.nodes_loading, __ATOMIC_RELAXED));
336
- rrddim_set_by_pointer(st_nodes, rd_archived, (collected_number)__atomic_load_n(&p.parent.nodes_archived, __ATOMIC_RELAXED));
337
- rrddim_set_by_pointer(st_nodes, rd_offline, (collected_number)__atomic_load_n(&p.parent.nodes_offline, __ATOMIC_RELAXED));
338
- rrddim_set_by_pointer(st_nodes, rd_waiting, (collected_number)__atomic_load_n(&p.parent.nodes_waiting, __ATOMIC_RELAXED));
339
- rrddim_set_by_pointer(st_nodes, rd_replication_waiting, (collected_number)__atomic_load_n(&p.parent.nodes_replication_waiting, __ATOMIC_RELAXED));
340
- rrddim_set_by_pointer(st_nodes, rd_replicating, (collected_number)__atomic_load_n(&p.parent.nodes_replicating, __ATOMIC_RELAXED));
341
- rrddim_set_by_pointer(st_nodes, rd_running, (collected_number)__atomic_load_n(&p.parent.nodes_running, __ATOMIC_RELAXED));
342
-
343
- rrdset_done(st_nodes);
376
+ rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_local, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_local, __ATOMIC_RELAXED));
377
+ rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_virtual, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_virtual, __ATOMIC_RELAXED));
378
+ rrddim_set_by_pointer(p.parent.type[idx].st_nodes,p.parent.type[idx].rd_loading, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_loading, __ATOMIC_RELAXED));
379
+ rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_archived, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_archived, __ATOMIC_RELAXED));
380
+ rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_offline, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_offline, __ATOMIC_RELAXED));
381
+ rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_waiting, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_waiting, __ATOMIC_RELAXED));
382
+ rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_replication_waiting, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_replication_waiting, __ATOMIC_RELAXED));
383
+ rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_replicating, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_replicating, __ATOMIC_RELAXED));
384
+ rrddim_set_by_pointer(p.parent.type[idx].st_nodes, p.parent.type[idx].rd_running, (collected_number)__atomic_load_n(&p.parent.type[idx].nodes_running, __ATOMIC_RELAXED));
385
+
386
+ rrdset_done(p.parent.type[idx].st_nodes);
387
}
388
389
if(extended) {
@@ -349,12 +392,14 @@ void pulse_parents_do(bool extended) {
392
"streaming_rejections_inbound",
393
"netdata.streaming_events_inbound",
394
"Inbound Streaming Events",
395
+ "rejections",
396
130151);
397
chart_by_reason(
398
&p.parent.disconnects_by_reason,
399
"streaming_disconnects_inbound",
400
"netdata.streaming_events_inbound",
401
"Inbound Streaming Events",
402
+ "disconnects",
403
130151);
404
}
405
}
@@ -369,6 +414,7 @@ void pulse_parents_do(bool extended) {
414
static RRDDIM *rd_replicating = NULL;
415
static RRDDIM *rd_running = NULL;
416
static RRDDIM *rd_no_dst = NULL;
417
+ static RRDDIM *rd_no_dst_failed = NULL;
418
419
if (unlikely(!st_nodes)) {
420
st_nodes = rrdset_create_localhost(
@@ -381,7 +427,7 @@ void pulse_parents_do(bool extended) {
427
, "nodes"
428
, "netdata"
429
, "pulse"
384
- , 130151
430
+ , 130153
431
, localhost->rrd_update_every
432
, RRDSET_TYPE_LINE
433
);
@@ -393,6 +439,7 @@ void pulse_parents_do(bool extended) {
439
rd_replicating = rrddim_add(st_nodes, "replicating", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
440
rd_running = rrddim_add(st_nodes, "running", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
441
rd_no_dst = rrddim_add(st_nodes, "no dst", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
442
+ rd_no_dst_failed = rrddim_add(st_nodes, "failed", NULL, 1, 1, RRD_ALGORITHM_ABSOLUTE);
443
}
444
445
rrddim_set_by_pointer(st_nodes, rd_connecting, (collected_number)__atomic_load_n(&p.sender.nodes_connecting, __ATOMIC_RELAXED));
@@ -402,6 +449,7 @@ void pulse_parents_do(bool extended) {
449
rrddim_set_by_pointer(st_nodes, rd_replicating, (collected_number)__atomic_load_n(&p.sender.nodes_replicating, __ATOMIC_RELAXED));
450
rrddim_set_by_pointer(st_nodes, rd_running, (collected_number)__atomic_load_n(&p.sender.nodes_running, __ATOMIC_RELAXED));
451
rrddim_set_by_pointer(st_nodes, rd_no_dst, (collected_number)__atomic_load_n(&p.sender.nodes_no_dst, __ATOMIC_RELAXED));
452
+ rrddim_set_by_pointer(st_nodes, rd_no_dst_failed, (collected_number)__atomic_load_n(&p.sender.nodes_no_dst_failed, __ATOMIC_RELAXED));
453
454
rrdset_done(st_nodes);
455
}
@@ -412,19 +460,22 @@ void pulse_parents_do(bool extended) {
460
"streaming_info_failed_outbound",
461
"netdata.streaming_events_outbound",
462
"Outbound Streaming Events",
415
- 130152);
463
+ "stream-info",
464
+ 130154);
465
chart_by_reason(
466
&p.sender.events_by_reason,
467
"streaming_rejections_outbound",
468
"netdata.streaming_events_outbound",
469
"Outbound Streaming Events",
421
- 130152);
470
+ "rejections",
471
+ 130154);
472
chart_by_reason(
473
&p.sender.disconnects_by_reason,
474
"streaming_disconnects_outbound",
475
"netdata.streaming_events_outbound",
476
"Outbound Streaming Events",
427
- 130152);
477
+ "disconnects",
478
+ 130154);
479
}
480
}
481
}
src/daemon/pulse/pulse-parents.h
+7
-4
@@ -21,10 +21,13 @@ typedef enum {
21
PULSE_HOST_STATUS_SND_PENDING = (1 << 10),
22
PULSE_HOST_STATUS_SND_CONNECTING = (1 << 11),
23
PULSE_HOST_STATUS_SND_NO_DST = (1 << 12),
24
- PULSE_HOST_STATUS_SND_WAITING = (1 << 13),
25
- PULSE_HOST_STATUS_SND_REPLICATING = (1 << 14),
26
- PULSE_HOST_STATUS_SND_RUNNING = (1 << 15),
27
- PULSE_HOST_STATUS_DELETED = (1 << 16),
24
+ PULSE_HOST_STATUS_SND_NO_DST_FAILED = (1 << 13),
25
+ PULSE_HOST_STATUS_SND_WAITING = (1 << 14),
26
+ PULSE_HOST_STATUS_SND_REPLICATING = (1 << 15),
27
+ PULSE_HOST_STATUS_SND_RUNNING = (1 << 16),
28
+ PULSE_HOST_STATUS_DELETED = (1 << 17),
29
+ PULSE_HOST_STATUS_EPHEMERAL = (1 << 18),
30
+ PULSE_HOST_STATUS_PERMANENT = (1 << 19),
31
} PULSE_HOST_STATUS;
32
33
void pulse_host_status(RRDHOST *host, PULSE_HOST_STATUS status, STREAM_HANDSHAKE reason);
src/daemon/service.c
+26
-22
@@ -223,41 +223,45 @@ static void svc_rrd_cleanup_obsolete_charts_from_all_hosts() {
223
224
static void svc_rrdhost_cleanup_orphan_hosts(RRDHOST *protected_host) {
225
worker_is_busy(WORKER_JOB_CLEANUP_ORPHAN_HOSTS);
226
- rrd_wrlock();
226
227
time_t now = now_realtime_sec();
228
230
- RRDHOST *host;
231
-
232
-restart_after_removal:
233
- rrdhost_foreach_write(host) {
234
- if(!rrdhost_should_be_removed(host, protected_host, now))
235
- continue;
236
-
237
- bool force = false;
238
- if (rrdhost_option_check(host, RRDHOST_OPTION_EPHEMERAL_HOST) &&
239
- now - host->stream.snd.status.last_connected > rrdhost_free_ephemeral_time_s)
240
- force = true;
229
+ rrd_wrlock();
230
+ RRDHOST *host, *next = localhost;
231
+ while((host = next) != NULL) {
232
+ next = host->next;
233
242
- bool is_archived = rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED);
243
- if (!force && is_archived)
234
+ if(!rrdhost_should_be_cleaned_up(host, protected_host, now))
235
continue;
236
246
- if (force) {
247
- netdata_log_info("Host '%s' with machine guid '%s' is archived, ephemeral clean up.", rrdhost_hostname(host), host->machine_guid);
237
+ bool delete = rrdhost_free_ephemeral_time_s &&
238
+ now - host->stream.rcv.status.last_disconnected > rrdhost_free_ephemeral_time_s &&
239
+ rrdhost_option_check(host, RRDHOST_OPTION_EPHEMERAL_HOST);
240
+
241
+ if (!delete && rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED)) {
242
+ // the node is archived, so the cleanup has already run
243
+ // however, the node may not have any retention now
244
+ // so it may still need to be needed
245
+ time_t from_s = 0, to_s = 0;
246
+ rrdhost_retention(host, now, rrdhost_is_online(host), &from_s, &to_s);
247
+ if(!from_s && !to_s)
248
+ delete = true;
249
+ else
250
+ continue;
251
}
252
253
worker_is_busy(WORKER_JOB_FREE_HOST);
251
- // in case we have cloud connection we inform cloud
252
- // a child disconnected
253
- if (force) {
254
+
255
+ if (delete) {
256
+ netdata_log_info("Host '%s' with machine guid '%s' is archived, ephemeral clean up.", rrdhost_hostname(host), host->machine_guid);
257
+ // we inform cloud a child has been removed
258
aclk_host_state_update(host, 0, 0);
259
unregister_node(host->machine_guid);
260
+ rrdhost_free___while_having_rrd_wrlock(host);
261
}
257
- rrdhost_free___while_having_rrd_wrlock(host, force);
258
- goto restart_after_removal;
262
+ else
263
+ rrdhost_cleanup_data_collection_and_health(host);
264
}
260
-
265
rrd_wrunlock();
266
}
267
src/database/contexts/rrdcontext.c
+1
@@ -312,6 +312,7 @@ void rrdcontext_hub_stop_streaming_command(void *ptr) {
312
rrdhost_flag_clear(host, RRDHOST_FLAG_ACLK_STREAM_CONTEXTS);
313
}
314
315
+ALWAYS_INLINE
316
bool rrdcontext_retention_match(RRDCONTEXT_ACQUIRED *rca, time_t after, time_t before) {
317
if(unlikely(!rca)) return false;
318
src/database/engine/dbengine-stresstest.c
+1
-1
@@ -185,7 +185,7 @@ void generate_dbengine_dataset(unsigned history_seconds)
185
}
186
freez(thread_info);
187
rrd_wrlock();
188
- rrdhost_free___while_having_rrd_wrlock(localhost, true);
188
+ rrdhost_free___while_having_rrd_wrlock(localhost);
189
rrd_wrunlock();
190
}
191
src/database/rrdfunctions.c
+1
@@ -178,6 +178,7 @@ void rrd_functions_host_init(RRDHOST *host) {
178
179
void rrd_functions_host_destroy(RRDHOST *host) {
180
dictionary_destroy(host->functions);
181
+ host->functions = NULL;
182
}
183
184
// ----------------------------------------------------------------------------
src/database/rrdhost-labels.c
+1
-1
@@ -71,7 +71,7 @@ static void rrdhost_load_auto_labels(void) {
71
// Their export seems to break exporting to Graphite, see https://github.com/netdata/netdata/issues/14084.
72
73
int is_ephemeral = inicfg_get_boolean(&netdata_config, CONFIG_SECTION_GLOBAL, "is ephemeral node", CONFIG_BOOLEAN_NO);
74
- rrdlabels_add(labels, "_is_ephemeral", is_ephemeral ? "true" : "false", RRDLABEL_SRC_AUTO);
74
+ rrdlabels_add(labels, HOST_LABEL_IS_EPHEMERAL, is_ephemeral ? "true" : "false", RRDLABEL_SRC_CONFIG);
75
76
int has_unstable_connection = inicfg_get_boolean(&netdata_config, CONFIG_SECTION_GLOBAL, "has unstable connection", CONFIG_BOOLEAN_NO);
77
rrdlabels_add(labels, "_has_unstable_connection", has_unstable_connection ? "true" : "false", RRDLABEL_SRC_AUTO);
src/database/rrdhost-status.c
+3
-4
@@ -150,7 +150,7 @@ static inline RRDHOST_INGEST_STATUS rrdhost_status_ingest(RRDHOST *host, RRDHOST
150
uint32_t replicating_instances = UINT32_MAX;
151
152
time_t since = MAX(host->stream.rcv.status.last_connected, host->stream.rcv.status.last_disconnected);
153
- STREAM_HANDSHAKE reason = (online) ? STREAM_HANDSHAKE_NEVER : host->stream.rcv.status.exit_reason;
153
+ STREAM_HANDSHAKE reason = host->stream.rcv.status.reason;
154
155
if (online) {
156
if (db_status == RRDHOST_DB_STATUS_INITIALIZING)
@@ -169,7 +169,7 @@ static inline RRDHOST_INGEST_STATUS rrdhost_status_ingest(RRDHOST *host, RRDHOST
169
status = RRDHOST_INGEST_STATUS_ONLINE;
170
}
171
else {
172
- if(!since)
172
+ if(!host->stream.rcv.status.connections)
173
status = RRDHOST_INGEST_STATUS_ARCHIVED;
174
else
175
status = RRDHOST_INGEST_STATUS_OFFLINE;
@@ -247,7 +247,6 @@ static void rrdhost_status_stream_internal(RRDHOST_STATUS *s) {
247
248
if (rrdhost_flag_check(host, RRDHOST_FLAG_STREAM_SENDER_CONNECTED)) {
249
s->stream.hops = host->sender->hops;
250
- s->stream.reason = STREAM_HANDSHAKE_NEVER;
250
s->stream.capabilities = host->sender->capabilities;
251
252
s->stream.replication.completion = rrdhost_sender_replication_completion_unsafe(host, now, &s->stream.replication.instances);
@@ -263,8 +262,8 @@ static void rrdhost_status_stream_internal(RRDHOST_STATUS *s) {
262
else {
263
s->stream.status = RRDHOST_STREAM_STATUS_OFFLINE;
264
s->stream.hops = (int16_t)(s->ingest.hops + 1);
266
- s->stream.reason = host->sender->exit.reason;
265
}
266
+ s->stream.reason = host->stream.snd.status.reason;
267
268
stream_sender_unlock(host->sender);
269
}
src/database/rrdhost.c
+56
-66
@@ -267,7 +267,7 @@ static RRDHOST *prepare_host_for_unittest(RRDHOST *host)
267
host->cache_dir);
268
269
rrd_wrlock();
270
- rrdhost_free___while_having_rrd_wrlock(host, true);
270
+ rrdhost_free___while_having_rrd_wrlock(host);
271
rrd_wrunlock();
272
return NULL;
273
}
@@ -445,7 +445,7 @@ RRDHOST *rrdhost_create(
445
rrdhost_hostname(host), host->machine_guid, rrdhost_hostname(t), t->machine_guid);
446
447
if (!is_localhost)
448
- rrdhost_free___while_having_rrd_wrlock(host, true);
448
+ rrdhost_free___while_having_rrd_wrlock(host);
449
450
rrd_wrunlock();
451
return NULL;
@@ -690,7 +690,7 @@ RRDHOST *rrdhost_find_or_create(
690
rrd_memory_mode_name(mode));
691
692
rrd_wrlock();
693
- rrdhost_free___while_having_rrd_wrlock(host, true);
693
+ rrdhost_free___while_having_rrd_wrlock(host);
694
host = NULL;
695
rrd_wrunlock();
696
}
@@ -755,7 +755,7 @@ RRDHOST *rrdhost_find_or_create(
755
return host;
756
}
757
758
-bool rrdhost_should_be_removed(RRDHOST *host, RRDHOST *protected_host, time_t now_s) {
758
+bool rrdhost_should_be_cleaned_up(RRDHOST *host, RRDHOST *protected_host, time_t now_s) {
759
if(host != protected_host
760
&& host != localhost
761
&& rrdhost_receiver_replicating_charts(host) == 0
@@ -764,7 +764,7 @@ bool rrdhost_should_be_removed(RRDHOST *host, RRDHOST *protected_host, time_t no
764
&& !rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD | RRDHOST_FLAG_COLLECTOR_ONLINE)
765
&& health_evloop_current_iteration() - rrdhost_health_evloop_last_iteration(host) > 10
766
&& host->stream.rcv.status.last_disconnected
767
- && host->stream.rcv.status.last_disconnected + rrdhost_free_orphan_time_s < now_s)
767
+ && host->stream.rcv.status.last_disconnected + rrdhost_cleanup_orphan_to_archive_time_s < now_s)
768
return true;
769
770
return false;
@@ -781,10 +781,40 @@ bool rrdhost_should_run_health(RRDHOST *host) {
781
// ----------------------------------------------------------------------------
782
// RRDHOST - free
783
784
-void rrdhost_free___while_having_rrd_wrlock(RRDHOST *host, bool force) {
784
+void rrdhost_cleanup_data_collection_and_health(RRDHOST *host) {
785
+ stream_receiver_signal_to_stop_and_wait(host, STREAM_HANDSHAKE_SND_DISCONNECT_HOST_CLEANUP);
786
+
787
+ rrdhost_pluginsd_send_chart_slots_free(host);
788
+ rrdhost_pluginsd_receive_chart_slots_free(host);
789
+
790
+ rrdcalc_delete_all(host);
791
+ rrdset_index_destroy(host);
792
+ rrdcalc_rrdhost_index_destroy(host);
793
+ health_alarm_log_free(host);
794
+
795
+ ml_host_delete(host);
796
+
797
+ freez(host->exporting_flags);
798
+ host->exporting_flags = NULL;
799
+
800
+ rrd_functions_host_destroy(host);
801
+ rrdvariables_destroy(host->rrdvars);
802
+ host->rrdvars = NULL;
803
+
804
+ rrdhost_stream_path_clear(host, true);
805
+ stream_sender_structures_free(host);
806
+
807
+ rrdhost_flag_set(host, RRDHOST_FLAG_ARCHIVED | RRDHOST_FLAG_ORPHAN);
808
+
809
+ nd_log(NDLS_DAEMON, NDLP_DEBUG,
810
+ "RRD: 'host:%s' is now in archive mode...",
811
+ rrdhost_hostname(host));
812
+}
813
+
814
+void rrdhost_free___while_having_rrd_wrlock(RRDHOST *host) {
815
if(!host) return;
816
787
- if (netdata_exit || force) {
817
+ if (netdata_exit) {
818
nd_log(NDLS_DAEMON, NDLP_DEBUG,
819
"RRD: 'host:%s' freeing memory...",
820
rrdhost_hostname(host));
@@ -801,78 +831,38 @@ void rrdhost_free___while_having_rrd_wrlock(RRDHOST *host, bool force) {
831
832
// ------------------------------------------------------------------------
833
804
- rrdhost_stream_path_clear(host, true);
805
-
806
- // ------------------------------------------------------------------------
807
- // clean up streaming chart slots
808
-
809
- rrdhost_pluginsd_send_chart_slots_free(host);
810
- rrdhost_pluginsd_receive_chart_slots_free(host);
811
-
812
- // ------------------------------------------------------------------------
813
- // clean up streaming
814
-
815
- stream_sender_structures_free(host);
816
-
817
- if (netdata_exit || force)
818
- stream_receiver_signal_to_stop_and_wait(host, STREAM_HANDSHAKE_SND_DISCONNECT_HOST_CLEANUP);
819
-
834
+ rrdhost_cleanup_data_collection_and_health(host);
835
+ if (!netdata_exit)
836
+ return;
837
838
// ------------------------------------------------------------------------
822
- // clean up alarms
823
-
824
- rrdcalc_delete_all(host);
825
-
826
- // delete all the RRDSETs of the host
827
- rrdset_index_destroy(host);
828
- rrdcalc_rrdhost_index_destroy(host);
829
-
830
- // cleanup ML resources
831
- ml_host_delete(host);
832
-
833
- freez(host->exporting_flags);
834
-
835
- health_alarm_log_free(host);
839
+ // free it
840
837
- if (!netdata_exit && !force) {
838
- nd_log(NDLS_DAEMON, NDLP_DEBUG,
839
- "RRD: 'host:%s' is now in archive mode...",
840
- rrdhost_hostname(host));
841
+ pulse_host_status(host, PULSE_HOST_STATUS_DELETED, 0);
842
+ __atomic_sub_fetch(&netdata_buffers_statistics.rrdhost_allocations_size, sizeof(RRDHOST), __ATOMIC_RELAXED);
843
842
- rrdhost_flag_set(host, RRDHOST_FLAG_ARCHIVED | RRDHOST_FLAG_ORPHAN);
843
- return;
844
- }
844
+ if (host == localhost)
845
+ health_plugin_destroy();
846
846
- // ------------------------------------------------------------------------
847
- // free it
847
+ freez(host->cache_dir);
848
+ rrdhost_stream_parents_free(host, false);
849
+ simple_pattern_free(host->stream.snd.charts_matching);
850
+ rrdhost_system_info_free(host->system_info);
851
852
+ rrdhost_destroy_rrdcontexts(host);
853
rrdlabels_destroy(host->rrdlabels);
854
+
855
+ string_freez(host->hostname);
856
string_freez(host->os);
857
string_freez(host->timezone);
858
string_freez(host->abbrev_timezone);
859
string_freez(host->program_name);
860
string_freez(host->program_version);
855
- rrdhost_system_info_free(host->system_info);
856
- freez(host->cache_dir);
857
- string_freez(host->stream.snd.api_key);
858
- string_freez(host->stream.snd.destination);
859
- rrdhost_stream_parents_free(host, false);
861
string_freez(host->health.default_exec);
862
string_freez(host->health.default_recipient);
863
string_freez(host->registry_hostname);
863
- simple_pattern_free(host->stream.snd.charts_matching);
864
-
865
- rrd_functions_host_destroy(host);
866
- rrdvariables_destroy(host->rrdvars);
867
- if (host == localhost)
868
- health_plugin_destroy();
869
-
870
- rrdhost_destroy_rrdcontexts(host);
871
-
872
- string_freez(host->hostname);
873
- __atomic_sub_fetch(&netdata_buffers_statistics.rrdhost_allocations_size, sizeof(RRDHOST), __ATOMIC_RELAXED);
874
-
875
- pulse_host_status(host, PULSE_HOST_STATUS_DELETED, 0);
864
+ string_freez(host->stream.snd.api_key);
865
+ string_freez(host->stream.snd.destination);
866
freez(host);
867
}
868
@@ -881,10 +871,10 @@ void rrdhost_free_all(void) {
871
872
/* Make sure child-hosts are released before the localhost. */
873
while(localhost && localhost->next)
884
- rrdhost_free___while_having_rrd_wrlock(localhost->next, true);
874
+ rrdhost_free___while_having_rrd_wrlock(localhost->next);
875
876
if(localhost)
887
- rrdhost_free___while_having_rrd_wrlock(localhost, true);
877
+ rrdhost_free___while_having_rrd_wrlock(localhost);
878
879
rrd_wrunlock();
880
}
src/database/rrdhost.h
+9
-5
@@ -5,6 +5,8 @@
5
6
#include "libnetdata/libnetdata.h"
7
8
+#define HOST_LABEL_IS_EPHEMERAL "_is_ephemeral"
9
+
10
struct stream_thread;
11
struct rrdset;
12
@@ -186,6 +188,7 @@ struct rrdhost {
188
189
time_t last_connected; // last time child connected (stored in db)
190
uint32_t connections; // the number of times this sender has connected
191
+ STREAM_HANDSHAKE reason; // the last receiver exit reason
192
193
struct {
194
uint32_t counter_in; // counts the number of replication statements we have received
@@ -219,10 +222,10 @@ struct rrdhost {
222
time_t last_disconnected; // the time the last sender was disconnected
223
time_t last_chart; // the time of the last CHART streaming command
224
bool check_obsolete; // set when child connects, will instruct parent to
222
- // trigger a check for obsoleted charts since previous connect
225
+ // trigger a check for obsoleted charts since previous connect
226
227
uint32_t connections; // the number of times this receiver has connected
225
- STREAM_HANDSHAKE exit_reason; // the last receiver exit reason
228
+ STREAM_HANDSHAKE reason; // the last receiver exit reason
229
230
struct {
231
uint32_t counter_in; // counts the number of replication statements we have received
@@ -442,9 +445,10 @@ RRDHOST *rrdhost_find_or_create(
445
446
void rrdhost_free_all(void);
447
445
-void rrdhost_free___while_having_rrd_wrlock(RRDHOST *host, bool force);
448
+void rrdhost_free___while_having_rrd_wrlock(RRDHOST *host);
449
+void rrdhost_cleanup_data_collection_and_health(RRDHOST *host);
450
447
-bool rrdhost_should_be_removed(RRDHOST *host, RRDHOST *protected_host, time_t now_s);
451
+bool rrdhost_should_be_cleaned_up(RRDHOST *host, RRDHOST *protected_host, time_t now_s);
452
bool rrdhost_should_run_health(RRDHOST *host);
453
454
void set_host_properties(
@@ -467,7 +471,7 @@ static inline void rrdhost_retention(RRDHOST *host, time_t now, bool online, tim
471
*to = online ? now : last_time_s;
472
}
473
470
-extern time_t rrdhost_free_orphan_time_s;
474
+extern time_t rrdhost_cleanup_orphan_to_archive_time_s;
475
extern time_t rrdhost_free_ephemeral_time_s;
476
477
#include "rrdhost-collection.h"
src/database/rrdlabels.c
+9
-2
@@ -395,7 +395,14 @@ void rrdlabels_key_to_buffer_array_item(RRDLABELS *labels, BUFFER *wb)
395
// ----------------------------------------------------------------------------
396
397
void rrdlabels_get_value_strcpyz(RRDLABELS *labels, char *dst, size_t dst_len, const char *key) {
398
- if(!labels) return;
398
+ if(!dst || !dst_len)
399
+ return;
400
+
401
+ // make sure the output is empty if we can't find it
402
+ *dst = '\0';
403
+
404
+ if(!labels)
405
+ return;
406
407
STRING *this_key = string_strdupz(key);
408
@@ -406,7 +413,7 @@ void rrdlabels_get_value_strcpyz(RRDLABELS *labels, char *dst, size_t dst_len, c
413
{
414
if (lb->index.key == this_key) {
415
if (lb->index.value)
409
- strncpyz(dst, string2str(lb->index.value), dst_len);
416
+ strncpyz(dst, string2str(lb->index.value), dst_len - 1);
417
else
418
dst[0] = '\0';
419
break;
src/database/sqlite/sqlite_aclk.c
+2
-1
@@ -131,7 +131,7 @@ static int create_host_callback(void *data, int argc, char **argv, char **column
131
char guid[UUID_STR_LEN];
132
uuid_unparse_lower(*(nd_uuid_t *)argv[IDX_HOST_ID], guid);
133
134
- if (is_ephemeral && age > rrdhost_free_ephemeral_time_s) {
134
+ if (is_ephemeral && rrdhost_free_ephemeral_time_s && age > rrdhost_free_ephemeral_time_s) {
135
netdata_log_info(
136
"%s ephemeral hostname \"%s\" with GUID \"%s\", age = %ld seconds (limit %ld seconds)",
137
is_registered ? "Loading registered" : "Skipping unregistered",
@@ -139,6 +139,7 @@ static int create_host_callback(void *data, int argc, char **argv, char **column
139
guid,
140
age,
141
rrdhost_free_ephemeral_time_s);
142
+
143
if (!is_registered)
144
return 0;
145
}
src/health/health.d/streaming.conf
new
+57
@@ -0,0 +1,57 @@
1
+
2
+# ---------------------------------------------------------------------------------------------------------------------
3
+# Alert when there are permanent nodes which never connected (during this session)
4
+
5
+# If these nodes nodes should actually be ephemeral, run:
6
+#
7
+# netdatacli mark-stale-nodes-ephemeral ALL_NODES
8
+#
9
+# or if you want to delete all the archived nodes, run:
10
+#
11
+# netdatacli remove-stale-node ALL_NODES
12
+#
13
+
14
+ template: streaming_never_connected
15
+ on: netdata.streaming_inbound
16
+ class: Availability
17
+ type: Streaming
18
+ component: Streaming
19
+ chart labels: type=permanent
20
+ calc: ${stale archived}
21
+ units: nodes
22
+ every: 10s
23
+ warn: $netdata.uptime.uptime > 30 * 60 AND $this > 0
24
+ delay: up 5m down 5m multiplier 1.5 max 30m
25
+ summary: Permanent streaming nodes never connected
26
+ info: Permanent child nodes never connected to this parent. \
27
+ If these nodes should actually be ephemeral, run: \
28
+ netdatacli mark-stale-nodes-ephemeral ALL_NODES
29
+ to: sysadmin
30
+
31
+# ---------------------------------------------------------------------------------------------------------------------
32
+# Alert when there are permanent nodes which disconnected (during this session)
33
+
34
+# If nodes are expected to disconnect, mark them as ephemeral, by editing their netdata.conf
35
+# and setting:
36
+#
37
+# [global]
38
+# is ephemeral node = yes
39
+#
40
+# and then restarting netdata
41
+
42
+ template: streaming_disconnected
43
+ on: netdata.streaming_inbound
44
+ class: Availability
45
+ type: Streaming
46
+ component: Streaming
47
+ chart labels: type=permanent
48
+ calc: ${stale disconnected}
49
+ units: nodes
50
+ every: 10s
51
+ warn: $netdata.uptime.uptime > 30 * 60 AND $this > 0
52
+ delay: up 5m down 5m multiplier 1.5 max 30m
53
+ summary: Permanent streaming nodes disconnected
54
+ info: Permanent child nodes disconnected from this parent. \
55
+ If nodes are expected to disconnect, mark them as ephemeral, by editing their netdata.conf \
56
+ and setting: [global].is ephemeral node = yes
57
+ to: sysadmin
src/libnetdata/dictionary/dictionary.c
+9
-3
@@ -112,6 +112,7 @@ void dictionary_register_delete_callback(DICTIONARY *dict, dict_cb_delete_t dele
112
// ----------------------------------------------------------------------------
113
// dictionary statistics API
114
115
+ALWAYS_INLINE
116
size_t dictionary_version(DICTIONARY *dict) {
117
if(unlikely(!dict)) return 0;
118
@@ -121,7 +122,8 @@ size_t dictionary_version(DICTIONARY *dict) {
122
return __atomic_load_n(&dict->version, __ATOMIC_RELAXED);
123
}
124
124
-ALWAYS_INLINE size_t dictionary_entries(DICTIONARY *dict) {
125
+ALWAYS_INLINE
126
+size_t dictionary_entries(DICTIONARY *dict) {
127
if(unlikely(!dict)) return 0;
128
129
// this is required for views to return the right number
@@ -717,6 +719,7 @@ void *dictionary_get_advanced(DICTIONARY *dict, const char *name, ssize_t name_l
719
// ----------------------------------------------------------------------------
720
// DUP/REL an item (increase/decrease its reference counter)
721
722
+ALWAYS_INLINE
723
DICT_ITEM_CONST DICTIONARY_ITEM *dictionary_acquired_item_dup(DICTIONARY *dict, DICT_ITEM_CONST DICTIONARY_ITEM *item) {
724
// we allow the item to be NULL here
725
api_internal_check(dict, item, false, true);
@@ -729,6 +732,7 @@ DICT_ITEM_CONST DICTIONARY_ITEM *dictionary_acquired_item_dup(DICTIONARY *dict,
732
return item;
733
}
734
735
+ALWAYS_INLINE
736
void dictionary_acquired_item_release(DICTIONARY *dict, DICT_ITEM_CONST DICTIONARY_ITEM *item) {
737
// we allow the item to be NULL here
738
api_internal_check(dict, item, false, true);
@@ -744,11 +748,13 @@ void dictionary_acquired_item_release(DICTIONARY *dict, DICT_ITEM_CONST DICTIONA
748
// ----------------------------------------------------------------------------
749
// get the name/value of an item
750
747
-ALWAYS_INLINE const char *dictionary_acquired_item_name(DICT_ITEM_CONST DICTIONARY_ITEM *item) {
751
+ALWAYS_INLINE
752
+const char *dictionary_acquired_item_name(DICT_ITEM_CONST DICTIONARY_ITEM *item) {
753
return item_get_name(item);
754
}
755
751
-ALWAYS_INLINE void *dictionary_acquired_item_value(DICT_ITEM_CONST DICTIONARY_ITEM *item) {
756
+ALWAYS_INLINE
757
+void *dictionary_acquired_item_value(DICT_ITEM_CONST DICTIONARY_ITEM *item) {
758
if(likely(item))
759
return item->shared->value;
760
src/libnetdata/simple_pattern/simple_pattern.c
+12
-6
@@ -168,7 +168,8 @@ SIMPLE_PATTERN *simple_pattern_create(const char *list, const char *separators,
168
return (SIMPLE_PATTERN *)root;
169
}
170
171
-static inline char *add_wildcarded(const char *matched, size_t matched_size, char *wildcarded, size_t *wildcarded_size) {
171
+ALWAYS_INLINE
172
+static char *add_wildcarded(const char *matched, size_t matched_size, char *wildcarded, size_t *wildcarded_size) {
173
//if(matched_size) {
174
// char buf[matched_size + 1];
175
// strncpyz(buf, matched, matched_size);
@@ -189,28 +190,32 @@ static inline char *add_wildcarded(const char *matched, size_t matched_size, cha
190
return wildcarded;
191
}
192
192
-static inline int sp_strcmp(const char *s1, const char *s2, bool case_sensitive) {
193
+ALWAYS_INLINE
194
+static int sp_strcmp(const char *s1, const char *s2, bool case_sensitive) {
195
if(case_sensitive)
196
return strcmp(s1, s2);
197
198
return strcasecmp(s1, s2);
199
}
200
199
-static inline int sp_strncmp(const char *s1, const char *s2, size_t n, bool case_sensitive) {
201
+ALWAYS_INLINE
202
+static int sp_strncmp(const char *s1, const char *s2, size_t n, bool case_sensitive) {
203
if(case_sensitive)
204
return strncmp(s1, s2, n);
205
206
return strncasecmp(s1, s2, n);
207
}
208
206
-static inline char *sp_strstr(const char *haystack, const char *needle, bool case_sensitive) {
209
+ALWAYS_INLINE
210
+static char *sp_strstr(const char *haystack, const char *needle, bool case_sensitive) {
211
if(case_sensitive)
212
return strstr(haystack, needle);
213
214
return strcasestr(haystack, needle);
215
}
216
213
-static inline bool match_pattern(struct simple_pattern *m, const char *str, size_t len, char *wildcarded, size_t *wildcarded_size) {
217
+ALWAYS_INLINE
218
+static bool match_pattern(struct simple_pattern *m, const char *str, size_t len, char *wildcarded, size_t *wildcarded_size) {
219
char *s;
220
221
bool loop = true;
@@ -276,7 +281,8 @@ static inline bool match_pattern(struct simple_pattern *m, const char *str, size
281
return false;
282
}
283
279
-static inline SIMPLE_PATTERN_RESULT simple_pattern_matches_extract_with_length(SIMPLE_PATTERN *list, const char *str, size_t len, char *wildcarded, size_t wildcarded_size) {
284
+ALWAYS_INLINE
285
+static SIMPLE_PATTERN_RESULT simple_pattern_matches_extract_with_length(SIMPLE_PATTERN *list, const char *str, size_t len, char *wildcarded, size_t wildcarded_size) {
286
struct simple_pattern *m, *root = (struct simple_pattern *)list;
287
288
for(m = root; m ; m = m->next) {
src/plugins.d/pluginsd_parser.c
+18
-13
@@ -174,6 +174,22 @@ static inline PARSER_RC pluginsd_host_labels(char **words, size_t num_words, PAR
174
PLUGINSD_KEYWORD_HOST_LABEL);
175
}
176
177
+static inline void pluginsd_update_host_ephemerality(RRDHOST *host) {
178
+ char value[64];
179
+ rrdlabels_get_value_strcpyz(host->rrdlabels, value, sizeof(value), HOST_LABEL_IS_EPHEMERAL);
180
+ if(value[0] && inicfg_test_boolean_value(value)) {
181
+ rrdhost_option_set(host, RRDHOST_OPTION_EPHEMERAL_HOST);
182
+ strncpyz(value, "true", sizeof(value) - 1);
183
+ }
184
+ else {
185
+ rrdhost_option_clear(host, RRDHOST_OPTION_EPHEMERAL_HOST);
186
+ strncpyz(value, "false", sizeof(value) - 1);
187
+ }
188
+
189
+ if(!rrdlabels_exist(host->rrdlabels, HOST_LABEL_IS_EPHEMERAL))
190
+ rrdlabels_add(host->rrdlabels, HOST_LABEL_IS_EPHEMERAL, value, RRDLABEL_SRC_CONFIG);
191
+}
192
+
193
static inline PARSER_RC pluginsd_host_define_end(char **words __maybe_unused, size_t num_words __maybe_unused, PARSER *parser) {
194
if(!parser->user.host_define.parsing_host)
195
return PLUGINSD_DISABLE_PLUGIN(parser, PLUGINSD_KEYWORD_HOST_DEFINE_END, "missing initialization, send " PLUGINSD_KEYWORD_HOST_DEFINE " before this");
@@ -217,6 +233,7 @@ static inline PARSER_RC pluginsd_host_define_end(char **words __maybe_unused, si
233
parser->user.host_define.rrdlabels = NULL;
234
}
235
236
+ pluginsd_update_host_ephemerality(host);
237
pluginsd_host_define_cleanup(parser);
238
239
parser->user.host = host;
@@ -608,17 +625,6 @@ static inline PARSER_RC pluginsd_label(char **words, size_t num_words, PARSER *p
625
if(unlikely(!(parser->user.new_host_labels)))
626
parser->user.new_host_labels = rrdlabels_create();
627
611
- if (strcmp(name,HOST_LABEL_IS_EPHEMERAL) == 0) {
612
- int is_ephemeral = inicfg_test_boolean_value((char *) value);
613
- RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_LABEL);
614
- if (host) {
615
- if (is_ephemeral)
616
- rrdhost_option_set(host, RRDHOST_OPTION_EPHEMERAL_HOST);
617
- else
618
- rrdhost_option_clear(host, RRDHOST_OPTION_EPHEMERAL_HOST);
619
- }
620
- }
621
-
628
rrdlabels_add(parser->user.new_host_labels, name, store, str2l(label_source));
629
630
if (allocated_store)
@@ -637,8 +643,7 @@ static inline PARSER_RC pluginsd_overwrite(char **words __maybe_unused, size_t n
643
host->rrdlabels = rrdlabels_create();
644
645
rrdlabels_migrate_to_these(host->rrdlabels, parser->user.new_host_labels);
640
- if (rrdhost_option_check(host, RRDHOST_OPTION_EPHEMERAL_HOST))
641
- rrdlabels_add(host->rrdlabels, HOST_LABEL_IS_EPHEMERAL, "true", RRDLABEL_SRC_CONFIG);
646
+ pluginsd_update_host_ephemerality(host);
647
648
if(!rrdlabels_exist(host->rrdlabels, "_os"))
649
rrdlabels_add(host->rrdlabels, "_os", string2str(host->os), RRDLABEL_SRC_AUTO);
src/plugins.d/pluginsd_parser.h
-1
@@ -19,7 +19,6 @@
19
20
#define PLUGINSD_MIN_RRDSET_POINTERS_CACHE 1024
21
22
-#define HOST_LABEL_IS_EPHEMERAL "_is_ephemeral"
22
// PARSER return codes
23
typedef enum __attribute__ ((__packed__)) parser_rc {
24
PARSER_RC_OK, // Callback was successful, go on
src/streaming/protocol/command-nodeid.c
+56
-20
@@ -7,7 +7,13 @@
7
8
// the child disconnected from the parent, and it has to clear the parent's claim id
9
void stream_sender_clear_parent_claim_id(RRDHOST *host) {
10
- host->aclk.claim_id_of_parent = UUID_ZERO;
10
+ if (!UUIDiszero(host->aclk.claim_id_of_parent)) {
11
+ nd_log(NDLS_DAEMON, NDLP_INFO,
12
+ "Host '%s' [PCLAIMID] cleared parent's claim id",
13
+ rrdhost_hostname(host));
14
+
15
+ host->aclk.claim_id_of_parent = UUID_ZERO;
16
+ }
17
}
18
19
// the parent sends to the child its claim id, node id and cloud url
@@ -45,10 +51,21 @@ void stream_sender_get_node_and_claim_id_from_parent(struct sender_state *s, con
51
bool claimed = is_agent_claimed();
52
bool update_node_id = false;
53
54
+ // ----------------------------------------------------------------------------------------------------------------
55
+ // validate the parameters
56
+
57
ND_UUID claim_id;
58
if (uuid_parse(claim_id_str ? claim_id_str : "", claim_id.uuid) != 0) {
59
nd_log(NDLS_DAEMON, NDLP_ERR,
51
- "STREAM SND '%s' [to %s] received invalid claim id '%s'",
60
+ "STREAM SND '%s' [to %s] [PCLAIMID]: received invalid claim id '%s'",
61
+ rrdhost_hostname(s->host), s->remote_ip,
62
+ claim_id_str ? claim_id_str : "(unset)");
63
+ return;
64
+ }
65
+
66
+ if(UUIDiszero(claim_id)) {
67
+ nd_log(NDLS_DAEMON, NDLP_DEBUG,
68
+ "STREAM SND '%s' [to %s] [PCLAIMID]: received zero claim id '%s'",
69
rrdhost_hostname(s->host), s->remote_ip,
70
claim_id_str ? claim_id_str : "(unset)");
71
return;
@@ -57,45 +74,64 @@ void stream_sender_get_node_and_claim_id_from_parent(struct sender_state *s, con
74
ND_UUID node_id;
75
if(uuid_parse(node_id_str ? node_id_str : "", node_id.uuid) != 0) {
76
nd_log(NDLS_DAEMON, NDLP_ERR,
60
- "STREAM SND '%s' [to %s] received an invalid node id '%s'",
77
+ "STREAM SND '%s' [to %s] [PCLAIMID] received an invalid node id '%s'",
78
rrdhost_hostname(s->host), s->remote_ip,
79
node_id_str ? node_id_str : "(unset)");
80
return;
81
}
82
66
- if (!UUIDiszero(s->host->aclk.claim_id_of_parent) && !UUIDeq(s->host->aclk.claim_id_of_parent, claim_id))
67
- nd_log(NDLS_DAEMON, NDLP_INFO,
68
- "STREAM SND '%s' [to %s] changed parent's claim id to %s",
83
+ if(UUIDiszero(node_id)) {
84
+ nd_log(NDLS_DAEMON, NDLP_DEBUG,
85
+ "STREAM SND '%s' [to %s] [PCLAIMID]: received zero node id '%s'",
86
rrdhost_hostname(s->host), s->remote_ip,
70
- claim_id_str ? claim_id_str : "(unset)");
87
+ node_id_str ? node_id_str : "(unset)");
88
+ return;
89
+ }
90
+
91
+ if(!url || !*url) {
92
+ nd_log(NDLS_DAEMON, NDLP_ERR,
93
+ "STREAM SND '%s' [to %s] [PCLAIMID] received an invalid cloud URL '%s'",
94
+ rrdhost_hostname(s->host), s->remote_ip,
95
+ url ? url : "(unset)");
96
+ return;
97
+ }
98
+
99
+ // ----------------------------------------------------------------------------------------------------------------
100
+ // the parameters are ok
101
+ // apply the changes
102
+
103
+ if (!UUIDeq(s->host->aclk.claim_id_of_parent, claim_id)) {
104
+ if(UUIDiszero(s->host->aclk.claim_id_of_parent))
105
+ nd_log(NDLS_DAEMON, NDLP_INFO,
106
+ "STREAM SND '%s' [to %s] [PCLAIMID] set parent's claim id to %s (was empty)",
107
+ rrdhost_hostname(s->host), s->remote_ip,
108
+ claim_id_str ? claim_id_str : "(unset)");
109
+ else
110
+ nd_log(NDLS_DAEMON, NDLP_INFO,
111
+ "STREAM SND '%s' [to %s] [PCLAIMID] changed parent's claim id to %s (was set)",
112
+ rrdhost_hostname(s->host), s->remote_ip,
113
+ claim_id_str ? claim_id_str : "(unset)");
114
+
115
+ s->host->aclk.claim_id_of_parent = claim_id;
116
+ }
117
118
if(!UUIDiszero(s->host->node_id) && !UUIDeq(s->host->node_id, node_id)) {
119
if(claimed) {
120
+ update_node_id = false;
121
nd_log(NDLS_DAEMON, NDLP_WARNING,
75
- "STREAM SND '%s' [to %s] parent reports different node id '%s', but we are claimed. Ignoring it.",
122
+ "STREAM SND '%s' [to %s] [PCLAIMID] parent reports different node id '%s', but we are claimed. Ignoring it.",
123
rrdhost_hostname(s->host), s->remote_ip,
124
node_id_str ? node_id_str : "(unset)");
78
- return;
125
}
126
else {
127
update_node_id = true;
128
nd_log(NDLS_DAEMON, NDLP_WARNING,
83
- "STREAM SND '%s' [to %s] changed node id to %s",
129
+ "STREAM SND '%s' [to %s] [PCLAIMID] changed node id to %s",
130
rrdhost_hostname(s->host), s->remote_ip,
131
node_id_str ? node_id_str : "(unset)");
132
}
133
}
134
89
- if(!url || !*url) {
90
- nd_log(NDLS_DAEMON, NDLP_ERR,
91
- "STREAM SND '%s' [to %s] received an invalid cloud URL '%s'",
92
- rrdhost_hostname(s->host), s->remote_ip,
93
- url ? url : "(unset)");
94
- return;
95
- }
96
-
97
- s->host->aclk.claim_id_of_parent = claim_id;
98
-
135
// There are some very strange corner cases here:
136
//
137
// - Agent is claimed but offline, and it receives node_id and cloud_url from a different Netdata Cloud.
src/streaming/stream-conf.c
-5
@@ -326,11 +326,6 @@ void stream_conf_receiver_config(struct receiver_state *rpt, struct stream_recei
326
&stream_config, machine_guid, "compression algorithms order",
327
inicfg_get(&stream_config, api_key, "compression algorithms order", STREAM_COMPRESSION_ALGORITHMS_ORDER)));
328
}
329
-
330
- config->ephemeral =
331
- inicfg_get_boolean(&stream_config, machine_guid, "is ephemeral node",
332
- inicfg_get_boolean(&stream_config, api_key, "is ephemeral node",
333
- CONFIG_BOOLEAN_NO));
329
}
330
331
bool stream_conf_is_key_type(const char *api_key, const char *type) {
src/streaming/stream-conf.h
-1
@@ -58,7 +58,6 @@ extern struct _stream_receive stream_receive;
58
59
struct stream_receiver_config {
60
RRD_DB_MODE mode;
61
- bool ephemeral;
61
int history;
62
int update_every;
63
src/streaming/stream-connector.c
+9
-13
@@ -224,9 +224,10 @@ stream_connect_validate_first_response(RRDHOST *host, struct sender_state *s, ch
224
}
225
226
if(version >= STREAM_HANDSHAKE_OK_V1) {
227
- stream_parent_set_reconnect_delay(host->stream.snd.parents.current, STREAM_HANDSHAKE_SP_CONNECTED,
228
- stream_send.parents.reconnect_delay_s);
227
+ stream_parent_set_host_reconnect_delay(
228
+ host, STREAM_HANDSHAKE_SP_CONNECTED, stream_send.parents.reconnect_delay_s);
229
s->capabilities = convert_stream_version_to_capabilities(version, host, true);
230
+ s->host->stream.snd.status.reason = (STREAM_HANDSHAKE)s->capabilities;
231
return true;
232
}
233
@@ -236,7 +237,7 @@ stream_connect_validate_first_response(RRDHOST *host, struct sender_state *s, ch
237
int delay = stream_responses[i].postpone_reconnect_seconds;
238
239
worker_is_busy(worker_job_id);
239
- stream_parent_set_reconnect_delay(host->stream.snd.parents.current, version, delay);
240
+ stream_parent_set_host_connect_failure_reason(host, version, delay);
241
242
ND_LOG_STACK lgs[] = {
243
ND_LOG_FIELD_I64(NDF_RESPONSE_CODE, stream_handshake_error_to_response_code(version)),
@@ -275,12 +276,9 @@ bool stream_connect(struct sender_state *s, uint16_t default_port, time_t timeou
276
&host->stream.snd.parents.current)) {
277
278
if(s->sock.error != ND_SOCK_ERR_NO_DESTINATION_AVAILABLE) {
278
- pulse_host_status(s->host, PULSE_HOST_STATUS_SND_OFFLINE, STREAM_HANDSHAKE_CONNECTION_FAILED);
279
nd_log(NDLS_DAEMON, NDLP_WARNING, "can't connect to a parent, last error: %s",
280
ND_SOCK_ERROR_2str(s->sock.error));
281
}
282
- else
283
- pulse_host_status(s->host, PULSE_HOST_STATUS_SND_NO_DST, 0);
282
283
nd_sock_close(&s->sock);
284
return false;
@@ -314,8 +312,7 @@ bool stream_connect(struct sender_state *s, uint16_t default_port, time_t timeou
312
if (s->parent_using_h2o && stream_connect_upgrade_prelude(host, s)) {
313
worker_is_busy(WORKER_SENDER_CONNECTOR_JOB_DISCONNECT_CANT_UPGRADE_CONNECTION);
314
nd_sock_close(&s->sock);
317
- stream_parent_set_reconnect_delay(
318
- host->stream.snd.parents.current, STREAM_HANDSHAKE_SND_DISCONNECT_HTTP_UPGRADE_FAILED, 60);
315
+ stream_parent_set_host_connect_failure_reason(host, STREAM_HANDSHAKE_SND_DISCONNECT_HTTP_UPGRADE_FAILED, 60);
316
return false;
317
}
318
@@ -335,8 +332,7 @@ bool stream_connect(struct sender_state *s, uint16_t default_port, time_t timeou
332
"STREAM CONNECT '%s' [to %s]: failed to send HTTP header to remote netdata.",
333
rrdhost_hostname(host), s->remote_ip);
334
338
- stream_parent_set_reconnect_delay(
339
- host->stream.snd.parents.current, STREAM_HANDSHAKE_CONNECT_SEND_TIMEOUT, 60);
335
+ stream_parent_set_host_connect_failure_reason(host, STREAM_HANDSHAKE_CONNECT_SEND_TIMEOUT, 60);
336
return false;
337
}
338
@@ -357,9 +353,7 @@ bool stream_connect(struct sender_state *s, uint16_t default_port, time_t timeou
353
"STREAM CONNECT '%s' [to %s]: remote netdata does not respond.",
354
rrdhost_hostname(host), s->remote_ip);
355
360
- stream_parent_set_reconnect_delay(
361
- host->stream.snd.parents.current, STREAM_HANDSHAKE_CONNECT_RECEIVE_TIMEOUT, 30);
362
-
356
+ stream_parent_set_host_connect_failure_reason(host, STREAM_HANDSHAKE_CONNECT_RECEIVE_TIMEOUT, 30);
357
return false;
358
}
359
response[bytes] = '\0';
@@ -498,6 +492,8 @@ void stream_connector_add(struct sender_state *s) {
492
nd_sock_close(&s->sock);
493
s->parent_using_h2o = stream_send.parents.h2o;
494
495
+ stream_parents_host_reset(s->host, 0);
496
+
497
// do not call this with any locks held
498
stream_connector_requeue(s, STRCNT_CMD_CONNECT);
499
}
src/streaming/stream-handshake.c
+5
-4
@@ -10,7 +10,7 @@ static struct {
10
{STREAM_HANDSHAKE_OK_V3, "CONNECTED", 200},
11
{STREAM_HANDSHAKE_OK_V2, "CONNECTED", 200},
12
{STREAM_HANDSHAKE_OK_V1, "CONNECTED", 200},
13
- {STREAM_HANDSHAKE_NEVER, "", 204}, // No Content
13
+ {STREAM_HANDSHAKE_NEVER, "NEVER CONNECTED", 204}, // No Content
14
{STREAM_HANDSHAKE_CONNECT_HANDSHAKE_FAILED, "BAD HANDSHAKE", 400}, // Bad Request
15
{STREAM_HANDSHAKE_PARENT_IS_LOCALHOST, "LOCALHOST", 101}, // Switching Protocols
16
{STREAM_HANDSHAKE_PARENT_NODE_ALREADY_CONNECTED, "ALREADY CONNECTED", 409}, // Conflict
@@ -24,12 +24,12 @@ static struct {
24
{STREAM_HANDSHAKE_PARENT_INTERNAL_ERROR, "INTERNAL ERROR", 500}, // Internal Server Error
25
{STREAM_HANDSHAKE_PARENT_IS_INITIALIZING, "REMOTE IS INITIALIZING", 102}, // Processing (WebDAV)
26
27
- // receiver only codes
27
+ // receiver-only codes
28
{STREAM_HANDSHAKE_RCV_DISCONNECT_PARSER_FAILED, "DISCONNECTED PARSE ERROR", 400}, // Bad Request
29
{STREAM_HANDSHAKE_RCV_DISCONNECT_STALE_RECEIVER, "DISCONNECTED STALE RECEIVER", 410}, // Gone
30
{STREAM_HANDSHAKE_RCV_DECOMPRESSION_FAILED, "DISCONNECTED DECOMPRESSION FAILED", 415}, // Unsupported Media Type
31
32
- // sender only codes
32
+ // sender-only codes
33
{STREAM_HANDSHAKE_SND_DISCONNECT_HOST_CLEANUP, "DISCONNECTED HOST CLEANUP", 202}, // Accepted
34
{STREAM_HANDSHAKE_SND_DISCONNECT_COMPRESSION_FAILED, "DISCONNECTED SND COMPRESSION FAILED", 415}, // Unsupported Media Type
35
{STREAM_HANDSHAKE_SND_DISCONNECT_HTTP_UPGRADE_FAILED, "HTTP UPGRADE ERROR", 426}, // Upgrade Required
@@ -53,8 +53,9 @@ static struct {
53
{STREAM_HANDSHAKE_SP_CONNECTION_REFUSED, "CONNECTION REFUSED", 403}, // Forbidden
54
{STREAM_HANDSHAKE_SP_CANT_RESOLVE_HOSTNAME, "CANT RESOLVE HOSTNAME", 400}, // Bad Request
55
{STREAM_HANDSHAKE_SP_CONNECTING, "CONNECTING", 102}, // Processing (WebDAV)
56
- {STREAM_HANDSHAKE_SP_CONNECTED, "CONNECTED", 200}, // OK
56
+ {STREAM_HANDSHAKE_SP_CONNECTED, "SOCKET CONNECTED", 200}, // OK
57
{STREAM_HANDSHAKE_SP_NO_STREAM_INFO, "NO STREAM INFO", 404}, // Not Found
58
+ {STREAM_HANDSHAKE_SP_NO_DESTINATION, "NO PARENT TO SEND TO", 502}, // Bad Gateway
59
60
{ 0, NULL, 0 },
61
};
src/streaming/stream-handshake.h
+4
-3
@@ -39,12 +39,12 @@ typedef enum {
39
40
// --- internal handshake reasons ---
41
42
- // receiver only codes
42
+ // receiver-only codes
43
STREAM_HANDSHAKE_RCV_DISCONNECT_PARSER_FAILED = -13,
44
STREAM_HANDSHAKE_RCV_DISCONNECT_STALE_RECEIVER = -14,
45
STREAM_HANDSHAKE_RCV_DECOMPRESSION_FAILED = -15,
46
47
- // sender only codes
47
+ // sender-only codes
48
STREAM_HANDSHAKE_SND_DISCONNECT_HOST_CLEANUP = -16,
49
STREAM_HANDSHAKE_SND_DISCONNECT_COMPRESSION_FAILED = -17,
50
STREAM_HANDSHAKE_SND_DISCONNECT_HTTP_UPGRADE_FAILED = -18,
@@ -70,9 +70,10 @@ typedef enum {
70
STREAM_HANDSHAKE_SP_CONNECTING = -34,
71
STREAM_HANDSHAKE_SP_CONNECTED = -35,
72
STREAM_HANDSHAKE_SP_NO_STREAM_INFO = -36,
73
+ STREAM_HANDSHAKE_SP_NO_DESTINATION = -37,
74
75
// terminator - keep this positive, bigger than all negative values
75
- STREAM_HANDSHAKE_NEGATIVE_MAX = 37,
76
+ STREAM_HANDSHAKE_NEGATIVE_MAX = 38,
77
} STREAM_HANDSHAKE;
78
79
const char *stream_handshake_error_to_string(STREAM_HANDSHAKE reason);
src/streaming/stream-parents.c
+68
-32
@@ -88,7 +88,9 @@ STREAM_HANDSHAKE stream_parent_get_disconnect_reason(STREAM_PARENT *d) {
88
return d->reason;
89
}
90
91
-void stream_parent_set_disconnect_reason(STREAM_PARENT *d, STREAM_HANDSHAKE reason, time_t since) {
91
+void stream_parent_set_host_disconnect_reason(RRDHOST *host, STREAM_HANDSHAKE reason, time_t since) {
92
+ host->stream.snd.status.reason = reason;
93
+ struct stream_parent *d = host->stream.snd.parents.current;
94
if(!d) return;
95
d->since_ut = since * USEC_PER_SEC;
96
d->reason = reason;
@@ -104,7 +106,7 @@ static inline usec_t randomize_wait_ut(time_t min, time_t max) {
106
return now_realtime_usec() + wait_ut;
107
}
108
107
-void rrdhost_stream_parents_reset(RRDHOST *host, STREAM_HANDSHAKE reason) {
109
+void stream_parents_host_reset(RRDHOST *host, STREAM_HANDSHAKE reason) {
110
usec_t until_ut = randomize_wait_ut(stream_send.parents.reconnect_delay_s / 2, stream_send.parents.reconnect_delay_s + 5);
111
rw_spinlock_write_lock(&host->stream.snd.parents.spinlock);
112
for (STREAM_PARENT *d = host->stream.snd.parents.all; d; d = d->next) {
@@ -115,12 +117,27 @@ void rrdhost_stream_parents_reset(RRDHOST *host, STREAM_HANDSHAKE reason) {
117
rw_spinlock_write_unlock(&host->stream.snd.parents.spinlock);
118
}
119
118
-void stream_parent_set_reconnect_delay(STREAM_PARENT *d, STREAM_HANDSHAKE reason, time_t secs) {
120
+static void stream_parent_set_reconnect_delay(STREAM_PARENT *d, STREAM_HANDSHAKE reason, time_t secs) {
121
if(!d) return;
122
d->reason = reason;
123
d->postpone_until_ut = randomize_wait_ut(5, secs);
124
}
125
126
+void stream_parent_set_host_reconnect_delay(RRDHOST *host, STREAM_HANDSHAKE reason, time_t secs) {
127
+ stream_parent_set_reconnect_delay(host->stream.snd.parents.current, reason, secs);
128
+}
129
+
130
+static void stream_parent_set_connect_failure_reason(RRDHOST *host, STREAM_PARENT *d, STREAM_HANDSHAKE reason, time_t secs) {
131
+ host->stream.snd.status.reason = reason;
132
+ pulse_host_status(host, PULSE_HOST_STATUS_SND_NO_DST_FAILED, reason);
133
+ pulse_sender_connection_failed(d ? string2str(d->destination) : NULL, reason);
134
+ stream_parent_set_reconnect_delay(d, reason, secs);
135
+}
136
+
137
+void stream_parent_set_host_connect_failure_reason(RRDHOST *host, STREAM_HANDSHAKE reason, time_t secs) {
138
+ stream_parent_set_connect_failure_reason(host, host->stream.snd.parents.current, reason, secs);
139
+}
140
+
141
usec_t stream_parent_get_reconnection_ut(STREAM_PARENT *d) {
142
return d ? d->postpone_until_ut : 0;
143
}
@@ -576,7 +593,7 @@ bool stream_parent_connect_to_one_unsafe(
593
usec_t now_ut = now_realtime_usec();
594
595
// fetch stream info for all of them and put them in the array
579
- size_t count = 0, skipped_but_useful = 0, skipped_not_useful = 0;
596
+ size_t count = 0, skipped_but_useful = 0, skipped_not_useful = 0, potential = 0;
597
for (STREAM_PARENT *d = host->stream.snd.parents.all; d && count < size ; d = d->next) {
598
if (nd_thread_signaled_to_cancel()) {
599
sender_sock->error = ND_SOCK_ERR_THREAD_CANCELLED;
@@ -584,16 +601,24 @@ bool stream_parent_connect_to_one_unsafe(
601
}
602
603
// make sure they all have a random number
587
- // this is taken from the parent, but if the stream_info call fails
604
+ // this is taken from the parent, but if the stream_info call fails,
605
// we generate a random number for every parent here
606
d->remote.nonce = os_random32();
607
d->banned_temporarily_erroneous = is_a_blocked_parent(d);
608
592
- if (d->banned_permanently || d->banned_for_this_session || d->banned_temporarily_erroneous)
609
+ if (d->banned_permanently || d->banned_for_this_session)
610
+ continue;
611
+
612
+ if (d->banned_temporarily_erroneous) {
613
+ potential++;
614
+ host->stream.snd.status.reason = d->reason;
615
continue;
616
+ }
617
618
if (d->postpone_until_ut > now_ut) {
619
skipped_but_useful++;
620
+ potential++;
621
+ host->stream.snd.status.reason = d->reason;
622
nd_log(NDLS_DAEMON, NDLP_DEBUG,
623
"STREAM PARENTS '%s': skipping useful parent '%s': POSTPONED FOR %ld SECS MORE: %s",
624
rrdhost_hostname(host),
@@ -603,7 +628,6 @@ bool stream_parent_connect_to_one_unsafe(
628
continue;
629
}
630
606
- bool skip = false;
631
if(stream_info_fetch(d, host->machine_guid, default_port,
632
sender_sock, stream_parent_is_ssl(d), rrdhost_hostname(host))) {
633
switch(d->remote.ingest_type) {
@@ -612,22 +636,23 @@ bool stream_parent_connect_to_one_unsafe(
636
d->reason = STREAM_HANDSHAKE_PARENT_IS_LOCALHOST;
637
d->since_ut = now_ut;
638
d->postpone_until_ut = randomize_wait_ut(3600, 7200);
639
+ d->banned_permanently = true;
640
+ skipped_not_useful++;
641
642
if(rrdhost_is_host_in_stream_path_before_us(host, d->remote.host_id, 1)) {
643
// we passed hops == 1, to make sure this succeeds only when the parent
644
// is the origin child of this node
619
- d->banned_permanently = true;
620
- skipped_not_useful++;
645
nd_log(NDLS_DAEMON, NDLP_INFO,
646
"STREAM PARENTS '%s': destination '%s' is banned permanently because it is the origin server",
647
rrdhost_hostname(host), string2str(d->destination));
624
- continue;
648
}
649
else {
627
- pulse_sender_stream_info_failed(string2str(d->destination), d->reason);
628
- skip = true;
650
+ nd_log(NDLS_DAEMON, NDLP_WARNING,
651
+ "STREAM PARENTS '%s': destination '%s' is banned permanently because it is the origin server, "
652
+ "but it is not in the stream path before us!",
653
+ rrdhost_hostname(host), string2str(d->destination));
654
}
630
- break;
655
+ continue;
656
657
default:
658
case RRDHOST_INGEST_TYPE_CHILD:
@@ -641,8 +666,14 @@ bool stream_parent_connect_to_one_unsafe(
666
d->since_ut = now_ut;
667
d->postpone_until_ut = randomize_wait_ut(30, 60);
668
pulse_sender_stream_info_failed(string2str(d->destination), d->reason);
644
- skip = true;
645
- break;
669
+ skipped_but_useful++;
670
+ potential++;
671
+ host->stream.snd.status.reason = d->reason;
672
+ nd_log(NDLS_DAEMON, NDLP_DEBUG,
673
+ "STREAM PARENTS '%s': skipping useful parent '%s': %s",
674
+ rrdhost_hostname(host), string2str(d->destination),
675
+ stream_handshake_error_to_string(d->reason));
676
+ continue;
677
678
case RRDHOST_INGEST_STATUS_REPLICATING:
679
case RRDHOST_INGEST_STATUS_ONLINE:
@@ -676,28 +707,26 @@ bool stream_parent_connect_to_one_unsafe(
707
else
708
pulse_sender_stream_info_failed(string2str(d->destination), d->reason);
709
679
- if(skip) {
680
- skipped_but_useful++;
681
- nd_log(NDLS_DAEMON, NDLP_DEBUG,
682
- "STREAM PARENTS '%s': skipping useful parent '%s': %s",
683
- rrdhost_hostname(host),
684
- string2str(d->destination),
685
- stream_handshake_error_to_string(d->reason));
686
- }
687
- else {
688
- d->selection.skipped = false;
689
- d->selection.batch = count + 1;
690
- d->selection.order = count + 1;
691
- array[count++] = d;
692
- }
710
+ d->selection.skipped = false;
711
+ d->selection.batch = count + 1;
712
+ d->selection.order = count + 1;
713
+ array[count++] = d;
714
}
715
716
// can we use any parent?
717
if(!count) {
718
nd_log(NDLS_DAEMON, NDLP_DEBUG,
698
- "STREAM PARENTS '%s': no parents available (%zu skipped but useful, %zu skipped not useful)",
699
- rrdhost_hostname(host),
700
- skipped_but_useful, skipped_not_useful);
719
+ "STREAM PARENTS '%s': no parents available (%zu skipped but useful, %zu skipped not useful, %zu potential)",
720
+ rrdhost_hostname(host), skipped_but_useful, skipped_not_useful, potential);
721
+
722
+ if(!potential) {
723
+ if(host->stream.snd.status.reason != STREAM_HANDSHAKE_SP_NO_DESTINATION) {
724
+ host->stream.snd.status.reason = STREAM_HANDSHAKE_SP_NO_DESTINATION;
725
+ pulse_sender_connection_failed(NULL, host->stream.snd.status.reason);
726
+ }
727
+ pulse_host_status(host, PULSE_HOST_STATUS_SND_NO_DST, 0);
728
+ }
729
+
730
return false;
731
}
732
@@ -787,6 +816,8 @@ bool stream_parent_connect_to_one_unsafe(
816
817
if(nd_thread_signaled_to_cancel()) {
818
sender_sock->error = ND_SOCK_ERR_THREAD_CANCELLED;
819
+ host->stream.snd.status.reason = STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP;
820
+ pulse_host_status(host, PULSE_HOST_STATUS_SND_OFFLINE, host->stream.snd.status.reason);
821
return false;
822
}
823
@@ -825,11 +856,15 @@ bool stream_parent_connect_to_one_unsafe(
856
sender_sock->fd);
857
858
sender_sock->error = ND_SOCK_ERR_NONE;
859
+ host->stream.snd.status.reason = STREAM_HANDSHAKE_SP_CONNECTED;
860
+ pulse_host_status(host, PULSE_HOST_STATUS_SND_CONNECTING, host->stream.snd.status.reason);
861
return true;
862
}
863
else {
864
stream_parent_nd_sock_error_to_reason(d, sender_sock);
865
+ host->stream.snd.status.reason = d->reason;
866
pulse_sender_connection_failed(string2str(d->destination), d->reason);
867
+ pulse_host_status(host, PULSE_HOST_STATUS_SND_CONNECTING, host->stream.snd.status.reason);
868
nd_log(NDLS_DAEMON, NDLP_DEBUG,
869
"STREAM PARENTS '%s': stream connection to '%s' failed (default port: %d): %s",
870
rrdhost_hostname(host),
@@ -838,6 +873,7 @@ bool stream_parent_connect_to_one_unsafe(
873
}
874
}
875
876
+ pulse_host_status(host, PULSE_HOST_STATUS_SND_OFFLINE, 0);
877
return false;
878
}
879
src/streaming/stream-parents.h
+5
-3
@@ -17,6 +17,7 @@ typedef struct rrdhost_stream_parents {
17
} RRDHOST_STREAM_PARENTS;
18
19
#include "stream-handshake.h"
20
+#include "database/rrdhost.h"
21
22
void rrdhost_stream_parent_ssl_init(struct sender_state *s);
23
@@ -33,14 +34,15 @@ bool stream_parent_connect_to_one(
34
35
void rrdhost_stream_parents_to_json(BUFFER *wb, struct rrdhost_status_t *s);
36
STREAM_HANDSHAKE stream_parent_get_disconnect_reason(STREAM_PARENT *d);
36
-void stream_parent_set_disconnect_reason(STREAM_PARENT *d, STREAM_HANDSHAKE reason, time_t since);
37
-void stream_parent_set_reconnect_delay(STREAM_PARENT *d, STREAM_HANDSHAKE reason, time_t secs);
37
+void stream_parent_set_host_disconnect_reason(RRDHOST *host, STREAM_HANDSHAKE reason, time_t since);
38
+void stream_parent_set_host_reconnect_delay(RRDHOST *host, STREAM_HANDSHAKE reason, time_t secs);
39
+void stream_parent_set_host_connect_failure_reason(RRDHOST *host, STREAM_HANDSHAKE reason, time_t secs);
40
usec_t stream_parent_get_reconnection_ut(STREAM_PARENT *d);
41
bool stream_parent_is_ssl(STREAM_PARENT *d);
42
43
usec_t stream_parent_handshake_error_to_json(BUFFER *wb, struct rrdhost *host);
44
43
-void rrdhost_stream_parents_reset(struct rrdhost *host, STREAM_HANDSHAKE reason);
45
+void stream_parents_host_reset(RRDHOST *host, STREAM_HANDSHAKE reason);
46
47
void rrdhost_stream_parents_update_from_destination(struct rrdhost *host);
48
void rrdhost_stream_parents_free(struct rrdhost *host, bool having_write_lock);
src/streaming/stream-receiver-connection.c
+1
-4
@@ -701,11 +701,8 @@ int stream_receiver_accept_connection(struct web_client *w, char *decoded_query_
701
schedule_node_state_update(rpt->host, 300);
702
rrdhost_set_is_parent_label();
703
704
- if (rpt->config.ephemeral)
705
- rrdhost_option_set(rpt->host, RRDHOST_OPTION_EPHEMERAL_HOST);
706
-
704
// let it reconnect to parents asap
708
- rrdhost_stream_parents_reset(rpt->host, STREAM_HANDSHAKE_SP_PREPARING);
705
+ stream_parents_host_reset(rpt->host, STREAM_HANDSHAKE_SP_PREPARING);
706
707
// add it to a stream thread queue
708
stream_receiver_add_to_queue(rpt);
src/streaming/stream-receiver.c
+6
-5
@@ -1116,6 +1116,7 @@ bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) {
1116
host->receiver = rpt;
1117
rpt->host = host;
1118
1119
+ host->stream.rcv.status.reason = (STREAM_HANDSHAKE)rpt->capabilities;
1120
rpt->exit.reason = 0;
1121
__atomic_store_n(&rpt->exit.shutdown, false, __ATOMIC_RELEASE);
1122
host->stream.rcv.status.last_connected = now_realtime_sec();
@@ -1146,7 +1147,7 @@ bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) {
1147
rrdhost_flag_set(rpt->host, RRDHOST_FLAG_COLLECTOR_ONLINE);
1148
aclk_queue_node_info(rpt->host, true);
1149
1149
- rrdhost_stream_parents_reset(host, STREAM_HANDSHAKE_SP_PREPARING);
1150
+ stream_parents_host_reset(host, STREAM_HANDSHAKE_SP_PREPARING);
1151
1152
set_this = true;
1153
}
@@ -1189,7 +1190,7 @@ void rrdhost_clear_receiver(struct receiver_state *rpt, STREAM_HANDSHAKE reason)
1190
if (rpt->config.health.enabled)
1191
rrdcalc_child_disconnected(host);
1192
1192
- rrdhost_stream_parents_reset(host, reason);
1193
+ stream_parents_host_reset(host, reason);
1194
}
1195
rrdhost_receiver_lock(host);
1196
@@ -1198,14 +1199,14 @@ void rrdhost_clear_receiver(struct receiver_state *rpt, STREAM_HANDSHAKE reason)
1199
stream_receiver_replication_reset(host);
1200
streaming_receiver_disconnected();
1201
1201
- host->receiver->exit.reason = 0;
1202
- __atomic_store_n(&host->receiver->exit.shutdown, false, __ATOMIC_RELEASE);
1202
+ host->stream.rcv.status.reason = rpt->exit.reason;
1203
+ rpt->exit.reason = 0;
1204
+ __atomic_store_n(&rpt->exit.shutdown, false, __ATOMIC_RELEASE);
1205
host->stream.rcv.status.check_obsolete = false;
1206
host->stream.rcv.status.last_connected = 0;
1207
host->stream.rcv.status.last_disconnected = now_realtime_sec();
1208
host->health.enabled = false;
1209
1208
- host->stream.rcv.status.exit_reason = rpt->exit.reason;
1210
rrdhost_flag_set(host, RRDHOST_FLAG_ORPHAN);
1211
host->receiver = NULL;
1212
}
src/streaming/stream-sender.c
+3
-3
@@ -366,12 +366,12 @@ void stream_sender_remove(struct sender_state *s, STREAM_HANDSHAKE reason) {
366
RRDHOST_FLAG_STREAM_SENDER_READY_4_METRICS);
367
368
s->last_state_since_t = now_realtime_sec();
369
- stream_parent_set_disconnect_reason(s->host->stream.snd.parents.current, reason, s->last_state_since_t);
369
+ stream_parent_set_host_disconnect_reason(s->host, reason, s->last_state_since_t);
370
s->connector.id = -1;
371
372
stream_sender_unlock(s);
373
374
- rrdhost_stream_parents_reset(s->host, reason);
374
+ stream_parents_host_reset(s->host, reason);
375
376
#ifdef NETDATA_LOG_STREAM_SENDER
377
spinlock_lock(&s->log.spinlock);
@@ -446,7 +446,7 @@ static void stream_sender_move_running_to_connector_or_remove(struct stream_thre
446
447
nd_sock_close(&s->sock);
448
449
- stream_parent_set_disconnect_reason(s->host->stream.snd.parents.current, reason, now_realtime_sec());
449
+ stream_parent_set_host_disconnect_reason(s->host, reason, now_realtime_sec());
450
stream_sender_clear_parent_claim_id(s->host);
451
sender_host_buffer_free(s->host);
452
src/web/api/formatters/charts2json.c
+1
-1
@@ -90,7 +90,7 @@ void charts2json(RRDHOST *host, BUFFER *wb) {
90
rrd_rdlock();
91
RRDHOST *h;
92
rrdhost_foreach_read(h) {
93
- if(!rrdhost_should_be_removed(h, host, now) /*&& !rrdhost_flag_check(h, RRDHOST_FLAG_ARCHIVED) */) {
93
+ if(!rrdhost_should_be_cleaned_up(h, host, now) /*&& !rrdhost_flag_check(h, RRDHOST_FLAG_ARCHIVED) */) {
94
buffer_json_add_array_item_object(wb);
95
buffer_json_member_add_string(wb, "hostname", rrdhost_hostname(h));
96
buffer_json_object_close(wb);
src/web/api/functions/function-streaming.c
+86
-1
@@ -59,6 +59,10 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
59
max_out_local_port = 0,
60
max_out_remote_port = 0;
61
62
+ uint32_t
63
+ max_in_connections = 0,
64
+ max_out_connections = 0;
65
+
66
{
67
RRDHOST *host;
68
dfe_start_read(rrdhost_root_index, host) {
@@ -91,6 +95,43 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
95
// Node
96
buffer_json_add_array_item_string(wb, rrdhost_hostname(s.host));
97
98
+ // rowOptions
99
+ buffer_json_add_array_item_object(wb);
100
+ {
101
+ const char *severity = NULL; // normal, debug, notice, warning, critical
102
+ if(!rrdhost_option_check(host, RRDHOST_OPTION_EPHEMERAL_HOST)) {
103
+ switch(s.ingest.status) {
104
+ case RRDHOST_INGEST_STATUS_OFFLINE:
105
+ case RRDHOST_INGEST_STATUS_ARCHIVED:
106
+ severity = "critical";
107
+ break;
108
+
109
+ default:
110
+ case RRDHOST_INGEST_STATUS_INITIALIZING:
111
+ case RRDHOST_INGEST_STATUS_ONLINE:
112
+ case RRDHOST_INGEST_STATUS_REPLICATING:
113
+ break;
114
+ }
115
+
116
+ switch(s.stream.status) {
117
+ case RRDHOST_STREAM_STATUS_OFFLINE:
118
+ if(!severity && s.stream.reason != STREAM_HANDSHAKE_SP_NO_DESTINATION)
119
+ severity = "warning";
120
+ break;
121
+
122
+ default:
123
+ case RRDHOST_STREAM_STATUS_REPLICATING:
124
+ case RRDHOST_STREAM_STATUS_ONLINE:
125
+ break;
126
+ }
127
+ }
128
+ buffer_json_member_add_string(wb, "severity", severity ? severity : "normal");
129
+ }
130
+ buffer_json_object_close(wb); // rowOptions
131
+
132
+ // Ephemerality
133
+ buffer_json_add_array_item_string(wb, rrdhost_option_check(s.host, RRDHOST_OPTION_EPHEMERAL_HOST) ? "ephemeral" : "permanent");
134
+
135
// AgentName and AgentVersion
136
buffer_json_add_array_item_string(wb, rrdhost_program_name(host));
137
buffer_json_add_array_item_string(wb, rrdhost_program_version(host));
@@ -123,6 +164,12 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
164
buffer_json_add_array_item_string(wb, rrdhost_ml_status_to_string(s.ml.status)); // MLStatus
165
166
// collection
167
+
168
+ // InConnections
169
+ buffer_json_add_array_item_uint64(wb, s.host->stream.rcv.status.connections);
170
+ if(s.host->stream.rcv.status.connections > max_in_connections)
171
+ max_in_connections = s.host->stream.rcv.status.connections;
172
+
173
if(s.ingest.since) {
174
uint64_t in_since = s.ingest.since * MSEC_PER_SEC;
175
buffer_json_add_array_item_uint64(wb, in_since); // InSince
@@ -136,7 +183,14 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
183
buffer_json_add_array_item_string(wb, NULL); // InSince
184
buffer_json_add_array_item_string(wb, NULL); // InAge
185
}
139
- buffer_json_add_array_item_string(wb, stream_handshake_error_to_string(s.ingest.reason)); // InReason
186
+
187
+ // InReason
188
+ if(s.ingest.type == RRDHOST_INGEST_TYPE_LOCALHOST)
189
+ buffer_json_add_array_item_string(wb, "LOCALHOST");
190
+ else if(s.ingest.type == RRDHOST_INGEST_TYPE_VIRTUAL)
191
+ buffer_json_add_array_item_string(wb, "VIRTUAL NODE");
192
+ else
193
+ buffer_json_add_array_item_string(wb, stream_handshake_error_to_string(s.ingest.reason));
194
195
buffer_json_add_array_item_int64(wb, s.ingest.hops); // InHops
196
if(s.ingest.hops > max_in_hops) max_in_hops = s.ingest.hops;
@@ -160,6 +214,12 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
214
buffer_json_add_array_item_uint64(wb, s.ingest.collected.contexts); // CollectedContexts
215
216
// streaming
217
+
218
+ // OutConnections
219
+ buffer_json_add_array_item_uint64(wb, s.host->stream.snd.status.connections);
220
+ if(s.host->stream.snd.status.connections > max_out_connections)
221
+ max_out_connections = s.host->stream.snd.status.connections;
222
+
223
if(s.stream.since) {
224
uint64_t out_since = s.stream.since * MSEC_PER_SEC;
225
buffer_json_add_array_item_uint64(wb, out_since); // OutSince
@@ -264,6 +324,19 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
324
RRDF_FIELD_OPTS_VISIBLE | RRDF_FIELD_OPTS_UNIQUE_KEY | RRDF_FIELD_OPTS_STICKY,
325
NULL);
326
327
+ buffer_rrdf_table_add_field(wb, field_id++, "rowOptions", "rowOptions",
328
+ RRDF_FIELD_TYPE_NONE, RRDR_FIELD_VISUAL_ROW_OPTIONS, RRDF_FIELD_TRANSFORM_NONE,
329
+ 0, NULL, NAN, RRDF_FIELD_SORT_FIXED, NULL,
330
+ RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_NONE, RRDF_FIELD_OPTS_DUMMY,
331
+ NULL);
332
+
333
+ buffer_rrdf_table_add_field(wb, field_id++, "Ephemerality", "The type of ephemerality for the node",
334
+ RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
335
+ 0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
336
+ RRDF_FIELD_SUMMARY_COUNT, RRDF_FIELD_FILTER_MULTISELECT,
337
+ RRDF_FIELD_OPTS_VISIBLE,
338
+ NULL);
339
+
340
buffer_rrdf_table_add_field(wb, field_id++, "AgentName", "The name of the Netdata agent",
341
RRDF_FIELD_TYPE_STRING, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
342
0, NULL, NAN, RRDF_FIELD_SORT_ASCENDING, NULL,
@@ -526,6 +599,12 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
599
600
// --- collection ---
601
602
+ buffer_rrdf_table_add_field(wb, field_id++, "InConnections", "Number of times this child connected",
603
+ RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
604
+ 0, NULL, (double)max_in_connections, RRDF_FIELD_SORT_DESCENDING, NULL,
605
+ RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
606
+ RRDF_FIELD_OPTS_NONE, NULL);
607
+
608
buffer_rrdf_table_add_field(wb, field_id++, "InSince", "Last Data Collection Status Change",
609
RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
610
0, NULL, (double)max_in_since, RRDF_FIELD_SORT_DESCENDING, NULL,
@@ -619,6 +698,12 @@ int function_streaming(BUFFER *wb, const char *function __maybe_unused, BUFFER *
698
699
// --- streaming ---
700
701
+ buffer_rrdf_table_add_field(wb, field_id++, "OutConnections", "Number of times connected to a parent",
702
+ RRDF_FIELD_TYPE_INTEGER, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_NONE,
703
+ 0, NULL, (double)max_out_connections, RRDF_FIELD_SORT_DESCENDING, NULL,
704
+ RRDF_FIELD_SUMMARY_SUM, RRDF_FIELD_FILTER_RANGE,
705
+ RRDF_FIELD_OPTS_NONE, NULL);
706
+
707
buffer_rrdf_table_add_field(wb, field_id++, "OutSince", "Last Streaming Status Change",
708
RRDF_FIELD_TYPE_TIMESTAMP, RRDF_FIELD_VISUAL_VALUE, RRDF_FIELD_TRANSFORM_DATETIME_MS,
709
0, NULL, (double)max_out_since, RRDF_FIELD_SORT_DESCENDING, NULL,