NM-297: Use single DB for Traffic logs to support multitenancy

clickhouse/clickhouse.go

@@ -23,7 +23,10 @@ var ErrConnNotFound = errors.New("no connection in context")
 //go:embed initdb.d/02_create_flows_table.sql
 var createFlowsTableScript string
 
-func Initialize() error {
+//go:embed initdb.d/03_alter_flows_table_tenant_id.sql
+var alterFlowsTableTenantIDScript string
+
+func Initialize(defaultTenantID string) error {
 	if ch != nil {
 		return nil
 	}
@@ -49,6 +52,11 @@ func Initialize() error {
 		return err
 	}
 
+	err = chConn.Exec(ctx, fmt.Sprintf(alterFlowsTableTenantIDScript, defaultTenantID))
+	if err != nil {
+		return err
+	}
+
 	ch = chConn
 	return nil
 }

clickhouse/initdb.d/02_create_flows_table.sql

@@ -1,6 +1,7 @@
 CREATE TABLE IF NOT EXISTS flows (
     -- Identity
     flow_id            String,
+    tenant_id          String,
     host_id            String,
     host_name          String,
     network_id         String,

clickhouse/initdb.d/03_alter_flows_table_tenant_id.sql

@@ -0,0 +1,2 @@
+ALTER TABLE flows
+ADD COLUMN IF NOT EXISTS tenant_id String DEFAULT '%s';

controllers/server.go

@@ -272,7 +272,7 @@ func updateSettings(w http.ResponseWriter, r *http.Request) {
 
 	if currSettings.EnableFlowLogs != req.EnableFlowLogs {
 		if req.EnableFlowLogs {
-			err := ch.Initialize()
+			err := ch.Initialize(servercfg.GetNetmakerTenantID())
 			if err != nil {
 				err = fmt.Errorf("failed to enable flow logs: %v", err)
 				logic.ReturnErrorResponse(w, r, logic.FormatError(err, logic.Internal))

pro/controllers/flows.go

@@ -13,6 +13,7 @@ import (
 	"github.com/gravitl/netmaker/database"
 	"github.com/gravitl/netmaker/logic"
 	proLogic "github.com/gravitl/netmaker/pro/logic"
+	"github.com/gravitl/netmaker/servercfg"
 )
 
 func FlowHandlers(r *mux.Router) {
@@ -22,7 +23,7 @@ func FlowHandlers(r *mux.Router) {
 const (
 	querySelect = `
 SELECT
-	flow_id, host_id, host_name, network_id,
+	flow_id, tenant_id, host_id, host_name, network_id,
 	protocol, src_port, dst_port,
 	icmp_type, icmp_code, direction,
 	src_ip, src_type, src_entity_id, src_entity_name,
@@ -40,6 +41,7 @@ LIMIT ? OFFSET ?`
 // FlowRow represents a single flow log entry
 type FlowRow struct {
 	FlowID        string    `ch:"flow_id" json:"flow_id"`
+	TenantID      string    `ch:"tenant_id" json:"tenant_id"`
 	HostID        string    `ch:"host_id" json:"host_id"`
 	HostName      string    `ch:"host_name" json:"host_name"`
 	NetworkID     string    `ch:"network_id" json:"network_id"`
@@ -217,6 +219,11 @@ func handleListFlows(w http.ResponseWriter, r *http.Request) {
 		args = append(args, srcTypeStr, srcEntity, dstTypeStr, dstEntity)
 	}
 
+	if servercfg.GetNetmakerTenantID() != "" {
+		whereParts = append(whereParts, "tenant_id = ?")
+		args = append(args, servercfg.GetNetmakerTenantID())
+	}
+
 	// Pagination
 	page := parseIntOrDefault(q.Get("page"), 1)
 	perPage := parseIntOrDefault(q.Get("per_page"), 100)

pro/initialize.go

@@ -118,7 +118,7 @@ func InitPro() {
 			addJitExpiryHookWithEmail()
 
 			if proLogic.GetFeatureFlags().EnableFlowLogs && logic.GetServerSettings().EnableFlowLogs {
-				err := ch.Initialize()
+				err := ch.Initialize(servercfg.GetNetmakerTenantID())
 				if err != nil {
 					logger.Log(0, "error connecting to clickhouse:", err.Error())
 				}