Filtered toptalker forwarding

examples/configurations/analysis/traffic-specific-toptalker.yml

@@ -0,0 +1,40 @@
+---
+###############################################################################
+# Consume flow messages, it's best to use an enriched topic as flowdump
+# printing involves interface descriptions.
+- segment: kafkaconsumer
+  config:
+    server: kafka01.example.com:9093
+    topic: flow-messages-enriched
+    group: myuser-flowdump
+    user: myuser
+    pass: $KAFKA_SASL_PASS
+
+###############################################################################
+# filter with traffic specific toptalker thresholds
+# forward traffic to all ips that are matched to a csv export 
+- segment: traffic_specific_toptalkers
+  config:
+    endpoint: ":8085"
+    traffic_specific_toptalkers:
+    - filter: "proto udp"
+      subfilter:
+      - filter: "port 0"
+        traffictype: "UDP Fragmented"
+        thresholdbps: 100
+      - filter: "port 53"
+        traffictype: "DNS"
+        thresholdbps: 100
+      - filter: "port 123"
+        traffictype: "NTP"
+        thresholdbps: 100
+      - filter: "port 389"
+        traffictype: "CLDAP"
+        thresholdbps: 100
+    - filter: "proto icmp or proto icmpv6"
+      traffictype: "ICMP"
+      thresholdpps: 100
+  matching_pipeline:
+  - segment: csv
+    config:
+      filename: "matched.csv"

pipeline/config/config.go

@@ -7,5 +7,5 @@ type Config struct {
 
 	//Define custom segment specific structured config params here
 	//The parameter MUST contain the segement name to not conflict with other existing config parameters
-	ThresholdMetricDefinition []*ThresholdMetricDefinition `yaml:"traffic_specific_toptalkers,omitempty"`
+	ThresholdMetricDefinition []*ThresholdMetricConfig `yaml:"traffic_specific_toptalkers,omitempty"`
 }

pipeline/config/evaluation_mode/evaluation_mode.go

@@ -0,0 +1,45 @@
+package evaluation_mode
+
+import "strings"
+
+type EvaluationMode uint8
+
+const (
+	Unknown EvaluationMode = iota
+	Destination
+	SourceAndDestination
+	Source
+	Connection
+)
+
+func (s EvaluationMode) String() string {
+	return [...]string{"", "destination", "source and destination", "source", "connection"}[s]
+}
+
+var (
+	evaluationModes = map[string]EvaluationMode{
+		"":                       Unknown,
+		"destination":            Destination,
+		"destination and source": SourceAndDestination, //bad input compability
+		"source and destination": SourceAndDestination,
+		"source":                 Source,
+		"connection":             Connection,
+	}
+)
+
+func ParseEvaluationMode(str string) EvaluationMode {
+	r, ok := evaluationModes[strings.ToLower(str)]
+	if ok {
+		return r
+	}
+	return Unknown
+}
+
+func (a EvaluationMode) MarshalText() ([]byte, error) {
+	return []byte(a.String()), nil
+}
+
+func (a *EvaluationMode) UnmarshalText(text []byte) error {
+	*a = ParseEvaluationMode(string(text))
+	return nil
+}

pipeline/config/segment_representation.go

@@ -12,8 +12,11 @@ type SegmentRepr struct {
 	Config Config `yaml:"config"`         // to be expanded by our instance
 	Jobs   int    `yaml:"jobs,omitempty"` // parallel jobs running the pipeline
 
-	//Adds if/then/else - not part of config for backwards compability
+	//Adds if/then/else
 	BranchOptions `yaml:",inline"`
+
+	//Adds matching_pipeline - used to process msg hows IPs are matching a filter
+	ThresholdMetricDefinition `yaml:",inline"`
 }
 
 // Returns the SegmentRepr's Config with all its variables expanded. It tries

pipeline/config/toptalker_config.go

@@ -1,18 +1,25 @@
 package config
 
+import a "github.com/BelWue/flowpipeline/pipeline/config/evaluation_mode"
+
 type ThresholdMetricDefinition struct {
+	MatchingPipeline []SegmentRepr `yaml:"matching_pipeline,omitempty"`
+}
+
+type ThresholdMetricConfig struct {
 	PrometheusMetricsParamsDefinition `yaml:",inline"`
-	FilterDefinition                  string                       `yaml:"filter,omitempty"`
-	SubDefinitions                    []*ThresholdMetricDefinition `yaml:"subfilter,omitempty"`
+	FilterDefinition                  string                   `yaml:"filter,omitempty"`
+	SubDefinitions                    []*ThresholdMetricConfig `yaml:"subfilter,omitempty"`
 }
 
 type PrometheusMetricsParamsDefinition struct {
-	TrafficType      string `yaml:"traffictype,omitempty"`      // optional, default is "", name for the traffic type (included as label)
-	Buckets          int    `yaml:"buckets,omitempty"`          // optional, default is 60, sets the number of seconds used as a sliding window size
-	ThresholdBuckets int    `yaml:"thresholdbuckets,omitempty"` // optional, use the last N buckets for calculation of averages, default: $Buckets
-	ReportBuckets    int    `yaml:"reportbuckets,omitempty"`    // optional, use the last N buckets to calculate averages that are reported as result, default: $Buckets
-	BucketDuration   int    `yaml:"bucketduration,omitempty"`   // optional, duration of a bucket, default is 1 second
-	ThresholdBps     uint64 `yaml:"thresholdbps,omitempty"`     // optional, default is 0, only log talkers with an average bits per second rate higher than this value
-	ThresholdPps     uint64 `yaml:"thresholdpps,omitempty"`     // optional, default is 0, only log talkers with an average packets per second rate higher than this value
-	RelevantAddress  string `yaml:"relevantaddress,omitempty"`  // optional, default is "destination", options are "destination", "source", "both", "connection"
+	TrafficType      string           `yaml:"traffictype,omitempty"`      // optional, default is "", name for the traffic type (included as label)
+	Buckets          int              `yaml:"buckets,omitempty"`          // optional, default is 60, sets the number of seconds used as a sliding window size
+	ThresholdBuckets int              `yaml:"thresholdbuckets,omitempty"` // optional, use the last N buckets for calculation of averages, default: $Buckets
+	ReportBuckets    int              `yaml:"reportbuckets,omitempty"`    // optional, use the last N buckets to calculate averages that are reported as result, default: $Buckets
+	BucketDuration   int              `yaml:"bucketduration,omitempty"`   // optional, duration of a bucket, default is 1 second
+	ThresholdBps     uint64           `yaml:"thresholdbps,omitempty"`     // optional, default is 0, only log talkers with an average bits per second rate higher than this value
+	ThresholdPps     uint64           `yaml:"thresholdpps,omitempty"`     // optional, default is 0, only log talkers with an average packets per second rate higher than this value
+	RelevantAddress  a.EvaluationMode `yaml:"relevantaddress,omitempty"`  // optional deprecated replaced by evaluationmode, default is "destination", options are "destination", "source", "both", "connection"
+	EvaluationMode   a.EvaluationMode `yaml:"evaluationmode,omitempty"`   // optional, default is "destination", options are "destination", "source", "source and destination", "connection"
 }

segments/analysis/toptalkers_metrics/toptalker_database.go

@@ -5,22 +5,40 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/BelWue/flowpipeline/pb"
+	"github.com/BelWue/flowpipeline/pipeline/config/evaluation_mode"
 )
 
-type Record struct {
-	FwdBytes       []uint64
-	FwdPackets     []uint64
-	DropBytes      []uint64
-	DropPackets    []uint64
-	capacity       int
-	pointer        int
-	aboveThreshold atomic.Bool
-	Address        string
+type Record interface {
+	Append(*pb.EnrichedFlow)
+	GetMetrics(buckets int, bucketDuration int) (float64, float64, float64, float64, string)
+	AboveThreshold() *atomic.Bool
+	tick(thresholdBuckets int, bucketDuration int, thresholdBps uint64, thresholdPps uint64)
+	isEmpty() bool
+}
+
+type DefaultRecord struct {
+	FwdBytes             []uint64
+	FwdPackets           []uint64
+	DropBytes            []uint64
+	DropPackets          []uint64
+	capacity             int
+	pointer              int
+	AboveThresholdAtomic atomic.Bool
+	Address              string
+
 	sync.RWMutex
 }
 
+type TwoWayRecord struct {
+	DefaultRecord
+	SrcAddr string
+	DstAddr string
+}
+
 type Database struct {
-	database           *map[string]*Record
+	database           *map[string]Record
 	TrafficType        string
 	thresholdBps       uint64
 	thresholdPps       uint64
@@ -31,14 +49,15 @@ type Database struct {
 	cleanupCounter     int
 	cleanupWindowSizes int
 	promExporter       *PrometheusExporter
+	evaluationMode     evaluation_mode.EvaluationMode
 	stopCleanupC       chan struct{}
 	stopClockC         chan struct{}
 	sync.RWMutex
 }
 
 func NewDatabase(params PrometheusMetricsParams, promExporter *PrometheusExporter) Database {
 	return Database{
-		database:           &map[string]*Record{},
+		database:           &map[string]Record{},
 		thresholdBps:       params.ThresholdBps,
 		thresholdPps:       params.ThresholdPps,
 		thresholdBuckets:   params.ThresholdBuckets,
@@ -49,29 +68,47 @@ func NewDatabase(params PrometheusMetricsParams, promExporter *PrometheusExporte
 		ReportBuckets:      params.ReportBuckets,
 		TrafficType:        params.TrafficType,
 		BucketDuration:     params.BucketDuration,
+		evaluationMode:     params.EvaluationMode,
 		stopCleanupC:       make(chan struct{}),
 		stopClockC:         make(chan struct{}),
 	}
 }
 
-func (db *Database) GetRecord(address string) *Record {
-	return db.GetTypedRecord("", address)
+func (db *Database) GetRecord(key string, srcAddr string, dstAddr string) Record {
+	return db.GetTypedRecord("", key, srcAddr, dstAddr)
 }
 
-func (db *Database) GetTypedRecord(typeLabel string, address string) *Record {
+func (db *Database) GetTypedRecord(typeLabel string, address string, srcAddr string, dstAddr string) Record {
 	key := fmt.Sprintf("%s: %s", typeLabel, address)
 	db.Lock()
 	defer db.Unlock()
 	record, found := (*db.database)[key]
 	if !found || record == nil {
-		record = NewRecord(db.ReportBuckets, address)
+		switch db.evaluationMode {
+		case evaluation_mode.SourceAndDestination:
+			record = NewTwoWayRecord(db.ReportBuckets, address, srcAddr, dstAddr)
+		case evaluation_mode.Connection:
+			record = NewTwoWayRecord(db.ReportBuckets, address, srcAddr, dstAddr)
+		default:
+			record = NewDefaultRecord(db.ReportBuckets, address)
+		}
+
 		(*db.database)[key] = record
 	}
 	return record
 }
 
-func NewRecord(windowSize int, address string) *Record {
-	record := &Record{
+func NewTwoWayRecord(windowSize int, address string, srcAddr string, dstAddr string) Record {
+	record := &TwoWayRecord{
+		DefaultRecord: *NewDefaultRecord(windowSize, address),
+		SrcAddr:       srcAddr,
+		DstAddr:       dstAddr,
+	}
+	return record
+}
+
+func NewDefaultRecord(windowSize int, address string) *DefaultRecord {
+	record := &DefaultRecord{
 		FwdBytes:    make([]uint64, windowSize),
 		FwdPackets:  make([]uint64, windowSize),
 		DropBytes:   make([]uint64, windowSize),
@@ -83,7 +120,14 @@ func NewRecord(windowSize int, address string) *Record {
 	return record
 }
 
-func (record *Record) Append(bytes uint64, packets uint64, statusFwd bool) {
+func (record *DefaultRecord) AboveThreshold() *atomic.Bool {
+	return &record.AboveThresholdAtomic
+}
+
+func (record *DefaultRecord) Append(msg *pb.EnrichedFlow) {
+	bytes := msg.Bytes
+	packets := msg.Packets
+	statusFwd := msg.IsForwarded()
 	record.Lock()
 	defer record.Unlock()
 	if statusFwd {
@@ -95,7 +139,7 @@ func (record *Record) Append(bytes uint64, packets uint64, statusFwd bool) {
 	}
 }
 
-func (record *Record) isEmpty() bool {
+func (record *DefaultRecord) isEmpty() bool {
 	record.RLock()
 	defer record.RUnlock()
 	for i := 0; i < record.capacity; i++ {
@@ -106,7 +150,7 @@ func (record *Record) isEmpty() bool {
 	return true
 }
 
-func (record *Record) GetMetrics(buckets int, bucketDuration int) (float64, float64, float64, float64, string) {
+func (record *DefaultRecord) GetMetrics(buckets int, bucketDuration int) (float64, float64, float64, float64, string) {
 	// buckets == 0 means "look at the whole window"
 	if buckets == 0 {
 		buckets = record.capacity
@@ -136,7 +180,7 @@ func (record *Record) GetMetrics(buckets int, bucketDuration int) (float64, floa
 	return sumFwdBps, sumFwdPps, sumDropBps, sumDropPps, record.Address
 }
 
-func (record *Record) tick(thresholdBuckets int, bucketDuration int, thresholdBps uint64, thresholdPps uint64) {
+func (record *DefaultRecord) tick(thresholdBuckets int, bucketDuration int, thresholdBps uint64, thresholdPps uint64) {
 	record.Lock()
 	defer record.Unlock()
 	// advance pointer to the next position
@@ -164,9 +208,9 @@ func (record *Record) tick(thresholdBuckets int, bucketDuration int, thresholdBp
 	bps := uint64(float64(sumBytes*8) / float64(bucketDuration*thresholdBuckets))
 	pps := uint64(float64(sumPackets) / float64(bucketDuration*thresholdBuckets))
 	if (bps > thresholdBps) && (pps > thresholdPps) {
-		record.aboveThreshold.Store(true)
+		record.AboveThresholdAtomic.Store(true)
 	} else {
-		record.aboveThreshold.Store(false)
+		record.AboveThresholdAtomic.Store(false)
 	}
 	// clear the current bucket
 	record.FwdBytes[record.pointer] = 0
@@ -223,11 +267,11 @@ func (db *Database) StopTimers() {
 
 func (db *Database) GetAllRecords() <-chan struct {
 	key    string
-	record *Record
+	record Record
 } {
 	out := make(chan struct {
 		key    string
-		record *Record
+		record Record
 	})
 	go func() {
 		db.Lock()
@@ -238,7 +282,7 @@ func (db *Database) GetAllRecords() <-chan struct {
 		for key, record := range *db.database {
 			out <- struct {
 				key    string
-				record *Record
+				record Record
 			}{key, record}
 		}
 	}()