ebpf memory cleanup (#10096)

Fix memory cleanup when process exit.

7 files changed, 180 insertions, 80 deletions

diff --git a/collectors/ebpf.plugin/ebpf.c b/collectors/ebpf.plugin/ebpf.c
index 4457723811..51ec7d9846 100644
--- a/collectors/ebpf.plugin/ebpf.c
+++ b/collectors/ebpf.plugin/ebpf.c
@@ -69,6 +69,7 @@ int running_on_kernel = 0;
 char kernel_string[64];
 int ebpf_nprocs;
 static int isrh;
+uint32_t finalized_threads = 1;
 
 pthread_mutex_t lock;
 pthread_mutex_t collect_data_mutex;
@@ -115,7 +116,6 @@ ebpf_module_t ebpf_modules[] = {
 };
 
 // Link with apps.plugin
-pid_t *pid_index;
 ebpf_process_stat_t *global_process_stat = NULL;
 
 //Network viewer
@@ -142,6 +142,7 @@ void clean_port_structure(ebpf_network_viewer_port_list_t **clean)
     ebpf_network_viewer_port_list_t *move = *clean;
     while (move) {
         ebpf_network_viewer_port_list_t *next = move->next;
+        freez(move->value);
         freez(move);
 
         move = next;
@@ -181,13 +182,13 @@ static void change_events()
  * Clean Loaded Events
  *
  * This function cleans the events previous loaded on Linux.
- */
 void clean_loaded_events()
 {
     int event_pid;
     for (event_pid = 0; ebpf_modules[event_pid].probes; event_pid++)
         clean_kprobe_events(NULL, (int)ebpf_modules[event_pid].thread_id, ebpf_modules[event_pid].probes);
 }
+ */
 
 /**
  * Close the collector gracefully
@@ -203,11 +204,9 @@ static void ebpf_exit(int sig)
         return;
     }
 
-    clean_apps_groups_target(apps_groups_root_target);
-
-    freez(pid_index);
     freez(global_process_stat);
 
+    /*
     int ret = fork();
     if (ret < 0) // error
         error("Cannot fork(), so I won't be able to clean %skprobe_events", NETDATA_DEBUGFS);
@@ -237,6 +236,7 @@ static void ebpf_exit(int sig)
     } else { // parent
         exit(0);
     }
+     */
 
     exit(sig);
 }
@@ -830,7 +830,6 @@ int ebpf_start_pthread_variables()
 static void ebpf_allocate_common_vectors()
 {
     all_pids = callocz((size_t)pid_max, sizeof(struct pid_stat *));
-    pid_index = callocz((size_t)pid_max, sizeof(pid_t));
     global_process_stat = callocz((size_t)ebpf_nprocs, sizeof(ebpf_process_stat_t));
 }
 
@@ -1969,7 +1968,7 @@ int main(int argc, char **argv)
     };
 
     change_events();
-    clean_loaded_events();
+    //clean_loaded_events();
 
     int i;
     for (i = 0; ebpf_threads[i].name != NULL; i++) {
diff --git a/collectors/ebpf.plugin/ebpf.h b/collectors/ebpf.plugin/ebpf.h
index fc7d31cab9..ebf6a4292a 100644
--- a/collectors/ebpf.plugin/ebpf.h
+++ b/collectors/ebpf.plugin/ebpf.h
@@ -31,12 +31,6 @@
 
 #include "ebpf_apps.h"
 
-typedef enum {
-    MODE_RETURN = 0, // This attaches kprobe when the function returns
-    MODE_DEVMODE,    // This stores log given description about the errors raised
-    MODE_ENTRY       // This attaches kprobe when the function is called
-} netdata_run_mode_t;
-
 typedef struct netdata_syscall_stat {
     unsigned long bytes;               // total number of bytes
     uint64_t call;                     // total number of calls
@@ -74,20 +68,6 @@ typedef struct netdata_error_report {
     int err;
 } netdata_error_report_t;
 
-typedef struct ebpf_module {
-    const char *thread_name;
-    const char *config_name;
-    int enabled;
-    void *(*start_routine)(void *);
-    int update_time;
-    int global_charts;
-    int apps_charts;
-    netdata_run_mode_t mode;
-    netdata_ebpf_events_t *probes;
-    uint32_t thread_id;
-    int optional;
-} ebpf_module_t;
-
 extern ebpf_module_t ebpf_modules[];
 #define EBPF_MODULE_PROCESS_IDX 0
 #define EBPF_MODULE_SOCKET_IDX 1
@@ -198,8 +178,8 @@ extern void write_end_chart();
 // Common variables
 extern char *ebpf_user_config_dir;
 extern char *ebpf_stock_config_dir;
-extern pid_t *pid_index;
 extern int debug_enabled;
+extern struct pid_stat *root_of_pids;
 
 // Socket functions and variables
 // Common functions
@@ -209,6 +189,7 @@ extern struct pid_stat *root_of_pids;
 extern ebpf_process_stat_t *global_process_stat;
 extern size_t all_pids_count;
 extern int update_every;
+extern uint32_t finalized_threads;
 
 #define EBPF_MAX_SYNCHRONIZATION_TIME 300
 
diff --git a/collectors/ebpf.plugin/ebpf_apps.c b/collectors/ebpf.plugin/ebpf_apps.c
index 11e72d7f6e..062c9a4e40 100644
--- a/collectors/ebpf.plugin/ebpf_apps.c
+++ b/collectors/ebpf.plugin/ebpf_apps.c
@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: GPL-3.0-or-later
 
 #include "ebpf.h"
+#include "ebpf_socket.h"
 #include "ebpf_apps.h"
 
 // ----------------------------------------------------------------------------
@@ -910,10 +911,8 @@ static inline void del_pid_entry(pid_t pid)
 
 /**
  * Remove PIDs when they are not running more.
- *
- * @param out is the structure where PIDs are stored.
  */
-void cleanup_exited_pids(ebpf_process_stat_t **out)
+void cleanup_exited_pids()
 {
     struct pid_stat *p = NULL;
 
@@ -926,10 +925,19 @@ void cleanup_exited_pids(ebpf_process_stat_t **out)
             p = p->next;
             del_pid_entry(r);
 
-            ebpf_process_stat_t *w = out[r];
-            if (w) {
-                freez(w);
-                out[r] = NULL;
+            // Clean process structure
+            freez(global_process_stats[r]);
+            global_process_stats[r] = NULL;
+
+            freez(current_apps_data[r]);
+            current_apps_data[r] = NULL;
+            prev_apps_data[r] = NULL;
+
+            // Clean socket structures
+            if (socket_bandwidth_curr) {
+                freez(socket_bandwidth_curr[r]);
+                socket_bandwidth_curr[r] = NULL;
+                socket_bandwidth_prev[r] = NULL;
             }
         } else {
             if (unlikely(p->keep))
@@ -1007,12 +1015,9 @@ static inline void aggregate_pid_on_target(struct target *w, struct pid_stat *p,
  * Read data from hash table and store it in appropriate vectors.
  * It also creates the link between targets and PIDs.
  *
- * @param out                   the output vector where we store data read from hash table.
- * @param index                 the vector to store the indexes read.
- * @param bpf_map_lookup_elem   A pointer to the function that reads the data.
  * @param tbl_pid_stats_fd      The mapped file descriptor for the hash table.
  */
-void collect_data_for_all_processes(ebpf_process_stat_t **out, pid_t *index, int tbl_pid_stats_fd)
+void collect_data_for_all_processes(int tbl_pid_stats_fd)
 {
     struct pid_stat *pids = root_of_pids; // global list of all processes running
     while (pids) {
@@ -1032,26 +1037,37 @@ void collect_data_for_all_processes(ebpf_process_stat_t **out, pid_t *index, int
 
     read_proc_filesystem();
 
-    int counter = 0;
     uint32_t key;
     pids = root_of_pids; // global list of all processes running
     // while (bpf_map_get_next_key(tbl_pid_stats_fd, &key, &next_key) == 0) {
     while (pids) {
         key = pids->pid;
-        ebpf_process_stat_t *w = out[key];
+        ebpf_process_stat_t *w = global_process_stats[key];
         if (!w) {
             w = mallocz(sizeof(ebpf_process_stat_t));
-            out[key] = w;
+            global_process_stats[key] = w;
         }
 
         if (bpf_map_lookup_elem(tbl_pid_stats_fd, &key, w)) {
+            // Clean Process structures
+            freez(w);
+            global_process_stats[key] = NULL;
+
+            freez(current_apps_data[key]);
+            current_apps_data[key] = NULL;
+            prev_apps_data[key] = NULL;
+
+            // Clean socket structures
+            if (socket_bandwidth_curr) {
+                freez(socket_bandwidth_curr[key]);
+                socket_bandwidth_curr[key] = NULL;
+                socket_bandwidth_prev[key] = NULL;
+            }
+
             pids = pids->next;
             continue;
         }
 
-        index[counter] = key;
-        counter++;
-
         pids = pids->next;
     }
 
@@ -1062,11 +1078,9 @@ void collect_data_for_all_processes(ebpf_process_stat_t **out, pid_t *index, int
     apps_groups_targets_count = zero_all_targets(apps_groups_root_target);
 
     // this has to be done, before the cleanup
-    struct pid_stat *p = NULL;
-
     // // concentrate everything on the targets
-    for (p = root_of_pids; p; p = p->next)
-        aggregate_pid_on_target(p->target, p, NULL);
+    for (pids = root_of_pids; pids; pids = pids->next)
+        aggregate_pid_on_target(pids->target, pids, NULL);
 
     post_aggregate_targets(apps_groups_root_target);
 }
diff --git a/collectors/ebpf.plugin/ebpf_apps.h b/collectors/ebpf.plugin/ebpf_apps.h
index c45632af76..8a73e699f4 100644
--- a/collectors/ebpf.plugin/ebpf_apps.h
+++ b/collectors/ebpf.plugin/ebpf_apps.h
@@ -14,6 +14,8 @@
 #define NETDATA_APPS_SYSCALL_GROUP "ebpf syscall"
 #define NETDATA_APPS_NET_GROUP "ebpf net"
 
+#include "ebpf_process.h"
+
 #define MAX_COMPARE_NAME 100
 #define MAX_NAME 100
 
@@ -404,7 +406,7 @@ extern size_t zero_all_targets(struct target *root);
 
 extern int am_i_running_as_root();
 
-extern void cleanup_exited_pids(ebpf_process_stat_t **out);
+extern void cleanup_exited_pids();
 
 extern int ebpf_read_hash_table(void *ep, int fd, uint32_t pid);
 
@@ -414,6 +416,10 @@ extern size_t read_processes_statistic_using_pid_on_target(ebpf_process_stat_t *
 
 extern size_t read_bandwidth_statistic_using_pid_on_target(ebpf_bandwidth_t **ep, int fd, struct pid_on_target *pids);
 
-extern void collect_data_for_all_processes(ebpf_process_stat_t **out, pid_t *index, int tbl_pid_stats_fd);
+extern void collect_data_for_all_processes(int tbl_pid_stats_fd);
+
+extern ebpf_process_stat_t **global_process_stats;
+extern ebpf_process_publish_apps_t **current_apps_data;
+extern ebpf_process_publish_apps_t **prev_apps_data;
 
 #endif /* NETDATA_EBPF_APPS_H */
diff --git a/collectors/ebpf.plugin/ebpf_process.c b/collectors/ebpf.plugin/ebpf_process.c
index d7072254c0..b9d431d10f 100644
--- a/collectors/ebpf.plugin/ebpf_process.c
+++ b/collectors/ebpf.plugin/ebpf_process.c
@@ -24,13 +24,15 @@ static netdata_publish_syscall_t *process_publish_aggregated = NULL;
 
 static ebpf_data_t process_data;
 
-static ebpf_process_stat_t **local_process_stats = NULL;
-static ebpf_process_publish_apps_t **current_apps_data = NULL;
-static ebpf_process_publish_apps_t **prev_apps_data = NULL;
+ebpf_process_stat_t **global_process_stats = NULL;
+ebpf_process_publish_apps_t **current_apps_data = NULL;
+ebpf_process_publish_apps_t **prev_apps_data = NULL;
 
 int process_enabled = 0;
 
 static int *map_fd = NULL;
+static struct bpf_object *objects = NULL;
+static struct bpf_link **probe_links = NULL;
 
 /*****************************************************************
  *
@@ -201,11 +203,11 @@ void ebpf_process_remove_pids()
     int pid_fd = map_fd[0];
     while (pids) {
         uint32_t pid = pids->pid;
-        ebpf_process_stat_t *w = local_process_stats[pid];
+        ebpf_process_stat_t *w = global_process_stats[pid];
         if (w) {
             if (w->removeme) {
                 freez(w);
-                local_process_stats[pid] = NULL;
+                global_process_stats[pid] = NULL;
                 bpf_map_delete_elem(pid_fd, &pid);
             }
         }
@@ -432,12 +434,14 @@ static void read_hash_global_tables()
  */
 static void ebpf_process_update_apps_data()
 {
-    size_t i;
-    for (i = 0; i < all_pids_count; i++) {
-        uint32_t current_pid = pid_index[i];
-        ebpf_process_stat_t *ps = local_process_stats[current_pid];
-        if (!ps)
+    struct pid_stat *pids = root_of_pids;
+    while (pids) {
+        uint32_t current_pid = pids->pid;
+        ebpf_process_stat_t *ps = global_process_stats[current_pid];
+        if (!ps) {
+            pids = pids->next;
             continue;
+        }
 
         ebpf_process_publish_apps_t *cad = current_apps_data[current_pid];
         ebpf_process_publish_apps_t *pad = prev_apps_data[current_pid];
@@ -477,6 +481,8 @@ static void ebpf_process_update_apps_data()
         cad->bytes_read = (uint64_t)ps->read_bytes + (uint64_t)ps->readv_bytes;
 
         ebpf_process_update_apps_publish(cad, pad, lstatus);
+
+        pids = pids->next;
     }
 }
 
@@ -832,8 +838,8 @@ static void process_collector(usec_t step, ebpf_module_t *em)
         read_hash_global_tables();
 
         pthread_mutex_lock(&collect_data_mutex);
-        cleanup_exited_pids(local_process_stats);
-        collect_data_for_all_processes(local_process_stats, pid_index, pid_fd);
+        cleanup_exited_pids();
+        collect_data_for_all_processes(pid_fd);
 
         ebpf_create_apps_charts(em, apps_groups_root_target);
 
@@ -866,6 +872,40 @@ static void process_collector(usec_t step, ebpf_module_t *em)
  *
  *****************************************************************/
 
+void clean_global_memory() {
+    int pid_fd = map_fd[0];
+    struct pid_stat *pids = root_of_pids;
+    while (pids) {
+        uint32_t pid = pids->pid;
+        freez(global_process_stats[pid]);
+
+        bpf_map_delete_elem(pid_fd, &pid);
+        freez(current_apps_data[pid]);
+
+        pids = pids->next;
+    }
+}
+
+void clean_pid_on_target(struct pid_on_target *ptr) {
+    while (ptr) {
+        struct pid_on_target *next = ptr->next;
+        freez(ptr);
+
+        ptr = next;
+    }
+}
+
+void clean_apps_structures(struct target *ptr) {
+    struct target *agdt = ptr;
+    while (agdt) {
+        struct target *next = agdt->next;
+        clean_pid_on_target(agdt->root_pid);
+        freez(agdt);
+
+        agdt = next;
+    }
+}
+
 /**
  * Clean up the main thread.
  *
@@ -873,17 +913,35 @@ static void process_collector(usec_t step, ebpf_module_t *em)
  */
 static void ebpf_process_cleanup(void *ptr)
 {
-    (void)ptr;
+    UNUSED(ptr);
+
+    heartbeat_t hb;
+    heartbeat_init(&hb);
+    uint32_t tick = 200*USEC_PER_MS;
+    while (!finalized_threads) {
+        usec_t dt = heartbeat_next(&hb, tick);
+        UNUSED(dt);
+    }
 
     freez(process_aggregated_data);
     freez(process_publish_aggregated);
     freez(process_hash_values);
 
-    freez(local_process_stats);
-
-    freez(process_data.map_fd);
+    clean_global_memory();
+    freez(global_process_stats);
     freez(current_apps_data);
     freez(prev_apps_data);
+
+    clean_apps_structures(apps_groups_root_target);
+    freez(process_data.map_fd);
+
+    struct bpf_program *prog;
+    size_t i = 0 ;
+    bpf_object__for_each_program(prog, objects) {
+        bpf_link__destroy(probe_links[i]);
+        i++;
+    }
+    bpf_object__close(objects);
 }
 
 /*****************************************************************
@@ -905,7 +963,7 @@ static void ebpf_process_allocate_global_vectors(size_t length)
     process_publish_aggregated = callocz(length, sizeof(netdata_publish_syscall_t));
     process_hash_values = callocz(ebpf_nprocs, sizeof(netdata_idx_t));
 
-    local_process_stats = callocz((size_t)pid_max, sizeof(ebpf_process_stat_t *));
+    global_process_stats = callocz((size_t)pid_max, sizeof(ebpf_process_stat_t *));
     current_apps_data = callocz((size_t)pid_max, sizeof(ebpf_process_publish_apps_t *));
     prev_apps_data = callocz((size_t)pid_max, sizeof(ebpf_process_publish_apps_t *));
 }
@@ -1000,8 +1058,8 @@ void *ebpf_process_thread(void *ptr)
     }
 
     set_local_pointers();
-    if (ebpf_load_program(
-            ebpf_plugin_dir, em->thread_id, em->mode, kernel_string, em->thread_name, process_data.map_fd)) {
+    probe_links = ebpf_load_program(ebpf_plugin_dir, em, kernel_string, &objects, process_data.map_fd);
+    if (!probe_links) {
         pthread_mutex_unlock(&lock);
         goto endprocess;
     }
diff --git a/collectors/ebpf.plugin/ebpf_socket.c b/collectors/ebpf.plugin/ebpf_socket.c
index a1614b2921..590346de96 100644
--- a/collectors/ebpf.plugin/ebpf_socket.c
+++ b/collectors/ebpf.plugin/ebpf_socket.c
@@ -22,13 +22,14 @@ static netdata_publish_syscall_t *socket_publish_aggregated = NULL;
 
 static ebpf_data_t socket_data;
 
-static ebpf_socket_publish_apps_t **socket_bandwidth_curr = NULL;
-static ebpf_socket_publish_apps_t **socket_bandwidth_prev = NULL;
+ebpf_socket_publish_apps_t **socket_bandwidth_curr = NULL;
+ebpf_socket_publish_apps_t **socket_bandwidth_prev = NULL;
 static ebpf_bandwidth_t *bandwidth_vector = NULL;
 
 static int socket_apps_created = 0;
 pthread_mutex_t nv_mutex;
 int wait_to_plot = 0;
+int read_thread_closed = 1;
 
 netdata_vector_plot_t inbound_vectors = { .plot = NULL, .next = 0, .last = 0 };
 netdata_vector_plot_t outbound_vectors = { .plot = NULL, .next = 0, .last = 0 };
@@ -37,6 +38,8 @@ netdata_socket_t *socket_values;
 ebpf_network_viewer_port_list_t *listen_ports = NULL;
 
 static int *map_fd = NULL;
+static struct bpf_object *objects = NULL;
+static struct bpf_link **probe_links = NULL;
 
 /*****************************************************************
  *
@@ -1341,6 +1344,7 @@ void *ebpf_socket_read_hash(void *ptr)
 {
     ebpf_module_t *em = (ebpf_module_t *)ptr;
 
+    read_thread_closed = 0;
     heartbeat_t hb;
     heartbeat_init(&hb);
     usec_t step = NETDATA_SOCKET_READ_SLEEP_MS;
@@ -1359,6 +1363,7 @@ void *ebpf_socket_read_hash(void *ptr)
         pthread_mutex_unlock(&nv_mutex);
     }
 
+    read_thread_closed = 1;
     return NULL;
 }
 
@@ -1480,6 +1485,10 @@ static void ebpf_socket_update_apps_data()
  *
  *****************************************************************/
 
+struct netdata_static_thread socket_threads = {"EBPF SOCKET READ",
+                                               NULL, NULL, 1, NULL,
+                                               NULL, ebpf_socket_read_hash };
+
 /**
  * Main loop for this collector.
  *
@@ -1493,10 +1502,7 @@ static void socket_collector(usec_t step, ebpf_module_t *em)
     heartbeat_t hb;
     heartbeat_init(&hb);
 
-    struct netdata_static_thread socket_threads = {"EBPF SOCKET READ",
-                                                    NULL, NULL, 1, NULL,
-                                                    NULL, ebpf_socket_read_hash };
-    socket_threads.thread = mallocz(sizeof(netdata_thread_t));;
+    socket_threads.thread = mallocz(sizeof(netdata_thread_t));
 
     netdata_thread_create(socket_threads.thread, socket_threads.name,
                           NETDATA_THREAD_OPTION_JOINABLE, ebpf_socket_read_hash, em);
@@ -1650,6 +1656,15 @@ static void clean_hostnames(ebpf_network_viewer_hostname_list_t *hostnames)
     }
 }
 
+void clean_thread_structures() {
+    struct pid_stat *pids = root_of_pids;
+    while (pids) {
+        freez(socket_bandwidth_curr[pids->pid]);
+
+        pids = pids->next;
+    }
+}
+
 /**
  * Clean up the main thread.
  *
@@ -1657,12 +1672,23 @@ static void clean_hostnames(ebpf_network_viewer_hostname_list_t *hostnames)
  */
 static void ebpf_socket_cleanup(void *ptr)
 {
-    UNUSED(ptr);
+    ebpf_module_t *em = (ebpf_module_t *)ptr;
+    if (!em->enabled)
+        return;
+
+    heartbeat_t hb;
+    heartbeat_init(&hb);
+    uint32_t tick = 200*USEC_PER_MS;
+    while (!read_thread_closed) {
+        usec_t dt = heartbeat_next(&hb, tick);
+        UNUSED(dt);
+    }
+
     freez(socket_aggregated_data);
     freez(socket_publish_aggregated);
     freez(socket_hash_values);
 
-    freez(socket_data.map_fd);
+    clean_thread_structures();
     freez(socket_bandwidth_curr);
     freez(socket_bandwidth_prev);
     freez(bandwidth_vector);
@@ -1683,6 +1709,18 @@ static void ebpf_socket_cleanup(void *ptr)
     clean_hostnames(network_viewer_opt.excluded_hostnames);
 
     pthread_mutex_destroy(&nv_mutex);
+    freez(socket_data.map_fd);
+
+    freez(socket_threads.thread);
+
+    struct bpf_program *prog;
+    size_t i = 0 ;
+    bpf_object__for_each_program(prog, objects) {
+        bpf_link__destroy(probe_links[i]);
+        i++;
+    }
+    bpf_object__close(objects);
+    finalized_threads = 1;
 }
 
 /*****************************************************************
@@ -1789,8 +1827,8 @@ void *ebpf_socket_thread(void *ptr)
     }
 
     set_local_pointers(em);
-    if (ebpf_load_program(
-            ebpf_plugin_dir, em->thread_id, em->mode, kernel_string, em->thread_name, socket_data.map_fd)) {
+    probe_links = ebpf_load_program(ebpf_plugin_dir, em, kernel_string, &objects, socket_data.map_fd);
+    if (!probe_links) {
         pthread_mutex_unlock(&lock);
         goto endsocket;
     }
@@ -1801,6 +1839,7 @@ void *ebpf_socket_thread(void *ptr)
 
     ebpf_create_global_charts(em);
 
+    finalized_threads = 0;
     pthread_mutex_unlock(&lock);
 
     socket_collector((usec_t)(em->update_time * USEC_PER_SEC), em);
diff --git a/collectors/ebpf.plugin/ebpf_socket.h b/collectors/ebpf.plugin/ebpf_socket.h
index 4c7dce8d38..f664925c75 100644
--- a/collectors/ebpf.plugin/ebpf_socket.h
+++ b/collectors/ebpf.plugin/ebpf_socket.h
@@ -255,4 +255,7 @@ extern void clean_port_structure(ebpf_network_viewer_port_list_t **clean);
 extern ebpf_network_viewer_port_list_t *listen_ports;
 extern void update_listen_table(uint16_t value, uint8_t proto);
 
+extern ebpf_socket_publish_apps_t **socket_bandwidth_curr;
+extern ebpf_socket_publish_apps_t **socket_bandwidth_prev;
+
 #endif