Ebpf swap (#11090)

Add new thread to ebpf.plugin.

14 files changed, 556 insertions, 7 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index ddce9882b9..3b33f89fda 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -496,6 +496,8 @@ set(EBPF_PROCESS_PLUGIN_FILES
         collectors/ebpf.plugin/ebpf_socket.h
         collectors/ebpf.plugin/ebpf_sync.c
         collectors/ebpf.plugin/ebpf_sync.h
+        collectors/ebpf.plugin/ebpf_swap.c
+        collectors/ebpf.plugin/ebpf_swap.h
         collectors/ebpf.plugin/ebpf_apps.c
         collectors/ebpf.plugin/ebpf_apps.h
         )
diff --git a/Makefile.am b/Makefile.am
index 43780959b2..88f8a3526a 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -295,6 +295,8 @@ EBPF_PLUGIN_FILES = \
     collectors/ebpf.plugin/ebpf_socket.h \
     collectors/ebpf.plugin/ebpf_sync.c \
     collectors/ebpf.plugin/ebpf_sync.h \
+    collectors/ebpf.plugin/ebpf_swap.c \
+    collectors/ebpf.plugin/ebpf_swap.h \
     collectors/ebpf.plugin/ebpf.h \
     collectors/ebpf.plugin/ebpf_apps.c \
     collectors/ebpf.plugin/ebpf_apps.h \
diff --git a/collectors/ebpf.plugin/Makefile.am b/collectors/ebpf.plugin/Makefile.am
index 18b1fc6c83..26bdc1fada 100644
--- a/collectors/ebpf.plugin/Makefile.am
+++ b/collectors/ebpf.plugin/Makefile.am
@@ -37,4 +37,5 @@ dist_ebpfconfig_DATA = \
     ebpf.d/network.conf \
     ebpf.d/process.conf \
     ebpf.d/sync.conf \
+    ebpf.d/swap.conf \
     $(NULL)
diff --git a/collectors/ebpf.plugin/ebpf.c b/collectors/ebpf.plugin/ebpf.c
index 5cc005f30b..e22dee59e6 100644
--- a/collectors/ebpf.plugin/ebpf.c
+++ b/collectors/ebpf.plugin/ebpf.c
@@ -93,7 +93,11 @@ ebpf_module_t ebpf_modules[] = {
     { .thread_name = "dc", .config_name = "dc", .enabled = 0, .start_routine = ebpf_dcstat_thread,
       .update_time = 1, .global_charts = 1, .apps_charts = 1, .mode = MODE_ENTRY,
       .optional = 0, .apps_routine = ebpf_dcstat_create_apps_charts, .maps = NULL,
-      .pid_map_size = ND_EBPF_DEFAULT_PID_SIZE },
+      .pid_map_size = ND_EBPF_DEFAULT_PID_SIZE, .names = NULL },
+    { .thread_name = "swap", .config_name = "swap", .enabled = 0, .start_routine = ebpf_swap_thread,
+      .update_time = 1, .global_charts = 1, .apps_charts = 1, .mode = MODE_ENTRY,
+      .optional = 0, .apps_routine = ebpf_swap_create_apps_charts, .maps = NULL,
+      .pid_map_size = ND_EBPF_DEFAULT_PID_SIZE, .names = NULL },
     { .thread_name = NULL, .enabled = 0, .start_routine = NULL, .update_time = 1,
       .global_charts = 0, .apps_charts = 1, .mode = MODE_ENTRY,
       .optional = 0, .apps_routine = NULL, .maps = NULL, .pid_map_size = 0, .names = NULL },
@@ -155,6 +159,12 @@ static void ebpf_exit(int sig)
         freez(dcstat_pid);
     }
 
+    if (ebpf_modules[EBPF_MODULE_SWAP_IDX].enabled) {
+        ebpf_modules[EBPF_MODULE_SWAP_IDX].enabled = 0;
+        clean_swap_pid_structures();
+        freez(swap_pid);
+    }
+
     /*
     int ret = fork();
     if (ret < 0) // error
@@ -605,6 +615,8 @@ void ebpf_print_help()
             "\n",
             " --sync or -s        Enable chart related to sync run time.\n"
             "\n"
+            " --swap or -w        Enable chart related to swap run time.\n"
+            "\n"
             VERSION,
             (year >= 116) ? year + 1900 : 2020);
 }
@@ -894,6 +906,13 @@ static void read_collector_values(int *disable_apps)
         started++;
     }
 
+    enabled = appconfig_get_boolean(&collector_config, EBPF_PROGRAMS_SECTION, "swap",
+                                    CONFIG_BOOLEAN_NO);
+    if (enabled) {
+        ebpf_enable_chart(EBPF_MODULE_SWAP_IDX, *disable_apps);
+        started++;
+    }
+
     if (!started){
         ebpf_enable_all_charts(*disable_apps);
         // Read network viewer section
@@ -982,6 +1001,7 @@ static void parse_args(int argc, char **argv)
         {"process",   no_argument,    0,  'p' },
         {"return",    no_argument,    0,  'r' },
         {"sync",      no_argument,    0,  's' },
+        {"swap",      no_argument,    0,  'w' },
         {0, 0, 0, 0}
     };
 
@@ -996,7 +1016,7 @@ static void parse_args(int argc, char **argv)
     }
 
     while (1) {
-        int c = getopt_long(argc, argv, "hvgacdnprs", long_options, &option_index);
+        int c = getopt_long(argc, argv, "hvgacdnprsw", long_options, &option_index);
         if (c == -1)
             break;
 
@@ -1075,6 +1095,14 @@ static void parse_args(int argc, char **argv)
 #endif
                 break;
             }
+            case 'w': {
+                enabled = 1;
+                ebpf_enable_chart(EBPF_MODULE_SWAP_IDX, disable_apps);
+#ifdef NETDATA_INTERNAL_CHECKS
+                info("EBPF enabling \"swap\" chart, because it was started with the option \"--swap\" or \"-w\".");
+#endif
+                break;
+            }
             default: {
                 break;
             }
@@ -1203,6 +1231,8 @@ int main(int argc, char **argv)
             NULL, NULL, ebpf_modules[EBPF_MODULE_SYNC_IDX].start_routine},
         {"EBPF DCSTAT" , NULL, NULL, 1,
             NULL, NULL, ebpf_modules[EBPF_MODULE_DCSTAT_IDX].start_routine},
+        {"EBPF SWAP" , NULL, NULL, 1,
+            NULL, NULL, ebpf_modules[EBPF_MODULE_SWAP_IDX].start_routine},
         {NULL          , NULL, NULL, 0,
           NULL, NULL, NULL}
     };
diff --git a/collectors/ebpf.plugin/ebpf.d.conf b/collectors/ebpf.plugin/ebpf.d.conf
index ef6ff8145d..df68994e64 100644
--- a/collectors/ebpf.plugin/ebpf.d.conf
+++ b/collectors/ebpf.plugin/ebpf.d.conf
@@ -31,11 +31,13 @@
 #  `socket`   : This eBPF program creates charts with information about `TCP` and `UDP` functions, including the
 #               bandwidth consumed by each.
 #  `sync`     : Montitor calls for syscall sync(2).             
+#  `swap`     : Monitor calls for internal swap functions.
 [ebpf programs]
     cachestat = no
     dcstat = no
     process = yes
     socket = yes
     sync = yes
+    swap = no
     network connections = no
 
diff --git a/collectors/ebpf.plugin/ebpf.d/swap.conf b/collectors/ebpf.plugin/ebpf.d/swap.conf
new file mode 100644
index 0000000000..2c7660840b
--- /dev/null
+++ b/collectors/ebpf.plugin/ebpf.d/swap.conf
@@ -0,0 +1,15 @@
+# The `ebpf load mode` option accepts the following values :
+#  `entry` : The eBPF collector only monitors calls for the functions, and does not show charts related to errors.
+#  `return : In the `return` mode, the eBPF collector monitors the same kernel functions as `entry`, but also creates
+#            new charts for the return of these functions, such as errors.
+#
+# The eBPF collector also creates charts for each running application through an integration with the `apps plugin`.
+# If you want to disable the integration with `apps.plugin` along with the above charts, change the setting `apps` to
+# 'no'.
+#
+#  
+[global]
+    ebpf load mode = entry
+    apps = yes
+    update every = 2
+    pid table size = 32768
diff --git a/collectors/ebpf.plugin/ebpf.h b/collectors/ebpf.plugin/ebpf.h
index 841701e20f..25575fb1bc 100644
--- a/collectors/ebpf.plugin/ebpf.h
+++ b/collectors/ebpf.plugin/ebpf.h
@@ -78,7 +78,8 @@ enum ebpf_module_indexes {
     EBPF_MODULE_SOCKET_IDX,
     EBPF_MODULE_CACHESTAT_IDX,
     EBPF_MODULE_SYNC_IDX,
-    EBPF_MODULE_DCSTAT_IDX
+    EBPF_MODULE_DCSTAT_IDX,
+    EBPF_MODULE_SWAP_IDX
 };
 
 // Copied from musl header
@@ -96,6 +97,8 @@ enum ebpf_module_indexes {
 #define NETDATA_EBPF_CHART_TYPE_LINE "line"
 #define NETDATA_EBPF_CHART_TYPE_STACKED "stacked"
 #define NETDATA_EBPF_MEMORY_GROUP "mem"
+#define NETDATA_EBPF_SYSTEM_GROUP "system"
+#define NETDATA_SYSTEM_SWAP_SUBMENU "swap"
 
 // Log file
 #define NETDATA_DEVELOPER_LOG_FILE "developer.log"
diff --git a/collectors/ebpf.plugin/ebpf_apps.c b/collectors/ebpf.plugin/ebpf_apps.c
index 6459bad0d3..7912114e1f 100644
--- a/collectors/ebpf.plugin/ebpf_apps.c
+++ b/collectors/ebpf.plugin/ebpf_apps.c
@@ -933,6 +933,12 @@ void cleanup_variables_from_other_threads(uint32_t pid)
         freez(dcstat_pid[pid]);
         dcstat_pid[pid] = NULL;
     }
+
+    // Clean swap structure
+    if (swap_pid) {
+        freez(swap_pid[pid]);
+        swap_pid[pid] = NULL;
+    }
 }
 
 /**
diff --git a/collectors/ebpf.plugin/ebpf_apps.h b/collectors/ebpf.plugin/ebpf_apps.h
index edcdef6057..f6355783c9 100644
--- a/collectors/ebpf.plugin/ebpf_apps.h
+++ b/collectors/ebpf.plugin/ebpf_apps.h
@@ -22,6 +22,7 @@
 #include "ebpf_dcstat.h"
 #include "ebpf_cachestat.h"
 #include "ebpf_sync.h"
+#include "ebpf_swap.h"
 
 #define MAX_COMPARE_NAME 100
 #define MAX_NAME 100
@@ -113,6 +114,7 @@ struct target {
     // Changes made to simplify integration between apps and eBPF.
     netdata_publish_cachestat_t cachestat;
     netdata_publish_dcstat_t dcstat;
+    netdata_publish_swap_t swap;
 
     /* These variables are not necessary for eBPF collector
     kernel_uint_t minflt;
diff --git a/collectors/ebpf.plugin/ebpf_swap.c b/collectors/ebpf.plugin/ebpf_swap.c
new file mode 100644
index 0000000000..122ef5229e
--- /dev/null
+++ b/collectors/ebpf.plugin/ebpf_swap.c
@@ -0,0 +1,437 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "ebpf.h"
+#include "ebpf_swap.h"
+
+static char *swap_dimension_name[NETDATA_SWAP_END] = { "read", "write" };
+static netdata_syscall_stat_t swap_aggregated_data[NETDATA_SWAP_END];
+static netdata_publish_syscall_t swap_publish_aggregated[NETDATA_SWAP_END];
+
+static int read_thread_closed = 1;
+static int *map_fd = NULL;
+netdata_publish_swap_t *swap_vector = NULL;
+
+static netdata_idx_t swap_hash_values[NETDATA_SWAP_END];
+
+netdata_publish_swap_t **swap_pid = NULL;
+
+static ebpf_data_t swap_data;
+struct config swap_config = { .first_section = NULL,
+    .last_section = NULL,
+    .mutex = NETDATA_MUTEX_INITIALIZER,
+    .index = { .avl_tree = { .root = NULL, .compar = appconfig_section_compare },
+        .rwlock = AVL_LOCK_INITIALIZER } };
+
+static ebpf_local_maps_t swap_maps[] = {{.name = "tbl_pid_swap", .internal_input = ND_EBPF_DEFAULT_PID_SIZE,
+                                         .user_input = 0},
+                                        {.name = NULL, .internal_input = 0, .user_input = 0}};
+
+static struct bpf_link **probe_links = NULL;
+static struct bpf_object *objects = NULL;
+
+struct netdata_static_thread swap_threads = {"SWAP KERNEL", NULL, NULL, 1,
+                                             NULL, NULL,  NULL};
+
+/*****************************************************************
+ *
+ *  FUNCTIONS TO CLOSE THE THREAD
+ *
+ *****************************************************************/
+
+/**
+ * Clean swap structure
+ */
+void clean_swap_pid_structures() {
+    struct pid_stat *pids = root_of_pids;
+    while (pids) {
+        freez(swap_pid[pids->pid]);
+
+        pids = pids->next;
+    }
+}
+
+/**
+ * Clean up the main thread.
+ *
+ * @param ptr thread data.
+ */
+static void ebpf_swap_cleanup(void *ptr)
+{
+    ebpf_module_t *em = (ebpf_module_t *)ptr;
+    if (!em->enabled)
+        return;
+
+    heartbeat_t hb;
+    heartbeat_init(&hb);
+    uint32_t tick = 2 * USEC_PER_MS;
+    while (!read_thread_closed) {
+        usec_t dt = heartbeat_next(&hb, tick);
+        UNUSED(dt);
+    }
+
+    ebpf_cleanup_publish_syscall(swap_publish_aggregated);
+
+    freez(swap_vector);
+
+    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);
+}
+
+/*****************************************************************
+ *
+ *  COLLECTOR THREAD
+ *
+ *****************************************************************/
+
+/**
+ * Apps Accumulator
+ *
+ * Sum all values read from kernel and store in the first address.
+ *
+ * @param out the vector with read values.
+ */
+static void swap_apps_accumulator(netdata_publish_swap_t *out)
+{
+    int i, end = (running_on_kernel >= NETDATA_KERNEL_V4_15) ? ebpf_nprocs : 1;
+    netdata_publish_swap_t *total = &out[0];
+    for (i = 1; i < end; i++) {
+        netdata_publish_swap_t *w = &out[i];
+        total->write += w->write;
+        total->read += w->read;
+    }
+}
+
+/**
+ * Fill PID
+ *
+ * Fill PID structures
+ *
+ * @param current_pid pid that we are collecting data
+ * @param out         values read from hash tables;
+ */
+static void swap_fill_pid(uint32_t current_pid, netdata_publish_swap_t *publish)
+{
+    netdata_publish_swap_t *curr = swap_pid[current_pid];
+    if (!curr) {
+        curr = callocz(1, sizeof(netdata_publish_swap_t));
+        swap_pid[current_pid] = curr;
+    }
+
+    memcpy(curr, publish, sizeof(netdata_publish_swap_t));
+}
+
+/**
+ * Read APPS table
+ *
+ * Read the apps table and store data inside the structure.
+ */
+static void read_apps_table()
+{
+    netdata_publish_swap_t *cv = swap_vector;
+    uint32_t key;
+    struct pid_stat *pids = root_of_pids;
+    int fd = map_fd[NETDATA_PID_SWAP_TABLE];
+    size_t length = sizeof(netdata_publish_swap_t)*ebpf_nprocs;
+    while (pids) {
+        key = pids->pid;
+
+        if (bpf_map_lookup_elem(fd, &key, cv)) {
+            pids = pids->next;
+            continue;
+        }
+
+        swap_apps_accumulator(cv);
+
+        swap_fill_pid(key, cv);
+
+        // We are cleaning to avoid passing data read from one process to other.
+        memset(cv, 0, length);
+
+        pids = pids->next;
+    }
+}
+
+/**
+* Send global
+*
+* Send global charts to Netdata
+*/
+static void swap_send_global()
+{
+    write_io_chart(NETDATA_MEM_SWAP_CHART, NETDATA_EBPF_SYSTEM_GROUP,
+                   swap_publish_aggregated[NETDATA_KEY_SWAP_WRITEPAGE_CALL].dimension,
+                   (long long) swap_hash_values[NETDATA_KEY_SWAP_WRITEPAGE_CALL],
+                   swap_publish_aggregated[NETDATA_KEY_SWAP_READPAGE_CALL].dimension,
+                   (long long) swap_hash_values[NETDATA_KEY_SWAP_READPAGE_CALL]);
+}
+
+/**
+ * Read global counter
+ *
+ * Read the table with number of calls for all functions
+ */
+static void read_global_table()
+{
+    uint64_t stored;
+    netdata_idx_t *val = swap_hash_values;
+    int fd = map_fd[NETDATA_SWAP_GLOBAL_TABLE];
+
+    uint32_t i, end = NETDATA_SWAP_END;
+    for (i = NETDATA_KEY_SWAP_READPAGE_CALL; i < end; i++) {
+        if (!bpf_map_lookup_elem(fd, &i, &stored)) {
+            val[i] = stored;
+        }
+    }
+}
+
+/**
+ * Socket read hash
+ *
+ * This is the thread callback.
+ * This thread is necessary, because we cannot freeze the whole plugin to read the data on very busy socket.
+ *
+ * @param ptr It is a NULL value for this thread.
+ *
+ * @return It always returns NULL.
+ */
+void *ebpf_swap_read_hash(void *ptr)
+{
+    read_thread_closed = 0;
+
+    heartbeat_t hb;
+    heartbeat_init(&hb);
+
+    ebpf_module_t *em = (ebpf_module_t *)ptr;
+    usec_t step = NETDATA_SWAP_SLEEP_MS * em->update_time;
+    while (!close_ebpf_plugin) {
+        usec_t dt = heartbeat_next(&hb, step);
+        (void)dt;
+
+        read_global_table();
+    }
+
+    read_thread_closed = 1;
+    return NULL;
+}
+
+/**
+ * Sum PIDs
+ *
+ * Sum values for all targets.
+ *
+ * @param swap
+ * @param root
+ */
+static void ebpf_swap_sum_pids(netdata_publish_swap_t *swap, struct pid_on_target *root)
+{
+    uint64_t local_read = 0;
+    uint64_t local_write = 0;
+
+    while (root) {
+        int32_t pid = root->pid;
+        netdata_publish_swap_t *w = swap_pid[pid];
+        if (w) {
+            local_write += w->write;
+            local_read += w->read;
+        }
+        root = root->next;
+    }
+
+    // These conditions were added, because we are using incremental algorithm
+    swap->write = (local_write >= swap->write) ? local_write : swap->write;
+    swap->read = (local_read >= swap->read) ? local_read : swap->read;
+}
+
+/**
+ * Send data to Netdata calling auxiliar functions.
+ *
+ * @param root the target list.
+*/
+void ebpf_swap_send_apps_data(struct target *root)
+{
+    struct target *w;
+    for (w = root; w; w = w->next) {
+        if (unlikely(w->exposed && w->processes)) {
+            ebpf_swap_sum_pids(&w->swap, w->root_pid);
+        }
+    }
+
+    write_begin_chart(NETDATA_APPS_FAMILY, NETDATA_MEM_SWAP_READ_CHART);
+    for (w = root; w; w = w->next) {
+        if (unlikely(w->exposed && w->processes)) {
+            write_chart_dimension(w->name, (long long) w->swap.read);
+        }
+    }
+    write_end_chart();
+
+    write_begin_chart(NETDATA_APPS_FAMILY, NETDATA_MEM_SWAP_WRITE_CHART);
+    for (w = root; w; w = w->next) {
+        if (unlikely(w->exposed && w->processes)) {
+            write_chart_dimension(w->name, (long long) w->swap.write);
+        }
+    }
+    write_end_chart();
+}
+
+/**
+* Main loop for this collector.
+*/
+static void swap_collector(ebpf_module_t *em)
+{
+    swap_threads.thread = mallocz(sizeof(netdata_thread_t));
+    swap_threads.start_routine = ebpf_swap_read_hash;
+
+    map_fd = swap_data.map_fd;
+
+    netdata_thread_create(swap_threads.thread, swap_threads.name, NETDATA_THREAD_OPTION_JOINABLE,
+                          ebpf_swap_read_hash, em);
+
+    int apps = em->apps_charts;
+    while (!close_ebpf_plugin) {
+        pthread_mutex_lock(&collect_data_mutex);
+        pthread_cond_wait(&collect_data_cond_var, &collect_data_mutex);
+
+        if (apps)
+            read_apps_table();
+
+        pthread_mutex_lock(&lock);
+
+        swap_send_global();
+
+        if (apps)
+            ebpf_swap_send_apps_data(apps_groups_root_target);
+
+        pthread_mutex_unlock(&lock);
+        pthread_mutex_unlock(&collect_data_mutex);
+    }
+}
+
+/*****************************************************************
+ *
+ *  INITIALIZE THREAD
+ *
+ *****************************************************************/
+
+/**
+ * Create apps charts
+ *
+ * Call ebpf_create_chart to create the charts on apps submenu.
+ *
+ * @param em a pointer to the structure with the default values.
+ */
+void ebpf_swap_create_apps_charts(struct ebpf_module *em, void *ptr)
+{
+    UNUSED(em);
+
+    struct target *root = ptr;
+    ebpf_create_charts_on_apps(NETDATA_MEM_SWAP_READ_CHART,
+                               "Calls for function <code>swap_readpage</code>.",
+                               EBPF_COMMON_DIMENSION_CALL,
+                               NETDATA_SWAP_SUBMENU,
+                               NETDATA_EBPF_CHART_TYPE_STACKED,
+                               20191,
+                               ebpf_algorithms[NETDATA_EBPF_INCREMENTAL_IDX],
+                               root);
+
+    ebpf_create_charts_on_apps(NETDATA_MEM_SWAP_WRITE_CHART,
+                               "Calls for function <code>swap_writepage</code>.",
+                               EBPF_COMMON_DIMENSION_CALL,
+                               NETDATA_SWAP_SUBMENU,
+                               NETDATA_EBPF_CHART_TYPE_STACKED,
+                               20192,
+                               ebpf_algorithms[NETDATA_EBPF_INCREMENTAL_IDX],
+                               root);
+}
+
+/**
+ * Allocate vectors used with this thread.
+ *
+ * We are not testing the return, because callocz does this and shutdown the software
+ * case it was not possible to allocate.
+ *
+ * @param length is the length for the vectors used inside the collector.
+ */
+static void ebpf_swap_allocate_global_vectors()
+{
+    swap_pid = callocz((size_t)pid_max, sizeof(netdata_publish_swap_t *));
+    swap_vector = callocz((size_t)ebpf_nprocs, sizeof(netdata_publish_swap_t))