@cryptotaxi247 / netdata-1 / commits / 33633ba17

Streaming improvements No 4 (#19186)

* ml logging about dimensions acquiring with rate limit * unify logs * backfilling thread for quickly backfilling charts before initiating replication * multiple backfill threads * fix for the multi-threaded backfill * another fix for the multi-threaded backfill * each backfilling thread is working on a dimension level * allocate memory without having the backfill lock * workers in backfill threads * use aral for backfilling threads; limit the number of backfilling threads to 16 * enable backfilling threads only on parents * leftover freez() * cleanup netdata startup * streaming receivers waiting list * use also replication to control the waiting list * when it is stable check on every iteration * make sure the right data are always set before adding the request to the queue * accept new nodes every 5 seconds

Costa Tsaousis committed Dec 12, 2024 at 13:16 UTC 33633ba17652ae3a6ce0162d0cb0d99dc3ebe915

41 files changed +768 -345

CMakeLists.txt

@@ -1551,6 +1551,8 @@ set(STREAMING_PLUGIN_FILES

1551 src/streaming/stream-circular-buffer.h

1552 src/streaming/stream-control.c

1553 src/streaming/stream-control.h

1554 + src/streaming/stream-waiting-list.c

1555 + src/streaming/stream-waiting-list.h

1556 )

1557

1558 set(WEB_PLUGIN_FILES

@@ -1564,6 +1566,8 @@ set(WEB_PLUGIN_FILES

1566 src/web/server/web_client_cache.h

1567 src/web/api/v3/api_v3_stream_info.c

1568 src/web/api/v3/api_v3_stream_path.c

1569 + src/web/api/queries/backfill.c

1570 + src/web/api/queries/backfill.h

1571 )

1572

1573 set(CLAIM_PLUGIN_FILES

src/daemon/analytics.c

+1 -1

@@ -911,7 +911,7 @@ void analytics_statistic_send(const analytics_statistic_t *statistic) {

911 }

912

913 void analytics_reset(void) {

914 - analytics_data.data_length = 0;

914 + analytics_data.data_length = 0;

915 analytics_set_data(&analytics_data.netdata_config_stream_enabled, "null");

916 analytics_set_data(&analytics_data.netdata_config_memory_mode, "null");

917 analytics_set_data(&analytics_data.netdata_config_exporting_enabled, "null");

src/daemon/config/netdata-conf-backwards-compatibility.c

-4

@@ -4,10 +4,6 @@

4 #include "database/engine/rrdengineapi.h"

6 void netdata_conf_backwards_compatibility(void) {

7 - static bool run = false;

8 - if(run) return;

9 - run = true;

10 -

7 // move [global] options to the [web] section

9 config_move(CONFIG_SECTION_GLOBAL, "http port listen backlog",

src/daemon/config/netdata-conf-global.c

+41 -2

@@ -18,9 +18,45 @@ static int get_hostname(char *buf, size_t buf_size) {

18 return rc;

19 }

21 -void netdata_conf_section_global(void) {

22 - netdata_conf_backwards_compatibility();

21 +static void glibc_initialize(void) {

22 + const char *pmax = config_get(CONFIG_SECTION_GLOBAL, "glibc malloc arena max for plugins", "1");

23 + if(pmax && *pmax)

24 + setenv("MALLOC_ARENA_MAX", pmax, 1);

25 +

26 +#if defined(HAVE_C_MALLOPT)

27 + int i = (int)config_get_number(CONFIG_SECTION_GLOBAL, "glibc malloc arena max for netdata", 1);

28 + if(i > 0)

29 + mallopt(M_ARENA_MAX, 1);

30 +

31 +#ifdef NETDATA_INTERNAL_CHECKS

32 + mallopt(M_PERTURB, 0x5A);

33 + // mallopt(M_MXFAST, 0);

34 +#endif

35 +#endif

36 +}

37 +

38 +static void libuv_initialize(void) {

39 + libuv_worker_threads = (int)get_netdata_cpus() * 6;

40 +

41 + if(libuv_worker_threads < MIN_LIBUV_WORKER_THREADS)

42 + libuv_worker_threads = MIN_LIBUV_WORKER_THREADS;

43 +

44 + if(libuv_worker_threads > MAX_LIBUV_WORKER_THREADS)

45 + libuv_worker_threads = MAX_LIBUV_WORKER_THREADS;

47 +

48 + libuv_worker_threads = config_get_number(CONFIG_SECTION_GLOBAL, "libuv worker threads", libuv_worker_threads);

49 + if(libuv_worker_threads < MIN_LIBUV_WORKER_THREADS) {

50 + libuv_worker_threads = MIN_LIBUV_WORKER_THREADS;

51 + config_set_number(CONFIG_SECTION_GLOBAL, "libuv worker threads", libuv_worker_threads);

52 + }

53 +

54 + char buf[20 + 1];

55 + snprintfz(buf, sizeof(buf) - 1, "%d", libuv_worker_threads);

56 + setenv("UV_THREADPOOL_SIZE", buf, 1);

57 +}

58 +

59 +void netdata_conf_section_global(void) {

60 // ------------------------------------------------------------------------

61 // get the hostname

@@ -42,6 +78,9 @@ void netdata_conf_section_global(void) {

79 os_get_system_cpus_uncached();

80 os_get_system_pid_max();

81 +

82 + glibc_initialize();

83 + libuv_initialize();

84 }

86 void netdata_conf_section_global_run_as_user(const char **user) {

src/daemon/config/netdata-conf-logs.c

+22

@@ -2,6 +2,27 @@

3 #include "netdata-conf-logs.h"

5 +static void debug_flags_initialize(void) {

6 + // --------------------------------------------------------------------

7 + // get the debugging flags from the configuration file

8 +

9 + const char *flags = config_get(CONFIG_SECTION_LOGS, "debug flags", "0x0000000000000000");

10 + nd_setenv("NETDATA_DEBUG_FLAGS", flags, 1);

11 +

12 + debug_flags = strtoull(flags, NULL, 0);

13 + netdata_log_debug(D_OPTIONS, "Debug flags set to '0x%" PRIX64 "'.", debug_flags);

14 +

15 + if(debug_flags != 0) {

16 + struct rlimit rl = { RLIM_INFINITY, RLIM_INFINITY };

17 + if(setrlimit(RLIMIT_CORE, &rl) != 0)

18 + netdata_log_error("Cannot request unlimited core dumps for debugging... Proceeding anyway...");

19 +

20 +#ifdef HAVE_SYS_PRCTL_H

21 + prctl(PR_SET_DUMPABLE, 1, 0, 0, 0);

22 +#endif

23 + }

24 +}

25 +

26 void netdata_conf_section_logs(void) {

27 static bool run = false;

28 if(run) return;

@@ -78,5 +99,6 @@ void netdata_conf_section_logs(void) {

99 nd_log_set_user_settings(NDLS_ACLK, config_get(CONFIG_SECTION_CLOUD, "conversation log file", filename));

100 }

101

102 + debug_flags_initialize();

103 aclk_config_get_query_scope();

104 }

src/daemon/config/netdata-conf.c

@@ -3,6 +3,10 @@

3 #include "netdata-conf.h"

5 bool netdata_conf_load(char *filename, char overwrite_used, const char **user) {

6 + static bool run = false;

7 + if(run) return false;

8 + run = true;

9 +

10 errno_clear();

12 int ret = 0;

@@ -29,6 +33,7 @@ bool netdata_conf_load(char *filename, char overwrite_used, const char **user) {

33 freez(filename);

34 }

36 + netdata_conf_backwards_compatibility();

37 netdata_conf_section_global_run_as_user(user);

38 return ret;

39 }

src/daemon/main.c

+50 -102

@@ -4,6 +4,7 @@

4 #include "buildinfo.h"

5 #include "daemon/watcher.h"

6 #include "static_threads.h"

7 +#include "web/api/queries/backfill.h"

9 #include "database/engine/page_test.h"

10 #include <curl/curl.h>

@@ -1326,108 +1327,68 @@ int netdata_main(int argc, char **argv) {

1327 cloud_conf_load(0);

1328 }

1329

1329 - // ------------------------------------------------------------------------

1330 - // initialize netdata

1331 - {

1332 - const char *pmax = config_get(CONFIG_SECTION_GLOBAL, "glibc malloc arena max for plugins", "1");

1333 - if(pmax && *pmax)

1334 - setenv("MALLOC_ARENA_MAX", pmax, 1);

1335 -

1336 -#if defined(HAVE_C_MALLOPT)

1337 - i = (int)config_get_number(CONFIG_SECTION_GLOBAL, "glibc malloc arena max for netdata", 1);

1338 - if(i > 0)

1339 - mallopt(M_ARENA_MAX, 1);

1330 + // ----------------------------------------------------------------------------------------------------------------

1331 + // initialize the logging system

1332 + // IMPORTANT: KEEP THIS FIRST SO THAT THE REST OF NETDATA WILL LOG PROPERLY

1333

1334 + netdata_conf_section_logs();

1335 + nd_log_limits_unlimited();

1336

1342 -#ifdef NETDATA_INTERNAL_CHECKS

1343 - mallopt(M_PERTURB, 0x5A);

1344 - // mallopt(M_MXFAST, 0);

1345 -#endif

1346 -#endif

1347 -

1348 - // set libuv worker threads

1349 - libuv_worker_threads = (int)get_netdata_cpus() * 6;

1350 -

1351 - if(libuv_worker_threads < MIN_LIBUV_WORKER_THREADS)

1352 - libuv_worker_threads = MIN_LIBUV_WORKER_THREADS;

1353 -

1354 - if(libuv_worker_threads > MAX_LIBUV_WORKER_THREADS)

1355 - libuv_worker_threads = MAX_LIBUV_WORKER_THREADS;

1337 + // initialize the log files

1338 + nd_log_initialize();

1339 + netdata_log_info("Netdata agent version '%s' is starting", NETDATA_VERSION);

1340

1341 + // ----------------------------------------------------------------------------------------------------------------

1342 + // global configuration

1343

1358 - libuv_worker_threads = config_get_number(CONFIG_SECTION_GLOBAL, "libuv worker threads", libuv_worker_threads);

1359 - if(libuv_worker_threads < MIN_LIBUV_WORKER_THREADS) {

1360 - libuv_worker_threads = MIN_LIBUV_WORKER_THREADS;

1361 - config_set_number(CONFIG_SECTION_GLOBAL, "libuv worker threads", libuv_worker_threads);

1362 - }

1344 + netdata_conf_section_global();

1345

1364 - {

1365 - char buf[20 + 1];

1366 - snprintfz(buf, sizeof(buf) - 1, "%d", libuv_worker_threads);

1367 - setenv("UV_THREADPOOL_SIZE", buf, 1);

1368 - }

1346 + // Get execution path before switching user to avoid permission issues

1347 + get_netdata_execution_path();

1348

1370 - // prepare configuration environment variables for the plugins

1371 - netdata_conf_section_global();

1372 - set_environment_for_plugins_and_scripts();

1373 - analytics_reset();

1349 + // ----------------------------------------------------------------------------------------------------------------

1350 + // analytics

1351

1375 - // work while we are cd into config_dir

1376 - // to allow the plugins refer to their config

1377 - // files using relative filenames

1378 - if(chdir(netdata_configured_user_config_dir) == -1)

1379 - fatal("Cannot cd to '%s'", netdata_configured_user_config_dir);

1352 + analytics_reset();

1353 + get_system_timezone();

1354

1381 - // Get execution path before switching user to avoid permission issues

1382 - get_netdata_execution_path();

1383 - }

1384 -

1385 - {

1386 - // --------------------------------------------------------------------

1387 - // get the debugging flags from the configuration file

1355 + // ----------------------------------------------------------------------------------------------------------------

1356 + // data collection plugins

1357

1389 - const char *flags = config_get(CONFIG_SECTION_LOGS, "debug flags", "0x0000000000000000");

1390 - nd_setenv("NETDATA_DEBUG_FLAGS", flags, 1);

1358 + // prepare configuration environment variables for the plugins

1359 + set_environment_for_plugins_and_scripts();

1360

1392 - debug_flags = strtoull(flags, NULL, 0);

1393 - netdata_log_debug(D_OPTIONS, "Debug flags set to '0x%" PRIX64 "'.", debug_flags);

1361 + // cd into config_dir to allow the plugins refer to their config files using relative filenames

1362 + if(chdir(netdata_configured_user_config_dir) == -1)

1363 + fatal("Cannot cd to '%s'", netdata_configured_user_config_dir);

1364

1395 - if(debug_flags != 0) {

1396 - struct rlimit rl = { RLIM_INFINITY, RLIM_INFINITY };

1397 - if(setrlimit(RLIMIT_CORE, &rl) != 0)

1398 - netdata_log_error("Cannot request unlimited core dumps for debugging... Proceeding anyway...");

1365 + // ----------------------------------------------------------------------------------------------------------------

1366 + // pulse (internal netdata instrumentation)

1367

1400 -#ifdef HAVE_SYS_PRCTL_H

1401 - prctl(PR_SET_DUMPABLE, 1, 0, 0, 0);

1368 +#ifdef NETDATA_INTERNAL_CHECKS

1369 + pulse_enabled = true;

1370 + pulse_extended_enabled = true;

1371 #endif

1403 - }

1404 -

1372

1406 - // --------------------------------------------------------------------

1407 - // get log filenames and settings

1408 -

1409 - netdata_conf_section_logs();

1410 - nd_log_limits_unlimited();

1411 -

1412 - // initialize the log files

1413 - nd_log_initialize();

1414 - netdata_log_info("Netdata agent version '%s' is starting", NETDATA_VERSION);

1373 + pulse_extended_enabled =

1374 + config_get_boolean(CONFIG_SECTION_PULSE, "extended", pulse_extended_enabled);

1375

1416 - check_local_streaming_capabilities();

1376 + if(pulse_extended_enabled)

1377 + // this has to run before starting any other threads that use workers

1378 + workers_utilization_enable();

1379

1418 - get_system_timezone();

1380 + // ----------------------------------------------------------------------------------------------------------------

1381 + // streaming, replication, backfilling

1382

1420 - replication_initialize();

1383 + stream_conf_load();

1384 + check_local_streaming_capabilities();

1385 + replication_initialize();

1386

1422 - rrd_functions_inflight_init();

1423 -

1424 - // --------------------------------------------------------------------

1425 - // get the certificate and start security

1426 -

1427 - netdata_conf_web_security_init();

1387 + rrd_functions_inflight_init();

1388

1389 + {

1390 // --------------------------------------------------------------------

1430 - // This is the safest place to start the SILENCERS structure

1391 + // alerts SILENCERS

1392

1393 health_set_silencers_filename();

1394 health_initialize_global_silencers();

@@ -1452,21 +1413,12 @@ int netdata_main(int argc, char **argv) {

1413 if (default_stacksize < 1 * 1024 * 1024)

1414 default_stacksize = 1 * 1024 * 1024;

1415

1455 -#ifdef NETDATA_INTERNAL_CHECKS

1456 - pulse_enabled = true;

1457 - pulse_extended_enabled = true;

1458 -#endif

1459 -

1460 - pulse_extended_enabled =

1461 - config_get_boolean(CONFIG_SECTION_PULSE, "extended", pulse_extended_enabled);

1462 -

1463 - if(pulse_extended_enabled)

1464 - // this has to run before starting any other threads that use workers

1465 - workers_utilization_enable();

1466 -

1416 for (i = 0; static_threads[i].name != NULL ; i++) {

1417 struct netdata_static_thread *st = &static_threads[i];

1418

1419 + if(st->enable_routine)

1420 + st->enabled = st->enable_routine();

1421 +

1422 if(st->config_name)

1423 st->enabled = config_get_boolean(st->config_section, st->config_name, st->enabled);

1424

@@ -1485,6 +1437,9 @@ int netdata_main(int argc, char **argv) {

1437

1438 delta_startup_time("initialize web server");

1439

1440 + // get the certificate and start security

1441 + netdata_conf_web_security_init();

1442 +

1443 nd_web_api_init();

1444 web_server_threading_selection();

1445

@@ -1502,14 +1457,6 @@ int netdata_main(int argc, char **argv) {

1457

1458 delta_startup_time("initialize ML");

1459 ml_init();

1505 -

1506 -#ifdef ENABLE_H2O

1507 - delta_startup_time("initialize h2o server");

1508 - for (int t = 0; static_threads[t].name; t++) {

1509 - if (static_threads[t].start_routine == h2o_main)

1510 - static_threads[t].enabled = httpd_is_enabled();

1511 - }

1512 -#endif

1460 }

1461

1462 delta_startup_time("set resource limits");

@@ -1632,6 +1579,7 @@ int netdata_main(int argc, char **argv) {

1579 delta_startup_time("start the static threads");

1580

1581 netdata_conf_section_web();

1582 + backfill_threads_detect_from_stream_conf();

1583

1584 set_late_analytics_variables(system_info);

1585 for (i = 0; static_threads[i].name != NULL ; i++) {

src/daemon/pulse/pulse-workers.c

@@ -142,6 +142,7 @@ static struct worker_utilization all_workers_utilization[] = {

142 { .name = "SERVICE", .family = "workers service", .priority = 1000000 },

143 { .name = "PROFILER", .family = "workers profile", .priority = 1000000 },

144 { .name = "PGCEVICT", .family = "workers dbengine eviction", .priority = 1000000 },

145 + { .name = "BACKFILL", .family = "workers backfill", .priority = 1000000 },

146

147 // has to be terminated with a NULL

148 { .name = NULL, .family = NULL }

src/daemon/static_threads.c

+12

@@ -1,6 +1,7 @@

1 // SPDX-License-Identifier: GPL-3.0-or-later

3 #include "common.h"

4 +#include "web/api/queries/backfill.h"

6 void *aclk_main(void *ptr);

7 void *analytics_main(void *ptr);

@@ -123,6 +124,7 @@ const struct netdata_static_thread static_threads_common[] = {

124 .name = "h2o",

125 .config_section = NULL,

126 .config_name = NULL,

127 + .enable_routine = httpd_is_enabled,

128 .enabled = 0,

129 .thread = NULL,

130 .init_routine = NULL,

@@ -168,6 +170,16 @@ const struct netdata_static_thread static_threads_common[] = {

170 .init_routine = NULL,

171 .start_routine = profile_main

172 },

173 + {

174 + .name = "BACKFILL",

175 + .config_section = NULL,

176 + .config_name = NULL,

177 + .enable_routine = backfill_threads_detect_from_stream_conf,

178 + .enabled = 0,

179 + .thread = NULL,

180 + .init_routine = NULL,

181 + .start_routine = backfill_thread

182 + },

183

184 // terminator

185 {

src/database/rrd.h

+3 -2

@@ -278,7 +278,7 @@ struct rrddim_tier {

278 STORAGE_COLLECT_HANDLE *sch; // the data collection handle

279 };

280

281 -void backfill_tier_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now_s);

281 +bool backfill_tier_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now_s);

282

283 // ----------------------------------------------------------------------------

284 // RRD DIMENSION - this is a metric

@@ -360,7 +360,7 @@ struct rrddim {

360

361 // ------------------------------------------------------------------------

362

363 - struct rrddim_tier tiers[]; // our tiers of databases

363 + struct rrddim_tier tiers[]; // our tiers of databases

364 };

365

366 size_t rrddim_size(void);

@@ -1238,6 +1238,7 @@ struct rrdhost {

1238

1239 struct {

1240 pid_t tid;

1241 + uint32_t state_id; // every time the receiver connects/disconnects, this is incremented

1242

1243 time_t last_connected; // the time the last sender was connected

1244 time_t last_disconnected; // the time the last sender was disconnected

src/database/rrdhost.c

-2

@@ -779,8 +779,6 @@ int rrd_init(const char *hostname, struct rrdhost_system_info *system_info, bool

779 dbengine_enabled = true;

780 }

781 else {

782 - stream_conf_init();

783 -

782 if (default_rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE || stream_conf_receiver_needs_dbengine()) {

783 nd_log(NDLS_DAEMON, NDLP_DEBUG,

784 "DBENGINE: Initializing ...");

src/libnetdata/locks/spinlock.h

@@ -29,7 +29,11 @@ typedef struct netdata_spinlock

29 #define spinlock_trylock(spinlock) (netdata_mutex_trylock(&((spinlock)->inner)) == 0)

30 #define spinlock_init(spinlock) netdata_mutex_init(&((spinlock)->inner)

31 #else

32 +#ifdef NETDATA_INTERNAL_CHECKS

33 +#define SPINLOCK_INITIALIZER { .locked = false, .locker_pid = 0, .spins = 0 }

34 +#else

35 #define SPINLOCK_INITIALIZER { .locked = false }

36 +#endif

38 void spinlock_init_with_trace(SPINLOCK *spinlock, const char *func);

39 #define spinlock_init(spinlock) spinlock_init_with_trace(spinlock, __FUNCTION__)

src/libnetdata/log/nd_log.h

+1 -1

@@ -163,7 +163,7 @@ typedef struct error_with_limit {

163 usec_t sleep_ut;

164 } ERROR_LIMIT;

165

166 -#define nd_log_limit_static_global_var(var, log_every_secs, sleep_usecs) static ERROR_LIMIT var = { .last_logged = 0, .count = 0, .log_every = (log_every_secs), .sleep_ut = (sleep_usecs) }

166 +#define nd_log_limit_static_global_var(var, log_every_secs, sleep_usecs) static ERROR_LIMIT var = { .spinlock = SPINLOCK_INITIALIZER, .log_every = (log_every_secs), .count = 0, .last_logged = 0, .sleep_ut = (sleep_usecs) }

167 #define nd_log_limit_static_thread_var(var, log_every_secs, sleep_usecs) static __thread ERROR_LIMIT var = { .last_logged = 0, .count = 0, .log_every = (log_every_secs), .sleep_ut = (sleep_usecs) }

168 void netdata_logger_with_limit(ERROR_LIMIT *erl, ND_LOG_SOURCES source, ND_LOG_FIELD_PRIORITY priority, const char *file, const char *function, unsigned long line, const char *fmt, ... ) PRINTFLIKE(7, 8);

169 #define nd_log_limit(erl, NDLS, NDLP, args...) netdata_logger_with_limit(erl, NDLS, NDLP, __FILE__, __FUNCTION__, __LINE__, ##args)

src/libnetdata/threads/threads.c

+9 -15

@@ -50,7 +50,7 @@ static struct {

50 ND_THREAD *list;

51 } running;

53 - pthread_attr_t *attr;

53 + pthread_attr_t attr;

54 } threads_globals = {

55 .exited = {

56 .spinlock = SPINLOCK_INITIALIZER,

@@ -60,7 +60,6 @@ static struct {

60 .spinlock = SPINLOCK_INITIALIZER,

61 .list = NULL,

62 },

63 - .attr = NULL,

63 };

65 static __thread ND_THREAD *_nd_thread_info = NULL;

@@ -186,20 +185,15 @@ void nd_thread_rwspinlock_write_unlocked(void) { if(_nd_thread_info) _nd_thread_

185 // early initialization

186

187 size_t netdata_threads_init(void) {

189 - int i;

188 + memset(&threads_globals.attr, 0, sizeof(threads_globals.attr));

189

191 - if(!threads_globals.attr) {

192 - threads_globals.attr = callocz(1, sizeof(pthread_attr_t));

193 - i = pthread_attr_init(threads_globals.attr);

194 - if (i != 0)

195 - fatal("pthread_attr_init() failed with code %d.", i);

196 - }

190 + if(pthread_attr_init(&threads_globals.attr) != 0)

191 + fatal("pthread_attr_init() failed.");

192

193 // get the required stack size of the threads of netdata

194 size_t stacksize = 0;

200 - i = pthread_attr_getstacksize(threads_globals.attr, &stacksize);

201 - if(i != 0)

202 - fatal("pthread_attr_getstacksize() failed with code %d.", i);

195 + if(pthread_attr_getstacksize(&threads_globals.attr, &stacksize) != 0)

196 + fatal("pthread_attr_getstacksize() failed with code.");

197

198 return stacksize;

199 }

@@ -211,8 +205,8 @@ void netdata_threads_init_after_fork(size_t stacksize) {

205 int i;

206

207 // set pthread stack size

214 - if(threads_globals.attr && stacksize > (size_t)PTHREAD_STACK_MIN) {

215 - i = pthread_attr_setstacksize(threads_globals.attr, stacksize);

208 + if(stacksize > (size_t)PTHREAD_STACK_MIN) {

209 + i = pthread_attr_setstacksize(&threads_globals.attr, stacksize);

210 if(i != 0)

211 nd_log(NDLS_DAEMON, NDLP_WARNING, "pthread_attr_setstacksize() to %zu bytes, failed with code %d.", stacksize, i);

212 else

@@ -371,7 +365,7 @@ ND_THREAD *nd_thread_create(const char *tag, NETDATA_THREAD_OPTIONS options, voi

365 DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(threads_globals.running.list, nti, prev, next);

366 spinlock_unlock(&threads_globals.running.spinlock);

367

374 - int ret = pthread_create(&nti->thread, threads_globals.attr, nd_thread_starting_point, nti);

368 + int ret = pthread_create(&nti->thread, &threads_globals.attr, nd_thread_starting_point, nti);

369 if(ret != 0) {

370 nd_log(NDLS_DAEMON, NDLP_ERR,

371 "failed to create new thread for %s. pthread_create() failed with code %d",

src/libnetdata/threads/threads.h

@@ -41,6 +41,9 @@ struct netdata_static_thread {

41 // internal use, to maintain a pointer to the created thread

42 ND_THREAD *thread;

44 + // a function to call to check it should be enabled or not

45 + bool (*enable_routine) (void);

46 +

47 // an initialization function to run before spawning the thread

48 void (*init_routine) (void);

src/ml/ml.cc

+12 -6

@@ -532,8 +532,10 @@ ml_dimension_deserialize_kmeans(const char *json_str)

532

533 AcquiredDimension AcqDim(DLI);

534 if (!AcqDim.acquired()) {

535 - netdata_log_error("Failed to deserialize kmeans: could not acquire dimension (machine-guid: %s, dimension: '%s.%s', reason: %s)",

536 - DLI.machineGuid(), DLI.chartId(), DLI.dimensionId(), AcqDim.acquire_failure());

535 + nd_log_limit_static_global_var(erl, 10, 0);

536 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,

537 + "ML: Failed to deserialize kmeans: could not acquire dimension (machine-guid: %s, dimension: '%s.%s', reason: %s)",

538 + DLI.machineGuid(), DLI.chartId(), DLI.dimensionId(), AcqDim.acquire_failure());

539 json_object_put(root);

540 return false;

541 }

@@ -1039,8 +1041,10 @@ static enum ml_worker_result ml_worker_create_new_model(ml_worker_t *worker, ml_

1041 AcquiredDimension AcqDim(req.DLI);

1042

1043 if (!AcqDim.acquired()) {

1042 - netdata_log_error("Failed to create new model: could not acquire dimension (machine-guid: %s, dimension: '%s.%s', reason: %s)",

1043 - req.DLI.machineGuid(), req.DLI.chartId(), req.DLI.dimensionId(), AcqDim.acquire_failure());

1044 + nd_log_limit_static_global_var(erl, 10, 0);

1045 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,

1046 + "ML: Failed to create new model: could not acquire dimension (machine-guid: %s, dimension: '%s.%s', reason: %s)",

1047 + req.DLI.machineGuid(), req.DLI.chartId(), req.DLI.dimensionId(), AcqDim.acquire_failure());

1048 return ML_WORKER_RESULT_NULL_ACQUIRED_DIMENSION;

1049 }

1050

@@ -1055,8 +1059,10 @@ static enum ml_worker_result ml_worker_add_existing_model(ml_worker_t *worker, m

1059 AcquiredDimension AcqDim(req.DLI);

1060

1061 if (!AcqDim.acquired()) {

1058 - netdata_log_error("Failed to add existing model: could not acquire dimension (machine-guid: %s, dimension: '%s.%s', reason: %s)",

1059 - req.DLI.machineGuid(), req.DLI.chartId(), req.DLI.dimensionId(), AcqDim.acquire_failure());

1062 + nd_log_limit_static_global_var(erl, 10, 0);

1063 + nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,

1064 + "ML: Failed to add existing model: could not acquire dimension (machine-guid: %s, dimension: '%s.%s', reason: %s)",

1065 + req.DLI.machineGuid(), req.DLI.chartId(), req.DLI.dimensionId(), AcqDim.acquire_failure());

1066 return ML_WORKER_RESULT_NULL_ACQUIRED_DIMENSION;

1067 }

1068

src/plugins.d/pluginsd_parser.c

+31 -3

@@ -2,6 +2,8 @@

3 #include "pluginsd_internals.h"

4 #include "streaming/replication.h"

5 +#include "streaming/stream-waiting-list.h"

6 +#include "web/api/queries/backfill.h"

8 static inline PARSER_RC pluginsd_set(char **words, size_t num_words, PARSER *parser) {

9 int idx = 1;

@@ -373,6 +375,19 @@ static inline PARSER_RC pluginsd_chart(char **words, size_t num_words, PARSER *p

375 return PARSER_RC_OK;

376 }

377

378 +static void backfill_callback(size_t successful_dims __maybe_unused, size_t failed_dims __maybe_unused, struct backfill_request_data *brd) {

379 + if (brd->rrdhost_receiver_state_id == __atomic_load_n(&brd->host->stream.rcv.status.state_id, __ATOMIC_RELAXED)) {

380 + if (!replicate_chart_request(send_to_plugin, brd->parser, brd->host, brd->st,

381 + brd->first_entry_child, brd->last_entry_child, brd->child_wall_clock_time,

382 + 0, 0)) {

383 + netdata_log_error(

384 + "PLUGINSD: 'host:%s' failed to initiate replication for 'chart:%s'",

385 + rrdhost_hostname(brd->host),

386 + rrdset_id(brd->st));

387 + }

388 + }

389 +}

390 +

391 static inline PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_words, PARSER *parser) {

392 const char *first_entry_txt = get_word(words, num_words, 1);

393 const char *last_entry_txt = get_word(words, num_words, 2);

@@ -401,9 +416,20 @@ static inline PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_w

416 rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);

417 rrdhost_receiver_replicating_charts_plus_one(st->rrdhost);

418

404 - ok = replicate_chart_request(send_to_plugin, parser, host, st,

405 - first_entry_child, last_entry_child, child_wall_clock_time,

406 - 0, 0);

419 + struct backfill_request_data brd = {

420 + .rrdhost_receiver_state_id =__atomic_load_n(&host->stream.rcv.status.state_id, __ATOMIC_RELAXED),

421 + .parser = parser,

422 + .host = host,

423 + .st = st,

424 + .first_entry_child = first_entry_child,

425 + .last_entry_child = last_entry_child,

426 + .child_wall_clock_time = child_wall_clock_time,

427 + };

428 +

429 + ok = backfill_request_add(st, backfill_callback, &brd);

430 + if(!ok)

431 + ok = replicate_chart_request(

432 + send_to_plugin, parser, host, st, first_entry_child, last_entry_child, child_wall_clock_time, 0, 0);

433 }

434 #ifdef NETDATA_LOG_REPLICATION_REQUESTS

435 else {

@@ -412,6 +438,8 @@ static inline PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_w

438 }

439 #endif

440

441 + stream_thread_received_metadata();

442 +

443 return ok ? PARSER_RC_OK : PARSER_RC_ERROR;

444 }

445

src/plugins.d/pluginsd_parser.h

+2 -2

@@ -5,10 +5,10 @@

6 #include "daemon/common.h"

8 -#define WORKER_PARSER_FIRST_JOB 34

8 +#define WORKER_PARSER_FIRST_JOB 35

10 // this has to be in-sync with the same at stream-thread.c

11 -#define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION (WORKER_PARSER_FIRST_JOB - 9)

11 +#define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION 25

13 // this controls the max response size of a function

14 #define PLUGINSD_MAX_DEFERRED_SIZE (100 * 1024 * 1024)

src/plugins.d/pluginsd_replication.c

@@ -3,6 +3,7 @@

3 #include "pluginsd_replication.h"

4 #include "streaming/stream-receiver-internals.h"

5 #include "streaming/replication.h"

6 +#include "streaming/stream-waiting-list.h"

8 PARSER_RC pluginsd_replay_begin(char **words, size_t num_words, PARSER *parser) {

9 int idx = 1;

@@ -359,6 +360,8 @@ PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARSER *parser) {

360 host->stream.rcv.status.replication.percent = 100.0;

361 worker_set_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, host->stream.rcv.status.replication.percent);

362

363 + stream_thread_received_replication();

364 +

365 return PARSER_RC_OK;

366 }

367

src/streaming/protocol/command-begin-set-end.c

+1 -1

@@ -30,7 +30,7 @@ stream_send_rrdset_metrics_v1_internal(BUFFER *wb, RRDSET *st, struct sender_sta

30 buffer_fast_strcat(wb, "\n", 1);

31 }

32 else {

33 - internal_error(true, "STREAM SEND '%s': 'chart:%s/dim:%s' flag 'exposed' is updated but not exposed",

33 + internal_error(true, "STREAM SND '%s': 'chart:%s/dim:%s' flag 'exposed' is updated but not exposed",

34 rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_id(rd));

35 // we will include it in the next iteration

36 rrddim_metadata_updated(rd);

src/streaming/protocol/command-nodeid.c

+6 -6

@@ -51,7 +51,7 @@ void stream_sender_get_node_and_claim_id_from_parent(struct sender_state *s) {

51 ND_UUID claim_id;

52 if (uuid_parse(claim_id_str ? claim_id_str : "", claim_id.uuid) != 0) {

53 nd_log(NDLS_DAEMON, NDLP_ERR,

54 - "STREAM SEND '%s' [to %s] received invalid claim id '%s'",

54 + "STREAM SND '%s' [to %s] received invalid claim id '%s'",

55 rrdhost_hostname(s->host), s->connected_to,

56 claim_id_str ? claim_id_str : "(unset)");

57 return;

@@ -60,7 +60,7 @@ void stream_sender_get_node_and_claim_id_from_parent(struct sender_state *s) {

60 ND_UUID node_id;

61 if(uuid_parse(node_id_str ? node_id_str : "", node_id.uuid) != 0) {

62 nd_log(NDLS_DAEMON, NDLP_ERR,

63 - "STREAM SEND '%s' [to %s] received an invalid node id '%s'",

63 + "STREAM SND '%s' [to %s] received an invalid node id '%s'",

64 rrdhost_hostname(s->host), s->connected_to,

65 node_id_str ? node_id_str : "(unset)");

66 return;

@@ -68,14 +68,14 @@ void stream_sender_get_node_and_claim_id_from_parent(struct sender_state *s) {

69 if (!UUIDiszero(s->host->aclk.claim_id_of_parent) && !UUIDeq(s->host->aclk.claim_id_of_parent, claim_id))

70 nd_log(NDLS_DAEMON, NDLP_INFO,

71 - "STREAM SEND '%s' [to %s] changed parent's claim id to %s",

71 + "STREAM SND '%s' [to %s] changed parent's claim id to %s",

72 rrdhost_hostname(s->host), s->connected_to,

73 claim_id_str ? claim_id_str : "(unset)");

75 if(!UUIDiszero(s->host->node_id) && !UUIDeq(s->host->node_id, node_id)) {

76 if(claimed) {

77 nd_log(NDLS_DAEMON, NDLP_WARNING,

78 - "STREAM SEND '%s' [to %s] parent reports different node id '%s', but we are claimed. Ignoring it.",

78 + "STREAM SND '%s' [to %s] parent reports different node id '%s', but we are claimed. Ignoring it.",

79 rrdhost_hostname(s->host), s->connected_to,

80 node_id_str ? node_id_str : "(unset)");

81 return;

@@ -83,7 +83,7 @@ void stream_sender_get_node_and_claim_id_from_parent(struct sender_state *s) {

83 else {

84 update_node_id = true;

85 nd_log(NDLS_DAEMON, NDLP_WARNING,

86 - "STREAM SEND '%s' [to %s] changed node id to %s",

86 + "STREAM SND '%s' [to %s] changed node id to %s",

87 rrdhost_hostname(s->host), s->connected_to,

88 node_id_str ? node_id_str : "(unset)");

89 }

@@ -91,7 +91,7 @@ void stream_sender_get_node_and_claim_id_from_parent(struct sender_state *s) {

92 if(!url || !*url) {

93 nd_log(NDLS_DAEMON, NDLP_ERR,

94 - "STREAM SEND '%s' [to %s] received an invalid cloud URL '%s'",

94 + "STREAM SND '%s' [to %s] received an invalid cloud URL '%s'",

95 rrdhost_hostname(s->host), s->connected_to,

96 url ? url : "(unset)");

97 return;

src/streaming/protocol/commands.c

+2 -2

@@ -31,7 +31,7 @@ RRDSET_STREAM_BUFFER stream_send_metrics_init(RRDSET *st, time_t wall_clock_time

31 // - the parent just disconnected, so local data are not streamed to parent

33 nd_log(NDLS_DAEMON, NDLP_INFO,

34 - "STREAM SEND '%s': streaming is not ready, not sending data to a parent...",

34 + "STREAM SND '%s': streaming is not ready, not sending data to a parent...",

35 rrdhost_hostname(host));

36 }

@@ -39,7 +39,7 @@ RRDSET_STREAM_BUFFER stream_send_metrics_init(RRDSET *st, time_t wall_clock_time

39 }

40 else if(unlikely(host_flags & RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS)) {

41 nd_log(NDLS_DAEMON, NDLP_INFO,

42 - "STREAM SEND '%s': streaming is ready, sending metrics to parent...",

42 + "STREAM SND '%s': streaming is ready, sending metrics to parent...",

43 rrdhost_hostname(host));

44 rrdhost_flag_clear(host, RRDHOST_FLAG_STREAM_SENDER_LOGGED_STATUS);

45 }

src/streaming/replication.c

+21 -21

@@ -155,7 +155,7 @@ static struct replication_query *replication_query_prepare(

155 if (st->last_updated.tv_sec > q->query.before) {

156 #ifdef NETDATA_LOG_REPLICATION_REQUESTS

157 internal_error(true,

158 - "STREAM SEND REPLAY: 'host:%s/chart:%s' "

158 + "STREAM SND REPLAY: 'host:%s/chart:%s' "

159 "has start_streaming = true, "

160 "adjusting replication before timestamp from %llu to %llu",

161 rrdhost_hostname(st->rrdhost), rrdset_id(st),

@@ -178,7 +178,7 @@ static struct replication_query *replication_query_prepare(

178

179 if (unlikely(rd_dfe.counter >= q->dimensions)) {

180 internal_error(true,

181 - "STREAM SEND REPLAY ERROR: 'host:%s/chart:%s' has more dimensions than the replicated ones",

181 + "STREAM SND REPLAY ERROR: 'host:%s/chart:%s' has more dimensions than the replicated ones",

182 rrdhost_hostname(st->rrdhost), rrdset_id(st));

183 break;

184 }

@@ -364,7 +364,7 @@ static bool replication_query_execute(BUFFER *wb, struct replication_query *q, s

364

365 nd_log_limit_static_global_var(erl, 1, 0);

366 nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR,

367 - "STREAM SEND REPLAY: 'host:%s/chart:%s/dim:%s': db does not advance the query "

367 + "STREAM SND REPLAY: 'host:%s/chart:%s/dim:%s': db does not advance the query "

368 "beyond time %llu (tried 1000 times to get the next point and always got back a point in the past)",

369 rrdhost_hostname(q->st->rrdhost), rrdset_id(q->st), rrddim_id(d->rd),

370 (unsigned long long) now);

@@ -414,7 +414,7 @@ static bool replication_query_execute(BUFFER *wb, struct replication_query *q, s

414 #ifdef NETDATA_INTERNAL_CHECKS

415 nd_log_limit_static_global_var(erl, 1, 0);

416 nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,

417 - "STREAM SEND REPLAY WARNING: 'host:%s/chart:%s' misaligned dimensions, "

417 + "STREAM SND REPLAY WARNING: 'host:%s/chart:%s' misaligned dimensions, "

418 "update every (min: %ld, max: %ld), "

419 "start time (min: %ld, max: %ld), "

420 "end time (min %ld, max %ld), "

@@ -450,7 +450,7 @@ static bool replication_query_execute(BUFFER *wb, struct replication_query *q, s

450 q->query.enable_streaming = false;

451

452 internal_error(true,

453 - "STREAM SEND REPLAY: current buffer size %zu is more than the "

453 + "STREAM SND REPLAY: current buffer size %zu is more than the "

454 "max message size %zu for chart '%s' of host '%s'. "

455 "Interrupting replication request (%ld to %ld, %s) at %ld to %ld, %s.",

456 buffer_strlen(wb), max_msg_size, rrdset_id(q->st), rrdhost_hostname(q->st->rrdhost),

@@ -530,14 +530,14 @@ static bool replication_query_execute(BUFFER *wb, struct replication_query *q, s

530 log_date(actual_after_buf, LOG_DATE_LENGTH, actual_after);

531 log_date(actual_before_buf, LOG_DATE_LENGTH, actual_before);

532 internal_error(true,

533 - "STREAM SEND REPLAY: 'host:%s/chart:%s': sending data %llu [%s] to %llu [%s] (requested %llu [delta %lld] to %llu [delta %lld])",

533 + "STREAM SND REPLAY: 'host:%s/chart:%s': sending data %llu [%s] to %llu [%s] (requested %llu [delta %lld] to %llu [delta %lld])",

534 rrdhost_hostname(q->st->rrdhost), rrdset_id(q->st),

535 (unsigned long long)actual_after, actual_after_buf, (unsigned long long)actual_before, actual_before_buf,

536 (unsigned long long)after, (long long)(actual_after - after), (unsigned long long)before, (long long)(actual_before - before));

537 }

538 else

539 internal_error(true,

540 - "STREAM SEND REPLAY: 'host:%s/chart:%s': nothing to send (requested %llu to %llu)",

540 + "STREAM SND REPLAY: 'host:%s/chart:%s': nothing to send (requested %llu to %llu)",

541 rrdhost_hostname(q->st->rrdhost), rrdset_id(q->st),

542 (unsigned long long)after, (unsigned long long)before);

543 #endif // NETDATA_LOG_REPLICATION_REQUESTS

@@ -708,13 +708,13 @@ bool replication_response_execute_and_finalize(struct replication_query *q, size

708 st->stream.snd.resync_time_s = 0;

709

710 #ifdef NETDATA_LOG_REPLICATION_REQUESTS

711 - internal_error(true, "STREAM SEND REPLAY: 'host:%s/chart:%s' streaming starts",

711 + internal_error(true, "STREAM SND REPLAY: 'host:%s/chart:%s' streaming starts",

712 rrdhost_hostname(st->rrdhost), rrdset_id(st));

713 #endif

714 }

715 else

716 internal_error(true,

717 - "STREAM SEND REPLAY ERROR: 'host:%s/chart:%s' "

717 + "STREAM SND REPLAY ERROR: 'host:%s/chart:%s' "

718 "received start streaming command, but the chart is not in progress replicating",

719 rrdhost_hostname(st->rrdhost), rrdset_id(st));

720 }

@@ -775,7 +775,7 @@ static void replicate_log_request(struct replication_request_details *r, const c

775 nd_log_limit_static_global_var(erl, 1, 0);

776 nd_log_limit(&erl, NDLS_DAEMON, NDLP_NOTICE,

777 #endif

778 - "STREAM SEND REPLAY ERROR: 'host:%s/chart:%s' child sent: "

778 + "STREAM SND REPLAY ERROR: 'host:%s/chart:%s' child sent: "

779 "db from %ld to %ld%s, wall clock time %ld, "

780 "last request from %ld to %ld, "

781 "issue: %s - "

@@ -813,7 +813,7 @@ static bool send_replay_chart_cmd(struct replication_request_details *r, const c

813 log_date(wanted_before_buf, LOG_DATE_LENGTH, r->wanted.before);

814

815 internal_error(true,

816 - "STREAM SEND REPLAY: 'host:%s/chart:%s' sending replication request %ld [%s] to %ld [%s], start streaming '%s': %s: "

816 + "STREAM SND REPLAY: 'host:%s/chart:%s' sending replication request %ld [%s] to %ld [%s], start streaming '%s': %s: "

817 "last[%ld - %ld] child[%ld - %ld, now %ld %s] local[%ld - %ld, now %ld] gap[%ld - %ld %s] %s"

818 , rrdhost_hostname(r->host), rrdset_id(r->st)

819 , r->wanted.after, wanted_after_buf

@@ -842,7 +842,7 @@ static bool send_replay_chart_cmd(struct replication_request_details *r, const c

842

843 ssize_t ret = r->caller.callback(buffer, r->caller.parser, STREAM_TRAFFIC_TYPE_REPLICATION);

844 if (ret < 0) {

845 - netdata_log_error("STREAM SEND REPLAY ERROR: 'host:%s/chart:%s' failed to send replication request to child (error %zd)",

845 + netdata_log_error("STREAM SND REPLAY ERROR: 'host:%s/chart:%s' failed to send replication request to child (error %zd)",

846 rrdhost_hostname(r->host), rrdset_id(r->st), ret);

847 return false;

848 }

@@ -1281,7 +1281,7 @@ static void replication_sort_entry_del(struct replication_request *rq, bool buff

1281 }

1282

1283 if (!rse_to_delete)

1284 - fatal("STREAM SEND REPLAY: 'host:%s/chart:%s' Cannot find sort entry to delete for time %ld.",

1284 + fatal("STREAM SND REPLAY: 'host:%s/chart:%s' Cannot find sort entry to delete for time %ld.",

1285 rrdhost_hostname(rq->sender->host), string2str(rq->chart_id), rq->after);

1286

1287 }

@@ -1384,7 +1384,7 @@ static bool replication_request_conflict_callback(const DICTIONARY_ITEM *item __

1384 // we can replace this command

1385 internal_error(

1386 true,

1387 - "STREAM SEND '%s' [to %s]: REPLAY: 'host:%s/chart:%s' replacing duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",

1387 + "STREAM SND '%s' [to %s]: REPLAY: 'host:%s/chart:%s' replacing duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",

1388 rrdhost_hostname(s->host), s->connected_to, rrdhost_hostname(s->host), dictionary_acquired_item_name(item),

1389 (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false",

1390 (unsigned long long)rq_new->after, (unsigned long long)rq_new->before, rq_new->start_streaming ? "true" : "false");

@@ -1397,7 +1397,7 @@ static bool replication_request_conflict_callback(const DICTIONARY_ITEM *item __

1397 replication_sort_entry_add(rq);

1398 internal_error(

1399 true,

1400 - "STREAM SEND '%s' [to %s]: REPLAY: 'host:%s/chart:%s' adding duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",

1400 + "STREAM SND '%s' [to %s]: REPLAY: 'host:%s/chart:%s' adding duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",

1401 rrdhost_hostname(s->host), s->connected_to, rrdhost_hostname(s->host), dictionary_acquired_item_name(item),

1402 (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false",

1403 (unsigned long long)rq_new->after, (unsigned long long)rq_new->before, rq_new->start_streaming ? "true" : "false");

@@ -1405,7 +1405,7 @@ static bool replication_request_conflict_callback(const DICTIONARY_ITEM *item __

1405 else {

1406 internal_error(

1407 true,

1408 - "STREAM SEND '%s' [to %s]: REPLAY: 'host:%s/chart:%s' ignoring duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",

1408 + "STREAM SND '%s' [to %s]: REPLAY: 'host:%s/chart:%s' ignoring duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",

1409 rrdhost_hostname(s->host), s->connected_to, rrdhost_hostname(s->host),

1410 dictionary_acquired_item_name(item),

1411 (unsigned long long) rq->after, (unsigned long long) rq->before, rq->start_streaming ? "true" : "false",

@@ -1449,7 +1449,7 @@ static bool replication_execute_request(struct replication_request *rq, bool wor

1449 }

1450

1451 if(!rq->st) {

1452 - internal_error(true, "STREAM SEND REPLAY ERROR: 'host:%s/chart:%s' not found",

1452 + internal_error(true, "STREAM SND REPLAY ERROR: 'host:%s/chart:%s' not found",

1453 rrdhost_hostname(rq->sender->host), string2str(rq->chart_id));

1454

1455 goto cleanup;

@@ -1577,7 +1577,7 @@ static size_t verify_host_charts_are_streaming_now(RRDHOST *host) {

1577 host->sender &&

1578 !stream_sender_pending_replication_requests(host->sender) &&

1579 dictionary_entries(host->sender->replication.requests) != 0,

1580 - "STREAM SEND REPLAY SUMMARY: 'host:%s' reports %zu pending replication requests, "

1580 + "STREAM SND REPLAY SUMMARY: 'host:%s' reports %zu pending replication requests, "

1581 "but its chart replication index says there are %zu charts pending replication",

1582 rrdhost_hostname(host),

1583 stream_sender_pending_replication_requests(host->sender),

@@ -1596,7 +1596,7 @@ static size_t verify_host_charts_are_streaming_now(RRDHOST *host) {

1596 if(!flags) {

1597 internal_error(

1598 true,

1599 - "STREAM SEND REPLAY SUMMARY: 'host:%s/chart:%s' is neither IN PROGRESS nor FINISHED",

1599 + "STREAM SND REPLAY SUMMARY: 'host:%s/chart:%s' is neither IN PROGRESS nor FINISHED",

1600 rrdhost_hostname(host), rrdset_id(st)

1601 );

1602 is_error = true;

@@ -1605,7 +1605,7 @@ static size_t verify_host_charts_are_streaming_now(RRDHOST *host) {

1605 if(!(flags & RRDSET_FLAG_SENDER_REPLICATION_FINISHED) || (flags & RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS)) {

1606 internal_error(

1607 true,

1608 - "STREAM SEND REPLAY SUMMARY: 'host:%s/chart:%s' is IN PROGRESS although replication is finished",

1608 + "STREAM SND REPLAY SUMMARY: 'host:%s/chart:%s' is IN PROGRESS although replication is finished",

1609 rrdhost_hostname(host), rrdset_id(st)

1610 );

1611 is_error = true;

@@ -1619,7 +1619,7 @@ static size_t verify_host_charts_are_streaming_now(RRDHOST *host) {

1619 rrdset_foreach_done(st);

1620

1621 internal_error(errors,

1622 - "STREAM SEND REPLAY SUMMARY: 'host:%s' finished replicating %zu charts, but %zu charts are still in progress although replication finished",

1622 + "STREAM SND REPLAY SUMMARY: 'host:%s' finished replicating %zu charts, but %zu charts are still in progress although replication finished",

1623 rrdhost_hostname(host), ok, errors);

1624

1625 return errors;

src/streaming/stream-capabilities.c

+2 -2

@@ -80,7 +80,7 @@ void log_receiver_capabilities(struct receiver_state *rpt) {

80 BUFFER *wb = buffer_create(100, NULL);

81 stream_capabilities_to_string(wb, rpt->capabilities);

83 - nd_log_daemon(NDLP_INFO, "STREAM RECEIVE '%s' [from [%s]:%s]: established link with negotiated capabilities: %s",

83 + nd_log_daemon(NDLP_INFO, "STREAM RCV '%s' [from [%s]:%s]: established link with negotiated capabilities: %s",

84 rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, buffer_tostring(wb));

86 buffer_free(wb);

@@ -90,7 +90,7 @@ void log_sender_capabilities(struct sender_state *s) {

90 BUFFER *wb = buffer_create(100, NULL);

91 stream_capabilities_to_string(wb, s->capabilities);

93 - nd_log_daemon(NDLP_INFO, "STREAM SEND '%s' [to %s]: established link with negotiated capabilities: %s",

93 + nd_log_daemon(NDLP_INFO, "STREAM SND '%s' [to %s]: established link with negotiated capabilities: %s",

94 rrdhost_hostname(s->host), s->connected_to, buffer_tostring(wb));

96 buffer_free(wb);

src/streaming/stream-conf.c

+8 -6

@@ -44,7 +44,7 @@ struct _stream_receive stream_receive = {

44 }

45 };

47 -static void stream_conf_load() {

47 +static void stream_conf_load_internal() {

48 errno_clear();

49 char *filename = filename_from_path_entry_strdupz(netdata_configured_user_config_dir, "stream.conf");

50 if(!appconfig_load(&stream_config, filename, 0, NULL)) {

@@ -88,8 +88,12 @@ bool stream_conf_receiver_needs_dbengine(void) {

88 return stream_conf_needs_dbengine(&stream_config);

89 }

91 -bool stream_conf_init() {

92 - stream_conf_load();

91 +void stream_conf_load() {

92 + static bool loaded = false;

93 + if(loaded) return;

94 + loaded = true;

95 +

96 + stream_conf_load_internal();

98 stream_send.enabled =

99 appconfig_get_boolean(&stream_config, CONFIG_SECTION_STREAM, "enabled", stream_send.enabled);

@@ -179,8 +183,6 @@ bool stream_conf_init() {

183 nd_log_daemon(NDLP_ERR, "STREAM [send]: cannot enable sending thread - information is missing.");

184 stream_send.enabled = false;

185 }

182 -

183 - return stream_send.enabled;

186 }

187

188 bool stream_conf_configured_as_parent() {

@@ -194,7 +196,7 @@ void stream_conf_receiver_config(struct receiver_state *rpt, struct stream_recei

196 rrd_memory_mode_name(default_rrd_memory_mode))));

197

198 if (unlikely(config->mode == RRD_MEMORY_MODE_DBENGINE && !dbengine_enabled)) {

197 - netdata_log_error("STREAM RECEIVE '%s' [from [%s]:%s]: "

199 + netdata_log_error("STREAM RCV '%s' [from [%s]:%s]: "

200 "dbengine is not enabled, falling back to default."

201 , rpt->hostname

202 , rpt->client_ip, rpt->client_port

src/streaming/stream-conf.h

+1 -1

@@ -83,7 +83,7 @@ struct stream_receiver_config {

84 void stream_conf_receiver_config(struct receiver_state *rpt, struct stream_receiver_config *config, const char *api_key, const char *machine_guid);

86 -bool stream_conf_init();

86 +void stream_conf_load();

87 bool stream_conf_receiver_needs_dbengine();

88 bool stream_conf_configured_as_parent();

src/streaming/stream-connector.c

+1 -1

@@ -204,7 +204,7 @@ static int stream_connect_upgrade_prelude(RRDHOST *host __maybe_unused, struct s

204 goto err_cleanup;

205 }

206

207 - netdata_log_debug(D_STREAM, "Stream sender upgrade to \"" NETDATA_STREAM_PROTO_NAME "\" successful");

207 + netdata_log_debug(D_STREAM, "STREAM SNDer upgrade to \"" NETDATA_STREAM_PROTO_NAME "\" successful");

208 rbuf_free(buf);

209 http_parse_ctx_destroy(&ctx);

210 return 0;

src/streaming/stream-receiver-connection.c

+15 -15

@@ -26,7 +26,7 @@ void stream_receiver_log_status(struct receiver_state *rpt, const char *msg, con

26 , (rpt->machine_guid && *rpt->machine_guid) ? rpt->machine_guid : ""

27 , msg);

29 - nd_log(NDLS_DAEMON, priority, "STREAM RECEIVE '%s' [from [%s]:%s]: %s %s%s%s"

29 + nd_log(NDLS_DAEMON, priority, "STREAM RCV '%s' [from [%s]:%s]: %s %s%s%s"

30 , (rpt->hostname && *rpt->hostname) ? rpt->hostname : ""

31 , rpt->client_ip, rpt->client_port

32 , msg

@@ -165,15 +165,15 @@ static bool stream_receiver_send_first_response(struct receiver_state *rpt) {

165 return false;

166 }

167

168 - if (unlikely(!stream_control_children_should_be_accepted())) {

169 - stream_receiver_log_status(

170 - rpt,

171 - "rejecting streaming connection; the system is backfilling higher tiers with high-resolution data, retry later",

172 - STREAM_STATUS_INITIALIZATION_IN_PROGRESS, NDLP_NOTICE);

173 -

174 - stream_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_INITIALIZATION);

175 - return false;

176 - }

168 +// if (unlikely(!stream_control_children_should_be_accepted())) {

169 +// stream_receiver_log_status(

170 +// rpt,

171 +// "rejecting streaming connection; the system is backfilling higher tiers with high-resolution data, retry later",

172 +// STREAM_STATUS_INITIALIZATION_IN_PROGRESS, NDLP_NOTICE);

173 +//

174 +// stream_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_INITIALIZATION);

175 +// return false;

176 +// }

177

178 if(!rrdhost_set_receiver(host, rpt)) {

179 stream_receiver_log_status(

@@ -187,7 +187,7 @@ static bool stream_receiver_send_first_response(struct receiver_state *rpt) {

187 }

188

189 #ifdef NETDATA_INTERNAL_CHECKS

190 - netdata_log_info("STREAM RECEIVE '%s' [from [%s]:%s]: "

190 + netdata_log_info("STREAM RCV '%s' [from [%s]:%s]: "

191 "client willing to stream metrics for host '%s' with machine_guid '%s': "

192 "update every = %d, history = %d, memory mode = %s, health %s,%s"

193 , rpt->hostname

@@ -235,7 +235,7 @@ static bool stream_receiver_send_first_response(struct receiver_state *rpt) {

235 // remove the non-blocking flag from the socket

236 if(sock_delnonblock(rpt->sock.fd) < 0)

237 nd_log(NDLS_DAEMON, NDLP_ERR,

238 - "STREAM RECEIVE '%s' [from [%s]:%s]: cannot remove the non-blocking flag from socket %d",

238 + "STREAM RCV '%s' [from [%s]:%s]: cannot remove the non-blocking flag from socket %d",

239 rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->sock.fd);

240

241 struct timeval timeout;

@@ -243,7 +243,7 @@ static bool stream_receiver_send_first_response(struct receiver_state *rpt) {

243 timeout.tv_usec = 0;

244 if (unlikely(setsockopt(rpt->sock.fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof timeout) != 0))

245 nd_log(NDLS_DAEMON, NDLP_ERR,

246 - "STREAM RECEIVE '%s' [from [%s]:%s]: cannot set timeout for socket %d",

246 + "STREAM RCV '%s' [from [%s]:%s]: cannot set timeout for socket %d",

247 rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->sock.fd);

248 }

249

@@ -378,7 +378,7 @@ int stream_receiver_accept_connection(struct web_client *w, char *decoded_query_

378 rpt->capabilities = convert_stream_version_to_capabilities(1, NULL, false);

379

380 if (unlikely(rrdhost_set_system_info_variable(rpt->system_info, name, value))) {

381 - nd_log_daemon(NDLP_NOTICE, "STREAM RECEIVE '%s' [from [%s]:%s]: "

381 + nd_log_daemon(NDLP_NOTICE, "STREAM RCV '%s' [from [%s]:%s]: "

382 "request has parameter '%s' = '%s', which is not used."

383 , (rpt->hostname && *rpt->hostname) ? rpt->hostname : "-"

384 , rpt->client_ip, rpt->client_port

@@ -540,7 +540,7 @@ int stream_receiver_accept_connection(struct web_client *w, char *decoded_query_

540 if(nd_sock_send_timeout(&rpt->sock, initial_response, strlen(initial_response), 0, 60) !=

541 (ssize_t)strlen(initial_response)) {

542

543 - nd_log_daemon(NDLP_ERR, "STREAM RECEIVE '%s' [from [%s]:%s]: failed to reply.",

543 + nd_log_daemon(NDLP_ERR, "STREAM RCV '%s' [from [%s]:%s]: failed to reply.",

544 rpt->hostname, rpt->client_ip, rpt->client_port

545 );

546 }

src/streaming/stream-receiver.c

+62 -54

@@ -148,14 +148,14 @@ static inline decompressor_status_t receiver_feed_decompressor(struct receiver_s

148

149 if (unlikely(!compressed_message_size)) {

150 nd_log(NDLS_DAEMON, NDLP_ERR,

151 - "STREAM RECEIVE[x] '%s' [from [%s]:%s]: multiplexed uncompressed data in compressed stream!",

151 + "STREAM RCV[x] '%s' [from [%s]:%s]: multiplexed uncompressed data in compressed stream!",

152 rrdhost_hostname(r->host), r->client_ip, r->client_port);

153 return DECOMPRESS_FAILED;

154 }

155

156 if(unlikely(compressed_message_size > COMPRESSION_MAX_MSG_SIZE)) {

157 nd_log(NDLS_DAEMON, NDLP_ERR,

158 - "STREAM RECEIVE[x] '%s' [from [%s]:%s]: received a compressed message of %zu bytes, "

158 + "STREAM RCV[x] '%s' [from [%s]:%s]: received a compressed message of %zu bytes, "

159 "which is bigger than the max compressed message "

160 "size supported of %zu. Ignoring message.",

161 rrdhost_hostname(r->host), r->client_ip, r->client_port,

@@ -174,7 +174,7 @@ static inline decompressor_status_t receiver_feed_decompressor(struct receiver_s

174

175 if (unlikely(!bytes_to_parse)) {

176 nd_log(NDLS_DAEMON, NDLP_ERR,

177 - "STREAM RECEIVE[x] '%s' [from [%s]:%s]: no bytes to decompress.",

177 + "STREAM RCV[x] '%s' [from [%s]:%s]: no bytes to decompress.",

178 rrdhost_hostname(r->host), r->client_ip, r->client_port);

179 return DECOMPRESS_FAILED;

180 }

@@ -265,7 +265,7 @@ void stream_receiver_handle_op(struct stream_thread *sth, struct receiver_state

265 STREAM_CIRCULAR_BUFFER_STATS stats = *stream_circular_buffer_stats_unsafe(rpt->thread.send_to_child.scb);

266 spinlock_unlock(&rpt->thread.send_to_child.spinlock);

267 nd_log(NDLS_DAEMON, NDLP_ERR,

268 - "STREAM RECEIVE[%zu] '%s' [from [%s]:%s]: send buffer is full (buffer size %u, max %u, used %u, available %u). "

268 + "STREAM RCV[%zu] '%s' [from [%s]:%s]: send buffer is full (buffer size %u, max %u, used %u, available %u). "

269 "Restarting connection.",

270 sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port,

271 stats.bytes_size, stats.bytes_max_size, stats.bytes_outstanding, stats.bytes_available);

@@ -275,7 +275,7 @@ void stream_receiver_handle_op(struct stream_thread *sth, struct receiver_state

275 }

276

277 nd_log(NDLS_DAEMON, NDLP_ERR,

278 - "STREAM RECEIVE[%zu]: invalid msg id %u", sth->id, (unsigned)msg->opcode);

278 + "STREAM RCV[%zu]: invalid msg id %u", sth->id, (unsigned)msg->opcode);

279 }

280

281 static ssize_t send_to_child(const char *txt, void *data, STREAM_TRAFFIC_TYPE type) {

@@ -400,62 +400,67 @@ static void stream_receive_log_database_gap(struct receiver_state *rpt) {

400 char buf[128];

401 duration_snprintf(buf, sizeof(buf), now - last_db_entry, "s", true);

402 nd_log(NDLS_DAEMON, NDLP_NOTICE,

403 - "STREAM RECEIVE '%s' [from [%s]:%s]: node connected; last sample in the database %s ago",

403 + "STREAM RCV '%s' [from [%s]:%s]: node connected; last sample in the database %s ago",

404 rrdhost_hostname(host), rpt->client_ip, rpt->client_port, buf);

405 }

406

407 -void stream_receiver_move_queue_to_running_unsafe(struct stream_thread *sth) {

407 +void stream_receiver_move_to_running_unsafe(struct stream_thread *sth, struct receiver_state *rpt) {

408 internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );

409

410 - // process the queue

411 - Word_t idx = 0;

412 - for(struct receiver_state *rpt = RECEIVERS_FIRST(&sth->queue.receivers, &idx);

413 - rpt;

414 - rpt = RECEIVERS_NEXT(&sth->queue.receivers, &idx)) {

415 - worker_is_busy(WORKER_STREAM_JOB_DEQUEUE);

410 + worker_is_busy(WORKER_STREAM_JOB_DEQUEUE);

411

417 - RECEIVERS_DEL(&sth->queue.receivers, (Word_t)rpt);

412 + ND_LOG_STACK lgs[] = {

413 + ND_LOG_FIELD_STR(NDF_NIDL_NODE, rpt->host->hostname),

414 + ND_LOG_FIELD_UUID(NDF_MESSAGE_ID, &streaming_to_parent_msgid),

415 + ND_LOG_FIELD_END(),

416 + };

417 + ND_LOG_STACK_PUSH(lgs);

418

419 - ND_LOG_STACK lgs[] = {

420 - ND_LOG_FIELD_STR(NDF_NIDL_NODE, rpt->host->hostname),

421 - ND_LOG_FIELD_UUID(NDF_MESSAGE_ID, &streaming_to_parent_msgid),

422 - ND_LOG_FIELD_END(),

423 - };

424 - ND_LOG_STACK_PUSH(lgs);

419 + nd_log(NDLS_DAEMON, NDLP_DEBUG,

420 + "STREAM RCV[%zu] '%s' [from [%s]:%s]: moving host from receiver queue to receiver running...",

421 + sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);

422

426 - nd_log(NDLS_DAEMON, NDLP_DEBUG,

427 - "STREAM RECEIVE[%zu] '%s' [from [%s]:%s]: moving host from receiver queue to receiver running...",

428 - sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);

423 + rpt->host->stream.rcv.status.tid = gettid_cached();

424 + rpt->thread.meta.type = POLLFD_TYPE_RECEIVER;

425 + rpt->thread.meta.rpt = rpt;

426

430 - rpt->host->stream.rcv.status.tid = gettid_cached();

431 - rpt->thread.meta.type = POLLFD_TYPE_RECEIVER;

432 - rpt->thread.meta.rpt = rpt;

427 + spinlock_lock(&rpt->thread.send_to_child.spinlock);

428 + rpt->thread.send_to_child.scb = stream_circular_buffer_create();

429 + rpt->thread.send_to_child.msg.thread_slot = (int32_t)sth->id;

430 + rpt->thread.send_to_child.msg.session = os_random32();

431 + rpt->thread.send_to_child.msg.meta = &rpt->thread.meta;

432 + spinlock_unlock(&rpt->thread.send_to_child.spinlock);

433

434 - spinlock_lock(&rpt->thread.send_to_child.spinlock);

435 - rpt->thread.send_to_child.scb = stream_circular_buffer_create();

436 - rpt->thread.send_to_child.msg.thread_slot = (int32_t)sth->id;

437 - rpt->thread.send_to_child.msg.session = os_random32();

438 - rpt->thread.send_to_child.msg.meta = &rpt->thread.meta;

439 - spinlock_unlock(&rpt->thread.send_to_child.spinlock);

434 + internal_fatal(META_GET(&sth->run.meta, (Word_t)&rpt->thread.meta) != NULL, "Receiver to be added is already in the list of receivers");

435 + META_SET(&sth->run.meta, (Word_t)&rpt->thread.meta, &rpt->thread.meta);

436

441 - internal_fatal(META_GET(&sth->run.meta, (Word_t)&rpt->thread.meta) != NULL, "Receiver to be added is already in the list of receivers");

442 - META_SET(&sth->run.meta, (Word_t)&rpt->thread.meta, &rpt->thread.meta);

437 + if(sock_setnonblock(rpt->sock.fd) < 0)

438 + nd_log(NDLS_DAEMON, NDLP_ERR,

439 + "STREAM RCV '%s' [from [%s]:%s]: cannot set the non-blocking flag from socket %d",

440 + rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->sock.fd);

441

444 - if(sock_setnonblock(rpt->sock.fd) < 0)

445 - nd_log(NDLS_DAEMON, NDLP_ERR,

446 - "STREAM RECEIVE '%s' [from [%s]:%s]: cannot set the non-blocking flag from socket %d",

447 - rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->sock.fd);

442 + if(!nd_poll_add(sth->run.ndpl, rpt->sock.fd, ND_POLL_READ, &rpt->thread.meta))

443 + nd_log(NDLS_DAEMON, NDLP_ERR,

444 + "STREAM RCV[%zu] '%s' [from [%s]:%s]:"

445 + "Failed to add receiver socket to nd_poll()",

446 + sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);

447

449 - if(!nd_poll_add(sth->run.ndpl, rpt->sock.fd, ND_POLL_READ, &rpt->thread.meta))

450 - nd_log(NDLS_DAEMON, NDLP_ERR,

451 - "STREAM RECEIVE[%zu] '%s' [from [%s]:%s]:"

452 - "Failed to add receiver socket to nd_poll()",

453 - sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);

448 + stream_receive_log_database_gap(rpt);

449

455 - stream_receive_log_database_gap(rpt);

450 + // keep this last, since it sends commands back to the child

451 + streaming_parser_init(rpt);

452 +}

453

457 - // keep this last, since it sends commands back to the child

458 - streaming_parser_init(rpt);

454 +void stream_receiver_move_entire_queue_to_running_unsafe(struct stream_thread *sth) {

455 + internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );

456 +

457 + // process the queue

458 + Word_t idx = 0;

459 + for(struct receiver_state *rpt = RECEIVERS_FIRST(&sth->queue.receivers, &idx);

460 + rpt;

461 + rpt = RECEIVERS_NEXT(&sth->queue.receivers, &idx)) {

462 + RECEIVERS_DEL(&sth->queue.receivers, idx);

463 + stream_receiver_move_to_running_unsafe(sth, rpt);

464 }

465 }

466

@@ -464,7 +469,7 @@ static void stream_receiver_remove(struct stream_thread *sth, struct receiver_st

469

470 errno_clear();

471 nd_log(NDLS_DAEMON, NDLP_ERR,

467 - "STREAM RECEIVE[%zu] '%s' [from [%s]:%s]: "

472 + "STREAM RCV[%zu] '%s' [from [%s]:%s]: "

473 "receiver disconnected: %s"

474 , sth->id

475 , rpt->hostname ? rpt->hostname : "-"

@@ -656,7 +661,7 @@ bool stream_receive_process_poll_events(struct stream_thread *sth, struct receiv

661 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SOCKET_ERROR);

662

663 nd_log(NDLS_DAEMON, NDLP_ERR,

659 - "STREAM RECEIVE[%zu] '%s' [from [%s]:%s]: %s - closing connection",

664 + "STREAM RCV[%zu] '%s' [from [%s]:%s]: %s - closing connection",

665 sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, error);

666

667 receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_SOCKET_ERROR, false);

@@ -683,7 +688,7 @@ bool stream_receive_process_poll_events(struct stream_thread *sth, struct receiv

688 if (!stats->bytes_outstanding) {

689 if (!nd_poll_upd(sth->run.ndpl, rpt->sock.fd, ND_POLL_READ, &rpt->thread.meta))

690 nd_log(NDLS_DAEMON, NDLP_ERR,

686 - "STREAM RECEIVE[%zu] '%s' [from [%s]:%s]: cannot update nd_poll()",

691 + "STREAM RCV[%zu] '%s' [from [%s]:%s]: cannot update nd_poll()",

692 sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);

693

694 // recreate the circular buffer if we have to

@@ -711,7 +716,7 @@ bool stream_receive_process_poll_events(struct stream_thread *sth, struct receiv

716 if (disconnect_reason) {

717 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SEND_ERROR);

718 nd_log(NDLS_DAEMON, NDLP_ERR,

714 - "STREAM RECEIVE[%zu] '%s' [from [%s]:%s]: %s (%zd, on fd %d) - closing connection - "

719 + "STREAM RCV[%zu] '%s' [from [%s]:%s]: %s (%zd, on fd %d) - closing connection - "

720 "we have sent %zu bytes in %zu operations.",

721 sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port,

722 disconnect_reason, rc, rpt->sock.fd, stats->bytes_sent, stats->sends);

@@ -745,7 +750,7 @@ bool stream_receive_process_poll_events(struct stream_thread *sth, struct receiv

750 else if (rc == 0 || errno == ECONNRESET) {

751 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_REMOTE_CLOSED);

752 nd_log(NDLS_DAEMON, NDLP_ERR,

748 - "STREAM RECEIVE[%zu] '%s' [from [%s]:%s]: socket %d reports EOF (closed by child).",

753 + "STREAM RCV[%zu] '%s' [from [%s]:%s]: socket %d reports EOF (closed by child).",

754 sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->sock.fd);

755 receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_SOCKET_CLOSED_BY_REMOTE_END, false);

756 stream_receiver_remove(sth, rpt, "socket reports EOF (closed by child)");

@@ -761,7 +766,7 @@ bool stream_receive_process_poll_events(struct stream_thread *sth, struct receiv

766 else {

767 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_RECEIVE_ERROR);

768 nd_log(NDLS_DAEMON, NDLP_ERR,

764 - "STREAM RECEIVE[%zu] '%s' [from [%s]:%s]: error during receive (%zd, on fd %d) - closing connection.",

769 + "STREAM RCV[%zu] '%s' [from [%s]:%s]: error during receive (%zd, on fd %d) - closing connection.",

770 sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rc, rpt->sock.fd);

771 receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_SOCKET_READ_FAILED, false);

772 stream_receiver_remove(sth, rpt, "error during receive");

@@ -801,6 +806,8 @@ bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) {

806 rrdhost_receiver_lock(host);

807

808 if (!host->receiver) {

809 + __atomic_add_fetch(&host->stream.rcv.status.state_id, 1, __ATOMIC_RELAXED);

810 +

811 rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);

812

813 host->stream.rcv.status.connections++;

@@ -819,7 +826,7 @@ bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) {

826 if (rpt->config.health.delay > 0) {

827 host->health.delay_up_to = now_realtime_sec() + rpt->config.health.delay;

828 nd_log(NDLS_DAEMON, NDLP_DEBUG,

822 - "STREAM RECEIVE '%s' [from [%s]:%s]: "

829 + "STREAM RCV '%s' [from [%s]:%s]: "

830 "Postponing health checks for %" PRId64 " seconds, because it was just connected.",

831 rrdhost_hostname(host), rpt->client_ip, rpt->client_port,

832 (int64_t) rpt->config.health.delay);

@@ -863,6 +870,7 @@ void rrdhost_clear_receiver(struct receiver_state *rpt) {

870 // Make sure that we detach this thread and don't kill a freshly arriving receiver

871

872 if (host->receiver == rpt) {

873 + __atomic_add_fetch(&host->stream.rcv.status.state_id, 1, __ATOMIC_RELAXED);

874 rrdhost_flag_clear(host, RRDHOST_FLAG_COLLECTOR_ONLINE);

875 rrdhost_receiver_unlock(host);

876 {

@@ -928,7 +936,7 @@ bool stream_receiver_signal_to_stop_and_wait(RRDHOST *host, STREAM_HANDSHAKE rea

936 }

937

938 if(host->receiver)

931 - netdata_log_error("STREAM RECEIVE[x] '%s' [from [%s]:%s]: "

939 + netdata_log_error("STREAM RCV[x] '%s' [from [%s]:%s]: "

940 "streaming thread takes too long to stop, giving up..."

941 , rrdhost_hostname(host)

942 , host->receiver->client_ip, host->receiver->client_port);

src/streaming/stream-sender-commit.c

+9 -9

@@ -21,7 +21,7 @@ void sender_commit_thread_buffer_free(void) {

21 // Collector thread starting a transmission

22 BUFFER *sender_commit_start_with_trace(struct sender_state *s __maybe_unused, struct sender_buffer *commit, const char *func) {

23 if(unlikely(commit->used))

24 - fatal("STREAM SEND '%s' [to %s]: thread buffer is used multiple times concurrently (%u). "

24 + fatal("STREAM SND '%s' [to %s]: thread buffer is used multiple times concurrently (%u). "

25 "It is already being used by '%s()', and now is called by '%s()'",

26 rrdhost_hostname(s->host), s->connected_to,

27 (unsigned)commit->used,

@@ -29,7 +29,7 @@ BUFFER *sender_commit_start_with_trace(struct sender_state *s __maybe_unused, st

29 func ? func : "(null)");

31 if(unlikely(commit->receiver_tid && commit->receiver_tid != gettid_cached()))

32 - fatal("STREAM SEND '%s' [to %s]: thread buffer is reserved for tid %d, but it used by thread %d function '%s()'.",

32 + fatal("STREAM SND '%s' [to %s]: thread buffer is reserved for tid %d, but it used by thread %d function '%s()'.",

33 rrdhost_hostname(s->host), s->connected_to,

34 commit->receiver_tid, gettid_cached(), func ? func : "(null)");

@@ -87,7 +87,7 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff

87 s->scb, src_len * STREAM_CIRCULAR_BUFFER_ADAPT_TO_TIMES_MAX_SIZE, false))) {

88 // adaptive sizing of the circular buffer

89 nd_log(NDLS_DAEMON, NDLP_NOTICE,

90 - "STREAM SEND '%s' [to %s]: Increased max buffer size to %u (message size %zu).",

90 + "STREAM SND '%s' [to %s]: Increased max buffer size to %u (message size %zu).",

91 rrdhost_hostname(s->host), s->connected_to, stats->bytes_max_size, src_len + 1);

92 }

@@ -126,7 +126,7 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff

126 size_t dst_len = stream_compress(&s->compressor, src, size_to_compress, &dst);

127 if (!dst_len) {

128 nd_log(NDLS_DAEMON, NDLP_ERR,

129 - "STREAM SEND '%s' [to %s]: COMPRESSION failed. Resetting compressor and re-trying",

129 + "STREAM SND '%s' [to %s]: COMPRESSION failed. Resetting compressor and re-trying",

130 rrdhost_hostname(s->host), s->connected_to);

131

132 stream_compression_initialize(s);

@@ -142,7 +142,7 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff

142 size_t decoded_dst_len = stream_decompress_decode_signature((const char *)&signature, sizeof(signature));

143 if (decoded_dst_len != dst_len)

144 fatal(

145 - "STREAM SEND '%s' [to %s]: invalid signature, original payload %zu bytes, "

145 + "STREAM SND '%s' [to %s]: invalid signature, original payload %zu bytes, "

146 "compressed payload length %zu bytes, but signature says payload is %zu bytes",

147 rrdhost_hostname(s->host), s->connected_to,

148 size_to_compress, dst_len, decoded_dst_len);

@@ -188,7 +188,7 @@ overflow_with_lock: {

188 stream_sender_send_opcode(s, msg);

189 nd_log_limit_static_global_var(erl, 1, 0);

190 nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR,

191 - "STREAM SEND '%s' [to %s]: buffer overflow (buffer size %u, max size %u, used %u, available %u). "

191 + "STREAM SND '%s' [to %s]: buffer overflow (buffer size %u, max size %u, used %u, available %u). "

192 "Restarting connection.",

193 rrdhost_hostname(s->host), s->connected_to,

194 stats->bytes_size, stats->bytes_max_size, stats->bytes_outstanding, stats->bytes_available);

@@ -203,7 +203,7 @@ compression_failed_with_lock: {

203 stream_sender_send_opcode(s, msg);

204 nd_log_limit_static_global_var(erl, 1, 0);

205 nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR,

206 - "STREAM SEND '%s' [to %s]: COMPRESSION failed (twice). "

206 + "STREAM SND '%s' [to %s]: COMPRESSION failed (twice). "

207 "Deactivating compression and restarting connection.",

208 rrdhost_hostname(s->host), s->connected_to);

209 }

@@ -213,11 +213,11 @@ void sender_thread_commit(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYP

213 struct sender_buffer *commit = (wb == commit___thread.wb) ? & commit___thread : &s->host->stream.snd.commit;

214

215 if (unlikely(wb != commit->wb))

216 - fatal("STREAM SEND '%s' [to %s]: function '%s()' is trying to commit an unknown commit buffer.",

216 + fatal("STREAM SND '%s' [to %s]: function '%s()' is trying to commit an unknown commit buffer.",

217 rrdhost_hostname(s->host), s->connected_to, func);

218

219 if (unlikely(!commit->used))

220 - fatal("STREAM SEND '%s' [to %s]: function '%s()' is committing a sender buffer twice.",

220 + fatal("STREAM SND '%s' [to %s]: function '%s()' is committing a sender buffer twice.",

221 rrdhost_hostname(s->host), s->connected_to, func);

222

223 commit->used = false;

src/streaming/stream-sender-execute.c

+5 -5

@@ -26,7 +26,7 @@ static void stream_execute_function_callback(BUFFER *func_wb, int code, void *da

27 sender_commit_clean_buffer(s, wb, STREAM_TRAFFIC_TYPE_FUNCTIONS);

29 - internal_error(true, "STREAM SEND '%s' [to %s]: FUNCTION transaction %s sending back response (%zu bytes, %"PRIu64" usec).",

29 + internal_error(true, "STREAM SND '%s' [to %s]: FUNCTION transaction %s sending back response (%zu bytes, %"PRIu64" usec).",

30 rrdhost_hostname(s->host), s->connected_to,

31 string2str(tmp->transaction),

32 buffer_strlen(func_wb),

@@ -57,7 +57,7 @@ static void execute_commands_function(struct sender_state *s, const char *comman

57 nd_log(NDLS_ACCESS, NDLP_INFO, NULL);

59 if(!transaction || !*transaction || !timeout_s || !*timeout_s || !function || !*function) {

60 - netdata_log_error("STREAM SEND '%s' [to %s]: %s execution command is incomplete (transaction = '%s', timeout = '%s', function = '%s'). Ignoring it.",

60 + netdata_log_error("STREAM SND '%s' [to %s]: %s execution command is incomplete (transaction = '%s', timeout = '%s', function = '%s'). Ignoring it.",

61 rrdhost_hostname(s->host), s->connected_to,

62 command,

63 transaction?transaction:"(unset)",

@@ -111,7 +111,7 @@ static void execute_deferred_json(struct sender_state *s, void *data) {

111 stream_path_set_from_json(s->host, buffer_tostring(s->defer.payload), true);

112 else

113 nd_log(NDLS_DAEMON, NDLP_ERR,

114 - "STREAM SEND '%s' [to %s]: unknown JSON keyword '%s' with payload: %s",

114 + "STREAM SND '%s' [to %s]: unknown JSON keyword '%s' with payload: %s",

115 rrdhost_hostname(s->host), s->connected_to,

116 keyword, buffer_tostring(s->defer.payload));

117 }

@@ -277,7 +277,7 @@ void stream_sender_execute_commands(struct sender_state *s) {

277 const char *before = get_word(s->rbuf.line.words, s->rbuf.line.num_words, 4);

278

279 if (!chart_id || !start_streaming || !after || !before) {

280 - netdata_log_error("STREAM SEND '%s' [to %s] %s command is incomplete"

280 + netdata_log_error("STREAM SND '%s' [to %s] %s command is incomplete"

281 " (chart=%s, start_streaming=%s, after=%s, before=%s)",

282 rrdhost_hostname(s->host), s->connected_to,

283 command,

@@ -313,7 +313,7 @@ void stream_sender_execute_commands(struct sender_state *s) {

313 s->defer.action_data = strdupz(keyword);

314 }

315 else {

316 - netdata_log_error("STREAM SEND '%s' [to %s] received unknown command over connection: %s",

316 + netdata_log_error("STREAM SND '%s' [to %s] received unknown command over connection: %s",

317 rrdhost_hostname(s->host), s->connected_to, s->rbuf.line.words[0]?s->rbuf.line.words[0]:"(unset)");

318 }

319

src/streaming/stream-sender.c

+19 -19

@@ -72,7 +72,7 @@ static void stream_sender_charts_and_replication_reset(struct sender_state *s) {

73 void stream_sender_on_connect(struct sender_state *s) {

74 nd_log(NDLS_DAEMON, NDLP_DEBUG,

75 - "STREAM SEND [%s]: running on-connect hooks...",

75 + "STREAM SND [%s]: running on-connect hooks...",

76 rrdhost_hostname(s->host));

78 rrdhost_flag_set(s->host, RRDHOST_FLAG_STREAM_SENDER_CONNECTED);

@@ -89,7 +89,7 @@ void stream_sender_on_connect(struct sender_state *s) {

90 static void stream_sender_on_ready_to_dispatch(struct sender_state *s) {

91 nd_log(NDLS_DAEMON, NDLP_DEBUG,

92 - "STREAM SEND '%s': running ready-to-dispatch hooks...",

92 + "STREAM SND '%s': running ready-to-dispatch hooks...",

93 rrdhost_hostname(s->host));

95 // set this flag before sending any data, or the data will not be sent

@@ -105,7 +105,7 @@ static void stream_sender_on_ready_to_dispatch(struct sender_state *s) {

105

106 static void stream_sender_on_disconnect(struct sender_state *s) {

107 nd_log(NDLS_DAEMON, NDLP_DEBUG,

108 - "STREAM SEND '%s': running on-disconnect hooks...",

108 + "STREAM SND '%s': running on-disconnect hooks...",

109 rrdhost_hostname(s->host));

110

111 stream_sender_lock(s);

@@ -182,7 +182,7 @@ void stream_sender_handle_op(struct stream_thread *sth, struct sender_state *s,

182 STREAM_CIRCULAR_BUFFER_STATS stats = *stream_circular_buffer_stats_unsafe(s->scb);

183 stream_sender_unlock(s);

184 nd_log(NDLS_DAEMON, NDLP_ERR,

185 - "STREAM SEND[%zu] '%s' [to %s]: send buffer is full (buffer size %u, max %u, used %u, available %u). "

185 + "STREAM SND[%zu] '%s' [to %s]: send buffer is full (buffer size %u, max %u, used %u, available %u). "

186 "Restarting connection.",

187 sth->id, rrdhost_hostname(s->host), s->connected_to,

188 stats.bytes_size, stats.bytes_max_size, stats.bytes_outstanding, stats.bytes_available);

@@ -203,7 +203,7 @@ void stream_sender_handle_op(struct stream_thread *sth, struct sender_state *s,

203 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_COMPRESSION_ERROR);

204 errno_clear();

205 nd_log(NDLS_DAEMON, NDLP_ERR,

206 - "STREAM SEND[%zu] '%s' [to %s]: restarting connection without compression.",

206 + "STREAM SND[%zu] '%s' [to %s]: restarting connection without compression.",

207 sth->id, rrdhost_hostname(s->host), s->connected_to);

208

209 stream_sender_move_running_to_connector_or_remove(

@@ -219,7 +219,7 @@ void stream_sender_handle_op(struct stream_thread *sth, struct sender_state *s,

219 }

220

221 nd_log(NDLS_DAEMON, NDLP_ERR,

222 - "STREAM SEND[%zu]: invalid msg id %u", sth->id, (unsigned)msg->opcode);

222 + "STREAM SND[%zu]: invalid msg id %u", sth->id, (unsigned)msg->opcode);

223 }

224

225

@@ -235,7 +235,7 @@ void stream_sender_move_queue_to_running_unsafe(struct stream_thread *sth) {

235 s = SENDERS_NEXT(&sth->queue.senders, &idx)) {

236 worker_is_busy(WORKER_STREAM_JOB_DEQUEUE);

237

238 - SENDERS_DEL(&sth->queue.senders, (Word_t)s);

238 + SENDERS_DEL(&sth->queue.senders, idx);

239

240 ND_LOG_STACK lgs[] = {

241 ND_LOG_FIELD_STR(NDF_NIDL_NODE, s->host->hostname),

@@ -245,7 +245,7 @@ void stream_sender_move_queue_to_running_unsafe(struct stream_thread *sth) {

245 ND_LOG_STACK_PUSH(lgs);

246

247 nd_log(NDLS_DAEMON, NDLP_DEBUG,

248 - "STREAM SEND[%zu] '%s' [to %s]: moving host from dispatcher queue to dispatcher running...",

248 + "STREAM SND[%zu] '%s' [to %s]: moving host from dispatcher queue to dispatcher running...",

249 sth->id, rrdhost_hostname(s->host), s->connected_to);

250

251 stream_sender_lock(s);

@@ -269,7 +269,7 @@ void stream_sender_move_queue_to_running_unsafe(struct stream_thread *sth) {

269

270 if(!nd_poll_add(sth->run.ndpl, s->sock.fd, ND_POLL_READ, &s->thread.meta))

271 nd_log(NDLS_DAEMON, NDLP_ERR,

272 - "STREAM SEND[%zu] '%s' [to %s]: failed to add sender socket to nd_poll()",

272 + "STREAM SND[%zu] '%s' [to %s]: failed to add sender socket to nd_poll()",

273 sth->id, rrdhost_hostname(s->host), s->connected_to);

274

275 stream_sender_on_ready_to_dispatch(s);

@@ -281,7 +281,7 @@ void stream_sender_remove(struct sender_state *s) {

281 // when it gives up on a certain node

282

283 nd_log(NDLS_DAEMON, NDLP_NOTICE,

284 - "STREAM SEND '%s' [to %s]: streaming sender removed host: %s",

284 + "STREAM SND '%s' [to %s]: streaming sender removed host: %s",

285 rrdhost_hostname(s->host), s->connected_to, stream_handshake_error_to_string(s->exit.reason));

286

287 stream_sender_lock(s);

@@ -319,7 +319,7 @@ static void stream_sender_move_running_to_connector_or_remove(struct stream_thre

319

320 if(!nd_poll_del(sth->run.ndpl, s->sock.fd))

321 nd_log(NDLS_DAEMON, NDLP_ERR,

322 - "STREAM SEND[%zu] '%s' [to %s]: failed to delete sender socket from nd_poll()",

322 + "STREAM SND[%zu] '%s' [to %s]: failed to delete sender socket from nd_poll()",

323 sth->id, rrdhost_hostname(s->host), s->connected_to);

324

325 // clear this flag asap, to stop other threads from pushing metrics for this node

@@ -335,7 +335,7 @@ static void stream_sender_move_running_to_connector_or_remove(struct stream_thre

335 stream_sender_unlock(s);

336

337 nd_log(NDLS_DAEMON, NDLP_NOTICE,

338 - "STREAM SEND[%zu] '%s' [to %s]: sender disconnected from parent, reason: %s",

338 + "STREAM SND[%zu] '%s' [to %s]: sender disconnected from parent, reason: %s",

339 sth->id, rrdhost_hostname(s->host), s->connected_to, stream_handshake_error_to_string(reason));

340

341 nd_sock_close(&s->sock);

@@ -402,7 +402,7 @@ void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth, usec_t n

402 size_snprintf(pending, sizeof(pending), stats.bytes_outstanding, "B", false);

403

404 nd_log(NDLS_DAEMON, NDLP_ERR,

405 - "STREAM SEND[%zu] '%s' [to %s]: could not send data for %ld seconds - closing connection - "

405 + "STREAM SND[%zu] '%s' [to %s]: could not send data for %ld seconds - closing connection - "

406 "we have sent %zu bytes in %zu operations, it is idle for %s, and we have %s pending to send "

407 "(buffer is used %.2f%%).",

408 sth->id, rrdhost_hostname(s->host), s->connected_to, stream_send.parents.timeout_s,

@@ -418,7 +418,7 @@ void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth, usec_t n

418

419 if(!nd_poll_upd(sth->run.ndpl, s->sock.fd, ND_POLL_READ | (stats.bytes_outstanding ? ND_POLL_WRITE : 0), &s->thread.meta))

420 nd_log(NDLS_DAEMON, NDLP_ERR,

421 - "STREAM SEND[%zu] '%s' [to %s]: failed to update nd_poll().",

421 + "STREAM SND[%zu] '%s' [to %s]: failed to update nd_poll().",

422 sth->id, rrdhost_hostname(s->host), s->connected_to);

423 }

424

@@ -470,7 +470,7 @@ bool stream_sender_process_poll_events(struct stream_thread *sth, struct sender_

470 stream_sender_unlock(s);

471

472 nd_log(NDLS_DAEMON, NDLP_ERR,

473 - "STREAM SEND[%zu] '%s' [to %s]: %s restarting connection - %zu bytes transmitted in %zu operations.",

473 + "STREAM SND[%zu] '%s' [to %s]: %s restarting connection - %zu bytes transmitted in %zu operations.",

474 sth->id, rrdhost_hostname(s->host), s->connected_to, error, stats.bytes_sent, stats.sends);

475

476 stream_sender_move_running_to_connector_or_remove(sth, s, STREAM_HANDSHAKE_DISCONNECT_SOCKET_ERROR, true);

@@ -502,7 +502,7 @@ bool stream_sender_process_poll_events(struct stream_thread *sth, struct sender_

502 // we sent them all - remove ND_POLL_WRITE

503 if (!nd_poll_upd(sth->run.ndpl, s->sock.fd, ND_POLL_READ, &s->thread.meta))

504 nd_log(NDLS_DAEMON, NDLP_ERR,

505 - "STREAM SEND[%zu] '%s' [to %s]: failed to update nd_poll().",

505 + "STREAM SND[%zu] '%s' [to %s]: failed to update nd_poll().",

506 sth->id, rrdhost_hostname(s->host), s->connected_to);

507

508 // recreate the circular buffer if we have to

@@ -531,7 +531,7 @@ bool stream_sender_process_poll_events(struct stream_thread *sth, struct sender_

531 if (disconnect_reason) {

532 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SEND_ERROR);

533 nd_log(NDLS_DAEMON, NDLP_ERR,

534 - "STREAM SEND[%zu] '%s' [to %s]: %s (%zd, on fd %d) - restarting connection - "

534 + "STREAM SND[%zu] '%s' [to %s]: %s (%zd, on fd %d) - restarting connection - "

535 "we have sent %zu bytes in %zu operations.",

536 sth->id, rrdhost_hostname(s->host), s->connected_to, disconnect_reason, rc, s->sock.fd,

537 stats->bytes_sent, stats->sends);

@@ -572,7 +572,7 @@ bool stream_sender_process_poll_events(struct stream_thread *sth, struct sender_

572 else if (rc == 0 || errno == ECONNRESET) {

573 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_REMOTE_CLOSED);

574 nd_log(NDLS_DAEMON, NDLP_ERR,

575 - "STREAM SEND[%zu] '%s' [to %s]: socket %d reports EOF (closed by parent).",

575 + "STREAM SND[%zu] '%s' [to %s]: socket %d reports EOF (closed by parent).",

576 sth->id, rrdhost_hostname(s->host), s->connected_to, s->sock.fd);

577 stream_sender_move_running_to_connector_or_remove(

578 sth, s, STREAM_HANDSHAKE_DISCONNECT_SOCKET_CLOSED_BY_REMOTE_END, true);

@@ -585,7 +585,7 @@ bool stream_sender_process_poll_events(struct stream_thread *sth, struct sender_

585 else {

586 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_RECEIVE_ERROR);

587 nd_log(NDLS_DAEMON, NDLP_ERR,

588 - "STREAM SEND[%zu] '%s' [to %s]: error during receive (%zd, on fd %d) - restarting connection.",

588 + "STREAM SND[%zu] '%s' [to %s]: error during receive (%zd, on fd %d) - restarting connection.",

589 sth->id, rrdhost_hostname(s->host), s->connected_to, rc, s->sock.fd);

590 stream_sender_move_running_to_connector_or_remove(

591 sth, s, STREAM_HANDSHAKE_DISCONNECT_SOCKET_READ_FAILED, true);

src/streaming/stream-thread.c

+29 -15

@@ -1,6 +1,8 @@

1 // SPDX-License-Identifier: GPL-3.0-or-later

3 +#define STREAM_INTERNALS

4 #include "stream-thread.h"

5 +#include "stream-waiting-list.h"

7 struct stream_thread_globals stream_thread_globals = {

8 .assign = {

@@ -30,7 +32,7 @@ static void stream_thread_handle_op(struct stream_thread *sth, struct stream_opc

32 if(!nd_poll_upd(sth->run.ndpl, m->s->sock.fd, ND_POLL_READ|ND_POLL_WRITE, m)) {

33 nd_log_limit_static_global_var(erl, 1, 0);

34 nd_log_limit(&erl, NDLS_DAEMON, NDLP_DEBUG,

33 - "STREAM SEND[%zu] '%s' [to %s]: cannot enable output on sender socket %d.",

35 + "STREAM SND[%zu] '%s' [to %s]: cannot enable output on sender socket %d.",

36 sth->id, rrdhost_hostname(m->s->host), m->s->connected_to, m->s->sock.fd);

37 }

38 msg->opcode &= ~(STREAM_OPCODE_SENDER_POLLOUT);

@@ -44,7 +46,7 @@ static void stream_thread_handle_op(struct stream_thread *sth, struct stream_opc

46 if (!nd_poll_upd(sth->run.ndpl, m->rpt->sock.fd, ND_POLL_READ | ND_POLL_WRITE, m)) {

47 nd_log_limit_static_global_var(erl, 1, 0);

48 nd_log_limit(&erl, NDLS_DAEMON, NDLP_DEBUG,

47 - "STREAM RECEIVE[%zu] '%s' [from [%s]:%s]: cannot enable output on receiver socket %d.",

49 + "STREAM RCV[%zu] '%s' [from [%s]:%s]: cannot enable output on receiver socket %d.",

50 sth->id, rrdhost_hostname(m->rpt->host), m->rpt->client_ip, m->rpt->client_port, m->rpt->sock.fd);

51 }

52 msg->opcode &= ~(STREAM_OPCODE_RECEIVER_POLLOUT);

@@ -81,14 +83,14 @@ void stream_receiver_send_opcode(struct receiver_state *rpt, struct stream_opcod

84 if(msg.meta != &rpt->thread.meta) {

85 nd_log(NDLS_DAEMON, NDLP_ERR,

84 - "STREAM RECEIVE '%s' [from [%s]:%s]: the receiver in the opcode the message does not match this receiver. "

86 + "STREAM RCV '%s' [from [%s]:%s]: the receiver in the opcode the message does not match this receiver. "

87 "Ignoring opcode.", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);

88 return;

89 }

90 struct stream_thread *sth = stream_thread_by_slot_id(msg.thread_slot);

91 if(!sth) {

92 nd_log(NDLS_DAEMON, NDLP_ERR,

91 - "STREAM RECEIVE '%s' [from [%s]:%s]: the opcode (%u) message cannot be verified. Ignoring it.",

93 + "STREAM RCV '%s' [from [%s]:%s]: the opcode (%u) message cannot be verified. Ignoring it.",

94 rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, msg.opcode);

95 return;

96 }

@@ -135,7 +137,7 @@ void stream_receiver_send_opcode(struct receiver_state *rpt, struct stream_opcod

137 }

138 #endif

139

138 - fatal("STREAM RECEIVE '%s' [from [%s]:%s]: The streaming opcode queue is full, but this should never happen...",

140 + fatal("STREAM RCV '%s' [from [%s]:%s]: The streaming opcode queue is full, but this should never happen...",

141 rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);

142 }

143

@@ -161,7 +163,7 @@ void stream_sender_send_opcode(struct sender_state *s, struct stream_opcode msg)

163

164 if(msg.meta != &s->thread.meta) {

165 nd_log(NDLS_DAEMON, NDLP_ERR,

164 - "STREAM SEND '%s' [to %s]: the opcode message does not match this sender. "

166 + "STREAM SND '%s' [to %s]: the opcode message does not match this sender. "

167 "Ignoring opcode.", rrdhost_hostname(s->host), s->connected_to);

168 return;

169 }

@@ -169,7 +171,7 @@ void stream_sender_send_opcode(struct sender_state *s, struct stream_opcode msg)

171 struct stream_thread *sth = stream_thread_by_slot_id(msg.thread_slot);

172 if(!sth) {

173 nd_log(NDLS_DAEMON, NDLP_ERR,

172 - "STREAM SEND[x] '%s' [to %s] the opcode (%u) message cannot be verified. Ignoring it.",

174 + "STREAM SND[x] '%s' [to %s] the opcode (%u) message cannot be verified. Ignoring it.",

175 rrdhost_hostname(s->host), s->connected_to, msg.opcode);

176 return;

177 }

@@ -216,7 +218,7 @@ void stream_sender_send_opcode(struct sender_state *s, struct stream_opcode msg)

218 }

219 #endif

220

219 - fatal("STREAM SEND '%s' [to %s]: The streaming opcode queue is full, but this should never happen...",

221 + fatal("STREAM SND '%s' [to %s]: The streaming opcode queue is full, but this should never happen...",

222 rrdhost_hostname(s->host), s->connected_to);

223 }

224

@@ -448,6 +450,10 @@ void *stream_thread(void *ptr) {

450 "ops processed", "messages",

451 WORKER_METRIC_INCREMENTAL_TOTAL);

452

453 + worker_register_job_custom_metric(WORKER_SENDER_JOB_RECEIVERS_WAITING_LIST_SIZE,

454 + "receivers waiting to be added", "nodes",

455 + WORKER_METRIC_ABSOLUTE);

456 +

457 if(pipe(sth->pipe.fds) != 0) {

458 nd_log(NDLS_DAEMON, NDLP_ERR, "STREAM THREAD[%zu]: cannot create required pipe.", sth->id);

459 sth->pipe.fds[PIPE_READ] = -1;

@@ -477,6 +483,7 @@ void *stream_thread(void *ptr) {

483

484 bool exit_thread = false;

485 size_t replay_entries = 0;

486 + size_t receivers_waiting = 0;

487 sth->snd.bytes_received = 0;

488 sth->snd.bytes_sent = 0;

489

@@ -488,9 +495,15 @@ void *stream_thread(void *ptr) {

495

496 // move any pending hosts in the inbound queue, to the running list

497 spinlock_lock(&sth->queue.spinlock);

498 +

499 stream_thread_messages_resize_unsafe(sth);

492 - stream_receiver_move_queue_to_running_unsafe(sth);

500 +

501 + stream_thread_process_waiting_list_unsafe(sth, now_ut);

502 + // stream_receiver_move_entire_queue_to_running_unsafe(sth);

503 +

504 stream_sender_move_queue_to_running_unsafe(sth);

505 +

506 + receivers_waiting = sth->queue.receivers_waiting;

507 spinlock_unlock(&sth->queue.spinlock);

508 last_dequeue_ut = now_ut;

509 }

@@ -507,6 +520,8 @@ void *stream_thread(void *ptr) {

520 worker_set_metric(WORKER_SENDER_JOB_BYTES_RECEIVED, (NETDATA_DOUBLE)sth->snd.bytes_received);

521 worker_set_metric(WORKER_SENDER_JOB_BYTES_SENT, (NETDATA_DOUBLE)sth->snd.bytes_sent);

522 worker_set_metric(WORKER_SENDER_JOB_REPLAY_DICT_SIZE, (NETDATA_DOUBLE)replay_entries);

523 +

524 + worker_set_metric(WORKER_SENDER_JOB_RECEIVERS_WAITING_LIST_SIZE, (NETDATA_DOUBLE)receivers_waiting);

525 replay_entries = 0;

526 sth->snd.bytes_received = 0;

527 sth->snd.bytes_sent = 0;

@@ -549,7 +564,7 @@ void *stream_thread(void *ptr) {

564 // dequeue

565 spinlock_lock(&sth->queue.spinlock);

566 stream_sender_move_queue_to_running_unsafe(sth);

552 - stream_receiver_move_queue_to_running_unsafe(sth);

567 + stream_receiver_move_entire_queue_to_running_unsafe(sth);

568 spinlock_unlock(&sth->queue.spinlock);

569

570 // cleanup receiver and dispatcher

@@ -685,12 +700,12 @@ void stream_receiver_add_to_queue(struct receiver_state *rpt) {

700 stream_thread_node_queued(rpt->host);

701

702 nd_log(NDLS_DAEMON, NDLP_DEBUG,

688 - "STREAM RECEIVE[%zu] '%s': moving host to receiver queue...",

703 + "STREAM RCV[%zu] '%s': moving host to receiver queue...",

704 sth->id, rrdhost_hostname(rpt->host));

705

706 spinlock_lock(&sth->queue.spinlock);

692 - internal_fatal(RECEIVERS_GET(&sth->queue.receivers, (Word_t)rpt) != NULL, "Receiver is already in the receivers queue");

693 - RECEIVERS_SET(&sth->queue.receivers, (Word_t)rpt, rpt);

707 + RECEIVERS_SET(&sth->queue.receivers, ++sth->queue.id, rpt);

708 + sth->queue.receivers_waiting++;

709 spinlock_unlock(&sth->queue.spinlock);

710 }

711

@@ -704,8 +719,7 @@ void stream_sender_add_to_queue(struct sender_state *s) {

719 sth->id, rrdhost_hostname(s->host));

720

721 spinlock_lock(&sth->queue.spinlock);

707 - internal_fatal(SENDERS_GET(&sth->queue.senders, (Word_t)s) != NULL, "Sender is already in the senders queue");

708 - SENDERS_SET(&sth->queue.senders, (Word_t)s, s);

722 + SENDERS_SET(&sth->queue.senders, ++sth->queue.id, s);

723 spinlock_unlock(&sth->queue.spinlock);

724 }

725

src/streaming/stream-thread.h

+53 -39

@@ -36,55 +36,58 @@ struct stream_opcode {

36 // IMPORTANT: to add workers, you have to edit WORKER_PARSER_FIRST_JOB accordingly

38 // stream thread events

39 -#define WORKER_STREAM_JOB_LIST (WORKER_PARSER_FIRST_JOB - 34)

40 -#define WORKER_STREAM_JOB_DEQUEUE (WORKER_PARSER_FIRST_JOB - 33)

41 -#define WORKER_STREAM_JOB_PREP (WORKER_PARSER_FIRST_JOB - 32)

42 -#define WORKER_STREAM_JOB_POLL_ERROR (WORKER_PARSER_FIRST_JOB - 31)

43 -#define WORKER_SENDER_JOB_PIPE_READ (WORKER_PARSER_FIRST_JOB - 30)

39 +#define WORKER_STREAM_JOB_LIST 0

40 +#define WORKER_STREAM_JOB_DEQUEUE 1

41 +#define WORKER_STREAM_JOB_PREP 2

42 +#define WORKER_STREAM_JOB_POLL_ERROR 3

43 +#define WORKER_SENDER_JOB_PIPE_READ 4

45 // socket operations

46 -#define WORKER_STREAM_JOB_SOCKET_RECEIVE (WORKER_PARSER_FIRST_JOB - 29)

47 -#define WORKER_STREAM_JOB_SOCKET_SEND (WORKER_PARSER_FIRST_JOB - 28)

48 -#define WORKER_STREAM_JOB_SOCKET_ERROR (WORKER_PARSER_FIRST_JOB - 27)

46 +#define WORKER_STREAM_JOB_SOCKET_RECEIVE 5

47 +#define WORKER_STREAM_JOB_SOCKET_SEND 6

48 +#define WORKER_STREAM_JOB_SOCKET_ERROR 7

50 // compression

51 -#define WORKER_STREAM_JOB_COMPRESS (WORKER_PARSER_FIRST_JOB - 26)

52 -#define WORKER_STREAM_JOB_DECOMPRESS (WORKER_PARSER_FIRST_JOB - 25)

51 +#define WORKER_STREAM_JOB_COMPRESS 8

52 +#define WORKER_STREAM_JOB_DECOMPRESS 9

54 // receiver events

55 -#define WORKER_RECEIVER_JOB_BYTES_READ (WORKER_PARSER_FIRST_JOB - 24)

56 -#define WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED (WORKER_PARSER_FIRST_JOB - 23)

55 +#define WORKER_RECEIVER_JOB_BYTES_READ 10

56 +#define WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED 11

58 // sender received commands

59 -#define WORKER_SENDER_JOB_EXECUTE (WORKER_PARSER_FIRST_JOB - 22)

60 -#define WORKER_SENDER_JOB_EXECUTE_REPLAY (WORKER_PARSER_FIRST_JOB - 21)

61 -#define WORKER_SENDER_JOB_EXECUTE_FUNCTION (WORKER_PARSER_FIRST_JOB - 20)

62 -#define WORKER_SENDER_JOB_EXECUTE_META (WORKER_PARSER_FIRST_JOB - 19)

63 -

64 -#define WORKER_SENDER_JOB_DISCONNECT_OVERFLOW (WORKER_PARSER_FIRST_JOB - 18)

65 -#define WORKER_SENDER_JOB_DISCONNECT_TIMEOUT (WORKER_PARSER_FIRST_JOB - 17)

66 -#define WORKER_SENDER_JOB_DISCONNECT_SOCKET_ERROR (WORKER_PARSER_FIRST_JOB - 16)

67 -#define WORKER_SENDER_JOB_DISCONNECT_REMOTE_CLOSED (WORKER_PARSER_FIRST_JOB - 15)

68 -#define WORKER_SENDER_JOB_DISCONNECT_RECEIVE_ERROR (WORKER_PARSER_FIRST_JOB - 14)

69 -#define WORKER_SENDER_JOB_DISCONNECT_SEND_ERROR (WORKER_PARSER_FIRST_JOB - 13)

70 -#define WORKER_SENDER_JOB_DISCONNECT_COMPRESSION_ERROR (WORKER_PARSER_FIRST_JOB - 12)

71 -#define WORKER_SENDER_JOB_DISCONNECT_RECEIVER_LEFT (WORKER_PARSER_FIRST_JOB - 11)

72 -#define WORKER_SENDER_JOB_DISCONNECT_HOST_CLEANUP (WORKER_PARSER_FIRST_JOB - 10)

59 +#define WORKER_SENDER_JOB_EXECUTE 12

60 +#define WORKER_SENDER_JOB_EXECUTE_REPLAY 13

61 +#define WORKER_SENDER_JOB_EXECUTE_FUNCTION 14

62 +#define WORKER_SENDER_JOB_EXECUTE_META 15

63 +

64 +#define WORKER_SENDER_JOB_DISCONNECT_OVERFLOW 16

65 +#define WORKER_SENDER_JOB_DISCONNECT_TIMEOUT 17

66 +#define WORKER_SENDER_JOB_DISCONNECT_SOCKET_ERROR 18

67 +#define WORKER_SENDER_JOB_DISCONNECT_REMOTE_CLOSED 19

68 +#define WORKER_SENDER_JOB_DISCONNECT_RECEIVE_ERROR 20

69 +#define WORKER_SENDER_JOB_DISCONNECT_SEND_ERROR 21

70 +#define WORKER_SENDER_JOB_DISCONNECT_COMPRESSION_ERROR 22

71 +#define WORKER_SENDER_JOB_DISCONNECT_RECEIVER_LEFT 23

72 +#define WORKER_SENDER_JOB_DISCONNECT_HOST_CLEANUP 24

74 // dispatcher metrics

75 // this has to be the same at pluginsd_parser.h

76 -#define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION (WORKER_PARSER_FIRST_JOB - 9)

77 -#define WORKER_STREAM_METRIC_NODES (WORKER_PARSER_FIRST_JOB - 8)

78 -#define WORKER_SENDER_JOB_BUFFER_RATIO (WORKER_PARSER_FIRST_JOB - 7)

79 -#define WORKER_SENDER_JOB_BYTES_RECEIVED (WORKER_PARSER_FIRST_JOB - 6)

80 -#define WORKER_SENDER_JOB_BYTES_SENT (WORKER_PARSER_FIRST_JOB - 5)

81 -#define WORKER_SENDER_JOB_BYTES_COMPRESSED (WORKER_PARSER_FIRST_JOB - 4)

82 -#define WORKER_SENDER_JOB_BYTES_UNCOMPRESSED (WORKER_PARSER_FIRST_JOB - 3)

83 -#define WORKER_SENDER_JOB_BYTES_COMPRESSION_RATIO (WORKER_PARSER_FIRST_JOB - 2)

84 -#define WORKER_SENDER_JOB_REPLAY_DICT_SIZE (WORKER_PARSER_FIRST_JOB - 1)

85 -#define WORKER_SENDER_JOB_MESSAGES (WORKER_PARSER_FIRST_JOB - 0)

86 -

87 -#if WORKER_UTILIZATION_MAX_JOB_TYPES < 35

76 +#define WORKER_RECEIVER_JOB_REPLICATION_COMPLETION 25

77 +#define WORKER_STREAM_METRIC_NODES 26

78 +#define WORKER_SENDER_JOB_BUFFER_RATIO 27

79 +#define WORKER_SENDER_JOB_BYTES_RECEIVED 28

80 +#define WORKER_SENDER_JOB_BYTES_SENT 29

81 +#define WORKER_SENDER_JOB_BYTES_COMPRESSED 30

82 +#define WORKER_SENDER_JOB_BYTES_UNCOMPRESSED 31

83 +#define WORKER_SENDER_JOB_BYTES_COMPRESSION_RATIO 32

84 +#define WORKER_SENDER_JOB_REPLAY_DICT_SIZE 33

85 +#define WORKER_SENDER_JOB_MESSAGES 34

86 +#define WORKER_SENDER_JOB_RECEIVERS_WAITING_LIST_SIZE 35

87 +

88 +// IMPORTANT: to add workers, you have to edit WORKER_PARSER_FIRST_JOB accordingly

89 +

90 +#if WORKER_UTILIZATION_MAX_JOB_TYPES < 36

91 #error WORKER_UTILIZATION_MAX_JOB_TYPES has to be at least 34

92 #endif

@@ -141,10 +144,19 @@ struct stream_thread {

144 // the incoming queue of the dispatcher thread

145 // the connector thread leaves the connected senders in this list, for the dispatcher to pick them up

146 SPINLOCK spinlock;

147 + Word_t id;

148 SENDERS_JudyLSet senders;

149 RECEIVERS_JudyLSet receivers;

150 +

151 + size_t receivers_waiting;

152 } queue;

153

154 + struct {

155 + usec_t last_accepted_ut;

156 + size_t metadata;

157 + size_t replication;

158 + } waiting_list;

159 +

160 struct {

161 SPINLOCK spinlock;

162 size_t added;

@@ -177,7 +189,7 @@ struct rrdhost;

189 extern struct stream_thread_globals stream_thread_globals;

190

191 void stream_sender_move_queue_to_running_unsafe(struct stream_thread *sth);

180 -void stream_receiver_move_queue_to_running_unsafe(struct stream_thread *sth);

192 +void stream_receiver_move_entire_queue_to_running_unsafe(struct stream_thread *sth);

193 void stream_sender_check_all_nodes_from_poll(struct stream_thread *sth, usec_t now_ut);

194

195 void stream_receiver_add_to_queue(struct receiver_state *rpt);

@@ -198,6 +210,8 @@ void stream_thread_node_removed(struct rrdhost *host);

210 // returns true if my_meta has received a message

211 bool stream_thread_process_opcodes(struct stream_thread *sth, struct pollfd_meta *my_meta);

212

213 +void stream_receiver_move_to_running_unsafe(struct stream_thread *sth, struct receiver_state *rpt);

214 +

215 #include "stream-sender-internals.h"

216 #include "stream-receiver-internals.h"

217 #include "plugins.d/pluginsd_parser.h"

src/streaming/stream-waiting-list.c new

+48

@@ -0,0 +1,48 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#define STREAM_INTERNALS

4 +#include "stream-waiting-list.h"

5 +

6 +#define ACCEPT_NODES_EVERY_UT (5 * USEC_PER_SEC)

7 +

8 +static __thread struct {

9 + size_t metadata;

10 + size_t replication;

11 +} throttle = { 0 };

12 +

13 +void stream_thread_received_metadata(void) {

14 + throttle.metadata++;

15 +}

16 +void stream_thread_received_replication(void) {

17 + throttle.replication++;

18 +}

19 +

20 +static inline size_t normalize_value(size_t v) {

21 + return (v / 100) * 100;

22 +}

23 +

24 +void stream_thread_process_waiting_list_unsafe(struct stream_thread *sth, usec_t now_ut) {

25 + internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );

26 +

27 + Word_t idx = 0;

28 + struct receiver_state *rpt = RECEIVERS_FIRST(&sth->queue.receivers, &idx);

29 + if(!rpt) return;

30 +

31 + if(sth->waiting_list.last_accepted_ut + ACCEPT_NODES_EVERY_UT > now_ut ||

32 + !stream_control_children_should_be_accepted())

33 + return;

34 +

35 + size_t n_metadata = normalize_value(throttle.metadata);

36 + size_t n_replication = normalize_value(throttle.replication);

37 +

38 + if(sth->waiting_list.metadata != n_metadata ||

39 + sth->waiting_list.replication != n_replication) {

40 + sth->waiting_list.metadata = n_metadata;

41 + sth->waiting_list.replication = n_replication;

42 + return;

43 + }

44 +

45 + RECEIVERS_DEL(&sth->queue.receivers, idx);

46 + stream_receiver_move_to_running_unsafe(sth, rpt);

47 + sth->queue.receivers_waiting--;

48 +}

src/streaming/stream-waiting-list.h new

+14

@@ -0,0 +1,14 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#ifndef NETDATA_STREAM_WAITING_LIST_H

4 +#define NETDATA_STREAM_WAITING_LIST_H

5 +

6 +void stream_thread_received_metadata(void);

7 +void stream_thread_received_replication(void);

8 +

9 +#ifdef STREAM_INTERNALS

10 +#include "stream-thread.h"

11 +void stream_thread_process_waiting_list_unsafe(struct stream_thread *sth, usec_t now_ut);

12 +#endif

13 +

14 +#endif //NETDATA_STREAM_WAITING_LIST_H

src/web/api/queries/backfill.c new

+231

@@ -0,0 +1,231 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#include "backfill.h"

4 +

5 +struct backfill_request {

6 + size_t rrdhost_receiver_state_id;

7 + RRDSET_ACQUIRED *rsa;

8 + uint32_t works;

9 + uint32_t successful;

10 + uint32_t failed;

11 + backfill_callback_t cb;

12 + struct backfill_request_data data;

13 +};

14 +

15 +struct backfill_dim_work {

16 + RRDDIM_ACQUIRED *rda;

17 + struct backfill_request *br;

18 +};

19 +

20 +DEFINE_JUDYL_TYPED(BACKFILL, struct backfill_dim_work *);

21 +

22 +static struct {

23 + struct completion completion;

24 +

25 + SPINLOCK spinlock;

26 + bool running;

27 + Word_t id;

28 + size_t queue_size;

29 + BACKFILL_JudyLSet queue;

30 +

31 + ARAL *ar_br;

32 + ARAL *ar_bdm;

33 +

34 +} backfill_globals = {

35 + .spinlock = SPINLOCK_INITIALIZER,

36 + .queue = { 0 },

37 +};

38 +

39 +bool backfill_request_add(RRDSET *st, backfill_callback_t cb, struct backfill_request_data *data) {

40 + bool rc = false;

41 + size_t dimensions = dictionary_entries(st->rrddim_root_index);

42 + if(!dimensions || dimensions > 200)

43 + return rc;

44 +

45 + size_t added = 0;

46 + struct backfill_dim_work *array[dimensions];

47 +

48 + if(backfill_globals.running) {

49 + struct backfill_request *br = aral_mallocz(backfill_globals.ar_br);

50 + br->data = *data;

51 + br->rrdhost_receiver_state_id =__atomic_load_n(&st->rrdhost->stream.rcv.status.state_id, __ATOMIC_RELAXED);

52 + br->rsa = rrdset_find_and_acquire(st->rrdhost, string2str(st->id));

53 + if(br->rsa) {

54 + br->cb = cb;

55 +

56 + RRDDIM *rd;

57 + rrddim_foreach_read(rd, st) {

58 + if(added >= dimensions)

59 + break;

60 +

61 + if (!rrddim_option_check(rd, RRDDIM_OPTION_BACKFILLED_HIGH_TIERS)) {

62 + struct backfill_dim_work *bdm = aral_mallocz(backfill_globals.ar_bdm);

63 + bdm->rda = (RRDDIM_ACQUIRED *)dictionary_acquired_item_dup(st->rrddim_root_index, rd_dfe.item);

64 + bdm->br = br;

65 + br->works++;

66 + array[added++] = bdm;

67 + }

68 + }

69 + rrddim_foreach_done(rd);

70 + }

71 +

72 + if(added) {

73 + spinlock_lock(&backfill_globals.spinlock);

74 +

75 + for(size_t i = 0; i < added ;i++) {

76 + backfill_globals.queue_size++;

77 + BACKFILL_SET(&backfill_globals.queue, backfill_globals.id++, array[i]);

78 + }

79 +

80 + spinlock_unlock(&backfill_globals.spinlock);

81 + completion_mark_complete_a_job(&backfill_globals.completion);

82 +

83 + rc = true;

84 + }

85 + else {

86 + // no dimensions added

87 + rrdset_acquired_release(br->rsa);

88 + aral_freez(backfill_globals.ar_br, br);

89 + }

90 + }

91 +

92 + return rc;

93 +}

94 +

95 +bool backfill_execute(struct backfill_dim_work *bdm) {

96 + RRDSET *st = rrdset_acquired_to_rrdset(bdm->br->rsa);

97 +

98 + if(bdm->br->rrdhost_receiver_state_id !=__atomic_load_n(&st->rrdhost->stream.rcv.status.state_id, __ATOMIC_RELAXED))

99 + return false;

100 +

101 + RRDDIM *rd = rrddim_acquired_to_rrddim(bdm->rda);

102 +

103 + size_t success = 0;

104 + for (size_t tier = 1; tier < storage_tiers; tier++)

105 + if(backfill_tier_from_smaller_tiers(rd, tier, now_realtime_sec()))

106 + success++;

107 +

108 + if(success > 0)

109 + rrddim_option_set(rd, RRDDIM_OPTION_BACKFILLED_HIGH_TIERS);

110 +

111 + return success > 0;

112 +}

113 +

114 +static void backfill_dim_work_free(bool successful, struct backfill_dim_work *bdm) {

115 + struct backfill_request *br = bdm->br;

116 +

117 + if(successful)

118 + __atomic_add_fetch(&br->successful, 1, __ATOMIC_RELAXED);

119 + else

120 + __atomic_add_fetch(&br->failed, 1, __ATOMIC_RELAXED);

121 +

122 + uint32_t works = __atomic_sub_fetch(&br->works, 1, __ATOMIC_RELAXED);

123 + if(!works) {

124 + if(br->cb)

125 + br->cb(__atomic_load_n(&br->successful, __ATOMIC_RELAXED),

126 + __atomic_load_n(&br->failed, __ATOMIC_RELAXED),

127 + &br->data);

128 +

129 + rrdset_acquired_release(br->rsa);

130 + aral_freez(backfill_globals.ar_br, br);

131 + }

132 +

133 + rrddim_acquired_release(bdm->rda);

134 + aral_freez(backfill_globals.ar_bdm, bdm);

135 +}

136 +

137 +void *backfill_worker_thread(void *ptr __maybe_unused) {

138 + worker_register("BACKFILL");

139 +

140 + worker_register_job_name(0, "get");

141 + worker_register_job_name(1, "backfill");

142 + worker_register_job_custom_metric(2, "backfill queue size", "dimensions", WORKER_METRIC_ABSOLUTE);

143 +

144 + size_t job_id = 0, queue_size = 0;

145 + while(!nd_thread_signaled_to_cancel() && service_running(SERVICE_COLLECTORS|SERVICE_STREAMING)) {

146 + worker_is_busy(0);

147 + spinlock_lock(&backfill_globals.spinlock);

148 + Word_t idx = 0;

149 + struct backfill_dim_work *bdm = BACKFILL_FIRST(&backfill_globals.queue, &idx);

150 + if(bdm) {

151 + backfill_globals.queue_size--;

152 + BACKFILL_DEL(&backfill_globals.queue, idx);

153 + }

154 + queue_size = backfill_globals.queue_size;

155 + spinlock_unlock(&backfill_globals.spinlock);

156 +

157 + if(bdm) {

158 + worker_is_busy(1);

159 + bool success = backfill_execute(bdm);

160 + backfill_dim_work_free(success, bdm);

161 + continue;

162 + }

163 +

164 + worker_set_metric(2, (NETDATA_DOUBLE)queue_size);

165 +

166 + worker_is_idle();

167 + job_id = completion_wait_for_a_job_with_timeout(&backfill_globals.completion, job_id, 1000);

168 + }

169 +

170 + worker_unregister();

171 +

172 + return NULL;

173 +}

174 +

175 +void *backfill_thread(void *ptr) {

176 + struct netdata_static_thread *static_thread = ptr;

177 + if(!static_thread) return NULL;

178 +

179 + nd_thread_tag_set("BACKFILL[0]");

180 +

181 + completion_init(&backfill_globals.completion);

182 + BACKFILL_INIT(&backfill_globals.queue);

183 + backfill_globals.ar_br = aral_by_size_acquire(sizeof(struct backfill_request));

184 + backfill_globals.ar_bdm = aral_by_size_acquire(sizeof(struct backfill_dim_work));

185 +

186 + spinlock_lock(&backfill_globals.spinlock);

187 + backfill_globals.running = true;

188 + spinlock_unlock(&backfill_globals.spinlock);

189 +

190 + size_t threads = get_netdata_cpus() / 2;

191 + if(threads < 2) threads = 2;

192 + if(threads > 16) threads = 16;

193 + ND_THREAD *th[threads - 1];

194 +

195 + for(size_t t = 0; t < threads - 1 ;t++) {

196 + char tag[15];

197 + snprintfz(tag, sizeof(tag), "BACKFILL[%zu]", t + 1);

198 + th[t] = nd_thread_create(tag, NETDATA_THREAD_OPTION_JOINABLE, backfill_worker_thread, NULL);

199 + }

200 +

201 + backfill_worker_thread(NULL);

202 + static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

203 +

204 + for(size_t t = 0; t < threads - 1 ;t++) {

205 + nd_thread_signal_cancel(th[t]);

206 + nd_thread_join(th[t]);

207 + }

208 +

209 + // cleanup

210 + spinlock_lock(&backfill_globals.spinlock);

211 + backfill_globals.running = false;

212 + Word_t idx = 0;

213 + for(struct backfill_dim_work *bdm = BACKFILL_FIRST(&backfill_globals.queue, &idx);

214 + bdm;

215 + bdm = BACKFILL_NEXT(&backfill_globals.queue, &idx)) {

216 + backfill_dim_work_free(false, bdm);

217 + }

218 + spinlock_unlock(&backfill_globals.spinlock);

219 +

220 + aral_by_size_release(backfill_globals.ar_br);

221 + aral_by_size_release(backfill_globals.ar_bdm);

222 + completion_destroy(&backfill_globals.completion);

223 +

224 + static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

225 +

226 + return NULL;

227 +}

228 +

229 +bool backfill_threads_detect_from_stream_conf(void) {

230 + return stream_conf_configured_as_parent();

231 +}

src/web/api/queries/backfill.h new

+26

@@ -0,0 +1,26 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#ifndef NETDATA_BACKFILL_H

4 +#define NETDATA_BACKFILL_H

5 +

6 +#include "database/rrd.h"

7 +

8 +struct parser;

9 +struct backfill_request_data {

10 + size_t rrdhost_receiver_state_id;

11 + struct parser *parser;

12 + RRDHOST *host;

13 + RRDSET *st;

14 + time_t first_entry_child;

15 + time_t last_entry_child;

16 + time_t child_wall_clock_time;

17 +};

18 +

19 +typedef void (*backfill_callback_t)(size_t successful_dims, size_t failed_dims, struct backfill_request_data *brd);

20 +

21 +void *backfill_thread(void *ptr);

22 +bool backfill_request_add(RRDSET *st, backfill_callback_t cb, struct backfill_request_data *data);

23 +

24 +bool backfill_threads_detect_from_stream_conf(void);

25 +

26 +#endif //NETDATA_BACKFILL_H

src/web/api/queries/query.c

+9 -7

@@ -1964,16 +1964,16 @@ static void rrd2rrdr_query_execute(RRDR *r, size_t dim_id_in_rrdr, QUERY_ENGINE_

1964

1965 void store_metric_at_tier(RRDDIM *rd, size_t tier, struct rrddim_tier *t, STORAGE_POINT sp, usec_t now_ut);

1966

1967 -void backfill_tier_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now_s) {

1968 - if(unlikely(tier >= storage_tiers)) return;

1967 +bool backfill_tier_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now_s) {

1968 + if(unlikely(tier >= storage_tiers)) return false;

1969 #ifdef ENABLE_DBENGINE

1970 - if(default_backfill == RRD_BACKFILL_NONE) return;

1970 + if(default_backfill == RRD_BACKFILL_NONE) return false;

1971 #else

1972 - return;

1972 + return false;

1973 #endif

1974

1975 struct rrddim_tier *t = &rd->tiers[tier];

1976 - if(unlikely(!t)) return;

1976 + if(unlikely(!t)) return false;

1977

1978 time_t latest_time_s = storage_engine_latest_time_s(t->seb, t->smh);

1979 time_t granularity = (time_t)t->tier_grouping * (time_t)rd->rrdset->update_every;

@@ -1981,13 +1981,13 @@ void backfill_tier_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now_s) {

1981

1982 // if the user wants only NEW backfilling, and we don't have any data

1983 #ifdef ENABLE_DBENGINE

1984 - if(default_backfill == RRD_BACKFILL_NEW && latest_time_s <= 0) return;

1984 + if(default_backfill == RRD_BACKFILL_NEW && latest_time_s <= 0) return false;

1985 #else

1986 return;

1987 #endif

1988

1989 // there is really nothing we can do

1990 - if(now_s <= latest_time_s || time_diff < granularity) return;

1990 + if(now_s <= latest_time_s || time_diff < granularity) return false;

1991

1992 stream_control_backfill_query_started();

1993

@@ -2026,6 +2026,8 @@ void backfill_tier_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now_s) {

2026 }

2027

2028 stream_control_backfill_query_finished();

2029 +

2030 + return true;

2031 }

2032

2033 // ----------------------------------------------------------------------------

src/web/server/h2o/http_server.c

+1 -1

@@ -423,6 +423,6 @@ void *h2o_main(void *ptr) {

423 return NULL;

424 }

425

426 -int httpd_is_enabled() {

426 +bool httpd_is_enabled() {

427 return config_get_boolean(HTTPD_CONFIG_SECTION, "enabled", HTTPD_ENABLED_DEFAULT);

428 }

src/web/server/h2o/http_server.h

+1 -1

@@ -10,6 +10,6 @@ void *h2o_main(void * ptr);

10 int h2o_stream_write(void *ctx, const char *data, size_t data_len);

11 size_t h2o_stream_read(void *ctx, char *buf, size_t read_bytes);

13 -int httpd_is_enabled();

13 +bool httpd_is_enabled();

15 #endif /* HTTP_SERVER_H */