netmaker/pro/controllers/flows.go

package controllers

import (
	"errors"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"

	"github.com/gorilla/mux"
	ch "github.com/gravitl/netmaker/clickhouse"
	"github.com/gravitl/netmaker/logic"
	"github.com/gravitl/netmaker/middleware"
	"github.com/gravitl/netmaker/scope"
	"gorm.io/gorm"
)

func FlowHandlers(r *mux.Router) {
	r.HandleFunc("/api/v1/flows", middleware.Scope(scope.TenantScope, logic.SecurityCheck(true, http.HandlerFunc(handleListFlows)))).Methods(http.MethodGet)
}

const (
	querySelect = `
SELECT
	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,
	dst_ip, dst_type, dst_entity_id, dst_entity_name,
	start_ts, end_ts,
	bytes_sent, bytes_recv,
	packets_sent, packets_recv,
	status, version
FROM flows`
	queryOrder = `
ORDER BY version DESC
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"`
	Protocol      uint16    `ch:"protocol" json:"protocol"`
	SrcPort       uint16    `ch:"src_port" json:"src_port"`
	DstPort       uint16    `ch:"dst_port" json:"dst_port"`
	ICMPType      uint8     `ch:"icmp_type" json:"icmp_type"`
	ICMPCode      uint8     `ch:"icmp_code" json:"icmp_code"`
	Direction     string    `ch:"direction" json:"direction"`
	SrcIP         string    `ch:"src_ip" json:"src_ip"`
	SrcType       string    `ch:"src_type" json:"src_type"`
	SrcEntityID   string    `ch:"src_entity_id" json:"src_entity_id"`
	SrcEntityName string    `ch:"src_entity_name" json:"src_entity_name"`
	DstIP         string    `ch:"dst_ip" json:"dst_ip"`
	DstType       string    `ch:"dst_type" json:"dst_type"`
	DstEntityID   string    `ch:"dst_entity_id" json:"dst_entity_id"`
	DstEntityName string    `ch:"dst_entity_name" json:"dst_entity_name"`
	StartTs       time.Time `ch:"start_ts" json:"start_ts"`
	EndTs         time.Time `ch:"end_ts" json:"end_ts"`
	BytesSent     uint64    `ch:"bytes_sent" json:"bytes_sent"`
	BytesRecv     uint64    `ch:"bytes_recv" json:"bytes_recv"`
	PacketsSent   uint64    `ch:"packets_sent" json:"packets_sent"`
	PacketsRecv   uint64    `ch:"packets_recv" json:"packets_recv"`
	Status        uint32    `ch:"status" json:"status"`
	Version       time.Time `ch:"version" json:"version"`
}

// @Summary     List flow logs
// @Router      /api/v1/flows [get]
// @Tags        Traffic Logs
// @Security    oauth
// @Produce     json
// @Param       network_id query string false "Filter by network ID"
// @Param       from query string false "Start time in RFC3339 format"
// @Param       to query string false "End time in RFC3339 format"
// @Param       src_type []query string false "Source type filter"
// @Param       src_entity_id query string false "Source entity ID filter"
// @Param       dst_type []query string false "Destination type filter"
// @Param       dst_entity_id query string false "Destination entity ID filter"
// @Param       protocol []query string false "Protocol filter"
// @Param       node_id query string false "Node ID filter"
// @Param       username query string false "Username filter"
// @Param       page query int false "Page number"
// @Param       per_page query int false "Items per page (max 1000)"
// @Success     200 {array} FlowRow
// @Failure     400 {object} models.ErrorResponse
// @Failure     500 {object} models.ErrorResponse
func handleListFlows(w http.ResponseWriter, r *http.Request) {
	if !logic.GetFeatureFlags(r.Context()).EnableFlowLogs || !logic.GetServerSettings(r.Context()).EnableFlowLogs {
		logic.ReturnErrorResponse(w, r, logic.FormatError(errors.New("flow logs not enabled"), logic.Forbidden))
		return
	}

	conn, err := ch.FromContext(r.Context())
	if err != nil {
		logic.ReturnErrorResponse(w, r,
			logic.FormatError(fmt.Errorf("clickhouse connection not available: %v", err), logic.Internal))
		return
	}

	q := r.URL.Query()

	// TODO: handle query filters better
	var (
		whereParts []string
		args       []any
	)

	// -1. Tenant filter.
	whereParts = append(whereParts, "tenant_id = ?")
	args = append(args, scope.ID(r.Context()))

	// 0. Network filter.
	networkID := q.Get("network_id")
	if networkID != "" {
		whereParts = append(whereParts, "network_id = ?")
		args = append(args, networkID)
	}

	// 1. Time filtering (start_ts: UInt64 timestamp in ms)
	fromStr := q.Get("from")
	toStr := q.Get("to")

	if fromStr != "" {
		fromVal, err := time.Parse(time.RFC3339, fromStr)
		if err != nil {
			logic.ReturnErrorResponse(w, r, logic.FormatError(fmt.Errorf("invalid 'from' timestamp: %v", err), logic.BadReq))
			return
		}
		whereParts = append(whereParts, "start_ts >= ?")
		args = append(args, fromVal)
	}

	if toStr != "" {
		toVal, err := time.Parse(time.RFC3339, toStr)
		if err != nil {
			logic.ReturnErrorResponse(w, r, logic.FormatError(fmt.Errorf("invalid 'to' timestamp: %v", err), logic.BadReq))
			return
		}
		whereParts = append(whereParts, "start_ts <= ?")
		args = append(args, toVal)
	}

	// 2. Source filters
	if q.Get("src_type") != "" {
		whereParts = append(whereParts, "src_type IN ?")
		args = append(args, q["src_type"])
	}

	srcEntity := q.Get("src_entity_id")
	if srcEntity != "" {
		whereParts = append(whereParts, "src_entity_id = ?")
		args = append(args, srcEntity)
	}

	// 3. Destination filters
	if q.Get("dst_type") != "" {
		whereParts = append(whereParts, "dst_type IN ?")
		args = append(args, q["dst_type"])
	}

	dstEntity := q.Get("dst_entity_id")
	if dstEntity != "" {
		whereParts = append(whereParts, "dst_entity_id = ?")
		args = append(args, dstEntity)
	}

	// 4. Protocol filter
	if q.Get("protocol") != "" {
		whereParts = append(whereParts, "protocol IN ?")
		args = append(args, q["protocol"])
	}

	// 5. Node filter
	nodeID := q.Get("node_id")
	if nodeID != "" {
		node, err := logic.GetNodeByID(nodeID)
		if err != nil {
			errType := logic.Internal
			if errors.Is(err, gorm.ErrRecordNotFound) {
				errType = logic.BadReq
			}
			logic.ReturnErrorResponse(w, r, logic.FormatError(fmt.Errorf("error fetching node with id %s: %v", nodeID, err), errType))
			return
		}

		if networkID == "" {
			whereParts = append(whereParts, "network_id = ?")
			args = append(args, node.Network)
		} else {
			if networkID != node.Network {
				logic.ReturnErrorResponse(w, r, logic.FormatError(fmt.Errorf("node with id %s does not belong to network %s", nodeID, networkID), logic.BadReq))
				return
			}
		}

		whereParts = append(whereParts, "host_id = ?")
		args = append(args, node.HostID)
	}

	// 6. User filter
	username := q.Get("username")
	if username != "" {
		if q.Has("src_type") || q.Has("dst_type") ||
			q.Has("src_entity_id") || q.Has("dst_entity_id") {
			logic.ReturnErrorResponse(w, r, logic.FormatError(errors.New("cannot provide username filter along with src/dst type and id filters"), logic.BadReq))
			return
		}

		srcTypeStr := "user"
		srcEntity := username
		dstTypeStr := "user"
		dstEntity := username

		whereParts = append(whereParts, "((src_type = ? AND src_entity_id = ?) OR (dst_type = ? AND dst_entity_id = ?))")
		args = append(args, srcTypeStr, srcEntity, dstTypeStr, dstEntity)
	}

	// 7. Ignore flow logs with zero end_ts.
	whereParts = append(whereParts, "end_ts <> ?")
	args = append(args, time.Unix(0, 0))

	// Pagination
	page := parseIntOrDefault(q.Get("page"), 1)
	perPage := parseIntOrDefault(q.Get("per_page"), 100)
	if perPage > 1000 {
		perPage = 1000
	}
	offset := (page - 1) * perPage

	whereSQL := ""
	if len(whereParts) > 0 {
		whereSQL = "WHERE " + strings.Join(whereParts, " AND ")
	}

	query := querySelect + "\n" + whereSQL + "\n" + queryOrder

	args = append(args, perPage, offset)

	rows, err := conn.Query(r.Context(), query, args...)
	if err != nil {
		logic.ReturnErrorResponse(w, r,
			logic.FormatError(fmt.Errorf("error fetching flows: %v", err), logic.Internal))
		return
	}
	defer rows.Close()

	result := make([]FlowRow, 0, 1000)

	for rows.Next() {
		var fr FlowRow
		if err := rows.ScanStruct(&fr); err != nil {
			logic.ReturnErrorResponse(w, r, logic.FormatError(fmt.Errorf("error fetching flows: %v", err), logic.Internal))
			return
		}
		result = append(result, fr)
	}

	logic.ReturnSuccessResponseWithJson(w, r, result, "flows retrieved successfully")
}

func parseIntOrDefault(s string, def int) int {
	if s == "" {
		return def
	}
	v, err := strconv.Atoi(s)
	if err != nil || v <= 0 {
		return def
	}
	return v
}