@cryptotaxi247 / netdata-1
Code Commits Issues Pulls Sessions CI/CD
master
master
c 613 lines 20.7 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "systemd-internals.h"
4 #include <sys/inotify.h>
5
6 #define INITIAL_WATCHES 256
7
8 #define WATCH_FOR (IN_CREATE | IN_MODIFY | IN_DELETE | IN_DELETE_SELF | IN_MOVED_FROM | IN_MOVED_TO | IN_UNMOUNT)
9
10 typedef uint32_t INOTIFY_MASK;
11
12 ENUM_STR_MAP_DEFINE(INOTIFY_MASK) = {
13 // helpers (combine multiple flags)
14 // must be first in the list
15 {.id = IN_ALL_EVENTS, .name = "IN_ALL_EVENTS"},
16 {.id = IN_CLOSE, .name = "IN_CLOSE"},
17 {.id = IN_MOVE, .name = "IN_MOVE"},
18
19 // individual flags
20 {.id = IN_ACCESS, .name = "IN_ACCESS"},
21 {.id = IN_MODIFY, .name = "IN_MODIFY"},
22 {.id = IN_ATTRIB, .name = "IN_ATTRIB"},
23 {.id = IN_CLOSE_WRITE, .name = "IN_CLOSE_WRITE"},
24 {.id = IN_CLOSE_NOWRITE, .name = "IN_CLOSE_NOWRITE"},
25 {.id = IN_OPEN, .name = "IN_OPEN"},
26 {.id = IN_MOVED_FROM, .name = "IN_MOVED_FROM"},
27 {.id = IN_MOVED_TO, .name = "IN_MOVED_TO"},
28 {.id = IN_CREATE, .name = "IN_CREATE"},
29 {.id = IN_DELETE, .name = "IN_DELETE"},
30 {.id = IN_DELETE_SELF, .name = "IN_DELETE_SELF"},
31 {.id = IN_MOVE_SELF, .name = "IN_MOVE_SELF"},
32 {.id = IN_UNMOUNT, .name = "IN_UNMOUNT"},
33 {.id = IN_Q_OVERFLOW, .name = "IN_Q_OVERFLOW"},
34 {.id = IN_IGNORED, .name = "IN_IGNORED"},
35 {.id = IN_ONLYDIR, .name = "IN_ONLYDIR"},
36 {.id = IN_DONT_FOLLOW, .name = "IN_DONT_FOLLOW"},
37 {.id = IN_EXCL_UNLINK, .name = "IN_EXCL_UNLINK"},
38 #ifdef IN_MASK_CREATE
39 {.id = IN_MASK_CREATE, .name = "IN_MASK_CREATE"},
40 #endif
41 {.id = IN_MASK_ADD, .name = "IN_MASK_ADD"},
42 {.id = IN_ISDIR, .name = "IN_ISDIR"},
43 {.id = IN_ONESHOT, .name = "IN_ONESHOT"},
44
45 // terminator
46 {.id = 0, .name = NULL}};
47
48 BITMAP_STR_DEFINE_FUNCTIONS(INOTIFY_MASK, 0, "UNKNOWN");
49
50 DEFINE_JUDYL_TYPED(SYMLINKED_DIRS, STRING *);
51
52 typedef struct watch_entry {
53 int slot;
54
55 int wd; // Watch descriptor
56 char *path; // Dynamically allocated path
57
58 struct watch_entry *next; // for the free list
59 } WatchEntry;
60
61 typedef struct {
62 WatchEntry *watchList;
63 WatchEntry *freeList;
64 int watchCount;
65 int watchListSize;
66
67 size_t errors;
68
69 SYMLINKED_DIRS_JudyLSet symlinkedDirs;
70 DICTIONARY *pending;
71 } Watcher;
72
73 static WatchEntry *get_slot(Watcher *watcher)
74 {
75 WatchEntry *t;
76
77 if (watcher->freeList != NULL) {
78 t = watcher->freeList;
79 watcher->freeList = t->next;
80 t->next = NULL;
81 return t;
82 }
83
84 if (watcher->watchCount == watcher->watchListSize) {
85 watcher->watchListSize *= 2;
86 watcher->watchList = reallocz(watcher->watchList, watcher->watchListSize * sizeof(WatchEntry));
87 }
88
89 watcher->watchList[watcher->watchCount] = (WatchEntry){
90 .slot = watcher->watchCount,
91 .wd = -1,
92 .path = NULL,
93 .next = NULL,
94 };
95 t = &watcher->watchList[watcher->watchCount];
96 watcher->watchCount++;
97
98 return t;
99 }
100
101 static void free_slot(Watcher *watcher, WatchEntry *t)
102 {
103 t->wd = -1;
104 freez(t->path);
105 t->path = NULL;
106
107 // link it to the free list
108 t->next = watcher->freeList;
109 watcher->freeList = t;
110 }
111
112 static int add_watch(Watcher *watcher, int inotifyFd, const char *path)
113 {
114 WatchEntry *t = get_slot(watcher);
115
116 errno_clear();
117 t->wd = inotify_add_watch(inotifyFd, path, WATCH_FOR);
118 if (t->wd == -1) {
119 nd_log(NDLS_COLLECTORS, NDLP_ERR, "JOURNAL WATCHER: cannot watch directory: '%s'", path);
120
121 free_slot(watcher, t);
122
123 struct stat info;
124 if (stat(path, &info) == 0 && S_ISDIR(info.st_mode)) {
125 // the directory exists, but we failed to add the watch
126 // increase errors
127 watcher->errors++;
128 }
129 } else {
130 t->path = strdupz(path);
131
132 nd_log(NDLS_COLLECTORS, NDLP_DEBUG, "JOURNAL WATCHER: watching directory: '%s'", path);
133 }
134 return t->wd;
135 }
136
137 static void remove_watch(Watcher *watcher, int inotifyFd, int wd)
138 {
139 errno_clear();
140
141 int i;
142 for (i = 0; i < watcher->watchCount; ++i) {
143 if (watcher->watchList[i].wd == wd) {
144 nd_log(
145 NDLS_COLLECTORS,
146 NDLP_DEBUG,
147 "JOURNAL WATCHER: removing watch from directory: '%s'",
148 watcher->watchList[i].path);
149
150 if (inotify_rm_watch(inotifyFd, watcher->watchList[i].wd) == -1)
151 nd_log(NDLS_COLLECTORS, NDLP_ERR, "JOURNAL WATCHER: inotify_rm_watch() returned -1");
152
153 free_slot(watcher, &watcher->watchList[i]);
154 return;
155 }
156 }
157
158 nd_log(NDLS_COLLECTORS, NDLP_WARNING, "JOURNAL WATCHER: cannot find directory watch %d to remove.", wd);
159 }
160
161 static void free_watches(Watcher *watcher, int inotifyFd)
162 {
163 for (int i = 0; i < watcher->watchCount; ++i) {
164 if (watcher->watchList[i].wd != -1) {
165 if (inotify_rm_watch(inotifyFd, watcher->watchList[i].wd) == -1)
166 nd_log(NDLS_COLLECTORS, NDLP_ERR, "JOURNAL WATCHER: inotify_rm_watch() returned -1");
167 free_slot(watcher, &watcher->watchList[i]);
168 }
169 }
170 freez(watcher->watchList);
171 watcher->watchList = NULL;
172
173 dictionary_destroy(watcher->pending);
174 watcher->pending = NULL;
175 }
176
177 static void free_symlinked_dirs(Watcher *watcher)
178 {
179 Word_t idx = 0;
180 STRING *value;
181 while ((value = SYMLINKED_DIRS_FIRST(&watcher->symlinkedDirs, &idx))) {
182 SYMLINKED_DIRS_DEL(&watcher->symlinkedDirs, idx);
183 STRING *key = (STRING *)idx;
184 string_freez(key);
185 string_freez(value);
186 }
187 }
188
189 static char *get_path_from_wd(Watcher *watcher, int wd)
190 {
191 for (int i = 0; i < watcher->watchCount; ++i) {
192 if (watcher->watchList[i].wd == wd)
193 return watcher->watchList[i].path;
194 }
195 return NULL;
196 }
197
198 static bool is_directory_watched(Watcher *watcher, const char *path)
199 {
200 for (int i = 0; i < watcher->watchCount; ++i) {
201 if (watcher->watchList[i].wd != -1 && strcmp(watcher->watchList[i].path, path) == 0) {
202 return true;
203 }
204 }
205 return false;
206 }
207
208 static void watch_directory_and_subdirectories(Watcher *watcher, int inotifyFd, const char *basePath)
209 {
210 DICTIONARY *dirs = dictionary_create(DICT_OPTION_SINGLE_THREADED | DICT_OPTION_DONT_OVERWRITE_VALUE);
211
212 // First resolve any symlinks in the base path
213 char real_path[PATH_MAX];
214 if (realpath(basePath, real_path) == NULL) {
215 // If realpath fails, try using the original path
216 strncpyz(real_path, basePath, sizeof(real_path));
217 }
218
219 nd_journal_directory_scan_recursively(NULL, dirs, real_path, 0);
220
221 void *x;
222 dfe_start_read(dirs, x)
223 {
224 const char *dirname = x_dfe.name;
225 char resolved_path[PATH_MAX];
226
227 // Resolve symlinks for each subdirectory
228 if (realpath(dirname, resolved_path) != NULL) {
229 // Check if this directory is already being watched
230 if (!is_directory_watched(watcher, resolved_path)) {
231 add_watch(watcher, inotifyFd, resolved_path);
232 }
233 } else {
234 // If realpath fails, try with original path
235 if (!is_directory_watched(watcher, dirname)) {
236 add_watch(watcher, inotifyFd, dirname);
237 }
238 }
239 }
240 dfe_done(x);
241
242 dictionary_destroy(dirs);
243 }
244
245 static bool is_subpath(const char *path, const char *subpath)
246 {
247 // Use strncmp to compare the paths
248 if (strncmp(path, subpath, strlen(path)) == 0) {
249 // Ensure that the next character is a '/' or '\0'
250 char next_char = subpath[strlen(path)];
251 return next_char == '/' || next_char == '\0';
252 }
253
254 return false;
255 }
256
257 void remove_directory_watch(Watcher *watcher, int inotifyFd, const char *dirPath)
258 {
259 for (int i = 0; i < watcher->watchCount; ++i) {
260 WatchEntry *t = &watcher->watchList[i];
261 if (t->wd != -1 && is_subpath(dirPath, t->path)) {
262 if (inotify_rm_watch(inotifyFd, t->wd) == -1)
263 nd_log(
264 NDLS_COLLECTORS, NDLP_ERR, "JOURNAL WATCHER: inotify_rm_watch() on path '%s' returned -1", t->path);
265 else
266 nd_log(NDLS_COLLECTORS, NDLP_DEBUG, "JOURNAL WATCHER: stopped watching directory '%s'", t->path);
267 free_slot(watcher, t);
268 }
269 }
270
271 struct nd_journal_file *njf;
272 dfe_start_write(nd_journal_files_registry, njf)
273 {
274 if (is_subpath(dirPath, njf->filename))
275 dictionary_del(nd_journal_files_registry, njf->filename);
276 }
277 dfe_done(njf);
278
279 dictionary_garbage_collect(nd_journal_files_registry);
280 }
281
282 void process_event(Watcher *watcher, int inotifyFd, struct inotify_event *event)
283 {
284 errno_clear();
285
286 if (!event->len) {
287 CLEAN_BUFFER *wb = buffer_create(0, NULL);
288 INOTIFY_MASK_2buffer(wb, event->mask, ", ");
289 nd_log(
290 NDLS_COLLECTORS,
291 NDLP_NOTICE,
292 "JOURNAL WATCHER: received event with mask %u (%s) and len %u (this is zero) - ignoring it.",
293 event->mask,
294 buffer_tostring(wb),
295 event->len);
296 return;
297 }
298
299 char *dirPath = get_path_from_wd(watcher, event->wd);
300 if (!dirPath) {
301 CLEAN_BUFFER *wb = buffer_create(0, NULL);
302 INOTIFY_MASK_2buffer(wb, event->mask, ", ");
303 nd_log(
304 NDLS_COLLECTORS,
305 NDLP_NOTICE,
306 "JOURNAL WATCHER: received event with mask %u (%s) and len %u for path: '%s' - "
307 "but we can't find its watch descriptor - ignoring it.",
308 event->mask,
309 buffer_tostring(wb),
310 event->len,
311 event->name);
312 return;
313 }
314
315 #if 0
316 {
317 CLEAN_BUFFER *wb = buffer_create(0, NULL);
318 INOTIFY_MASK_2buffer(wb, event->mask, ", ");
319 nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
320 "JOURNAL WATCHER: received event with mask %u (%s) for path: '%s' inside '%s'",
321 event->mask, buffer_tostring(wb), event->name, dirPath);
322 }
323 #endif
324
325 if (event->mask & IN_DELETE_SELF) {
326 remove_watch(watcher, inotifyFd, event->wd);
327 return;
328 }
329
330 static __thread char fullPath[PATH_MAX];
331 snprintfz(fullPath, sizeof(fullPath), "%s/%s", dirPath, event->name);
332
333 bool is_dir = event->mask & IN_ISDIR;
334 char resolved_path[PATH_MAX];
335 const char *path_to_use = fullPath;
336
337 if (event->mask & (IN_CREATE | IN_MOVED_TO)) {
338 // Give the system a moment to establish the symlink
339 sleep_usec(1000); // 1ms sleep
340
341 struct stat st;
342 if (lstat(fullPath, &st) == 0) {
343 if (S_ISLNK(st.st_mode)) {
344 // It's a symlink - resolve it
345 if (realpath(fullPath, resolved_path) != NULL) {
346 path_to_use = resolved_path;
347
348 // Check if it points to a directory
349 if (stat(resolved_path, &st) == 0 && S_ISDIR(st.st_mode)) {
350 is_dir = true;
351
352 STRING *fullPathString = string_strdupz(fullPath);
353 STRING *symlinked = SYMLINKED_DIRS_GET(&watcher->symlinkedDirs, (uintptr_t)fullPathString);
354 if (!symlinked) {
355 SYMLINKED_DIRS_SET(
356 &watcher->symlinkedDirs, (uintptr_t)fullPathString, string_strdupz(resolved_path));
357
358 // we leave fullPathString allocated, as it's now in the JudyL set
359
360 nd_log(
361 NDLS_COLLECTORS,
362 NDLP_DEBUG,
363 "JOURNAL WATCHER: New symlinked directory created: '%s' -> '%s'",
364 fullPath,
365 resolved_path);
366 } else if (string_strcmp(symlinked, resolved_path) != 0) {
367 SYMLINKED_DIRS_SET(
368 &watcher->symlinkedDirs, (uintptr_t)fullPathString, string_strdupz(resolved_path));
369
370 nd_log(
371 NDLS_COLLECTORS,
372 NDLP_DEBUG,
373 "JOURNAL WATCHER: Updated symlinked directory: '%s' -> '%s' (was '%s')",
374 fullPath,
375 resolved_path,
376 string2str(symlinked));
377
378 string_freez(symlinked);
379
380 // we need to free this, since it was already in the JudyL set
381 string_freez(fullPathString);
382 } else
383 string_freez(fullPathString); // we don't need it anymore
384 }
385 }
386 }
387 }
388 } else if (event->mask & IN_DELETE) {
389 // Check if it was a symlink
390 STRING *fullPathString = string_strdupz(fullPath);
391 STRING *symlinked = SYMLINKED_DIRS_GET(&watcher->symlinkedDirs, (uintptr_t)fullPathString);
392 if (symlinked) {
393 strncpyz(resolved_path, string2str(symlinked), sizeof(resolved_path) - 1);
394 path_to_use = resolved_path;
395 SYMLINKED_DIRS_DEL(&watcher->symlinkedDirs, (uintptr_t)fullPathString);
396 string_freez(fullPathString); // to remove also the one referenced in the JudyL set
397 string_freez(symlinked);
398 is_dir = true;
399
400 nd_log(
401 NDLS_COLLECTORS,
402 NDLP_DEBUG,
403 "JOURNAL WATCHER: Deleted symlinked directory: '%s' -> '%s'",
404 fullPath,
405 resolved_path);
406 }
407 string_freez(fullPathString); // the one we allocated above
408 }
409
410 if (is_dir) {
411 if (event->mask & (IN_DELETE | IN_MOVED_FROM)) {
412 nd_log(NDLS_COLLECTORS, NDLP_DEBUG, "JOURNAL WATCHER: Directory deleted or moved out: '%s'", path_to_use);
413
414 remove_directory_watch(watcher, inotifyFd, path_to_use);
415 } else if (event->mask & (IN_CREATE | IN_MOVED_TO)) {
416 nd_log(
417 NDLS_COLLECTORS, NDLP_DEBUG, "JOURNAL WATCHER: New directory created or moved in: '%s'", path_to_use);
418
419 watch_directory_and_subdirectories(watcher, inotifyFd, path_to_use);
420 } else {
421 CLEAN_BUFFER *wb = buffer_create(0, NULL);
422 INOTIFY_MASK_2buffer(wb, event->mask, ", ");
423 nd_log(
424 NDLS_COLLECTORS,
425 NDLP_WARNING,
426 "JOURNAL WATCHER: Received unhandled event with mask %u (%s) for directory '%s'",
427 event->mask,
428 buffer_tostring(wb),
429 path_to_use);
430 }
431 } else if (is_journal_file(event->name, (ssize_t)strlen(event->name), NULL)) {
432 dictionary_set(watcher->pending, path_to_use, NULL, 0);
433 } else {
434 CLEAN_BUFFER *wb = buffer_create(0, NULL);
435 INOTIFY_MASK_2buffer(wb, event->mask, ", ");
436 nd_log(
437 NDLS_COLLECTORS,
438 NDLP_DEBUG,
439 "JOURNAL WATCHER: ignoring event with mask %u (%s) for file '%s' ('%s')",
440 event->mask,
441 buffer_tostring(wb),
442 path_to_use,
443 fullPath);
444 }
445 }
446
447 static void process_pending(Watcher *watcher)
448 {
449 errno_clear();
450
451 void *x;
452 dfe_start_write(watcher->pending, x)
453 {
454 struct stat info;
455 const char *fullPath = x_dfe.name;
456
457 if (stat(fullPath, &info) != 0) {
458 nd_log(
459 NDLS_COLLECTORS,
460 NDLP_DEBUG,
461 "JOURNAL WATCHER: file '%s' no longer exists, removing it from the registry",
462 fullPath);
463
464 dictionary_del(nd_journal_files_registry, fullPath);
465 } else if (S_ISREG(info.st_mode)) {
466 // nd_log(NDLS_COLLECTORS, NDLP_DEBUG,
467 // "JOURNAL WATCHER: file '%s' has been added/updated, updating the registry",
468 // fullPath);
469
470 struct nd_journal_file t = {
471 .file_last_modified_ut = info.st_mtim.tv_sec * USEC_PER_SEC + info.st_mtim.tv_nsec / NSEC_PER_USEC,
472 .last_scan_monotonic_ut = now_monotonic_usec(),
473 .size = info.st_size,
474 .max_journal_vs_realtime_delta_ut = JOURNAL_VS_REALTIME_DELTA_DEFAULT_UT,
475 };
476 struct nd_journal_file *jf = dictionary_set(nd_journal_files_registry, fullPath, &t, sizeof(t));
477 nd_journal_file_update_header(jf->filename, jf);
478 }
479
480 dictionary_del(watcher->pending, fullPath);
481 }
482 dfe_done(x);
483
484 dictionary_garbage_collect(watcher->pending);
485 }
486
487 size_t journal_watcher_wanted_session_id = 0;
488
489 void nd_journal_watcher_restart(void)
490 {
491 __atomic_add_fetch(&journal_watcher_wanted_session_id, 1, __ATOMIC_RELAXED);
492 }
493
494 static bool process_inotify_events(struct buffered_reader *reader, Watcher *watcher, int inotifyFd)
495 {
496 errno_clear();
497
498 bool unmount_event = false;
499 ssize_t processed = 0;
500
501 // Process as many complete events as we can
502 while (processed + (ssize_t)sizeof(struct inotify_event) <= reader->read_len) {
503 struct inotify_event *event = (struct inotify_event *)(reader->read_buffer + processed);
504
505 if (event->len > NAME_MAX + 1) {
506 // The event length is impossibly large
507 nd_log(
508 NDLS_COLLECTORS,
509 NDLP_ERR,
510 "JOURNAL WATCHER: received impossibly large event length %u - restarting",
511 event->len);
512 return true; // force a restart
513 }
514
515 // Check if we have the complete event including the name
516 ssize_t total_size = (ssize_t)sizeof(struct inotify_event) + event->len;
517 if (processed + total_size > reader->read_len)
518 break; // Wait for more data
519
520 if (event->mask & IN_UNMOUNT) {
521 unmount_event = true;
522 break;
523 }
524
525 process_event(watcher, inotifyFd, event);
526 processed += total_size;
527 }
528
529 // If we have unprocessed data, move it to the start
530 if (processed < reader->read_len) {
531 memmove(reader->read_buffer, reader->read_buffer + processed, reader->read_len - processed);
532 reader->read_len -= processed;
533 } else
534 reader->read_len = 0;
535
536 reader->read_buffer[reader->read_len] = '\0';
537 return unmount_event;
538 }
539
540 void nd_journal_watcher_main(void *arg __maybe_unused)
541 {
542 while (1) {
543 size_t journal_watcher_session_id = __atomic_load_n(&journal_watcher_wanted_session_id, __ATOMIC_RELAXED);
544
545 Watcher watcher = {
546 .watchList = mallocz(INITIAL_WATCHES * sizeof(WatchEntry)),
547 .freeList = NULL,
548 .watchCount = 0,
549 .watchListSize = INITIAL_WATCHES,
550 .pending = dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_SINGLE_THREADED),
551 .errors = 0,
552 };
553
554 int inotifyFd = inotify_init();
555 if (inotifyFd < 0) {
556 nd_log(NDLS_COLLECTORS, NDLP_ERR, "inotify_init() failed.");
557 free_watches(&watcher, inotifyFd);
558 return;
559 }
560
561 for (unsigned i = 0; i < MAX_JOURNAL_DIRECTORIES; i++) {
562 if (!journal_directories[i].path)
563 break;
564 watch_directory_and_subdirectories(&watcher, inotifyFd, string2str(journal_directories[i].path));
565 }
566
567 usec_t last_headers_update_ut = now_monotonic_usec();
568 struct buffered_reader reader;
569 buffered_reader_init(&reader);
570
571 while (journal_watcher_session_id == __atomic_load_n(&journal_watcher_wanted_session_id, __ATOMIC_RELAXED)) {
572 buffered_reader_ret_t rc =
573 buffered_reader_read_timeout(&reader, inotifyFd, ND_SD_JOURNAL_EXECUTE_WATCHER_PENDING_EVERY_MS, false);
574
575 if (rc == BUFFERED_READER_READ_OK || rc == BUFFERED_READER_READ_BUFFER_FULL) {
576 if (process_inotify_events(&reader, &watcher, inotifyFd))
577 break;
578 } else if (rc != BUFFERED_READER_READ_POLL_TIMEOUT) {
579 nd_log(
580 NDLS_COLLECTORS,
581 NDLP_ERR,
582 "JOURNAL WATCHER: cannot read inotify events, buffered_reader_read_timeout() returned %d - "
583 "restarting the watcher.",
584 rc);
585 break;
586 }
587
588 usec_t ut = now_monotonic_usec();
589 if (dictionary_entries(watcher.pending) &&
590 (rc == BUFFERED_READER_READ_POLL_TIMEOUT ||
591 last_headers_update_ut + (ND_SD_JOURNAL_EXECUTE_WATCHER_PENDING_EVERY_MS * USEC_PER_MS) <= ut)) {
592 process_pending(&watcher);
593 last_headers_update_ut = ut;
594 }
595
596 if (watcher.errors) {
597 nd_log(
598 NDLS_COLLECTORS,
599 NDLP_NOTICE,
600 "JOURNAL WATCHER: there were errors in setting up inotify watches - restarting the watcher.");
601 }
602 }
603
604 close(inotifyFd);
605 free_watches(&watcher, inotifyFd);
606 free_symlinked_dirs(&watcher);
607
608 // this will scan the directories and cleanup the registry
609 nd_journal_files_registry_update();
610
611 sleep_usec(2 * USEC_PER_SEC);
612 }
613 }