Add rate-limiting for notifications

Add token-bucket based rate limiting for falco notifications.

The token bucket is implemented in token_bucket.cpp and maintains a
current count of tokens (i.e. right to send a notification). Its main
method is claim(), which attemps to claim a token and returns true if
one was claimed successfully. It has a configurable configurable max
burst size and rate. The token bucket gains "rate" tokens per second, up
to a maximum of max_burst tokens.

These parameters are configurable in falco.yaml via the config
options (defaults shown):

outputs:
  rate: 1
  max_burst: 1000

In falco_outputs::handle_event(), try to claim a token, and if
unsuccessful log a debug message and return immediately.

falco.yaml

@@ -9,6 +9,21 @@ json_output: false
 log_stderr: true
 log_syslog: true
 
+# A throttling mechanism implemented as a token bucket limits the
+# rate of falco notifications. This throttling is controlled by the following configuration
+# options:
+#  - rate: the number of tokens (i.e. right to send a notification)
+#    gained per second. Defaults to 1.
+#  - max_burst: the maximum number of tokens outstanding. Defaults to 1000.
+#
+# With these defaults, falco could send up to 1000 notifications after
+# an initial quiet period, and then up to 1 notification per second
+# afterward. It would gain the full burst back after 1000 seconds of
+# no activity.
+
+outputs:
+  rate: 1
+  max_burst: 1000
 
 # Where security notifications should go.
 # Multiple outputs can be enabled.

userspace/falco/CMakeLists.txt

@@ -9,7 +9,7 @@ include_directories("${CURL_INCLUDE_DIR}")
 include_directories("${YAMLCPP_INCLUDE_DIR}")
 include_directories("${DRAIOS_DEPENDENCIES_DIR}/yaml-${DRAIOS_YAML_VERSION}/target/include")
 
-add_executable(falco configuration.cpp logger.cpp falco_outputs.cpp statsfilewriter.cpp falco.cpp)
+add_executable(falco configuration.cpp logger.cpp token_bucket.cpp falco_outputs.cpp statsfilewriter.cpp falco.cpp)
 
 target_link_libraries(falco falco_engine sinsp)
 target_link_libraries(falco

userspace/falco/configuration.cpp

@@ -101,6 +101,9 @@ void falco_configuration::init(string conf_filename, list<string> &cmdline_optio
 		throw invalid_argument("Error reading config file (" + m_config_file + "): No outputs configured. Please configure at least one output file output enabled but no filename in configuration block");
 	}
 
+	m_notifications_rate = m_config->get_scalar<uint32_t>("outputs", "rate", 1);
+	m_notifications_max_burst = m_config->get_scalar<uint32_t>("outputs", "max_burst", 1000);
+
 	falco_logger::log_stderr = m_config->get_scalar<bool>("log_stderr", false);
 	falco_logger::log_syslog = m_config->get_scalar<bool>("log_syslog", true);
 }

userspace/falco/configuration.h

@@ -144,6 +144,8 @@ class falco_configuration
 	std::list<std::string> m_rules_filenames;
 	bool m_json_output;
 	std::vector<falco_outputs::output_config> m_outputs;
+	uint32_t m_notifications_rate;
+	uint32_t m_notifications_max_burst;
  private:
 	void init_cmdline_options(std::list<std::string> &cmdline_options);
 

userspace/falco/falco.cpp

@@ -427,7 +427,7 @@ int falco_init(int argc, char **argv)
 			engine->enable_rule(pattern, false);
 		}
 
-		outputs->init(config.m_json_output);
+		outputs->init(config.m_json_output, config.m_notifications_rate, config.m_notifications_max_burst);
 
 		if(!all_events)
 		{

userspace/falco/falco_outputs.cpp

@@ -51,7 +51,7 @@ falco_outputs::~falco_outputs()
 	}
 }
 
-void falco_outputs::init(bool json_output)
+void falco_outputs::init(bool json_output, uint32_t rate, uint32_t max_burst)
 {
 	// The engine must have been given an inspector by now.
 	if(! m_inspector)
@@ -68,6 +68,8 @@ void falco_outputs::init(bool json_output)
 
 	falco_logger::init(m_ls);
 
+	m_notifications_tb.init(rate, max_burst);
+
 	m_initialized = true;
 }
 
@@ -105,6 +107,12 @@ void falco_outputs::add_output(output_config oc)
 
 void falco_outputs::handle_event(sinsp_evt *ev, string &rule, string &priority, string &format)
 {
+	if(!m_notifications_tb.claim())
+	{
+		falco_logger::log(LOG_DEBUG, "Skipping rate-limited notification for rule " + rule + "\n");
+		return;
+	}
+
 	lua_getglobal(m_ls, m_lua_output_event.c_str());
 
 	if(lua_isfunction(m_ls, -1))

userspace/falco/falco_outputs.h

@@ -19,6 +19,7 @@ along with falco.  If not, see <http://www.gnu.org/licenses/>.
 #pragma once
 
 #include "falco_common.h"
+#include "token_bucket.h"
 
 //
 // This class acts as the primary interface between a program and the
@@ -40,7 +41,7 @@ class falco_outputs : public falco_common
 		std::map<std::string, std::string> options;
 	};
 
-	void init(bool json_output);
+	void init(bool json_output, uint32_t rate, uint32_t max_burst);
 
 	void add_output(output_config oc);
 
@@ -53,6 +54,9 @@ class falco_outputs : public falco_common
 private:
 	bool m_initialized;
 
+	// Rate limits notifications
+	token_bucket m_notifications_tb;
+
 	std::string m_lua_add_output = "add_output";
 	std::string m_lua_output_event = "output_event";
 	std::string m_lua_output_cleanup = "output_cleanup";

userspace/falco/token_bucket.cpp

@@ -0,0 +1,78 @@
+/*
+Copyright (C) 2016 Draios inc.
+
+This file is part of falco.
+
+falco is free software; you can redistribute it and/or modify
+it under the terms of the GNU General Public License version 2 as
+published by the Free Software Foundation.
+
+falco is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with falco.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <cstddef>
+#include <sys/time.h>
+
+#include "token_bucket.h"
+
+token_bucket::token_bucket()
+{
+	init(1, 1);
+}
+
+token_bucket::~token_bucket()
+{
+}
+
+void token_bucket::init(uint32_t rate, uint32_t max_tokens)
+{
+	m_rate = rate;
+	m_max_tokens = max_tokens;
+	m_tokens = max_tokens;
+	m_last_seen = get_epoch_ns();
+}
+
+bool token_bucket::claim()
+{
+	// Determine the number of tokens gained. Delta between
+	// last_seen and now, divided by the rate.
+	uint64_t now = get_epoch_ns();
+	uint64_t tokens_gained = (now - m_last_seen) / (m_rate * 1000000000);
+	m_last_seen = now;
+
+	m_tokens += tokens_gained;
+
+	//
+	// Cap at max_tokens
+	//
+	if(m_tokens > m_max_tokens)
+	{
+		m_tokens = m_max_tokens;
+	}
+
+	//
+	// If tokens is < 1, can't claim.
+	//
+	if(m_tokens < 1)
+	{
+		return false;
+	}
+
+	m_tokens--;
+
+	return true;
+}
+
+uint64_t token_bucket::get_epoch_ns()
+{
+    struct timeval tv;
+    gettimeofday(&tv, NULL);
+
+    return tv.tv_sec * (uint64_t) 1000000000 + (tv.tv_usec * 1000);
+}

userspace/falco/token_bucket.h

@@ -0,0 +1,68 @@
+/*
+Copyright (C) 2016 Draios inc.
+
+This file is part of falco.
+
+falco is free software; you can redistribute it and/or modify
+it under the terms of the GNU General Public License version 2 as
+published by the Free Software Foundation.
+
+falco is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU General Public License
+along with falco.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#pragma once
+
+#include <cstdint>
+
+// A simple token bucket that accumulates tokens at a fixed rate and allows
+// for limited bursting in the form of "banked" tokens.
+class token_bucket
+{
+public:
+	token_bucket();
+	virtual ~token_bucket();
+
+	//
+	// Initialize the token bucket and start accumulating tokens
+	//
+	void init(uint32_t rate, uint32_t max_tokens);
+
+	//
+	// Returns true if a token can be claimed. Also updates
+	// internal metrics.
+	//
+	bool claim();
+private:
+
+	// Utility function to get the time in nanoseconds since the epoch.
+	uint64_t get_epoch_ns();
+
+	//
+	// The number of tokens generated per second.
+	//
+	uint64_t m_rate;
+
+	//
+	// The maximum number of tokens that can be banked for future
+	// claim()s.
+	//
+	uint64_t m_max_tokens;
+
+	//
+	// The current number of tokens
+	//
+	uint64_t m_tokens;
+
+	//
+	// The last time claim() was called (or the object was created).
+	// Nanoseconds since the epoch.
+	//
+	uint64_t m_last_seen;
+};
+