bidirectional gRPC outputs + fix server streaming gRPC outputs

proposals/20190826-grpc-outputs.md

@@ -1,4 +1,4 @@
-# gRPC Falco Output
+# Falco gRPC Outputs
 
 <!-- toc -->
 
@@ -25,7 +25,7 @@ An alert is an "output" when it goes over a transport, and it is emitted by Falc
 
 At the current moment, however, Falco can deliver alerts in a very basic way, for example by dumping them to standard output.
 
-For this reason, many Falco users asked, with issues - eg., [falco#528](https://github.com/falcosecurity/falco/issues/528) - or in the [slack channel](https://sysdig.slack.com) if we can find a more consumable way to implement Falco outputs in an extensible way.
+For this reason, many Falco users asked, with issues - eg., [falco#528](https://github.com/falcosecurity/falco/issues/528) - or in the [slack channel](https://slack.k8s.io) if we can find a more consumable way to implement Falco outputs in an extensible way.
 
 The motivation behind this proposal is to design a new output implementation that can meet our user's needs.
 
@@ -39,7 +39,10 @@ The motivation behind this proposal is to design a new output implementation tha
 - To continue supporting the old output formats by implementing their same interface
 - To be secure by default (**mutual TLS** authentication)
 - To be **asynchronous** and **non-blocking**
-- To implement a Go SDK
+- To provide a connection over unix socket (no authentication)
+- To implement a Go client
+- To implement a Rust client
+- To implement a Python client
 
 ### Non-Goals
 
@@ -77,26 +80,25 @@ syntax = "proto3";
 import "google/protobuf/timestamp.proto";
 import "schema.proto";
 
-package falco.output;
+package falco.outputs;
 
-option go_package = "github.com/falcosecurity/client-go/pkg/api/output";
+option go_package = "github.com/falcosecurity/client-go/pkg/api/outputs";
 
-// The `subscribe` service defines the RPC call
-// to perform an output `request` which will lead to obtain an output `response`.
+// This service defines the RPC methods
+// to `request` a stream of output `response`s.
 service service {
-  rpc subscribe(request) returns (stream response);
+  // Subscribe to a stream of Falco outputs by sending a stream of requests.
+  rpc sub(stream request) returns (stream response);
+  // Get all the Falco outputs present in the system up to this call.
+  rpc get(request) returns (stream response);
 }
 
 // The `request` message is the logical representation of the request model.
-// It is the input of the `subscribe` service.
-// It is used to configure the kind of subscription to the gRPC streaming server.
+// It is the input of the `output.service` service.
 message request {
-  bool keepalive = 1;
-  // string duration = 2; // TODO(leodido, fntlnz): not handled yet but keeping for reference.
-  // repeated string tags = 3; // TODO(leodido, fntlnz): not handled yet but keeping for reference.
 }
 
-// The `response` message is the logical representation of the output model.
+// The `response` message is the representation of the output model.
 // It contains all the elements that Falco emits in an output along with the
 // definitions for priorities and source.
 message response {
@@ -106,7 +108,7 @@ message response {
   string rule = 4;
   string output = 5;
   map<string, string> output_fields = 6;
-  // repeated string tags = 7; // TODO(leodido,fntlnz): tags not supported yet, keeping for reference
+  string hostname = 7;
 }
 ```
 

userspace/falco/CMakeLists.txt

@@ -19,23 +19,24 @@ add_custom_command(
     ${CMAKE_CURRENT_BINARY_DIR}/version.grpc.pb.h
     ${CMAKE_CURRENT_BINARY_DIR}/version.pb.cc
     ${CMAKE_CURRENT_BINARY_DIR}/version.pb.h
-    ${CMAKE_CURRENT_BINARY_DIR}/output.grpc.pb.cc
-    ${CMAKE_CURRENT_BINARY_DIR}/output.grpc.pb.h
-    ${CMAKE_CURRENT_BINARY_DIR}/output.pb.cc
-    ${CMAKE_CURRENT_BINARY_DIR}/output.pb.h
+    ${CMAKE_CURRENT_BINARY_DIR}/outputs.grpc.pb.cc
+    ${CMAKE_CURRENT_BINARY_DIR}/outputs.grpc.pb.h
+    ${CMAKE_CURRENT_BINARY_DIR}/outputs.pb.cc
+    ${CMAKE_CURRENT_BINARY_DIR}/outputs.pb.h
     ${CMAKE_CURRENT_BINARY_DIR}/schema.pb.cc
     ${CMAKE_CURRENT_BINARY_DIR}/schema.pb.h
-  COMMENT "Generate gRPC version API"
+  COMMENT "Generate gRPC API"
+  # Falco gRPC Version API
   DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/version.proto
   COMMAND ${PROTOC} -I ${CMAKE_CURRENT_SOURCE_DIR} --cpp_out=. ${CMAKE_CURRENT_SOURCE_DIR}/version.proto
   COMMAND ${PROTOC} -I ${CMAKE_CURRENT_SOURCE_DIR} --grpc_out=. --plugin=protoc-gen-grpc=${GRPC_CPP_PLUGIN}
           ${CMAKE_CURRENT_SOURCE_DIR}/version.proto
-  COMMENT "Generate gRPC outputs API"
-  DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/output.proto
-  COMMAND ${PROTOC} -I ${CMAKE_CURRENT_SOURCE_DIR} --cpp_out=. ${CMAKE_CURRENT_SOURCE_DIR}/output.proto
+  # Falco gRPC Outputs API
+  DEPENDS ${CMAKE_CURRENT_SOURCE_DIR}/outputs.proto
+  COMMAND ${PROTOC} -I ${CMAKE_CURRENT_SOURCE_DIR} --cpp_out=. ${CMAKE_CURRENT_SOURCE_DIR}/outputs.proto
           ${CMAKE_CURRENT_SOURCE_DIR}/schema.proto
   COMMAND ${PROTOC} -I ${CMAKE_CURRENT_SOURCE_DIR} --grpc_out=. --plugin=protoc-gen-grpc=${GRPC_CPP_PLUGIN}
-          ${CMAKE_CURRENT_SOURCE_DIR}/output.proto
+          ${CMAKE_CURRENT_SOURCE_DIR}/outputs.proto
   WORKING_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR})
 
 add_executable(
@@ -54,8 +55,8 @@ add_executable(
   grpc_server.cpp
   ${CMAKE_CURRENT_BINARY_DIR}/version.grpc.pb.cc
   ${CMAKE_CURRENT_BINARY_DIR}/version.pb.cc
-  ${CMAKE_CURRENT_BINARY_DIR}/output.grpc.pb.cc
-  ${CMAKE_CURRENT_BINARY_DIR}/output.pb.cc
+  ${CMAKE_CURRENT_BINARY_DIR}/outputs.grpc.pb.cc
+  ${CMAKE_CURRENT_BINARY_DIR}/outputs.pb.cc
   ${CMAKE_CURRENT_BINARY_DIR}/schema.pb.cc)
 
 add_dependencies(falco civetweb string-view-lite)

userspace/falco/falco_outputs.cpp

@@ -22,11 +22,10 @@ limitations under the License.
 
 #include "formats.h"
 #include "logger.h"
-#include "falco_output_queue.h"
+#include "falco_outputs_queue.h"
 #include "banned.h" // This raises a compilation error when certain functions are used
 
 using namespace std;
-using namespace falco::output;
 
 const static struct luaL_reg ll_falco_outputs [] =
 {
@@ -316,7 +315,7 @@ int falco_outputs::handle_grpc(lua_State *ls)
 		lua_error(ls);
 	}
 
-	response grpc_res = response();
+	falco::outputs::response grpc_res;
 
 	// time
 	gen_event *evt = (gen_event *)lua_topointer(ls, 1);
@@ -366,7 +365,7 @@ int falco_outputs::handle_grpc(lua_State *ls)
 	auto host = grpc_res.mutable_hostname();
 	*host = (char *)lua_tostring(ls, 7);
 
-	falco::output::queue::get().push(grpc_res);
+	falco::outputs::queue::get().push(grpc_res);
 
 	return 1;
 }

userspace/falco/falco_output_queue.h → userspace/falco/falco_outputs_queue.h

@@ -16,12 +16,12 @@ limitations under the License.
 
 #pragma once
 
-#include "output.pb.h"
+#include "outputs.pb.h"
 #include "tbb/concurrent_queue.h"
 
 namespace falco
 {
-namespace output
+namespace outputs
 {
 typedef tbb::concurrent_queue<response> response_cq;
 

userspace/falco/grpc_context.h

@@ -36,7 +36,7 @@ class context
 {
 public:
 	context(::grpc::ServerContext* ctx);
-	~context() = default;
+	virtual ~context() = default;
 
 	void get_metadata(std::string key, std::string& val);
 
@@ -50,7 +50,7 @@ class stream_context : public context
 public:
 	stream_context(::grpc::ServerContext* ctx):
 		context(ctx){};
-	~stream_context() = default;
+	virtual ~stream_context() = default;
 
 	enum : char
 	{
@@ -61,6 +61,15 @@ class stream_context : public context
 
 	mutable void* m_stream = nullptr; // todo(fntlnz, leodido) > useful in the future
 	mutable bool m_has_more = false;
+	mutable bool m_is_running = true;
+};
+
+class bidi_context : public stream_context
+{
+public:
+	bidi_context(::grpc::ServerContext* ctx):
+		stream_context(ctx){};
+	virtual ~bidi_context() = default;
 };
 
 } // namespace grpc

userspace/falco/grpc_request_context.cpp

@@ -24,12 +24,12 @@ namespace grpc
 {
 
 template<>
-void request_stream_context<falco::output::service, falco::output::request, falco::output::response>::start(server* srv)
+void request_stream_context<outputs::service, outputs::request, outputs::response>::start(server* srv)
 {
 	m_state = request_context_base::REQUEST;
 	m_srv_ctx.reset(new ::grpc::ServerContext);
 	auto srvctx = m_srv_ctx.get();
-	m_res_writer.reset(new ::grpc::ServerAsyncWriter<output::response>(srvctx));
+	m_res_writer.reset(new ::grpc::ServerAsyncWriter<outputs::response>(srvctx));
 	m_stream_ctx.reset();
 	m_req.Clear();
 	auto cq = srv->m_completion_queue.get();
@@ -38,7 +38,7 @@ void request_stream_context<falco::output::service, falco::output::request, falc
 }
 
 template<>
-void request_stream_context<falco::output::service, falco::output::request, falco::output::response>::process(server* srv)
+void request_stream_context<outputs::service, outputs::request, outputs::response>::process(server* srv)
 {
 	// When it is the 1st process call
 	if(m_state == request_context_base::REQUEST)
@@ -48,40 +48,46 @@ void request_stream_context<falco::output::service, falco::output::request, falc
 	}
 
 	// Processing
-	output::response res;
-	(srv->*m_process_func)(*m_stream_ctx, m_req, res); // subscribe()
+	outputs::response res;
+	(srv->*m_process_func)(*m_stream_ctx, m_req, res); // get()
+
+	if(!m_stream_ctx->m_is_running)
+	{
+		m_state = request_context_base::FINISH;
+		m_res_writer->Finish(::grpc::Status::OK, this);
+		return;
+	}
 
 	// When there are still more responses to stream
 	if(m_stream_ctx->m_has_more)
 	{
 		// todo(leodido) > log "write: tag=this, state=m_state"
 		m_res_writer->Write(res, this);
+		return;
 	}
+
 	// No more responses to stream
-	else
-	{
-		// Communicate to the gRPC runtime that we have finished.
-		// The memory address of "this" instance uniquely identifies the event.
-		m_state = request_context_base::FINISH;
-		// todo(leodido) > log "finish: tag=this, state=m_state"
-		m_res_writer->Finish(::grpc::Status::OK, this);
-	}
+	// Communicate to the gRPC runtime that we have finished.
+	// The memory address of "this" instance uniquely identifies the event.
+	m_state = request_context_base::FINISH;
+	// todo(leodido) > log "finish: tag=this, state=m_state"
+	m_res_writer->Finish(::grpc::Status::OK, this);
 }
 
 template<>
-void request_stream_context<falco::output::service, falco::output::request, falco::output::response>::end(server* srv, bool errored)
+void request_stream_context<outputs::service, outputs::request, outputs::response>::end(server* srv, bool error)
 {
 	if(m_stream_ctx)
 	{
-		if(errored)
+		if(error)
 		{
 			// todo(leodido) > log error "error streaming: tag=this, state=m_state, stream=m_stream_ctx->m_stream"
 		}
-		m_stream_ctx->m_status = errored ? stream_context::ERROR : stream_context::SUCCESS;
+		m_stream_ctx->m_status = error ? stream_context::ERROR : stream_context::SUCCESS;
 
 		// Complete the processing
-		output::response res;
-		(srv->*m_process_func)(*m_stream_ctx, m_req, res); // subscribe()
+		outputs::response res;
+		(srv->*m_process_func)(*m_stream_ctx, m_req, res); // get()
 	}
 	else
 	{
@@ -98,7 +104,7 @@ void request_stream_context<falco::output::service, falco::output::request, falc
 }
 
 template<>
-void falco::grpc::request_context<falco::version::service, falco::version::request, falco::version::response>::start(server* srv)
+void request_context<version::service, version::request, version::response>::start(server* srv)
 {
 	m_state = request_context_base::REQUEST;
 	m_srv_ctx.reset(new ::grpc::ServerContext);
@@ -113,7 +119,7 @@ void falco::grpc::request_context<falco::version::service, falco::version::reque
 }
 
 template<>
-void falco::grpc::request_context<falco::version::service, falco::version::request, falco::version::response>::process(server* srv)
+void request_context<version::service, version::request, version::response>::process(server* srv)
 {
 	version::response res;
 	(srv->*m_process_func)(m_srv_ctx.get(), m_req, res);
@@ -125,13 +131,85 @@ void falco::grpc::request_context<falco::version::service, falco::version::reque
 }
 
 template<>
-void falco::grpc::request_context<falco::version::service, falco::version::request, falco::version::response>::end(server* srv, bool errored)
+void request_context<version::service, version::request, version::response>::end(server* srv, bool error)
 {
 	// todo(leodido) > handle processing errors here
-	
+
 	// Ask to start processing requests
 	start(srv);
 }
 
+template<>
+void request_bidi_context<outputs::service, outputs::request, outputs::response>::start(server* srv)
+{
+	m_state = request_context_base::REQUEST;
+	m_srv_ctx.reset(new ::grpc::ServerContext);
+	auto srvctx = m_srv_ctx.get();
+	m_reader_writer.reset(new ::grpc::ServerAsyncReaderWriter<outputs::response, outputs::request>(srvctx));
+	m_req.Clear();
+	auto cq = srv->m_completion_queue.get();
+	// Request to start processing given requests.
+	// Using "this" - ie., the memory address of this context - as the tag that uniquely identifies the request.
+	// In this way, different contexts can serve different requests concurrently.
+	(srv->m_output_svc.*m_request_func)(srvctx, m_reader_writer.get(), cq, cq, this);
+};
+
+template<>
+void request_bidi_context<outputs::service, outputs::request, outputs::response>::process(server* srv)
+{
+	switch(m_state)
+	{
+	case request_context_base::REQUEST:
+		m_bidi_ctx.reset(new bidi_context(m_srv_ctx.get()));
+		m_bidi_ctx->m_status = bidi_context::STREAMING;
+		m_state = request_context_base::WRITE;
+		m_reader_writer->Read(&m_req, this);
+		return;
+	case request_context_base::WRITE:
+		// Processing
+		{
+			outputs::response res;
+			(srv->*m_process_func)(*m_bidi_ctx, m_req, res); // sub()
+
+			if(!m_bidi_ctx->m_is_running)
+			{
+				m_state = request_context_base::FINISH;
+				m_reader_writer->Finish(::grpc::Status::OK, this);
+				return;
+			}
+
+			if(m_bidi_ctx->m_has_more)
+			{
+				m_state = request_context_base::WRITE;
+				m_reader_writer->Write(res, this);
+				return;
+			}
+
+			m_state = request_context_base::WRITE;
+			m_reader_writer->Read(&m_req, this);
+		}
+
+		return;
+	default:
+		return;
+	}
+};
+
+template<>
+void request_bidi_context<outputs::service, outputs::request, outputs::response>::end(server* srv, bool error)
+{
+	if(m_bidi_ctx)
+	{
+		m_bidi_ctx->m_status = error ? bidi_context::ERROR : bidi_context::SUCCESS;
+
+		// Complete the processing
+		outputs::response res;
+		(srv->*m_process_func)(*m_bidi_ctx, m_req, res); // sub()
+	}
+
+	// Ask to start processing requests
+	start(srv);
+};
+
 } // namespace grpc
-} // namespace falco
+} // namespace falco