diff --git a/examples/configurations/analysis/traffic-specific-toptalker.yml b/examples/configurations/analysis/traffic-specific-toptalker.yml new file mode 100644 index 00000000..e65f4e10 --- /dev/null +++ b/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" diff --git a/pipeline/config/config.go b/pipeline/config/config.go index 1befd1bb..4cfa4fa3 100644 --- a/pipeline/config/config.go +++ b/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"` } diff --git a/pipeline/config/evaluation_mode/evaluation_mode.go b/pipeline/config/evaluation_mode/evaluation_mode.go new file mode 100644 index 00000000..aab93708 --- /dev/null +++ b/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 +} diff --git a/pipeline/config/segment_representation.go b/pipeline/config/segment_representation.go index 6f46dd3a..ce5fe3ff 100644 --- a/pipeline/config/segment_representation.go +++ b/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 diff --git a/pipeline/config/toptalker_config.go b/pipeline/config/toptalker_config.go index 087ed25f..478c8ca5 100644 --- a/pipeline/config/toptalker_config.go +++ b/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" } diff --git a/segments/analysis/toptalkers_metrics/toptalker_database.go b/segments/analysis/toptalkers_metrics/toptalker_database.go index b054b8e8..efca836f 100644 --- a/segments/analysis/toptalkers_metrics/toptalker_database.go +++ b/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,6 +49,7 @@ type Database struct { cleanupCounter int cleanupWindowSizes int promExporter *PrometheusExporter + evaluationMode evaluation_mode.EvaluationMode stopCleanupC chan struct{} stopClockC chan struct{} sync.RWMutex @@ -38,7 +57,7 @@ type Database struct { 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} } }() diff --git a/segments/analysis/toptalkers_metrics/toptalker_prometheus.go b/segments/analysis/toptalkers_metrics/toptalker_prometheus.go index aa047be3..57597f22 100644 --- a/segments/analysis/toptalkers_metrics/toptalker_prometheus.go +++ b/segments/analysis/toptalkers_metrics/toptalker_prometheus.go @@ -7,6 +7,7 @@ import ( "strconv" "github.com/BelWue/flowpipeline/pipeline/config" + "github.com/BelWue/flowpipeline/pipeline/config/evaluation_mode" "github.com/rs/zerolog/log" "github.com/prometheus/client_golang/prometheus" @@ -66,8 +67,8 @@ func (params *PrometheusMetricsParams) InitDefaultPrometheusMetricParams() { if params.BucketDuration == 0 { params.BucketDuration = 1 } - if params.RelevantAddress == "" { - params.RelevantAddress = "destination" + if params.EvaluationMode == evaluation_mode.Unknown { + params.EvaluationMode = evaluation_mode.Destination } if params.CleanupWindowSizes == 0 { params.CleanupWindowSizes = 5 @@ -150,17 +151,25 @@ func (prometheusParams *PrometheusMetricsParams) ParsePrometheusConfig(config ma log.Info().Msg("ToptalkersMetrics: 'thresholdpps' set to default '0'.") } - switch config["relevantaddress"] { - case - "destination", - "source", - "both", - "connection": - prometheusParams.RelevantAddress = config["relevantaddress"] - case "": - log.Info().Msg("ToptalkersMetrics: 'relevantaddress' set to default 'destination'.") - default: - log.Error().Msg("ToptalkersMetrics: Could not parse 'relevantaddress', using default value 'destination'.") + if config["evaluationmode"] == "" && config["relevantaddress"] != "" { + log.Warn().Msg("ToptalkersMetrics: Using deprecated parameter 'relevantaddress' - please use evaluationmode instead") + config["evaluationmode"] = config["relevantaddress"] + } + + if config["evaluationmode"] == "" { + log.Info().Msg("ToptalkersMetrics: 'evaluationmode' set to default 'destination'.") + } else { + if config["evaluationmode"] == "both" { + log.Warn().Msg("ToptalkersMetrics: using depected evaluation mode 'both' - please use 'Source and Destination' instead") + } + evaluationMode := evaluation_mode.ParseEvaluationMode(config["evaluationmode"]) + + if evaluationMode == evaluation_mode.Unknown { + log.Error().Msg("ToptalkersMetrics: Could not parse 'evaluationmode', using default value 'destination'.") + evaluationMode = evaluation_mode.Destination + } + prometheusParams.EvaluationMode = evaluationMode + } return nil } @@ -176,7 +185,7 @@ func (collector *PrometheusCollector) Collect(ch chan<- prometheus.Metric) { // check if thresholds are exceeded buckets := db.ReportBuckets bucketDuration := db.BucketDuration - if record.aboveThreshold.Load() { + if record.AboveThreshold().Load() { sumFwdBps, sumFwdPps, sumDropBps, sumDropPps, address := record.GetMetrics(buckets, bucketDuration) ch <- prometheus.MustNewConstMetric( collector.trafficBpsDesc, diff --git a/segments/analysis/toptalkers_metrics/toptalkers_metrics.go b/segments/analysis/toptalkers_metrics/toptalkers_metrics.go index 4d42653b..cc4f1c54 100644 --- a/segments/analysis/toptalkers_metrics/toptalkers_metrics.go +++ b/segments/analysis/toptalkers_metrics/toptalkers_metrics.go @@ -27,6 +27,7 @@ package toptalkers_metrics import ( "sync" + "github.com/BelWue/flowpipeline/pipeline/config/evaluation_mode" "github.com/BelWue/flowpipeline/segments" "github.com/rs/zerolog/log" ) @@ -75,8 +76,8 @@ func (segment *ToptalkersMetrics) Run(wg *sync.WaitGroup) { var promExporter = PrometheusExporter{} database := NewDatabase(segment.PrometheusMetricsParams, &promExporter) - collector := NewPrometheusCollector([]*Database{&database}) promExporter.Initialize() + collector := NewPrometheusCollector([]*Database{&database}) promExporter.FlowReg.MustRegister(collector) promExporter.ServeEndpoints(&segment.PrometheusParams) @@ -86,19 +87,19 @@ func (segment *ToptalkersMetrics) Run(wg *sync.WaitGroup) { for msg := range segment.In { promExporter.KafkaMessageCount.Inc() var keys []string - switch segment.RelevantAddress { - case "source": + switch segment.EvaluationMode { + case evaluation_mode.Source: keys = []string{msg.SrcAddrObj().String()} - case "destination": + case evaluation_mode.Destination: keys = []string{msg.DstAddrObj().String()} - case "both": + case evaluation_mode.SourceAndDestination: keys = []string{msg.SrcAddrObj().String(), msg.DstAddrObj().String()} } forward := false for _, key := range keys { - record := database.GetRecord(key) - record.Append(msg.Bytes, msg.Packets, msg.IsForwarded()) - if record.aboveThreshold.Load() { + record := database.GetRecord(key, msg.SrcAddrObj().String(), msg.DstAddrObj().String()) + record.Append(msg) + if record.AboveThreshold().Load() { forward = true } } diff --git a/segments/analysis/traffic_specific_toptalkers/traffic-specific-toptalkers.go b/segments/analysis/traffic_specific_toptalkers/traffic-specific-toptalkers.go index 190f9607..16012818 100644 --- a/segments/analysis/traffic_specific_toptalkers/traffic-specific-toptalkers.go +++ b/segments/analysis/traffic_specific_toptalkers/traffic-specific-toptalkers.go @@ -1,8 +1,17 @@ -// The traffic specific toptalker metrics segement is simmilar to the `toptalker-metrics` segment, +// The `traffic_specific_toptalkers` segement is simmilar to the `toptalker-metrics` segment, // but allows filtering for specific protocols. The use of nested filters is supported to // allow for a more efficient filtering. // // Filters with a specified `traffictyp` will be exported if they reach the configured thresholds. +// The traffic specific toptalker metrics segement is simmilar to the `toptalker-metrics` segment, +// but allows filtering for specific protocols. The use of nested filters is supported to +// allow for a more efficient filtering. + +// Filters with a specified `traffictyp` will be exported if they reach the configured thresholds. +// The segment allows forwarding all traffic to a matched ip to a subpipeline. +// See the [example configuration](https://github.com/BelWue/flowpipeline/tree/master/examples/configuration/analysis/traffic-specific-toptalker.yml) +// for an example using that functionality + package traffic_specific_toptalkers import ( @@ -13,7 +22,9 @@ import ( "github.com/BelWue/flowfilter/parser" "github.com/BelWue/flowpipeline/pb" + "github.com/BelWue/flowpipeline/pipeline" "github.com/BelWue/flowpipeline/pipeline/config" + "github.com/BelWue/flowpipeline/pipeline/config/evaluation_mode" "github.com/BelWue/flowpipeline/segments" "github.com/BelWue/flowpipeline/segments/analysis/toptalkers_metrics" "github.com/BelWue/flowpipeline/segments/filter/flowfilter" @@ -23,7 +34,8 @@ type TrafficSpecificToptalkers struct { segments.BaseSegment toptalkers_metrics.PrometheusParams ThresholdMetricDefinition []*ThresholdMetric - RelevantAddress string // optional, default is "destination", options are "destination", "source", "both", "connection" + EvaluationMode evaluation_mode.EvaluationMode // optional, default is "destination", options are "destination", "source", "both", "connection" + MatchingPipeline *pipeline.Pipeline } type ThresholdMetric struct { @@ -54,10 +66,26 @@ func (segment TrafficSpecificToptalkers) New(config map[string]string) segments. } else { newSegment.FlowdataPath = config["flowdatapath"] } - if config["relevantaddress"] != "" { - newSegment.RelevantAddress = config["relevantaddress"] + if config["evaluationmode"] == "" && config["relevantaddress"] != "" { + log.Warn().Msg("ThresholdToptalkersMetrics: Using deprecated parameter 'relevantaddress' - please use evaluationmode instead") + config["evaluationmode"] = config["relevantaddress"] + } + + if config["evaluationmode"] == "" { + log.Info().Msg("ThresholdToptalkersMetrics: 'evaluationmode' set to default 'destination'.") + newSegment.EvaluationMode = evaluation_mode.Destination } else { - newSegment.RelevantAddress = "" + if config["evaluationmode"] == "both" { + log.Warn().Msg("ThresholdToptalkersMetrics: using depected evaluation mode 'both' - please use 'Source and Destination' instead") + } + evaluationMode := evaluation_mode.ParseEvaluationMode(config["evaluationmode"]) + + if evaluationMode == evaluation_mode.Unknown { + log.Error().Msg("ThresholdToptalkersMetrics: Could not parse 'evaluationmode', using default value 'destination'.") + evaluationMode = evaluation_mode.Destination + } + newSegment.EvaluationMode = evaluationMode + } return newSegment @@ -71,17 +99,21 @@ func (segment *TrafficSpecificToptalkers) AddCustomConfig(segmentReprs config.Se } segment.ThresholdMetricDefinition = append(segment.ThresholdMetricDefinition, metric) } + if segmentReprs.MatchingPipeline != nil { + segments := pipeline.SegmentsFromRepr(segmentReprs.MatchingPipeline) + segment.MatchingPipeline = pipeline.New(segments...) + } } -func (segment *TrafficSpecificToptalkers) metricFromDefinition(definition *config.ThresholdMetricDefinition) (*ThresholdMetric, error) { +func (segment *TrafficSpecificToptalkers) metricFromDefinition(definition *config.ThresholdMetricConfig) (*ThresholdMetric, error) { var err error metric := ThresholdMetric{} metric.PrometheusMetricsParamsDefinition = definition.PrometheusMetricsParamsDefinition metric.FilterDefinition = definition.FilterDefinition metric.InitDefaultPrometheusMetricParams() - if segment.RelevantAddress != "" { - metric.RelevantAddress = segment.RelevantAddress + if metric.EvaluationMode == evaluation_mode.Unknown { + metric.EvaluationMode = segment.EvaluationMode } metric.Expression, err = parser.Parse(definition.FilterDefinition) @@ -104,12 +136,19 @@ func (segment *TrafficSpecificToptalkers) metricFromDefinition(definition *confi func (segment *TrafficSpecificToptalkers) Run(wg *sync.WaitGroup) { var allDatabases *[]*toptalkers_metrics.Database defer func() { + if segment.MatchingPipeline != nil { + segment.MatchingPipeline.Close() + } close(segment.Out) for _, db := range *allDatabases { db.StopTimers() } wg.Done() }() + if segment.MatchingPipeline != nil { + segment.MatchingPipeline.AutoDrain() + segment.MatchingPipeline.Start() + } var promExporter = toptalkers_metrics.PrometheusExporter{} promExporter.Initialize() @@ -125,13 +164,72 @@ func (segment *TrafficSpecificToptalkers) Run(wg *sync.WaitGroup) { filter := &flowfilter.Filter{} log.Info().Msgf("Threshold Metric Report runing on %s", segment.Endpoint) - for msg := range segment.In { - promExporter.KafkaMessageCount.Inc() - for _, filterDef := range segment.ThresholdMetricDefinition { - addMessageToMatchingToptalkers(msg, filterDef, filter) + if segment.MatchingPipeline != nil { + for msg := range segment.In { + promExporter.KafkaMessageCount.Inc() + for _, filterDef := range segment.ThresholdMetricDefinition { + addMessageToMatchingToptalkers(msg, filterDef, filter) + } + if segment.MatchingPipeline != nil && segment.IpInToptalkers(msg, filter) { + segment.MatchingPipeline.In <- msg + } + segment.Out <- msg + } + } else { + for msg := range segment.In { + promExporter.KafkaMessageCount.Inc() + for _, filterDef := range segment.ThresholdMetricDefinition { + addMessageToMatchingToptalkers(msg, filterDef, filter) + } + segment.Out <- msg + } + } +} + +func (segment *TrafficSpecificToptalkers) IpInToptalkers(msg *pb.EnrichedFlow, filter *flowfilter.Filter) bool { + for _, metricDef := range segment.ThresholdMetricDefinition { + if segment.IpInToptalkersOfMetric(metricDef, msg, filter) { + return true + } + } + return false +} + +func (segment *TrafficSpecificToptalkers) IpInToptalkersOfMetric(metricDef *ThresholdMetric, msg *pb.EnrichedFlow, filter *flowfilter.Filter) bool { + if metricDef.PrometheusMetricsParams.TrafficType != "" { + source := msg.SrcAddrObj().String() + destination := msg.DstAddrObj().String() + + var keys []string + switch metricDef.PrometheusMetricsParams.RelevantAddress { + case evaluation_mode.Source: + keys = []string{source} + case evaluation_mode.Destination: + keys = []string{destination} + case evaluation_mode.SourceAndDestination: + keys = []string{source, destination} + case evaluation_mode.Connection: + keys = []string{fmt.Sprintf("%s -> %s", source, destination)} + case evaluation_mode.Unknown: + //default = Destination + keys = []string{destination} + } + + for _, key := range keys { + record := metricDef.Database.GetTypedRecord(metricDef.PrometheusMetricsParams.TrafficType, key, source, destination) + if record.AboveThreshold().Load() { + return true + } + } + } + + //also check subfilters + for _, subdefinition := range metricDef.SubDefinitions { + if segment.IpInToptalkersOfMetric(subdefinition, msg, filter) { + return true } - segment.Out <- msg } + return false } func initDatabasesAndCollector(promExporter toptalkers_metrics.PrometheusExporter, segment *TrafficSpecificToptalkers) *[]*toptalkers_metrics.Database { @@ -165,19 +263,19 @@ func addMessageToMatchingToptalkers(msg *pb.EnrichedFlow, definition *ThresholdM // Update Counters if definition has a prometheus label defined if definition.PrometheusMetricsParams.TrafficType != "" { var keys []string - switch definition.PrometheusMetricsParams.RelevantAddress { - case "source": + switch definition.PrometheusMetricsParams.EvaluationMode { + case evaluation_mode.Source: keys = []string{msg.SrcAddrObj().String()} - case "destination": + case evaluation_mode.Destination: keys = []string{msg.DstAddrObj().String()} - case "both": + case evaluation_mode.SourceAndDestination: keys = []string{msg.SrcAddrObj().String(), msg.DstAddrObj().String()} - case "connection": + case evaluation_mode.Connection: keys = []string{fmt.Sprintf("%s -> %s", msg.SrcAddrObj().String(), msg.DstAddrObj().String())} } for _, key := range keys { - record := definition.Database.GetTypedRecord(definition.PrometheusMetricsParams.TrafficType, key) - record.Append(msg.Bytes, msg.Packets, msg.IsForwarded()) + record := definition.Database.GetTypedRecord(definition.PrometheusMetricsParams.TrafficType, key, msg.SrcAddrObj().String(), msg.DstAddrObj().String()) + record.Append(msg) } } diff --git a/segments/analysis/traffic_specific_toptalkers/traffic-specific-toptalkers_test.go b/segments/analysis/traffic_specific_toptalkers/traffic-specific-toptalkers_test.go index 132f997a..5ff677a9 100644 --- a/segments/analysis/traffic_specific_toptalkers/traffic-specific-toptalkers_test.go +++ b/segments/analysis/traffic_specific_toptalkers/traffic-specific-toptalkers_test.go @@ -3,15 +3,18 @@ package traffic_specific_toptalkers import ( "sync" "testing" + "time" "github.com/rs/zerolog/log" "github.com/BelWue/flowpipeline/pb" + "github.com/BelWue/flowpipeline/pipeline" "github.com/BelWue/flowpipeline/pipeline/config" "github.com/BelWue/flowpipeline/segments" + "github.com/BelWue/flowpipeline/segments/testing/counter" ) -func TestSegment_Branch_passthrough(t *testing.T) { +func TestSegment_TrafficSpecificToptalkers_passthrough(t *testing.T) { msg := &pb.EnrichedFlow{SrcAddr: []byte{192, 168, 88, 142}, DstAddr: []byte{192, 168, 88, 123}, DstPort: 123, Packets: 1000, Bytes: 230000, Proto: 17} //Ntp (udp) msg2 := &pb.EnrichedFlow{SrcAddr: []byte{192, 168, 88, 142}, DstAddr: []byte{192, 168, 88, 123}, DstPort: 443, Packets: 1, Bytes: 100, Proto: 6} @@ -19,10 +22,10 @@ func TestSegment_Branch_passthrough(t *testing.T) { //normally done via config segment.AddCustomConfig(config.SegmentRepr{ Config: config.Config{ - ThresholdMetricDefinition: []*config.ThresholdMetricDefinition{ + ThresholdMetricDefinition: []*config.ThresholdMetricConfig{ { FilterDefinition: "proto udp", - SubDefinitions: []*config.ThresholdMetricDefinition{ + SubDefinitions: []*config.ThresholdMetricConfig{ { FilterDefinition: "port 123", PrometheusMetricsParamsDefinition: config.PrometheusMetricsParamsDefinition{ @@ -64,3 +67,82 @@ func TestSegment_Branch_passthrough(t *testing.T) { t.Error("Segment traffic_specific_toptalkers is not passing through flows.") } } + +func TestSegment_TrafficSpecificToptalkers_matching_passthrough(t *testing.T) { + pipeline := pipeline.NewFromConfig([]byte(`--- +- segment: traffic_specific_toptalkers + config: + endpoint: ":8085" + traffic_specific_toptalkers: + - filter: "proto udp" + traffictype: "UDP" + matching_pipeline: + - segment: counter +`)) + + var ( + counterSegment *counter.Counter + ) + + //Test if pipeline is setup correctly + segment0 := pipeline.SegmentList[0] + switch segment := segment0.(type) { + case *TrafficSpecificToptalkers: + if segment.MatchingPipeline == nil { + t.Error("[error] subpipeline not initialized correctly") + t.Fail() + } else { + subsegment0 := segment.MatchingPipeline.SegmentList[0] + switch subsegment := subsegment0.(type) { + case *counter.Counter: + counterSegment = subsegment + default: + t.Error("[error] subpipeline not initialized correctly") + t.Fail() + } + } + default: + t.Error("[error] pipeline not initialized correctly") + t.Fail() + } + + if counterSegment.Counter != 0 { + t.Error("counter not initialized correctly") + t.Fail() + } + pipeline.AutoDrain() + pipeline.Start() + + //tcp - matched by the filter the filter + pipeline.In <- &pb.EnrichedFlow{Proto: 17, InIf: 1, OutIf: 1, DstAddr: []byte{192, 168, 88, 142}, Packets: 15000, Bytes: 3000000} + time.Sleep(2 * time.Second) // takes 1 db tick to get above threashold + + //tcp - not in filter + pipeline.In <- &pb.EnrichedFlow{Proto: 6, InIf: 1, OutIf: 1, DstAddr: []byte{192, 168, 88, 143}, Packets: 15000, Bytes: 3000000} + time.Sleep(2 * time.Second) + if counterSegment.Counter != 0 { + t.Error("[error] tcp packet was counted from upd matcher") + } + + //upd in Filter + pipeline.In <- &pb.EnrichedFlow{Proto: 17, InIf: 1, OutIf: 1, DstAddr: []byte{192, 168, 88, 142}, Packets: 15000, Bytes: 3000000} + time.Sleep(2 * time.Second) + if counterSegment.Counter != 1 { + t.Error("[error] udp packet was not counted from upd matcher") + } + + //tcp - still not in filter + pipeline.In <- &pb.EnrichedFlow{Proto: 6, InIf: 1, OutIf: 1, DstAddr: []byte{192, 168, 88, 143}, Packets: 15000, Bytes: 3000000} + time.Sleep(2 * time.Second) + if counterSegment.Counter != 1 { + t.Error("[error] tcp packet was counted from upd matcher") + } + + //tcp, but should now be counted since the address was matched previously + pipeline.In <- &pb.EnrichedFlow{Proto: 6, InIf: 1, OutIf: 1, DstAddr: []byte{192, 168, 88, 142}, Packets: 15000, Bytes: 3000000} + time.Sleep(2 * time.Second) + if counterSegment.Counter != 2 { + t.Error("[error] tcp packet was not counted from upd matcher") + } + pipeline.Close() +} diff --git a/segments/testing/counter/counter.go b/segments/testing/counter/counter.go new file mode 100644 index 00000000..dee61a29 --- /dev/null +++ b/segments/testing/counter/counter.go @@ -0,0 +1,35 @@ +package counter + +import ( + "sync" + + "github.com/BelWue/flowpipeline/segments" +) + +type Counter struct { + segments.BaseSegment + Counter int +} + +func (segment Counter) New(config map[string]string) segments.Segment { + return &Counter{ + Counter: 0, + } +} + +func (segment *Counter) Run(wg *sync.WaitGroup) { + defer func() { + close(segment.Out) + wg.Done() + }() + + for msg := range segment.In { + segment.Counter += 1 + segment.Out <- msg + } +} + +func init() { + segment := &Counter{} + segments.RegisterSegment("counter", segment) +}