fix(userspace/falco): handle grpc server thread stop gracefully

Co-authored-by: Lorenzo Fontana <lo@linux.com>
Signed-off-by: Leonardo Di Donato <leodidonato@gmail.com>

userspace/falco/falco.cpp

@@ -51,8 +51,6 @@ 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;
@@ -458,6 +456,8 @@ int falco_init(int argc, char **argv)
 	scap_stats cstats;
 
 	falco_webserver webserver;
+	falco_grpc_server grpc_server;
+	std::thread grpc_server_thread;
 
 	static struct option long_options[] =
 	{
@@ -1173,14 +1173,11 @@ 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?
-			// TODO(fntlnz, leodido): make sure we clean it gracefully
+			grpc_server.init(config.m_grpc_bind_address, config.m_grpc_threadiness);
-			// grpc_server_thread = std::thread(start_grpc_server, config.m_grpc_bind_address, config.m_grpc_threadiness);
 			grpc_server_thread = std::thread([&grpc_server] {
 				grpc_server.run();
 			});
@@ -1229,7 +1226,11 @@ int falco_init(int argc, char **argv)
 		engine->print_stats();
 		sdropmgr.print_stats();
 		webserver.stop();
-		grpc_server.shutdown();
+		if(grpc_server_thread.joinable())
+		{
+			grpc_server.shutdown();
+			grpc_server_thread.join();
+		}
 	}
 	catch(exception &e)
 	{
@@ -1238,6 +1239,10 @@ int falco_init(int argc, char **argv)
 		result = EXIT_FAILURE;
 
 		webserver.stop();
+		if (grpc_server_thread.joinable()) {
+			grpc_server.shutdown();
+			grpc_server_thread.join();
+		}
 	}
 
 exit:

userspace/falco/grpc_server.cpp

@@ -22,7 +22,6 @@ limitations under the License.
 #include <grpc++/grpc++.h>
 #endif
 #include <signal.h> // pthread_sigmask
-#include <unistd.h> // sleep, _exit
 
 #include "logger.h"
 #include "grpc_server.h"
@@ -189,28 +188,14 @@ void read(const std::string& filename, std::string& data)
 	return;
 }
 
-extern "C" void handle_signal(int signum)
+void falco_grpc_server::init(std::string server_addr, int threadiness)
 {
-	// todo > print "Got signal << sigenum << : exiting...");
+	m_server_addr = server_addr;
-	// exit(1);
+	m_threadiness = threadiness;
 }
 
 void falco_grpc_server::run()
 {
-	// Handle SIGHUP and SIGINT in the main process.
-	// Not in the spawned threads.
-	// sigset_t set;
-	// sigemptyset(&set);
-	// sigaddset(&set, SIGTERM);
-	// // sigaddset(&set, SIGHUP); // todo > SIGHUP should restart Falco, what to do?
-	// sigaddset(&set, SIGINT);
-	// pthread_sigmask(SIG_UNBLOCK, &set, nullptr);
-
-	// struct sigaction action;
-	// action.sa_handler = handle_signal;
-	// sigaction(SIGHUP, &action, nullptr);
-	// sigaction(SIGINT, &action, nullptr);
-
 	string private_key;
 	string cert_chain;
 	string root_certs;
@@ -254,7 +239,6 @@ void falco_grpc_server::run()
 
 	while(is_running())
 	{
-		sleep(1); // todo > do we want to sleep here?
 	}
 
 	stop();

userspace/falco/grpc_server.h

@@ -28,6 +28,9 @@ limitations under the License.
 class falco_grpc_server : public falco_grpc_server_impl
 {
 public:
+	falco_grpc_server()
+	{
+	}
 	falco_grpc_server(std::string server_addr, int threadiness):
 		m_server_addr(server_addr),
 		m_threadiness(threadiness)
@@ -35,6 +38,7 @@ public:
 	}
 	virtual ~falco_grpc_server() = default;
 
+	void init(std::string server_addr, int threadiness);
 	void thread_process(int thread_index);
 	void run();
 	void stop();