update(anomalydetection): ability to reset data structures w/ timers

Signed-off-by: Melissa Kilby <[email protected]>

plugins/anomalydetection/README.md

@@ -60,7 +60,9 @@ plugins:
           {
             "fields": "%container.id %proc.name %proc.aname[1] %proc.aname[2] %proc.aname[3] %proc.exepath %proc.tty %proc.vpgid.name %proc.sname",
             # execve, execveat
-            "event_codes": [293, 331]
+            "event_codes": [293, 331],
+            # optional config `reset_timer_ms`, resets the data structure every x milliseconds, here one hour as example
+            "reset_timer_ms": 3600000
           },
           {
             "fields": "%container.id %proc.name %proc.aname[1] %proc.aname[2] %proc.aname[3] %proc.exepath %proc.tty %proc.vpgid.name %proc.sname %fd.name",

plugins/anomalydetection/src/num/cms.h

@@ -15,6 +15,8 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
+#pragma once
+
 #include "xxhash_ext.h"
 
 #include <iostream>
@@ -96,6 +98,15 @@ class cms
         }
     }
 
+    void reset()
+    {
+        // Reset data structure
+        for (uint64_t i = 0; i < d_; ++i) 
+        {
+            std::fill(sketch[i].get(), sketch[i].get() + w_, static_cast<T>(0));
+        }
+    }
+
     uint64_t hash_XXH3_seed(std::string value, uint64_t seed) const
     {
         // using https://raw.githubusercontent.com/Cyan4973/xxHash/v0.8.2/xxhash.h

plugins/anomalydetection/src/plugin.cpp

@@ -99,6 +99,10 @@ falcosecurity::init_schema anomalydetection::get_init_schema()
                   "type": "number",
                   "description": "PPME event codes supported by Falco."
                 }
+              },
+              "reset_timer_ms": {
+                "type": "number",
+                "description": "The anomaly detection behavior profile timer, in milliseconds (ms), is used to reset the sketch counts."
               }
             },
             "required": [
@@ -134,16 +138,17 @@ void anomalydetection::parse_init_config(nlohmann::json& config_json)
 
         // If used, config JSON schema enforces a minimum of 1 items and 2-d sub-arrays
         auto gamma_eps_pointer = nlohmann::json::json_pointer("/count_min_sketch/gamma_eps");
+        m_gamma_eps.clear();
         if (config_json.contains(gamma_eps_pointer) && config_json[gamma_eps_pointer].is_array())
         {
-            int i = 0;
+            int n = 1;
             for (const auto& array : config_json[gamma_eps_pointer])
             {
                 if (array.is_array() && array.size() == 2)
                 {
                     std::vector<double> sub_array = {array[0].get<double>(), array[1].get<double>()};
                     log_error("Count min sketch data structure number (" 
-                    + std::to_string(i+1) + ") loaded with gamma and eps values (" 
+                    + std::to_string(n) + ") loaded with gamma and eps values (" 
                     + std::to_string(sub_array[0]) + ","
                     + std::to_string(sub_array[1])
                     + ") equivalent to sketch dimensions ("
@@ -154,15 +159,16 @@ void anomalydetection::parse_init_config(nlohmann::json& config_json)
                     + ") bytes of constant memory allocation on the heap");
                     m_gamma_eps.emplace_back(sub_array);     
                 }
-                i++;
+                n++;
             }
         }
 
         // If used, config JSON schema enforces a minimum of 1 items and 2-d sub-arrays
         auto rows_cols_pointer = nlohmann::json::json_pointer("/count_min_sketch/rows_cols");
+        m_rows_cols.clear();
         if (config_json.contains(rows_cols_pointer) && config_json[rows_cols_pointer].is_array())
         {
-            int i = 0;
+            int n = 1;
             if (config_json.contains(gamma_eps_pointer) && config_json[gamma_eps_pointer].is_array())
             {
                 log_error("[Override Notice] Count min sketch data structures will be overriden with below settings as 'rows_cols' config overrides any previous setting");
@@ -173,7 +179,7 @@ void anomalydetection::parse_init_config(nlohmann::json& config_json)
                 {
                     std::vector<uint64_t> sub_array = {array[0].get<uint64_t>(), array[1].get<uint64_t>()};
                     log_error("Count min sketch data structure number (" 
-                    + std::to_string(i+1) + ") loaded with d and w/buckets values (" 
+                    + std::to_string(n) + ") loaded with d and w/buckets values (" 
                     + std::to_string(sub_array[0]) + ","
                     + std::to_string(sub_array[1])
                     + ") equivalent to sketch error probability and relative error tolerances ("
@@ -184,7 +190,7 @@ void anomalydetection::parse_init_config(nlohmann::json& config_json)
                     + ") bytes of constant memory allocation on the heap");
                     m_rows_cols.emplace_back(sub_array);
                 }
-                i++;
+                n++;
             }
         }
 
@@ -215,6 +221,9 @@ void anomalydetection::parse_init_config(nlohmann::json& config_json)
                 supported_codes_fd_profile.end()
             );
 
+            m_reset_timers.clear();
+            m_behavior_profiles_fields.clear();
+            m_behavior_profiles_event_codes.clear();
             int n = 1;
             for (const auto& profile : behavior_profiles)
             {
@@ -260,6 +269,21 @@ void anomalydetection::parse_init_config(nlohmann::json& config_json)
                         }
                     }
                 }
+                if (profile.contains("reset_timer_ms"))
+                {
+                    uint64_t interval = profile["reset_timer_ms"].get<uint64_t>();
+                    if (interval > 100)
+                    {
+                        m_reset_timers.emplace_back(interval);
+                    } else
+                    {
+                        m_reset_timers.emplace_back(uint64_t(0));
+                    }
+                    log_error("Behavior profile number (" + std::to_string(n) + ") resets the counts to zero every (" + std::to_string(interval) + ") ms");
+                } else
+                {
+                    m_reset_timers.emplace_back(uint64_t(0));
+                }
                 m_behavior_profiles_fields.emplace_back(filter_check_fields);
                 m_behavior_profiles_event_codes.emplace_back(std::move(codes));
                 n++;
@@ -375,32 +399,41 @@ bool anomalydetection::init(falcosecurity::init_input& in)
         return false;
     }
 
-    //////////////////////////
+    ////////////////////
     // Init sketches
-    //////////////////////////
+    ////////////////////
 
     // Init the plugin managed state table holding the count min sketch estimates for each behavior profile
+    m_thread_manager.stop_threads(); // Important for reloading configs conditions
+    m_count_min_sketches.lock()->clear();
     if (m_rows_cols.size() == m_n_sketches)
     {
         for (uint32_t i = 0; i < m_n_sketches; ++i)
         {
             uint64_t rows = m_rows_cols[i][0];
             uint64_t cols = m_rows_cols[i][1];
-            m_count_min_sketches.lock()->push_back(std::make_unique<plugin::anomalydetection::num::cms<uint64_t>>(rows, cols));
+            m_count_min_sketches.lock()->push_back(std::make_shared<plugin::anomalydetection::num::cms<uint64_t>>(rows, cols));
         }
     } else if (m_gamma_eps.size() == m_n_sketches && m_rows_cols.empty())
     {
         for (uint32_t i = 0; i < m_n_sketches; ++i)
         {
             double gamma = m_gamma_eps[i][0];
             double eps = m_gamma_eps[i][1];
-            m_count_min_sketches.lock()->push_back(std::make_unique<plugin::anomalydetection::num::cms<uint64_t>>(gamma, eps));
+            m_count_min_sketches.lock()->push_back(std::make_shared<plugin::anomalydetection::num::cms<uint64_t>>(gamma, eps));
         }
     } else
     {
         return false;
     }
 
+    // Launch threads to periodically reset the data structures (if applicable)
+    m_thread_manager.m_stop_requested = false;
+    for (uint32_t i = 0; i < m_n_sketches; ++i)
+    {
+        m_thread_manager.start_periodic_count_min_sketch_reset_worker<uint64_t>(i, (uint64_t)m_reset_timers[i], m_count_min_sketches);
+    }
+
     return true;
 }
 

plugins/anomalydetection/src/plugin.h

@@ -21,6 +21,7 @@ limitations under the License.
 #include "plugin_consts.h"
 #include "plugin_utils.h"
 #include "plugin_mutex.h"
+#include "plugin_thread_manager.h"
 #include "plugin_sinsp_filterchecks.h"
 
 #include <falcosecurity/sdk.h>
@@ -45,6 +46,8 @@ struct sinsp_param
 class anomalydetection
 {
     public:
+    anomalydetection() : m_thread_manager() {}
+
     // Keep this aligned with `get_fields`
     enum anomalydetection_fields
     {
@@ -132,15 +135,19 @@ class anomalydetection
 
     private:
 
+    // Manages plugin side threads, such as resetting the count min sketch data structures
+    ThreadManager m_thread_manager;
+
     bool m_count_min_sketch_enabled = false;
     uint32_t m_n_sketches = 0;
     std::vector<std::vector<double>> m_gamma_eps;
     std::vector<std::vector<uint64_t>> m_rows_cols; // If set supersedes m_gamma_eps
     std::vector<std::vector<plugin_sinsp_filterchecks_field>> m_behavior_profiles_fields;
     std::vector<std::unordered_set<ppm_event_code>> m_behavior_profiles_event_codes;
+    std::vector<uint64_t> m_reset_timers;
 
     // Plugin managed state table
-    plugin_anomalydetection::Mutex<std::vector<std::unique_ptr<plugin::anomalydetection::num::cms<uint64_t>>>> m_count_min_sketches;
+    plugin_anomalydetection::Mutex<std::vector<std::shared_ptr<plugin::anomalydetection::num::cms<uint64_t>>>> m_count_min_sketches;
 
     // required; standard plugin API
     std::string m_lasterr;

plugins/anomalydetection/src/plugin_thread_manager.h

@@ -0,0 +1,117 @@
+// SPDX-License-Identifier: Apache-2.0
+/*
+Copyright (C) 2024 The Falco Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+*/
+
+#pragma once
+
+#include "num/cms.h"
+#include "plugin_mutex.h"
+
+#include <iostream>
+#include <thread>
+#include <chrono>
+#include <atomic>
+#include <vector>
+#include <memory>
+
+class ThreadManager {
+public:
+    ThreadManager() : m_stop_requested(false) {}
+
+    ~ThreadManager()
+    {
+        stop_threads();
+    }
+
+    void stop_threads()
+    {
+        {
+            std::lock_guard<std::mutex> lock(m_thread_mutex);
+            m_stop_requested = true;
+        }
+
+        {
+            std::lock_guard<std::mutex> lock(m_thread_mutex);
+            for (auto& t : m_threads)
+            {
+                if (t.joinable())
+                {
+                    t.join();
+                }
+            }
+            m_threads.clear();
+        }
+    }
+
+    template<typename T>
+    void start_periodic_count_min_sketch_reset_worker(int id, uint64_t interval_ms, plugin_anomalydetection::Mutex<std::vector<std::shared_ptr<plugin::anomalydetection::num::cms<T>>>>& count_min_sketches)
+    {
+        if (interval_ms > 100)
+        {
+            auto worker = [id, interval_ms, &count_min_sketches, this]() {
+                periodic_count_min_sketch_reset_worker<T>(id, interval_ms, count_min_sketches);
+            };
+
+            std::thread worker_thread(worker);
+            {
+                std::lock_guard<std::mutex> lock(m_thread_mutex);
+                m_threads.push_back(std::move(worker_thread));
+            }
+        }
+    }
+    std::atomic<bool> m_stop_requested;
+
+private:
+    std::vector<std::thread> m_threads;
+    std::mutex m_thread_mutex;
+
+    template<typename T>
+    void reset_sketches_worker(int id, plugin_anomalydetection::Mutex<std::vector<std::shared_ptr<plugin::anomalydetection::num::cms<T>>>>& count_min_sketches)
+    {
+        auto sketches = count_min_sketches.lock();
+        if (id >= 0 && id < sketches->size())
+        {
+            auto& sketch_ptr = sketches->at(id);
+            if (sketch_ptr)
+            {
+                sketch_ptr->reset();
+            }
+        }
+    }
+
+    template<typename T>
+    void periodic_count_min_sketch_reset_worker(int id, uint64_t interval_ms, plugin_anomalydetection::Mutex<std::vector<std::shared_ptr<plugin::anomalydetection::num::cms<T>>>>& count_min_sketches)
+    {
+        std::chrono::milliseconds interval(interval_ms);
+        while (true)
+        {
+            std::this_thread::sleep_for(interval);
+            {
+                std::lock_guard<std::mutex> lock(m_thread_mutex);
+                if (m_stop_requested)
+                    break;
+            }
+
+            try
+            {
+                reset_sketches_worker<T>(id, count_min_sketches);
+            } catch (const std::exception& e)
+            {
+            }
+        }
+    }
+};

plugins/anomalydetection/src/plugin_utils.cpp

@@ -17,6 +17,8 @@ limitations under the License.
 
 #include "plugin_utils.h"
 
+#define SCAP_MAX_PATH_SIZE 1024
+
 static const filtercheck_field_info sinsp_filter_check_fields[] =
 {
 	{PT_CHARBUF, EPF_ANOMALY_PLUGIN | EPF_NONE, PF_NA, "proc.exe", "First Argument", "The first command-line argument (i.e., argv[0]), typically the executable name or a custom string as specified by the user. It is primarily obtained from syscall arguments, truncated after 4096 bytes, or, as a fallback, by reading /proc/PID/cmdline, in which case it may be truncated after 1024 bytes. This field may differ from the last component of proc.exepath, reflecting how command invocation and execution paths can vary."},

plugins/anomalydetection/src/plugin_utils.h

@@ -25,8 +25,6 @@ limitations under the License.
 #include <regex>
 #include <unordered_set>
 
-#define SCAP_MAX_PATH_SIZE 1024
-
 typedef struct plugin_sinsp_filterchecks_field
 {
     plugin_sinsp_filterchecks::check_type id;