fix(userspace/falco): correcly log SIGINT handling (fixes #791)

Signed-off-by: Leonardo Di Donato <leodidonato@gmail.com>

userspace/falco/configuration.cpp

@@ -150,7 +150,7 @@ void falco_configuration::init(string conf_filename, list<string> &cmdline_optio
 
 	m_grpc_enabled = m_config->get_scalar<bool>("grpc", "enabled", false);
 	m_grpc_bind_address = m_config->get_scalar<string>("grpc", "bind_address", "0.0.0.0:5060");
-	m_grpc_threadiness = m_config->get_scalar<uint32_t>("grpc", "threadiness", 8);
+	m_grpc_threadiness = m_config->get_scalar<uint32_t>("grpc", "threadiness", 8); // todo > limit it to avoid overshubscription? std::thread::hardware_concurrency()
 	// todo(fntlnz,leodido) > chose correct paths
 	m_grpc_private_key = m_config->get_scalar<string>("grpc", "private_key", "");
 	m_grpc_cert_chain = m_config->get_scalar<string>("grpc", "cert_chain", "");

userspace/falco/falco.cpp

@@ -51,6 +51,8 @@ limitations under the License.
 #include "grpc_server.h"
 #include "falco_output_queue.h"
 
+#include <future>
+
 typedef function<void(sinsp* inspector)> open_t;
 
 bool g_terminate = false;
@@ -272,7 +274,12 @@ uint64_t do_inspect(falco_engine *engine,
 			g_reopen_outputs = false;
 		}
 
-		if (g_terminate || g_restart)
+		if(g_terminate)
+		{
+			falco_logger::log(LOG_INFO, "SIGINT received, exiting...\n");
+			break;
+		}
+		else if (g_restart)
 		{
 			falco_logger::log(LOG_INFO, "SIGHUP Received, restarting...\n");
 			break;
@@ -1166,13 +1173,18 @@ int falco_init(int argc, char **argv)
 		}
 
 		// grpc server
-
+		std::thread grpc_server_thread;
+		falco_grpc_server grpc_server(config.m_grpc_bind_address, config.m_grpc_threadiness);
+		if(config.m_grpc_enabled)
+		{
 			// TODO(fntlnz,leodido): when we want to spawn multiple threads we need to have a queue per thread, or implement
 			// different queuing mechanisms, round robin, fanout? What we want to achieve?
-		int threadiness = 1; // TODO(fntlnz, leodido): make this configurable
+			// TODO(fntlnz, leodido): make sure we clean it gracefully
-
+			// grpc_server_thread = std::thread(start_grpc_server, config.m_grpc_bind_address, config.m_grpc_threadiness);
-		// TODO(fntlnz): do any handling, make sure we handle signals in the GRPC server and we clean it gracefully
+			grpc_server_thread = std::thread([&grpc_server] {
-		std::thread grpc_server_thread (start_grpc_server, "0.0.0.0:5060", threadiness);
+				grpc_server.run();
+			});
+		}
 
 		if(!trace_filename.empty() && !trace_is_scap)
 		{
@@ -1217,6 +1229,7 @@ int falco_init(int argc, char **argv)
 		engine->print_stats();
 		sdropmgr.print_stats();
 		webserver.stop();
+		grpc_server.shutdown();
 	}
 	catch(exception &e)
 	{

userspace/falco/grpc_server.cpp

@@ -99,11 +99,11 @@ void request_stream_context<request, response>::end(falco_grpc_server* srv, bool
 
 void falco_grpc_server::thread_process(int thread_index)
 {
-	// TODO: is this right? That's what we want?
 	// Tell pthread to not handle termination signals in the current thread
 	sigset_t set;
 	sigemptyset(&set);
-	sigaddset(&set, SIGHUP);
+	sigaddset(&set, SIGTERM);
+	// sigaddset(&set, SIGHUP); // todo > SIGHUP should restart Falco, what to do?
 	sigaddset(&set, SIGINT);
 	pthread_sigmask(SIG_BLOCK, &set, nullptr);
 
@@ -192,22 +192,24 @@ void read(const std::string& filename, std::string& data)
 extern "C" void handle_signal(int signum)
 {
 	// todo > print "Got signal << sigenum << : exiting...");
-	exit(1);
+	// exit(1);
 }
 
 void falco_grpc_server::run()
 {
 	// Handle SIGHUP and SIGINT in the main process.
 	// Not in the spawned threads.
-    sigset_t set;
+	// sigset_t set;
-    sigemptyset(&set);
+	// sigemptyset(&set);
-    sigaddset(&set, SIGHUP);
+	// sigaddset(&set, SIGTERM);
-    sigaddset(&set, SIGINT);
+	// // sigaddset(&set, SIGHUP); // todo > SIGHUP should restart Falco, what to do?
-    pthread_sigmask(SIG_UNBLOCK, &set, nullptr);
+	// sigaddset(&set, SIGINT);
-	struct sigaction action;
+	// pthread_sigmask(SIG_UNBLOCK, &set, nullptr);
-	action.sa_handler = handle_signal;
+
-	sigaction(SIGHUP, &action, nullptr);
+	// struct sigaction action;
-	sigaction(SIGINT, &action, nullptr);
+	// action.sa_handler = handle_signal;
+	// sigaction(SIGHUP, &action, nullptr);
+	// sigaction(SIGINT, &action, nullptr);
 
 	string private_key;
 	string cert_chain;
@@ -236,7 +238,11 @@ void falco_grpc_server::run()
 	m_server = builder.BuildAndStart();
 	falco_logger::log(LOG_INFO, "Starting gRPC webserver at " + m_server_addr + "\n");
 
-	int context_count = m_threadiness * 1; // todo > 10 or 100?
+	// Create context for server threads
+	// The number of contexts is multiple of the number of threads
+	// This defines the number of simultaneous completion queue requests of the same type (service::AsyncService::Request##RPC)
+	// For this approach to be sufficient falco_grpc_server::IMPL have to be fast
+	int context_count = m_threadiness * 10;
 	PROCESS_STREAM(request, response, subscribe, subscribe, context_count)
 
 	m_threads.resize(m_threadiness);