refactor(userspace/engine): strengthen and document thread-safety guarantees of falco_engine::process_event

userspace/engine/falco_engine.cpp

@@ -194,6 +194,15 @@ std::unique_ptr<load_result> falco_engine::load_rules(const std::string &rules_c
 		m_rule_loader.compile(cfg, m_rules);
 	}
 
+	if (cfg.res->successful())
+	{
+		m_rule_stats_manager.clear();
+		for (const auto &r : m_rules)
+		{
+			m_rule_stats_manager.on_rule_loaded(r);
+		}
+	}
+
 	return std::move(cfg.res);
 }
 
@@ -323,6 +332,12 @@ std::shared_ptr<gen_event_formatter> falco_engine::create_formatter(const std::s
 unique_ptr<falco_engine::rule_result> falco_engine::process_event(std::size_t source_idx, gen_event *ev, uint16_t ruleset_id)
 {
 	falco_rule rule;
+
+	// note: there are no thread-safety guarantees on the filter_ruleset::run()
+	// method, but the thread-safety assumptions of falco_engine::process_event()
+	// imply that concurrent invokers use different and non-switchable values of
+	// source_idx, which means that at any time each filter_ruleset will only
+	// be accessed by a single thread.
 	if(should_drop_evt() || !find_source(source_idx)->ruleset->run(ev, rule, ruleset_id))
 	{
 		return unique_ptr<struct rule_result>();

userspace/engine/falco_engine.h

@@ -165,18 +165,33 @@ class falco_engine
 	//
 	// Given an event, check it against the set of rules in the
 	// engine and if a matching rule is found, return details on
-	// the rule that matched. If no rule matched, returns NULL.
-	//
-	// When ruleset_id is provided, use the enabled/disabled status
-	// associated with the provided ruleset. This is only useful
-	// when you have previously called enable_rule/enable_rule_by_tag
-	// with a ruleset string.
-	//
-	// the returned rule_result is allocated and must be delete()d.
+	// the rule that matched. If no rule matched, returns nullptr.
+	//
+	// This method should be invoked only after having initialized and
+	// configured the engine. In particular, invoking this with a source_idx
+	// not previosly-returned by a call to add_source() would cause a
+	// falco_exception to be thrown.
+	// 
+	// This method is thread-safe only with the assumption that every invoker
+	// uses a different source_idx. Moreover, each invoker must not switch
+	// source_idx in subsequent invocations of this method.
+	// Considering that each invoker is related to a unique event source, it
+	// is safe to assume that each invoker will pass a different event
+	// to this method too, since two distinct sources cannot possibly produce
+	// the same event. Lastly, filterchecks and formatters (and their factories)
+	// that are used to populate the conditions for a given event-source
+	// ruleset must not be reused across rulesets of other event sources.
+	// These assumptions guarantee thread-safety because internally the engine
+	// is partitioned by event sources. However, each ruleset assigned to each
+	// event source is not thread-safe of its own, so invoking this method
+	// concurrently with the same source_idx would inherently cause data races
+	// and lead to undefined behavior.
 	std::unique_ptr<rule_result> process_event(std::size_t source_idx, gen_event *ev, uint16_t ruleset_id);
 
 	//
-	// Wrapper assuming the default ruleset
+	// Wrapper assuming the default ruleset.
+	//
+	// This inherits the same thread-safety guarantees.
 	//
 	std::unique_ptr<rule_result> process_event(std::size_t source_idx, gen_event *ev);
 

userspace/engine/stats_manager.cpp

@@ -45,37 +45,48 @@ void stats_manager::format(
 	out += "Rule counts by severity:\n";
 	for (size_t i = 0; i < m_by_priority.size(); i++)
 	{
-		if (m_by_priority[i] > 0)
+		auto val = m_by_priority[i].get()->load();
+		if (val > 0)
 		{
 			falco_common::format_priority(
 				(falco_common::priority_type) i, fmt, true);
 			transform(fmt.begin(), fmt.end(), fmt.begin(), ::toupper);
-			out += "   " + fmt;
-			out += ": " + to_string(m_by_priority[i]) + "\n";
+			out += "   " + fmt + ": " + to_string(val) + "\n";
 		}
 	}
 	out += "Triggered rules by rule name:\n";
 	for (size_t i = 0; i < m_by_rule_id.size(); i++)
 	{
-		if (m_by_rule_id[i] > 0)
+		auto val = m_by_rule_id[i].get()->load();
+		if (val > 0)
 		{
-			out += "   " + rules.at(i)->name;
-			out += ": " + to_string(m_by_rule_id[i]) + "\n";
+			out += "   " + rules.at(i)->name + ": " + to_string(val) + "\n";
 		}
 	}
 }
 
-void stats_manager::on_event(const falco_rule& rule)
+void stats_manager::on_rule_loaded(const falco_rule& rule)
 {
-	if (m_by_rule_id.size() <= rule.id)
+	while (m_by_rule_id.size() <= rule.id)
+	{
+		m_by_rule_id.emplace_back();
+		m_by_rule_id[m_by_rule_id.size() - 1].reset(new atomic<uint64_t>(0));
+	}
+	while (m_by_priority.size() <= (size_t) rule.priority)
 	{
-		m_by_rule_id.resize(rule.id + 1, (uint64_t) 0);
+		m_by_priority.emplace_back();
+		m_by_priority[m_by_priority.size() - 1].reset(new atomic<uint64_t>(0));
 	}
-	if (m_by_priority.size() <= (size_t) rule.priority)
+}
+
+void stats_manager::on_event(const falco_rule& rule)
+{
+	if (m_by_rule_id.size() <= rule.id
+		|| m_by_priority.size() <= (size_t) rule.priority)
 	{
-		m_by_priority.resize((size_t) rule.priority + 1, (uint64_t) 0);
+		throw falco_exception("rule id or priority out of bounds");
 	}
-	m_total++;
-	m_by_rule_id[rule.id]++;
-	m_by_priority[(size_t) rule.priority]++;
+	m_total.fetch_add(1, std::memory_order_relaxed);
+	m_by_rule_id[rule.id]->fetch_add(1, std::memory_order_relaxed);
+	m_by_priority[(size_t) rule.priority]->fetch_add(1, std::memory_order_relaxed);
 }

userspace/engine/stats_manager.h

@@ -18,11 +18,16 @@ limitations under the License.
 
 #include <vector>
 #include <string>
+#include <atomic>
+#include <memory>
 #include "falco_rule.h"
 #include "indexed_vector.h"
 
 /*!
-	\brief Manager for the internal statistics of the rule engine
+	\brief Manager for the internal statistics of the rule engine.
+	The on_event() is thread-safe and non-blocking, and it can be used
+	concurrently across many callers in parallel.
+	All the other methods are not thread safe.
 */
 class stats_manager
 {
@@ -36,19 +41,29 @@ class stats_manager
 	virtual void clear();
 
 	/*!
-		\brief Callback for when a given rule matches an event
+		\brief Callback for when a new rule is loaded in the engine.
+		Rules must be passed through this method before submitting them as
+		an argument of on_event().
+	*/
+	virtual void on_rule_loaded(const falco_rule& rule);
+
+	/*!
+		\brief Callback for when a given rule matches an event.
+		This method is thread-safe.
+		\throws falco_exception if rule has not been passed to
+		on_rule_loaded() first
 	*/
 	virtual void on_event(const falco_rule& rule);
 
 	/*!
-		\brief Formats the internal statistics into the out string
+		\brief Formats the internal statistics into the out string.
 	*/
 	virtual void format(
 		const indexed_vector<falco_rule>& rules,
 		std::string& out) const;
 
 private:
-	uint64_t m_total;
-	std::vector<uint64_t> m_by_priority;
-	std::vector<uint64_t> m_by_rule_id;
+	atomic<uint64_t> m_total;
+	std::vector<std::unique_ptr<atomic<uint64_t>>> m_by_priority;
+	std::vector<std::unique_ptr<atomic<uint64_t>>> m_by_rule_id;
 };