add processed bytes/metrics for nftables

cmd/root.go

@@ -228,6 +228,34 @@ func metricsUpdater(met *models.RemediationComponentsMetrics) {
 					Unit: ptr.Of("packet"),
 				})
 				metrics.LastDroppedPacketsValue[key] = value
+			case metrics.ProcessedBytesMetricName:
+				labels := metric.GetLabel()
+				value := metric.GetGauge().GetValue()
+				ipType := getLabelValue(labels, "ip_type")
+				log.Debugf("Sending processed bytes for %s %f | current value: %f | previous value: %f\n", ipType, value-metrics.LastProcessedBytesValue[ipType], value, metrics.LastProcessedBytesValue[ipType])
+				met.Metrics[0].Items = append(met.Metrics[0].Items, &models.MetricsDetailItem{
+					Name:  ptr.Of("processed"),
+					Value: ptr.Of(value - metrics.LastProcessedBytesValue[ipType]),
+					Labels: map[string]string{
+						"ip_type": ipType,
+					},
+					Unit: ptr.Of("byte"),
+				})
+				metrics.LastProcessedBytesValue[ipType] = value
+			case metrics.ProcessedPacketsMetricName:
+				labels := metric.GetLabel()
+				value := metric.GetGauge().GetValue()
+				ipType := getLabelValue(labels, "ip_type")
+				log.Debugf("Sending processed packets for %s %f | current value: %f | previous value: %f\n", ipType, value-metrics.LastProcessedPacketsValue[ipType], value, metrics.LastProcessedPacketsValue[ipType])
+				met.Metrics[0].Items = append(met.Metrics[0].Items, &models.MetricsDetailItem{
+					Name:  ptr.Of("processed"),
+					Value: ptr.Of(value - metrics.LastProcessedPacketsValue[ipType]),
+					Labels: map[string]string{
+						"ip_type": ipType,
+					},
+					Unit: ptr.Of("packet"),
+				})
+				metrics.LastProcessedPacketsValue[ipType] = value
 			}
 		}
 	}
@@ -330,7 +358,7 @@ func Execute() error {
 		if config.Mode == cfg.IpsetMode {
 			prometheus.MustRegister(metrics.TotalActiveBannedIPs)
 		} else {
-			prometheus.MustRegister(metrics.TotalDroppedBytes, metrics.TotalDroppedPackets, metrics.TotalActiveBannedIPs)
+			prometheus.MustRegister(metrics.TotalDroppedBytes, metrics.TotalDroppedPackets, metrics.TotalActiveBannedIPs, metrics.TotalProcessedBytes, metrics.TotalProcessedPackets)
 		}
 	}
 

pkg/metrics/metrics.go

@@ -9,9 +9,11 @@ import (
 const MetricCollectionInterval = time.Second * 10
 
 const (
-	DroppedPacketsMetricName  = "fw_bouncer_dropped_packets"
-	DroppedBytesMetricName    = "fw_bouncer_dropped_bytes"
-	ActiveBannedIPsMetricName = "fw_bouncer_banned_ips"
+	DroppedPacketsMetricName   = "fw_bouncer_dropped_packets"
+	DroppedBytesMetricName     = "fw_bouncer_dropped_bytes"
+	ProcessedPacketsMetricName = "fw_bouncer_processed_packets"
+	ProcessedBytesMetricName   = "fw_bouncer_processed_bytes"
+	ActiveBannedIPsMetricName  = "fw_bouncer_banned_ips"
 )
 
 var TotalDroppedPackets = prometheus.NewGaugeVec(prometheus.GaugeOpts{
@@ -31,3 +33,15 @@ var TotalActiveBannedIPs = prometheus.NewGaugeVec(prometheus.GaugeOpts{
 	Help: "Denotes the number of IPs which are currently banned",
 }, []string{"origin", "ip_type"})
 var LastActiveBannedIPsValue map[string]float64 = make(map[string]float64)
+
+var TotalProcessedPackets = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+	Name: ProcessedPacketsMetricName,
+	Help: "Denotes the number of total processed packets by the rules created by crowdsec",
+}, []string{"ip_type"})
+var LastProcessedPacketsValue map[string]float64 = make(map[string]float64)
+
+var TotalProcessedBytes = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+	Name: ProcessedBytesMetricName,
+	Help: "Denotes the number of total processed bytes by the rules created by crowdsec",
+}, []string{"ip_type"})
+var LastProcessedBytesValue map[string]float64 = make(map[string]float64)

pkg/nftables/metrics.go

@@ -15,39 +15,40 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-func (c *nftContext) collectDroppedPackets() (map[string]int, map[string]int, error) {
+func (c *nftContext) collectDroppedPackets() (map[string]int, map[string]int, int, int, error) {
 	droppedPackets := make(map[string]int)
 	droppedBytes := make(map[string]int)
+	processedPackets := 0
+	processedBytes := 0
 	//setName := ""
 	for chainName, chain := range c.chains {
 		rules, err := c.conn.GetRules(c.table, chain)
 		if err != nil {
-			log.Errorf("can't get rules for ip4 chain %s: %s", chainName, err)
+			log.Errorf("can't get rules for chain %s: %s", chainName, err)
 			continue
 		}
 		for _, rule := range rules {
-			origin := ""
-			pkts := 0
-			bytes := 0
 			for _, xpr := range rule.Exprs {
 				switch obj := xpr.(type) {
 				case *expr.Counter:
-					log.Debugf("rule %d (%s): packets %d, bytes %d", rule.Position, rule.Table.Name, obj.Packets, obj.Bytes)
-					pkts += int(obj.Packets)
-					bytes += int(obj.Bytes)
-				case *expr.Lookup:
-					log.Debugf("rule %d (%s): lookup %s", rule.Position, rule.Table.Name, obj.SetName)
-					origin, _ = strings.CutPrefix(obj.SetName, fmt.Sprintf("%s-", c.blacklists))
+					log.Debugf("rule %d (%s): packets %d, bytes %d (%s)", rule.Position, rule.Table.Name, obj.Packets, obj.Bytes, rule.UserData)
+					if string(rule.UserData) == "processed" {
+						processedPackets += int(obj.Packets)
+						processedBytes += int(obj.Bytes)
+						continue
+					}
+					origin, _ := strings.CutPrefix(string(rule.UserData), c.blacklists+"-")
+					if origin == "" {
+						continue
+					}
+					droppedPackets[origin] += int(obj.Packets)
+					droppedBytes[origin] += int(obj.Bytes)
 				}
 			}
-			if origin != "" {
-				droppedPackets[origin] += pkts
-				droppedBytes[origin] += bytes
-			}
 		}
 	}
 
-	return droppedPackets, droppedBytes, nil
+	return droppedPackets, droppedBytes, processedPackets, processedBytes, nil
 }
 
 func (c *nftContext) collectActiveBannedIPs() (map[string]int, error) {
@@ -70,12 +71,12 @@ func (c *nftContext) collectActiveBannedIPs() (map[string]int, error) {
 	return ret, nil
 }
 
-func (c *nftContext) collectDropped() (map[string]int, map[string]int, map[string]int) {
+func (c *nftContext) collectDropped() (map[string]int, map[string]int, int, int, map[string]int) {
 	if c.conn == nil {
-		return nil, nil, nil
+		return nil, nil, 0, 0, nil
 	}
 
-	droppedPackets, droppedBytes, err := c.collectDroppedPackets()
+	droppedPackets, droppedBytes, processedPackets, processedBytes, err := c.collectDroppedPackets()
 
 	if err != nil {
 		log.Errorf("can't collect dropped packets for ip%s from nft: %s", c.version, err)
@@ -86,7 +87,7 @@ func (c *nftContext) collectDropped() (map[string]int, map[string]int, map[strin
 		log.Errorf("can't collect total banned IPs for ip%s from nft: %s", c.version, err)
 	}
 
-	return droppedPackets, droppedBytes, banned
+	return droppedPackets, droppedBytes, processedPackets, processedBytes, banned
 }
 
 func getOriginForList(origin string) string {
@@ -103,12 +104,18 @@ func (n *nft) CollectMetrics() {
 
 	for range t.C {
 		startTime := time.Now()
-		ip4DroppedPackets, ip4DroppedBytes, bannedIP4 := n.v4.collectDropped()
-		ip6DroppedPackets, ip6DroppedBytes, bannedIP6 := n.v6.collectDropped()
+		ip4DroppedPackets, ip4DroppedBytes, ip4ProcessedPackets, ip4ProcessedBytes, bannedIP4 := n.v4.collectDropped()
+		ip6DroppedPackets, ip6DroppedBytes, ip6ProcessedPackets, ip6ProcessedBytes, bannedIP6 := n.v6.collectDropped()
 
 		log.Debugf("metrics collection took %s", time.Since(startTime))
-		log.Debugf("ip4: dropped packets: %+v, dropped bytes: %+v, banned IPs: %+v", ip4DroppedPackets, ip4DroppedBytes, bannedIP4)
-		log.Debugf("ip6: dropped packets: %+v, dropped bytes: %+v, banned IPs: %+v", ip6DroppedPackets, ip6DroppedBytes, bannedIP6)
+		log.Debugf("ip4: dropped packets: %+v, dropped bytes: %+v, banned IPs: %+v, proccessed packets: %d, processed bytes: %d", ip4DroppedPackets, ip4DroppedBytes, bannedIP4, ip4ProcessedPackets, ip4ProcessedBytes)
+		log.Debugf("ip6: dropped packets: %+v, dropped bytes: %+v, banned IPs: %+v, proccessed packets: %d, processed bytes: %d", ip6DroppedPackets, ip6DroppedBytes, bannedIP6, ip6ProcessedPackets, ip6ProcessedBytes)
+
+		metrics.TotalProcessedPackets.With(prometheus.Labels{"ip_type": "ipv4"}).Set(float64(ip4ProcessedPackets))
+		metrics.TotalProcessedBytes.With(prometheus.Labels{"ip_type": "ipv4"}).Set(float64(ip4ProcessedBytes))
+
+		metrics.TotalProcessedPackets.With(prometheus.Labels{"ip_type": "ipv6"}).Set(float64(ip6ProcessedPackets))
+		metrics.TotalProcessedBytes.With(prometheus.Labels{"ip_type": "ipv6"}).Set(float64(ip6ProcessedBytes))
 
 		for origin, count := range bannedIP4 {
 			origin = getOriginForList(origin)

pkg/nftables/nftables_context.go

@@ -188,6 +188,17 @@ func (c *nftContext) initOwnTable(hooks []string) error {
 			Priority: &priority,
 		})
 
+		r := &nftables.Rule{
+			Table: c.table,
+			Chain: chain,
+			Exprs: []expr.Any{
+				&expr.Counter{},
+			},
+			UserData: []byte("processed"),
+		}
+
+		c.conn.AddRule(r)
+
 		c.chains[hook] = chain
 
 		log.Debugf("nftables: ip%s chain '%s' created", c.version, chain.Name)
@@ -250,9 +261,10 @@ func (c *nftContext) createRule(chain *nftables.Chain, set *nftables.Set,
 	denyLog bool, denyLogPrefix string, denyAction string,
 ) (*nftables.Rule, error) {
 	r := &nftables.Rule{
-		Table: c.table,
-		Chain: chain,
-		Exprs: []expr.Any{},
+		Table:    c.table,
+		Chain:    chain,
+		Exprs:    []expr.Any{},
+		UserData: []byte(set.Name),
 	}
 	// [ payload load 4b @ network header + 16 => reg 1 ]
 	r.Exprs = append(r.Exprs, &expr.Payload{