feat(conntrack-metrics): move implementation to conntrack gc, remove …

…labels

pkg/metrics/metrics.go

@@ -162,28 +162,24 @@ func InitializeMetrics() {
 		exporter.DefaultRegistry,
 		utils.ConntrackPacketsForwardGaugeName,
 		ConntrackPacketForwardDescription,
-		utils.ConntrackGaugeLabels...,
 	)
 
 	ConntrackPacketsReply = exporter.CreatePrometheusGaugeVecForMetric(
 		exporter.DefaultRegistry,
 		utils.ConntrackPacketsReplyGaugeName,
 		ConntrackPacketReplyDescription,
-		utils.ConntrackGaugeLabels...,
 	)
 
 	ConntrackBytesForward = exporter.CreatePrometheusGaugeVecForMetric(
 		exporter.DefaultRegistry,
 		utils.ConntrackBytesForwardGaugeName,
 		ConntrackBytesForwardDescription,
-		utils.ConntrackGaugeLabels...,
 	)
 
 	ConntrackBytesReply = exporter.CreatePrometheusGaugeVecForMetric(
 		exporter.DefaultRegistry,
 		utils.ConntrackBytesReplyGaugeName,
 		ConntrackBytesReplyDescription,
-		utils.ConntrackGaugeLabels...,
 	)
 
 	isInitialized = true

pkg/plugin/conntrack/conntrack_linux.go

@@ -16,13 +16,16 @@ import (
 	"github.com/microsoft/retina/internal/ktime"
 	"github.com/microsoft/retina/pkg/loader"
 	"github.com/microsoft/retina/pkg/log"
+	"github.com/microsoft/retina/pkg/metrics"
 	plugincommon "github.com/microsoft/retina/pkg/plugin/common"
 	_ "github.com/microsoft/retina/pkg/plugin/conntrack/_cprog" // nolint // This is needed so cprog is included when vendoring
 	"github.com/microsoft/retina/pkg/utils"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
 )
 
+var conntrackMetricsEnabled = false // global variable to enable conntrack metrics
+
 //go:generate go run github.com/cilium/ebpf/cmd/bpf2go@master -cflags "-g -O2 -Wall -D__TARGET_ARCH_${GOARCH} -Wall" -target ${GOARCH} -type ct_v4_key conntrack ./_cprog/conntrack.c -- -I../lib/_${GOARCH} -I../lib/common/libbpf/_src -I../lib/common/libbpf/_include/linux -I../lib/common/libbpf/_include/uapi/linux -I../lib/common/libbpf/_include/asm
 
 // Init initializes the conntrack eBPF map in the kernel for the first time.
@@ -88,6 +91,10 @@ func GenerateDynamic(ctx context.Context, dynamicHeaderPath string, conntrackMet
 	if err != nil {
 		return errors.Wrap(err, "failed to write conntrack dynamic header")
 	}
+	// set a global variable to enable conntrack metrics
+	if conntrackMetrics == 1 {
+		conntrackMetricsEnabled = true
+	}
 	return nil
 }
 
@@ -118,6 +125,10 @@ func (ct *Conntrack) Run(ctx context.Context) error {
 			// List of keys to be deleted
 			var keysToDelete []conntrackCtV4Key
 
+			// metrics counters
+			var packetsCountForward, packetsCountReply uint32
+			var bytesCountForward, bytesCountReply uint64
+
 			iter := ct.ctMap.Iterate()
 			for iter.Next(&key, &value) {
 				noOfCtEntries++
@@ -133,6 +144,18 @@ func (ct *Conntrack) Run(ctx context.Context) error {
 				dstIP := utils.Int2ip(key.DstIp).To4()
 				sourcePortShort := uint32(utils.HostToNetShort(key.SrcPort))
 				destinationPortShort := uint32(utils.HostToNetShort(key.DstPort))
+
+				// Add conntrack metrics.
+				if conntrackMetricsEnabled {
+					// Basic metrics, node-level
+					// for each ct_entry increment counters
+					ctMeta := value.ConntrackMetadata
+					packetsCountForward += ctMeta.PacketsForwardCount
+					packetsCountReply += ctMeta.PacketsReplyCount
+					bytesCountForward += ctMeta.BytesForwardCount
+					bytesCountReply += ctMeta.BytesReplyCount
+				}
+
 				ct.l.Debug("conntrack entry",
 					zap.String("src_ip", srcIP.String()),
 					zap.Uint32("src_port", sourcePortShort),
@@ -151,6 +174,15 @@ func (ct *Conntrack) Run(ctx context.Context) error {
 			if err := iter.Err(); err != nil {
 				ct.l.Error("Iterate failed", zap.Error(err))
 			}
+
+			// create metrics
+			if conntrackMetricsEnabled {
+				metrics.ConntrackPacketsForward.WithLabelValues().Set(float64(packetsCountForward))
+				metrics.ConntrackBytesForward.WithLabelValues().Set(float64(bytesCountForward))
+				metrics.ConntrackPacketsReply.WithLabelValues().Set(float64(packetsCountReply))
+				metrics.ConntrackBytesReply.WithLabelValues().Set(float64(bytesCountReply))
+			}
+
 			// Delete the conntrack entries
 			for _, key := range keysToDelete {
 				if err := ct.ctMap.Delete(key); err != nil {

pkg/plugin/packetparser/packetparser_linux.go

@@ -637,19 +637,6 @@ func (p *packetParser) processRecord(ctx context.Context, id int) {
 				p.enricher.Write(ev)
 			}
 
-			// Add conntrack metrics.
-			if p.cfg.EnableConntrackMetrics {
-				labels := []string{
-					protoToString(bpfEvent.Proto),
-					fl.GetTrafficDirection().String(),
-				}
-				// Basic metrics, node-level
-				metrics.ConntrackPacketsForward.WithLabelValues(labels...).Set(float64(bpfEvent.ConntrackMetadata.PacketsForwardCount))
-				metrics.ConntrackBytesForward.WithLabelValues(labels...).Set(float64(bpfEvent.ConntrackMetadata.BytesForwardCount))
-				metrics.ConntrackPacketsReply.WithLabelValues(labels...).Set(float64(bpfEvent.ConntrackMetadata.PacketsReplyCount))
-				metrics.ConntrackBytesReply.WithLabelValues(labels...).Set(float64(bpfEvent.ConntrackMetadata.BytesReplyCount))
-			}
-
 			// Write the event to the external channel.
 			if p.externalChannel != nil {
 				select {
@@ -717,13 +704,3 @@ func absPath() (string, error) {
 	dir := path.Dir(filename)
 	return dir, nil
 }
-
-func protoToString(bpfEventProto uint8) string {
-	var proto string
-	if bpfEventProto == 6 {
-		proto = "tcp"
-	} else if bpfEventProto == 17 {
-		proto = "udp"
-	}
-	return proto
-}

pkg/utils/attr_utils.go

@@ -85,9 +85,6 @@ var (
 	// DNS labels.
 	DNSRequestLabels  = []string{"query_type", "query"}
 	DNSResponseLabels = []string{"return_code", "query_type", "query", "response", "num_response"}
-
-	// Flow labels.
-	ConntrackGaugeLabels = []string{"protocol", "traffic_direction"}
 )
 
 func GetPluginEventAttributes(attrs []attribute.KeyValue, pluginName, eventName, timestamp string) []attribute.KeyValue {