@cryptotaxi247 / netdata-1 / commits / ef816f830

Rework maintenance thread (#20694)

* Call maintenance functions from metadata event loop Remove service (MAINTENANCE) thread * run_maintenace every 10 seconds

Stelios Fragkakis committed Oct 16, 2025 at 18:32 UTC ef816f8300b30146efdd2b335b66d59cb34d0484
8 files changed +57 -125
src/daemon/daemon-service.c
-4
@@ -96,14 +96,10 @@ void service_signal_exit(SERVICE_TYPE service) {
96 }
97
98 static void service_to_buffer(BUFFER *wb, SERVICE_TYPE service) {
99 - if(service & SERVICE_MAINTENANCE)
100 - buffer_strcat(wb, "MAINTENANCE ");
99 if(service & SERVICE_COLLECTORS)
100 buffer_strcat(wb, "COLLECTORS ");
101 if(service & SERVICE_REPLICATION)
102 buffer_strcat(wb, "REPLICATION ");
105 - if(service & ABILITY_DATA_QUERIES)
106 - buffer_strcat(wb, "DATA_QUERIES ");
103 if(service & ABILITY_WEB_REQUESTS)
104 buffer_strcat(wb, "WEB_REQUESTS ");
105 if(service & SERVICE_WEB_SERVER)
src/daemon/daemon-service.h
+14 -16
@@ -6,22 +6,20 @@
6 #include "libnetdata/libnetdata.h"
7
8 typedef enum {
9 - ABILITY_DATA_QUERIES = (1 << 0),
10 - ABILITY_WEB_REQUESTS = (1 << 1),
11 - ABILITY_STREAMING_CONNECTIONS = (1 << 2),
12 - SERVICE_MAINTENANCE = (1 << 3),
13 - SERVICE_COLLECTORS = (1 << 4),
14 - SERVICE_REPLICATION = (1 << 5),
15 - SERVICE_WEB_SERVER = (1 << 6),
16 - SERVICE_ACLK = (1 << 7),
17 - SERVICE_HEALTH = (1 << 8),
18 - SERVICE_STREAMING = (1 << 9),
19 - SERVICE_STREAMING_CONNECTOR = (1 << 10),
20 - SERVICE_CONTEXT = (1 << 11),
21 - SERVICE_ANALYTICS = (1 << 12),
22 - SERVICE_EXPORTERS = (1 << 13),
23 - SERVICE_HTTPD = (1 << 14),
24 - SERVICE_SYSTEMD = (1 << 15),
9 + ABILITY_WEB_REQUESTS = (1 << 0),
10 + ABILITY_STREAMING_CONNECTIONS = (1 << 1),
11 + SERVICE_COLLECTORS = (1 << 2),
12 + SERVICE_REPLICATION = (1 << 3),
13 + SERVICE_WEB_SERVER = (1 << 4),
14 + SERVICE_ACLK = (1 << 5),
15 + SERVICE_HEALTH = (1 << 6),
16 + SERVICE_STREAMING = (1 << 7),
17 + SERVICE_STREAMING_CONNECTOR = (1 << 8),
18 + SERVICE_CONTEXT = (1 << 9),
19 + SERVICE_ANALYTICS = (1 << 10),
20 + SERVICE_EXPORTERS = (1 << 11),
21 + SERVICE_HTTPD = (1 << 12),
22 + SERVICE_SYSTEMD = (1 << 13),
23 } SERVICE_TYPE;
24
25 typedef void (*force_quit_t)(void *data);
src/daemon/daemon-shutdown.c
+2 -3
@@ -204,14 +204,13 @@ static void netdata_cleanup_and_exit(EXIT_REASON reason, bool abnormal, bool exi
204 webrtc_close_all_connections();
205 watcher_step_complete(WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS);
206
207 - service_signal_exit(SERVICE_MAINTENANCE | ABILITY_DATA_QUERIES | ABILITY_WEB_REQUESTS | SERVICE_ACLK |
208 - ABILITY_STREAMING_CONNECTIONS | SERVICE_SYSTEMD);
207 + service_signal_exit(ABILITY_WEB_REQUESTS | SERVICE_ACLK | ABILITY_STREAMING_CONNECTIONS | SERVICE_SYSTEMD);
208
209 service_signal_exit(SERVICE_EXPORTERS | SERVICE_HEALTH | SERVICE_WEB_SERVER | SERVICE_HTTPD);
210
211 watcher_step_complete(WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS);
212
214 - service_wait_exit(SERVICE_MAINTENANCE | SERVICE_SYSTEMD, 5 * USEC_PER_SEC);
213 + service_wait_exit(SERVICE_SYSTEMD, 5 * USEC_PER_SEC);
214 watcher_step_complete(WATCHER_STEP_ID_STOP_MAINTENANCE_THREAD);
215
216 service_wait_exit(SERVICE_EXPORTERS | SERVICE_HEALTH | SERVICE_WEB_SERVER | SERVICE_HTTPD, 3 * USEC_PER_SEC);
src/daemon/libuv_workers.c
+10
@@ -86,6 +86,16 @@ static void register_libuv_worker_jobs_internal(void) {
86 // netdatacli
87 worker_register_job_name(UV_EVENT_SCHEDULE_CMD, "schedule command");
88
89 + // maintenance
90 + worker_register_job_name(UV_EVENT_CLEANUP_OBSOLETE_CHARTS, "cleanup obsolete charts");
91 + worker_register_job_name(UV_EVENT_ARCHIVE_CHART_DIMENSIONS, "archive chart dimensions");
92 + worker_register_job_name(UV_EVENT_ARCHIVE_DIMENSION, "archive dimension");
93 + worker_register_job_name(UV_EVENT_CLEANUP_ORPHAN_HOSTS, "cleanup orphan hosts");
94 + worker_register_job_name(UV_EVENT_CLEANUP_OBSOLETE_CHARTS_ON_HOSTS, "cleanup obsolete charts on all hosts");
95 + worker_register_job_name(UV_EVENT_FREE_HOST, "free host");
96 + worker_register_job_name(UV_EVENT_FREE_CHART, "free chart");
97 + worker_register_job_name(UV_EVENT_FREE_DIMENSION, "free dimension");
98 +
99 // make sure we have the right thread id
100 gettid_uncached();
101
src/daemon/libuv_workers.h
+10
@@ -82,6 +82,16 @@ enum event_loop_job {
82 UV_EVENT_CREATE_NODE_INSTANCE,
83 UV_EVENT_UNREGISTER_NODE,
84
85 + // maintenance
86 + UV_EVENT_CLEANUP_OBSOLETE_CHARTS,
87 + UV_EVENT_ARCHIVE_CHART_DIMENSIONS,
88 + UV_EVENT_ARCHIVE_DIMENSION,
89 + UV_EVENT_CLEANUP_ORPHAN_HOSTS,
90 + UV_EVENT_CLEANUP_OBSOLETE_CHARTS_ON_HOSTS,
91 + UV_EVENT_FREE_HOST,
92 + UV_EVENT_FREE_CHART,
93 + UV_EVENT_FREE_DIMENSION,
94 +
95 // netdatacli
96 UV_EVENT_SCHEDULE_CMD,
97 };
src/daemon/service.c
+11 -93
@@ -2,24 +2,6 @@
2
3 #include "common.h"
4
5 -/* Run service jobs every X seconds */
6 -#define SERVICE_HEARTBEAT 10
7 -
8 -#define WORKER_JOB_CHILD_CHART_OBSOLETION_CHECK 1
9 -#define WORKER_JOB_CLEANUP_OBSOLETE_CHARTS 2
10 -#define WORKER_JOB_ARCHIVE_CHART 3
11 -#define WORKER_JOB_ARCHIVE_CHART_DIMENSIONS 4
12 -#define WORKER_JOB_ARCHIVE_DIMENSION 5
13 -#define WORKER_JOB_CLEANUP_ORPHAN_HOSTS 6
14 -#define WORKER_JOB_CLEANUP_OBSOLETE_CHARTS_ON_HOSTS 7
15 -#define WORKER_JOB_FREE_HOST 9
16 -#define WORKER_JOB_FREE_CHART 12
17 -#define WORKER_JOB_FREE_DIMENSION 15
18 -#define WORKER_JOB_PGC_MAIN_EVICT 16
19 -#define WORKER_JOB_PGC_MAIN_FLUSH 17
20 -#define WORKER_JOB_PGC_OPEN_EVICT 18
21 -#define WORKER_JOB_PGC_OPEN_FLUSH 19
22 -
5 static bool svc_rrddim_obsolete_to_archive(RRDDIM *rd) {
6 RRDSET *st = rd->rrdset;
7
@@ -32,7 +14,7 @@ static bool svc_rrddim_obsolete_to_archive(RRDDIM *rd) {
14 else
15 return false;
16
35 - worker_is_busy(WORKER_JOB_ARCHIVE_DIMENSION);
17 + worker_is_busy(UV_EVENT_ARCHIVE_DIMENSION);
18
19 if (rd->rrd_memory_mode == RRD_DB_MODE_DBENGINE) {
20 if (!rrddim_finalize_collection_and_check_retention(rd)) {
@@ -41,7 +23,7 @@ static bool svc_rrddim_obsolete_to_archive(RRDDIM *rd) {
23 }
24 }
25
44 - worker_is_busy(WORKER_JOB_FREE_DIMENSION);
26 + worker_is_busy(UV_EVENT_FREE_DIMENSION);
27 rrddim_free(st, rd);
28 return true;
29 }
@@ -50,7 +32,7 @@ static inline bool svc_rrdset_archive_obsolete_dimensions(RRDSET *st, bool all_d
32 if(!all_dimensions && !rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE_DIMENSIONS))
33 return true;
34
53 - worker_is_busy(WORKER_JOB_ARCHIVE_CHART_DIMENSIONS);
35 + worker_is_busy(UV_EVENT_ARCHIVE_CHART_DIMENSIONS);
36
37 rrdset_flag_clear(st, RRDSET_FLAG_OBSOLETE_DIMENSIONS);
38
@@ -69,14 +51,9 @@ static inline bool svc_rrdset_archive_obsolete_dimensions(RRDSET *st, bool all_d
51 if(rd->collector.last_collected_time.tv_sec + rrdset_free_obsolete_time_s < now) {
52 size_t references = dictionary_acquired_item_references(rd_dfe.item);
53 if(references == 1) {
72 -// netdata_log_info("Removing obsolete dimension 'host:%s/chart:%s/dim:%s'",
73 -// rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_id(rd));
54 if(svc_rrddim_obsolete_to_archive(rd))
55 dim_archives++;
56 }
77 -// else
78 -// netdata_log_info("Cannot remove obsolete dimension 'host:%s/chart:%s/dim:%s'",
79 -// rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_id(rd));
57 }
58 }
59 }
@@ -109,7 +86,7 @@ static inline void svc_rrdhost_cleanup_charts_marked_obsolete(RRDHOST *host) {
86 if(!rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_OBSOLETE_CHARTS|RRDHOST_FLAG_PENDING_OBSOLETE_DIMENSIONS))
87 return;
88
112 - worker_is_busy(WORKER_JOB_CLEANUP_OBSOLETE_CHARTS);
89 + worker_is_busy(UV_EVENT_CLEANUP_OBSOLETE_CHARTS);
90
91 rrdhost_flag_clear(host, RRDHOST_FLAG_PENDING_OBSOLETE_CHARTS|RRDHOST_FLAG_PENDING_OBSOLETE_DIMENSIONS);
92
@@ -140,7 +117,7 @@ static inline void svc_rrdhost_cleanup_charts_marked_obsolete(RRDHOST *host) {
117 if(svc_rrdset_archive_obsolete_dimensions(st, true)) {
118 full_archives++;
119
143 - worker_is_busy(WORKER_JOB_FREE_CHART);
120 + worker_is_busy(UV_EVENT_FREE_CHART);
121 rrdset_free(st);
122 }
123 else
@@ -170,16 +147,12 @@ void svc_rrdhost_obsolete_all_charts(RRDHOST *host) {
147 }
148
149 static void svc_rrd_cleanup_obsolete_charts_from_all_hosts() {
173 - worker_is_busy(WORKER_JOB_CLEANUP_OBSOLETE_CHARTS_ON_HOSTS);
150 + worker_is_busy(UV_EVENT_CLEANUP_OBSOLETE_CHARTS_ON_HOSTS);
151
152 rrd_rdlock();
153
154 RRDHOST *host;
155 rrdhost_foreach_read(host) {
179 -
180 - if (!service_running(SERVICE_MAINTENANCE))
181 - break;
182 -
156 if(rrdhost_receiver_replicating_charts(host) || rrdhost_sender_replicating_charts(host))
157 continue;
158
@@ -205,7 +178,7 @@ static void svc_rrd_cleanup_obsolete_charts_from_all_hosts() {
178 }
179
180 static void svc_rrdhost_cleanup_orphan_hosts(RRDHOST *protected_host) {
208 - worker_is_busy(WORKER_JOB_CLEANUP_ORPHAN_HOSTS);
181 + worker_is_busy(UV_EVENT_CLEANUP_ORPHAN_HOSTS);
182
183 time_t now = now_realtime_sec();
184
@@ -233,7 +206,7 @@ static void svc_rrdhost_cleanup_orphan_hosts(RRDHOST *protected_host) {
206 continue;
207 }
208
236 - worker_is_busy(WORKER_JOB_FREE_HOST);
209 + worker_is_busy(UV_EVENT_FREE_HOST);
210
211 if (delete) {
212 netdata_log_info("Host '%s' with machine guid '%s' is archived, ephemeral clean up.", rrdhost_hostname(host), host->machine_guid);
@@ -248,62 +221,7 @@ static void svc_rrdhost_cleanup_orphan_hosts(RRDHOST *protected_host) {
221 rrd_wrunlock();
222 }
223
251 -static void service_main_cleanup(void *pptr)
252 -{
253 - struct netdata_static_thread *static_thread = CLEANUP_FUNCTION_GET_PTR(pptr);
254 - if(!static_thread) return;
255 -
256 - static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
257 -
258 - worker_unregister();
259 -
260 - static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
261 -}
262 -
263 -/*
264 - * The service thread.
265 - */
266 -void *service_main(void *ptr)
267 -{
268 - worker_register("SERVICE");
269 - worker_register_job_name(WORKER_JOB_CHILD_CHART_OBSOLETION_CHECK, "child chart obsoletion check");
270 - worker_register_job_name(WORKER_JOB_CLEANUP_OBSOLETE_CHARTS, "cleanup obsolete charts");
271 - worker_register_job_name(WORKER_JOB_ARCHIVE_CHART, "archive chart");
272 - worker_register_job_name(WORKER_JOB_ARCHIVE_CHART_DIMENSIONS, "archive chart dimensions");
273 - worker_register_job_name(WORKER_JOB_ARCHIVE_DIMENSION, "archive dimension");
274 - worker_register_job_name(WORKER_JOB_CLEANUP_ORPHAN_HOSTS, "cleanup orphan hosts");
275 - worker_register_job_name(WORKER_JOB_CLEANUP_OBSOLETE_CHARTS_ON_HOSTS, "cleanup obsolete charts on all hosts");
276 - worker_register_job_name(WORKER_JOB_FREE_HOST, "free host");
277 - worker_register_job_name(WORKER_JOB_FREE_CHART, "free chart");
278 - worker_register_job_name(WORKER_JOB_FREE_DIMENSION, "free dimension");
279 - worker_register_job_name(WORKER_JOB_PGC_MAIN_EVICT, "main cache evictions");
280 - worker_register_job_name(WORKER_JOB_PGC_MAIN_FLUSH, "main cache flushes");
281 - worker_register_job_name(WORKER_JOB_PGC_OPEN_EVICT, "open cache evictions");
282 - worker_register_job_name(WORKER_JOB_PGC_OPEN_FLUSH, "open cache flushes");
283 -
284 - CLEANUP_FUNCTION_REGISTER(service_main_cleanup) cleanup_ptr = ptr;
285 -
286 - heartbeat_t hb;
287 - heartbeat_init(&hb, USEC_PER_SEC);
288 - usec_t step = USEC_PER_SEC * SERVICE_HEARTBEAT;
289 - usec_t real_step = USEC_PER_SEC;
290 -
291 - netdata_log_debug(D_SYSTEM, "Service thread starts");
292 -
293 - while (service_running(SERVICE_MAINTENANCE)) {
294 - worker_is_idle();
295 - heartbeat_next(&hb);
296 - if (real_step < step) {
297 - real_step += USEC_PER_SEC;
298 - continue;
299 - }
300 - real_step = USEC_PER_SEC;
301 -
302 - svc_rrd_cleanup_obsolete_charts_from_all_hosts();
303 -
304 - if (service_running(SERVICE_MAINTENANCE))
305 - svc_rrdhost_cleanup_orphan_hosts(localhost);
306 - }
307 -
308 - return NULL;
224 +void run_maintenace() {
225 + svc_rrd_cleanup_obsolete_charts_from_all_hosts();
226 + svc_rrdhost_cleanup_orphan_hosts(localhost);
227 }
src/daemon/static_threads.c
-9
@@ -97,15 +97,6 @@ const struct netdata_static_thread static_threads_common[] = {
97 .init_routine = NULL,
98 .start_routine = pluginsd_main
99 },
100 - {
101 - .name = "SERVICE",
102 - .config_section = NULL,
103 - .config_name = NULL,
104 - .enabled = 1,
105 - .thread = NULL,
106 - .init_routine = NULL,
107 - .start_routine = service_main
108 - },
100 {
101 .name = "STATSD_FLUSH",
102 .config_section = NULL,
src/database/sqlite/sqlite_metadata.c
+10
@@ -2429,11 +2429,21 @@ static void store_hosts_metadata(struct meta_config_s *config, BUFFER *work_buff
2429 }
2430 }
2431
2432 +#define SERVICE_HEARTBEAT 10
2433 +void run_maintenace();
2434 +
2435 // Worker thread to scan hosts for pending metadata to store
2436 static void start_metadata_hosts(uv_work_t *req)
2437 {
2438 + static time_t next_maintenance_check = 0;
2439 register_libuv_worker_jobs();
2440
2441 + time_t now = now_realtime_sec();
2442 + if (now> next_maintenance_check) {
2443 + run_maintenace();
2444 + next_maintenance_check = now + SERVICE_HEARTBEAT;
2445 + }
2446 +
2447 worker_data_t *worker = req->data;
2448 struct meta_config_s *config = worker->config;
2449