@cryptotaxi247 / netdata-1 / commits / 161ba1592

Fix streaming scaling (#8375)

* Disallow multiple streaming connections to the same master agent * Reject multiple streaming connections quickly without blocking * Increase timeout for systemd service shutdown to give time to flush the db. * Optimize page correlation ID to use atomic counter instead of locks * Reduce contention in global configuration mutex * Optimize complexity of inserting configuration sections from O(N) to O(1) * Reduce overhead of clockgettime() by utilizing CLOCK_MONOTONIC_COARSE when applicable. * Fix unit test compile errors

Markos Fountoulakis committed Mar 16, 2020 at 13:39 UTC 161ba1592f5412bd378ace104979ddb66cd33973
15 files changed +107 -114
collectors/ebpf_process.plugin/ebpf_process.c
+1 -1
@@ -809,7 +809,7 @@ static inline void set_log_file(char *ptr) {
809 }
810
811 static void set_global_values() {
812 - struct section *sec = collector_config.sections;
812 + struct section *sec = collector_config.first_section;
813 while(sec) {
814 if(!strcasecmp(sec->name, "global")) {
815 struct config_option *values = sec->values;
daemon/main.c
+3 -1
@@ -6,7 +6,8 @@ int netdata_zero_metrics_enabled;
6 int netdata_anonymous_statistics_enabled;
7
8 struct config netdata_config = {
9 - .sections = NULL,
9 + .first_section = NULL,
10 + .last_section = NULL,
11 .mutex = NETDATA_MUTEX_INITIALIZER,
12 .index = {
13 .avl_tree = {
@@ -1148,6 +1149,7 @@ int main(int argc, char **argv) {
1149 mallopt(M_ARENA_MAX, 1);
1150 #endif
1151 test_clock_boottime();
1152 + test_clock_monotonic_coarse();
1153
1154 // prepare configuration environment variables for the plugins
1155
database/engine/rrdengineapi.c
+1 -3
@@ -175,9 +175,7 @@ void rrdeng_store_metric_next(RRDDIM *rd, usec_t point_in_time, storage_number n
175
176 handle->descr = descr;
177
178 - uv_rwlock_wrlock(&pg_cache->committed_page_index.lock);
179 - handle->page_correlation_id = pg_cache->committed_page_index.latest_corr_id++;
180 - uv_rwlock_wrunlock(&pg_cache->committed_page_index.lock);
178 + handle->page_correlation_id = rrd_atomic_fetch_add(&pg_cache->committed_page_index.latest_corr_id, 1);
179
180 if (0 == rd->rrdset->rrddim_page_alignment) {
181 /* this is the leading dimension that defines chart alignment */
database/engine/rrdenginelib.h
+4 -2
@@ -27,11 +27,13 @@ struct rrdengine_instance;
27 typedef uintptr_t rrdeng_stats_t;
28
29 #ifdef __ATOMIC_RELAXED
30 -#define rrd_stat_atomic_add(p, n) do {(void) __atomic_fetch_add(p, n, __ATOMIC_RELAXED);} while(0)
30 +#define rrd_atomic_fetch_add(p, n) __atomic_fetch_add(p, n, __ATOMIC_RELAXED)
31 #else
32 -#define rrd_stat_atomic_add(p, n) do {(void) __sync_fetch_and_add(p, n);} while(0)
32 +#define rrd_atomic_fetch_add(p, n) __sync_fetch_and_add(p, n)
33 #endif
34
35 +#define rrd_stat_atomic_add(p, n) rrd_atomic_fetch_add(p, n)
36 +
37 #define RRDENG_PATH_MAX (4096)
38
39 /* returns old *ptr value */
database/rrdset.c
+12 -10
@@ -550,19 +550,21 @@ RRDSET *rrdset_create_custom(
550 // ------------------------------------------------------------------------
551 // get the options from the config, we need to create it
552
553 - long rentries = config_get_number(config_section, "history", history_entries);
554 - long entries = align_entries_to_pagesize(memory_mode, rentries);
555 - if(entries != rentries) entries = config_set_number(config_section, "history", entries);
556 -
557 - if(memory_mode == RRD_MEMORY_MODE_NONE && entries != rentries)
558 - entries = config_set_number(config_section, "history", 10);
559 -
553 + long entries;
554 + if(memory_mode == RRD_MEMORY_MODE_DBENGINE) {
555 + // only sets it the first time
556 + entries = config_get_number(config_section, "history", 5);
557 + } else {
558 + long rentries = config_get_number(config_section, "history", history_entries);
559 + entries = align_entries_to_pagesize(memory_mode, rentries);
560 + if (entries != rentries) entries = config_set_number(config_section, "history", entries);
561 +
562 + if (memory_mode == RRD_MEMORY_MODE_NONE && entries != rentries)
563 + entries = config_set_number(config_section, "history", 10);
564 + }
565 int enabled = config_get_boolean(config_section, "enabled", 1);
566 if(!enabled) entries = 5;
567
563 - if(memory_mode == RRD_MEMORY_MODE_DBENGINE)
564 - entries = config_set_number(config_section, "history", 5);
565 -
568 unsigned long size = sizeof(RRDSET);
569 char *cache_dir = rrdset_cache_dir(host, fullid, config_section);
570
exporting/read_config.c
+2 -1
@@ -2,7 +2,8 @@
2
3 #include "exporting_engine.h"
4
5 -struct config exporting_config = {.sections = NULL,
5 +struct config exporting_config = {.first_section = NULL,
6 + .last_section = NULL,
7 .mutex = NETDATA_MUTEX_INITIALIZER,
8 .index = {.avl_tree = {.root = NULL, .compar = appconfig_section_compare},
9 .rwlock = AVL_LOCK_INITIALIZER}};
libnetdata/clocks/clocks.c
+17 -6
@@ -3,6 +3,7 @@
3 #include "../libnetdata.h"
4
5 static int clock_boottime_valid = 1;
6 +static int clock_monotonic_coarse_valid = 1;
7
8 #ifndef HAVE_CLOCK_GETTIME
9 inline int clock_gettime(clockid_t clk_id, struct timespec *ts) {
@@ -23,6 +24,12 @@ void test_clock_boottime(void) {
24 clock_boottime_valid = 0;
25 }
26
27 +void test_clock_monotonic_coarse(void) {
28 + struct timespec ts;
29 + if(clock_gettime(CLOCK_MONOTONIC_COARSE, &ts) == -1 && errno == EINVAL)
30 + clock_monotonic_coarse_valid = 0;
31 +}
32 +
33 static inline time_t now_sec(clockid_t clk_id) {
34 struct timespec ts;
35 if(unlikely(clock_gettime(clk_id, &ts) == -1)) {
@@ -69,27 +76,31 @@ inline int now_realtime_timeval(struct timeval *tv) {
76 }
77
78 inline time_t now_monotonic_sec(void) {
72 - return now_sec(CLOCK_MONOTONIC);
79 + return now_sec(likely(clock_monotonic_coarse_valid) ? CLOCK_MONOTONIC_COARSE : CLOCK_MONOTONIC);
80 }
81
82 inline usec_t now_monotonic_usec(void) {
76 - return now_usec(CLOCK_MONOTONIC);
83 + return now_usec(likely(clock_monotonic_coarse_valid) ? CLOCK_MONOTONIC_COARSE : CLOCK_MONOTONIC);
84 }
85
86 inline int now_monotonic_timeval(struct timeval *tv) {
80 - return now_timeval(CLOCK_MONOTONIC, tv);
87 + return now_timeval(likely(clock_monotonic_coarse_valid) ? CLOCK_MONOTONIC_COARSE : CLOCK_MONOTONIC, tv);
88 }
89
90 inline time_t now_boottime_sec(void) {
84 - return now_sec(likely(clock_boottime_valid) ? CLOCK_BOOTTIME : CLOCK_MONOTONIC);
91 + return now_sec(likely(clock_boottime_valid) ? CLOCK_BOOTTIME :
92 + likely(clock_monotonic_coarse_valid) ? CLOCK_MONOTONIC_COARSE : CLOCK_MONOTONIC);
93 }
94
95 inline usec_t now_boottime_usec(void) {
88 - return now_usec(likely(clock_boottime_valid) ? CLOCK_BOOTTIME : CLOCK_MONOTONIC);
96 + return now_usec(likely(clock_boottime_valid) ? CLOCK_BOOTTIME :
97 + likely(clock_monotonic_coarse_valid) ? CLOCK_MONOTONIC_COARSE : CLOCK_MONOTONIC);
98 }
99
100 inline int now_boottime_timeval(struct timeval *tv) {
92 - return now_timeval(likely(clock_boottime_valid) ? CLOCK_BOOTTIME : CLOCK_MONOTONIC, tv);
101 + return now_timeval(likely(clock_boottime_valid) ? CLOCK_BOOTTIME :
102 + likely(clock_monotonic_coarse_valid) ? CLOCK_MONOTONIC_COARSE : CLOCK_MONOTONIC,
103 + tv);
104 }
105
106 inline usec_t timeval_usec(struct timeval *tv) {
libnetdata/clocks/clocks.h
+13 -1
@@ -36,6 +36,12 @@ typedef struct heartbeat {
36 #define CLOCK_MONOTONIC CLOCK_REALTIME
37 #endif
38
39 +/* Prefer CLOCK_MONOTONIC_COARSE where available to reduce overhead. It has the same semantics as CLOCK_MONOTONIC */
40 +#ifndef CLOCK_MONOTONIC_COARSE
41 +/* fallback to CLOCK_MONOTONIC if not available */
42 +#define CLOCK_MONOTONIC_COARSE CLOCK_MONOTONIC
43 +#endif
44 +
45 #ifndef CLOCK_BOOTTIME
46
47 #ifdef CLOCK_UPTIME
@@ -43,7 +49,7 @@ typedef struct heartbeat {
49 #define CLOCK_BOOTTIME CLOCK_UPTIME
50 #else // CLOCK_UPTIME
51 /* CLOCK_BOOTTIME falls back to CLOCK_MONOTONIC */
46 -#define CLOCK_BOOTTIME CLOCK_MONOTONIC
52 +#define CLOCK_BOOTTIME CLOCK_MONOTONIC_COARSE
53 #endif // CLOCK_UPTIME
54
55 #else // CLOCK_BOOTTIME
@@ -136,6 +142,12 @@ extern int sleep_usec(usec_t usec);
142 */
143 void test_clock_boottime(void);
144
145 +/*
146 + * When running a binary with CLOCK_MONOTONIC_COARSE defined on a system with a linux kernel older than Linux 2.6.32 the
147 + * clock_gettime(2) system call fails with EINVAL. In that case it must fall-back to CLOCK_MONOTONIC.
148 + */
149 +void test_clock_monotonic_coarse(void);
150 +
151 extern collected_number uptime_msec(char *filename);
152
153 #endif /* NETDATA_CLOCKS_H */
libnetdata/config/appconfig.c
+5 -4
@@ -169,12 +169,13 @@ static inline struct section *appconfig_section_create(struct config *root, cons
169 error("INTERNAL ERROR: indexing of section '%s', already exists.", co->name);
170
171 appconfig_wrlock(root);
172 - struct section *co2 = root->sections;
172 + struct section *co2 = root->last_section;
173 if(co2) {
174 - while (co2->next) co2 = co2->next;
174 co2->next = co;
175 + } else {
176 + root->first_section = co;
177 }
177 - else root->sections = co;
178 + root->last_section = co;
179 appconfig_unlock(root);
180
181 return co;
@@ -678,7 +679,7 @@ void appconfig_generate(struct config *root, BUFFER *wb, int only_changed)
679 }
680
681 appconfig_wrlock(root);
681 - for(co = root->sections; co ; co = co->next) {
682 + for(co = root->first_section; co ; co = co->next) {
683 if(!strcmp(co->name, CONFIG_SECTION_GLOBAL)
684 || !strcmp(co->name, CONFIG_SECTION_WEB)
685 || !strcmp(co->name, CONFIG_SECTION_STATSD)
libnetdata/config/appconfig.h
+2 -1
@@ -141,7 +141,8 @@ struct section {
141 };
142
143 struct config {
144 - struct section *sections;
144 + struct section *first_section;
145 + struct section *last_section; // optimize inserting at the end
146 netdata_mutex_t mutex;
147 avl_tree_lock index;
148 };
streaming/rrdpush.c
+42 -67
@@ -44,7 +44,8 @@ typedef struct {
44 } stream_encoded_t;
45
46 static struct config stream_config = {
47 - .sections = NULL,
47 + .first_section = NULL,
48 + .last_section = NULL,
49 .mutex = NETDATA_MUTEX_INITIALIZER,
50 .index = {
51 .avl_tree = {
@@ -168,7 +169,7 @@ int configured_as_master() {
169 int is_master = 0;
170
171 appconfig_wrlock(&stream_config);
171 - for (section = stream_config.sections; section; section = section->next) {
172 + for (section = stream_config.first_section; section; section = section->next) {
173 uuid_t uuid;
174
175 if (uuid_parse(section->name, uuid) != -1 &&
@@ -1080,35 +1081,6 @@ static void log_stream_connection(const char *client_ip, const char *client_port
1081 log_access("STREAM: %d '[%s]:%s' '%s' host '%s' api key '%s' machine guid '%s'", gettid(), client_ip, client_port, msg, host, api_key, machine_guid);
1082 }
1083
1083 -static RRDPUSH_MULTIPLE_CONNECTIONS_STRATEGY get_multiple_connections_strategy(struct config *c, const char *section, const char *name, RRDPUSH_MULTIPLE_CONNECTIONS_STRATEGY def) {
1084 - char *value;
1085 - switch(def) {
1086 - default:
1087 - case RRDPUSH_MULTIPLE_CONNECTIONS_ALLOW:
1088 - value = "allow";
1089 - break;
1090 -
1091 - case RRDPUSH_MULTIPLE_CONNECTIONS_DENY_NEW:
1092 - value = "deny";
1093 - break;
1094 - }
1095 -
1096 - value = appconfig_get(c, section, name, value);
1097 -
1098 - RRDPUSH_MULTIPLE_CONNECTIONS_STRATEGY ret = def;
1099 -
1100 - if(strcasecmp(value, "allow") == 0 || strcasecmp(value, "permit") == 0 || strcasecmp(value, "accept") == 0)
1101 - ret = RRDPUSH_MULTIPLE_CONNECTIONS_ALLOW;
1102 -
1103 - else if(strcasecmp(value, "deny") == 0 || strcasecmp(value, "reject") == 0 || strcasecmp(value, "block") == 0)
1104 - ret = RRDPUSH_MULTIPLE_CONNECTIONS_DENY_NEW;
1105 -
1106 - else
1107 - error("Invalid stream config value at section [%s], setting '%s', value '%s'", section, name, value);
1108 -
1109 - return ret;
1110 -}
1111 -
1084 static int rrdpush_receive(int fd
1085 , const char *key
1086 , const char *hostname
@@ -1137,7 +1109,6 @@ static int rrdpush_receive(int fd
1109 char *rrdpush_api_key = default_rrdpush_api_key;
1110 char *rrdpush_send_charts_matching = default_rrdpush_send_charts_matching;
1111 time_t alarms_delay = 60;
1140 - RRDPUSH_MULTIPLE_CONNECTIONS_STRATEGY rrdpush_multiple_connections_strategy = RRDPUSH_MULTIPLE_CONNECTIONS_ALLOW;
1112
1113 update_every = (int)appconfig_get_number(&stream_config, machine_guid, "update every", update_every);
1114 if(update_every < 0) update_every = 1;
@@ -1164,9 +1135,6 @@ static int rrdpush_receive(int fd
1135 rrdpush_api_key = appconfig_get(&stream_config, key, "default proxy api key", rrdpush_api_key);
1136 rrdpush_api_key = appconfig_get(&stream_config, machine_guid, "proxy api key", rrdpush_api_key);
1137
1167 - rrdpush_multiple_connections_strategy = get_multiple_connections_strategy(&stream_config, key, "multiple connections", rrdpush_multiple_connections_strategy);
1168 - rrdpush_multiple_connections_strategy = get_multiple_connections_strategy(&stream_config, machine_guid, "multiple connections", rrdpush_multiple_connections_strategy);
1169 -
1138 rrdpush_send_charts_matching = appconfig_get(&stream_config, key, "default proxy send charts matching", rrdpush_send_charts_matching);
1139 rrdpush_send_charts_matching = appconfig_get(&stream_config, machine_guid, "proxy send charts matching", rrdpush_send_charts_matching);
1140
@@ -1179,26 +1147,40 @@ static int rrdpush_receive(int fd
1147 close(fd);
1148 return 1;
1149 }
1182 - else
1183 - host = rrdhost_find_or_create(
1184 - hostname
1185 - , registry_hostname
1186 - , machine_guid
1187 - , os
1188 - , timezone
1189 - , tags
1190 - , program_name
1191 - , program_version
1192 - , update_every
1193 - , history
1194 - , mode
1195 - , (unsigned int)(health_enabled != CONFIG_BOOLEAN_NO)
1196 - , (unsigned int)(rrdpush_enabled && rrdpush_destination && *rrdpush_destination && rrdpush_api_key && *rrdpush_api_key)
1197 - , rrdpush_destination
1198 - , rrdpush_api_key
1199 - , rrdpush_send_charts_matching
1200 - , system_info
1201 - );
1150 +
1151 + /*
1152 + * Quick path for rejecting multiple connections. Don't take any locks so that progress is made. The same
1153 + * condition will be checked again below, while holding the global and host writer locks. Any potential false
1154 + * positives will not cause harm. Data hazards with host deconstruction will be handled when reference counting
1155 + * is implemented.
1156 + */
1157 + host = rrdhost_find_by_guid(machine_guid, 0);
1158 + if(host && host->connected_senders > 0) {
1159 + log_stream_connection(client_ip, client_port, key, host->machine_guid, host->hostname, "REJECTED - ALREADY CONNECTED");
1160 + info("STREAM %s [receive from [%s]:%s]: multiple streaming connections for the same host detected. Rejecting new connection.", host->hostname, client_ip, client_port);
1161 + close(fd);
1162 + return 0;
1163 + }
1164 +
1165 + host = rrdhost_find_or_create(
1166 + hostname
1167 + , registry_hostname
1168 + , machine_guid
1169 + , os
1170 + , timezone
1171 + , tags
1172 + , program_name
1173 + , program_version
1174 + , update_every
1175 + , history
1176 + , mode
1177 + , (unsigned int)(health_enabled != CONFIG_BOOLEAN_NO)
1178 + , (unsigned int)(rrdpush_enabled && rrdpush_destination && *rrdpush_destination && rrdpush_api_key && *rrdpush_api_key)
1179 + , rrdpush_destination
1180 + , rrdpush_api_key
1181 + , rrdpush_send_charts_matching
1182 + , system_info
1183 + );
1184
1185 if(!host) {
1186 close(fd);
@@ -1279,18 +1261,11 @@ static int rrdpush_receive(int fd
1261
1262 rrdhost_wrlock(host);
1263 if(host->connected_senders > 0) {
1282 - switch(rrdpush_multiple_connections_strategy) {
1283 - case RRDPUSH_MULTIPLE_CONNECTIONS_ALLOW:
1284 - info("STREAM %s [receive from [%s]:%s]: multiple streaming connections for the same host detected. If multiple netdata are pushing metrics for the same charts, at the same time, the result is unexpected.", host->hostname, client_ip, client_port);
1285 - break;
1286 -
1287 - case RRDPUSH_MULTIPLE_CONNECTIONS_DENY_NEW:
1288 - rrdhost_unlock(host);
1289 - log_stream_connection(client_ip, client_port, key, host->machine_guid, host->hostname, "REJECTED - ALREADY CONNECTED");
1290 - info("STREAM %s [receive from [%s]:%s]: multiple streaming connections for the same host detected. Rejecting new connection.", host->hostname, client_ip, client_port);
1291 - fclose(fp);
1292 - return 0;
1293 - }
1264 + rrdhost_unlock(host);
1265 + log_stream_connection(client_ip, client_port, key, host->machine_guid, host->hostname, "REJECTED - ALREADY CONNECTED");
1266 + info("STREAM %s [receive from [%s]:%s]: multiple streaming connections for the same host detected. Rejecting new connection.", host->hostname, client_ip, client_port);
1267 + fclose(fp);
1268 + return 0;
1269 }
1270
1271 rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);
streaming/stream.conf
-14
@@ -149,13 +149,6 @@
149 # postpone alarms for a short period after the sender is connected
150 default postpone alarms on connect seconds = 60
151
152 - # allow or deny multiple connections for the same host?
153 - # If you are sure all your netdata have their own machine GUID,
154 - # set this to 'allow', since it allows faster reconnects.
155 - # When set to 'deny', new connections for a host will not be
156 - # accepted until an existing connection is cleared.
157 - multiple connections = allow
158 -
152 # need to route metrics differently? set these.
153 # the defaults are the ones at the [stream] section (above)
154 #default proxy enabled = yes | no
@@ -204,13 +197,6 @@
197 # postpone alarms when the sender connects
198 postpone alarms on connect seconds = 60
199
207 - # allow or deny multiple connections for the same host?
208 - # If you are sure all your netdata have their own machine GUID,
209 - # set this to 'allow', since it allows faster reconnects.
210 - # When set to 'deny', new connections for a host will not be
211 - # accepted until an existing connection is cleared.
212 - multiple connections = allow
213 -
200 # need to route metrics differently?
201 # the defaults are the ones at the [API KEY] section
202 #proxy enabled = yes | no
system/netdata.service.in
+1 -1
@@ -21,7 +21,7 @@ ExecStartPre=/bin/chown -R netdata:netdata @localstatedir_POST@/run/netdata
21 PermissionsStartOnly=true
22
23 # saving a big db on slow disks may need some time
24 -TimeoutStopSec=60
24 +TimeoutStopSec=150
25
26 # restart netdata if it crashes
27 Restart=on-failure
web/api/tests/valid_urls.c
+2 -1
@@ -182,7 +182,8 @@ WEB_SERVER_MODE web_server_mode = WEB_SERVER_MODE_STATIC_THREADED;
182 char *netdata_configured_web_dir = "UNKNOWN FIXME";
183 RRDHOST *localhost = NULL;
184
185 -struct config netdata_config = { .sections = NULL,
185 +struct config netdata_config = { .first_section = NULL,
186 + .last_section = NULL,
187 .mutex = NETDATA_MUTEX_INITIALIZER,
188 .index = { .avl_tree = { .root = NULL, .compar = appconfig_section_compare },
189 .rwlock = AVL_LOCK_INITIALIZER } };
web/api/tests/web_api.c
+2 -1
@@ -184,7 +184,8 @@ WEB_SERVER_MODE web_server_mode = WEB_SERVER_MODE_STATIC_THREADED;
184 char *netdata_configured_web_dir = "UNKNOWN FIXME";
185 RRDHOST *localhost = NULL;
186
187 -struct config netdata_config = { .sections = NULL,
187 +struct config netdata_config = { .first_section = NULL,
188 + .last_section = NULL,
189 .mutex = NETDATA_MUTEX_INITIALIZER,
190 .index = { .avl_tree = { .root = NULL, .compar = appconfig_section_compare },
191 .rwlock = AVL_LOCK_INITIALIZER } };