feat(standalone): Create new metrics module that processes flows and update HNS plugin to emit relevant metrics

pkg/config/config.go

@@ -68,6 +68,8 @@ type Config struct {
 	EnableTelemetry          bool          `yaml:"enableTelemetry"`
 	EnableRetinaEndpoint     bool          `yaml:"enableRetinaEndpoint"`
 	EnablePodLevel           bool          `yaml:"enablePodLevel"`
+	EnableStandalone         bool          `yaml:"enableStandalone"`
+	EnableCrictl             bool          `yaml:"enableCrictl"`
 	EnableConntrackMetrics   bool          `yaml:"enableConntrackMetrics"`
 	RemoteContext            bool          `yaml:"remoteContext"`
 	EnableAnnotations        bool          `yaml:"enableAnnotations"`

pkg/metrics/metrics.go

@@ -42,8 +42,8 @@ func InitializeMetrics() {
 		utils.Direction)
 	HNSStatsGauge = exporter.CreatePrometheusGaugeVecForMetric(
 		exporter.DefaultRegistry,
-		hnsStats,
-		hnsStatsDescription,
+		HNSStats,
+		HNSStatsDescription,
 		utils.Direction,
 	)
 	NodeConnectivityStatusGauge = exporter.CreatePrometheusGaugeVecForMetric(
@@ -74,14 +74,14 @@ func InitializeMetrics() {
 	TCPConnectionStatsGauge = exporter.CreatePrometheusGaugeVecForMetric(
 		exporter.DefaultRegistry,
 		utils.TCPConnectionStatsName,
-		tcpConnectionStatsGaugeDescription,
+		TCPConnectionStatsGaugeDescription,
 		utils.StatName,
 	)
 
 	TCPFlagGauge = exporter.CreatePrometheusGaugeVecForMetric(
 		exporter.DefaultRegistry,
 		utils.TCPFlagGauge,
-		tcpFlagGaugeDescription,
+		TCPFlagGaugeDescription,
 		utils.Direction,
 		utils.Flag,
 	)

pkg/metrics/types.go

@@ -15,8 +15,8 @@ const (
 	parsedPacketsCounterName                  = "parsed_packets_counter"
 
 	// Windows
-	hnsStats            = "windows_hns_stats"
-	hnsStatsDescription = "Include many different metrics from packets sent/received to closed connections"
+	HNSStats            = "windows_hns_stats"
+	HNSStatsDescription = "Include many different metrics from packets sent/received to closed connections"
 
 	// Linux only metrics (for now).
 	nodeApiServerHandshakeLatencyHistName = "node_apiserver_handshake_latency_ms"
@@ -30,8 +30,8 @@ const (
 	nodeConnectivityLatencySecondsGaugeDescription = "The last observed latency between the current Cilium agent and other Cilium nodes in seconds"
 	tcpStateGaugeDescription                       = "Number of active TCP connections by state"
 	tcpConnectionRemoteGaugeDescription            = "Number of active TCP connections by remote address"
-	tcpConnectionStatsGaugeDescription             = "TCP connections statistics"
-	tcpFlagGaugeDescription                        = "TCP gauges by flag"
+	TCPConnectionStatsGaugeDescription             = "TCP connections statistics"
+	TCPFlagGaugeDescription                        = "TCP gauges by flag"
 	ipConnectionStatsGaugeDescription              = "IP connections statistics"
 	udpConnectionStatsGaugeDescription             = "UDP connections statistics"
 	interfaceStatsGaugeDescription                 = "Interface statistics"

pkg/module/metrics/basemetricsobject.go

@@ -49,5 +49,5 @@ func (b *baseMetricObject) populateCtxOptions(ctxOptions *api.MetricsContextOpti
 }
 
 func (b *baseMetricObject) isLocalContext() bool {
-	return b.contextMode == localContext
+	return b.contextMode == LocalContext
 }

pkg/module/metrics/dns.go

@@ -238,5 +238,6 @@ func (d *DNSMetrics) processLocalCtxFlow(flow *v1.Flow) {
 }
 
 func (d *DNSMetrics) Clean() {
+	d.l.Info("Cleaning metric", zap.String("name", d.metricName))
 	exporter.UnregisterMetric(exporter.AdvancedRegistry, metricsinit.ToPrometheusType(d.dnsMetrics))
 }

pkg/module/metrics/drops.go

@@ -25,11 +25,12 @@ const (
 
 type DropCountMetrics struct {
 	baseMetricObject
-	dropMetric metrics.GaugeVec
-	metricName string
+	dropMetric       metrics.GaugeVec
+	metricName       string
+	enableStandalone bool
 }
 
-func NewDropCountMetrics(ctxOptions *api.MetricsContextOptions, fl *log.ZapLogger, isLocalContext enrichmentContext) *DropCountMetrics {
+func NewDropCountMetrics(ctxOptions *api.MetricsContextOptions, fl *log.ZapLogger, isLocalContext enrichmentContext, enableStandalone bool) *DropCountMetrics {
 	if ctxOptions == nil || !strings.Contains(strings.ToLower(ctxOptions.MetricName), "drop") {
 		return nil
 	}
@@ -38,6 +39,7 @@ func NewDropCountMetrics(ctxOptions *api.MetricsContextOptions, fl *log.ZapLogge
 	fl.Info("Creating drop count metrics", zap.Any("options", ctxOptions))
 	return &DropCountMetrics{
 		baseMetricObject: newBaseMetricsObject(ctxOptions, fl, isLocalContext),
+		enableStandalone: enableStandalone,
 	}
 }
 
@@ -83,6 +85,7 @@ func (d *DropCountMetrics) getLabels() []string {
 }
 
 func (d *DropCountMetrics) Clean() {
+	d.l.Info("Cleaning metric", zap.String("name", d.metricName))
 	exporter.UnregisterMetric(exporter.AdvancedRegistry, metrics.ToPrometheusType(d.dropMetric))
 }
 
@@ -95,6 +98,11 @@ func (d *DropCountMetrics) ProcessFlow(flow *v1.Flow) {
 		return
 	}
 
+	if d.enableStandalone {
+		d.processStandaloneFlow(flow)
+		return
+	}
+
 	if flow.Verdict != v1.Verdict_DROPPED {
 		return
 	}
@@ -169,3 +177,36 @@ func (d *DropCountMetrics) update(fl *v1.Flow, labels []string) {
 		d.dropMetric.WithLabelValues(labels...).Add(float64(utils.PacketSize(fl)))
 	}
 }
+
+func (d *DropCountMetrics) processStandaloneFlow(fl *v1.Flow) {
+	// Ingress values
+	ingressLbls := []string{
+		ingress,
+		fl.GetIP().GetSource(),
+		fl.GetSource().GetNamespace(),
+		fl.GetSource().GetPodName(),
+		"",
+		"",
+	}
+	// Egress values
+	egressLbls := []string{
+		egress,
+		fl.GetIP().GetSource(),
+		fl.GetSource().GetNamespace(),
+		fl.GetSource().GetPodName(),
+		"",
+		"",
+	}
+
+	endpointStats := GetHNSMetadata(fl).GetEndpointStats()
+	d.dropMetric.WithLabelValues(append([]string{utils.Endpoint}, ingressLbls...)...).Set(float64(endpointStats.GetDroppedPacketsIncoming()))
+	d.dropMetric.WithLabelValues(append([]string{utils.Endpoint}, egressLbls...)...).Set(float64(endpointStats.GetDroppedPacketsOutgoing()))
+
+	if GetHNSMetadata(fl).GetVfpPortStatsData() == nil {
+		return
+	}
+
+	vfpStats := GetHNSMetadata(fl).GetVfpPortStatsData()
+	d.dropMetric.WithLabelValues(append([]string{utils.AclRule}, ingressLbls...)...).Set(float64(vfpStats.GetIn().GetDropCounters().GetAclDropPacketCount()))
+	d.dropMetric.WithLabelValues(append([]string{utils.AclRule}, egressLbls...)...).Set(float64(vfpStats.GetOut().GetDropCounters().GetAclDropPacketCount()))
+}

pkg/module/metrics/drops_test.go

@@ -5,12 +5,16 @@
 package metrics
 
 import (
+	"strings"
 	"testing"
 
 	"github.com/cilium/cilium/api/v1/flow"
 	"github.com/microsoft/retina/crd/api/v1alpha1"
+	api "github.com/microsoft/retina/crd/api/v1alpha1"
+	"github.com/microsoft/retina/pkg/exporter"
 	"github.com/microsoft/retina/pkg/log"
 	metricsinit "github.com/microsoft/retina/pkg/metrics"
+	"github.com/microsoft/retina/pkg/utils"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
 	"go.uber.org/mock/gomock"
@@ -209,7 +213,7 @@ func TestNewDrop(t *testing.T) {
 			},
 			metricCall:   1,
 			nilObj:       false,
-			localContext: localContext,
+			localContext: LocalContext,
 		},
 		{
 			name: "drop source opts with destination flow in localcontext",
@@ -235,7 +239,7 @@ func TestNewDrop(t *testing.T) {
 			},
 			metricCall:   1,
 			nilObj:       false,
-			localContext: localContext,
+			localContext: LocalContext,
 		},
 		{
 			name: "drop source opts with source and destination flow in localcontext",
@@ -262,15 +266,15 @@ func TestNewDrop(t *testing.T) {
 			},
 			metricCall:   2,
 			nilObj:       false,
-			localContext: localContext,
+			localContext: LocalContext,
 		},
 	}
 
 	for _, tc := range tt {
 		for _, metricName := range []string{"drop_count", "drop_bytes"} {
 			log.Logger().Info("Running test name", zap.String("name", tc.name), zap.String("metricName", metricName))
 			ctrl := gomock.NewController(t)
-			f := NewDropCountMetrics(tc.opts, log.Logger(), tc.localContext)
+			f := NewDropCountMetrics(tc.opts, log.Logger(), tc.localContext, false)
 			if tc.nilObj {
 				assert.Nil(t, f, "drop metrics should be nil Test Name: %s", tc.name)
 				continue
@@ -297,3 +301,89 @@ func TestNewDrop(t *testing.T) {
 		}
 	}
 }
+
+func TestStandaloneDropMetrics(t *testing.T) {
+	logger, err := log.SetupZapLogger(log.GetDefaultLogOpts())
+	assert.NoError(t, err)
+
+	ctxOptions := &api.MetricsContextOptions{
+		MetricName:   utils.DroppedPacketsGaugeName,
+		SourceLabels: append(DefaultCtxOptions(), utils.Reason, utils.Direction),
+	}
+
+	drop := NewDropCountMetrics(ctxOptions, logger, LocalContext, true)
+	drop.Init(ctxOptions.MetricName)
+
+	originalGetHNS := GetHNSMetadata
+	GetHNSMetadata = func(_ *flow.Flow) *utils.HNSStatsMetadata {
+		return &utils.HNSStatsMetadata{
+			EndpointStats: &utils.EndpointStats{
+				DroppedPacketsIncoming: 0,
+				DroppedPacketsOutgoing: 99,
+			},
+			VfpPortStatsData: &utils.VfpPortStatsData{
+				In: &utils.VfpDirectedPortCounters{
+					DropCounters: &utils.VfpPacketDropStats{
+						AclDropPacketCount: 100,
+					},
+				},
+				Out: &utils.VfpDirectedPortCounters{
+					DropCounters: &utils.VfpPacketDropStats{
+						AclDropPacketCount: 199,
+					},
+				},
+			},
+		}
+	}
+	defer func() { GetHNSMetadata = originalGetHNS }()
+
+	testFlow := &flow.Flow{
+		IP: &flow.IP{Source: "1.1.1.1"},
+		Source: &flow.Endpoint{
+			Namespace: "default",
+			PodName:   "test-pod",
+		},
+	}
+
+	drop.ProcessFlow(testFlow)
+
+	mfs, err := exporter.AdvancedRegistry.Gather()
+	assert.NoError(t, err)
+	var validMetricCount int
+
+	for _, mf := range mfs {
+		if !strings.Contains(mf.GetName(), utils.DroppedPacketsGaugeName) {
+			continue
+		}
+		t.Logf("Metric Family: %s", mf.GetName())
+
+		for _, m := range mf.GetMetric() {
+			labelMap := map[string]string{}
+			for _, label := range m.GetLabel() {
+				labelMap[label.GetName()] = label.GetValue()
+			}
+			assert.Equal(t, "1.1.1.1", labelMap["ip"])
+			assert.Equal(t, "default", labelMap["namespace"])
+			assert.Equal(t, "test-pod", labelMap["podname"])
+			assert.Empty(t, labelMap["workload_kind"])
+			assert.Empty(t, labelMap["workload_name"])
+
+			if labelMap["direction"] == "ingress" && labelMap["reason"] == utils.Endpoint {
+				assert.Equal(t, float64(0), m.GetGauge().GetValue())
+				validMetricCount++
+			} else if labelMap["direction"] == "ingress" && labelMap["reason"] == utils.AclRule {
+				assert.Equal(t, float64(100), m.GetGauge().GetValue())
+				validMetricCount++
+			} else if labelMap["direction"] == "egress" && labelMap["reason"] == utils.Endpoint {
+				assert.Equal(t, float64(99), m.GetGauge().GetValue())
+				validMetricCount++
+			} else if labelMap["direction"] == "egress" && labelMap["reason"] == utils.AclRule {
+				assert.Equal(t, float64(199), m.GetGauge().GetValue())
+				validMetricCount++
+			}
+		}
+	}
+
+	assert.Equal(t, 4, validMetricCount, "Expected 4 metric samples with correct labels and values")
+
+}

pkg/module/metrics/forward.go

@@ -30,19 +30,21 @@ const (
 type ForwardMetrics struct {
 	baseMetricObject
 	forwardMetric metricsinit.GaugeVec
-	// bytesMetric metricsinit.IGaugeVec
-	metricName string
+	// bytesMetric      metricsinit.GaugeVec
+	metricName       string
+	enableStandalone bool
 }
 
-func NewForwardCountMetrics(ctxOptions *api.MetricsContextOptions, fl *log.ZapLogger, isLocalContext enrichmentContext) *ForwardMetrics {
+func NewForwardCountMetrics(ctxOptions *api.MetricsContextOptions, fl *log.ZapLogger, isLocalContext enrichmentContext, enableStandalone bool) *ForwardMetrics {
 	if ctxOptions == nil || !strings.Contains(strings.ToLower(ctxOptions.MetricName), "forward") {
 		return nil
 	}
 
-	l := fl.Named("forward-metricsmodule")
-	l.Info("Creating forward count metrics", zap.Any("options", ctxOptions))
+	fl = fl.Named("forward-metricsmodule")
+	fl.Info("Creating forward count metrics", zap.Any("options", ctxOptions))
 	return &ForwardMetrics{
 		baseMetricObject: newBaseMetricsObject(ctxOptions, fl, isLocalContext),
+		enableStandalone: enableStandalone,
 	}
 }
 
@@ -54,12 +56,14 @@ func (f *ForwardMetrics) Init(metricName string) {
 			TotalCountName,
 			TotalCountDesc,
 			f.getLabels()...)
+		f.l.Info("Initialized forward packets metric")
 	case utils.ForwardBytesGaugeName:
 		f.forwardMetric = exporter.CreatePrometheusGaugeVecForMetric(
 			exporter.AdvancedRegistry,
 			TotalBytesName,
 			TotalBytesDesc,
 			f.getLabels()...)
+		f.l.Info("Initialized forward bytes metric")
 	default:
 		f.l.Error("unknown metric name", zap.String("name", metricName))
 	}
@@ -93,6 +97,7 @@ func (f *ForwardMetrics) getLabels() []string {
 }
 
 func (f *ForwardMetrics) Clean() {
+	f.l.Info("Cleaning metric", zap.String("name", f.metricName))
 	exporter.UnregisterMetric(exporter.AdvancedRegistry, metricsinit.ToPrometheusType(f.forwardMetric))
 }
 
@@ -105,6 +110,11 @@ func (f *ForwardMetrics) ProcessFlow(flow *v1.Flow) {
 		return
 	}
 
+	if f.enableStandalone {
+		f.processStandaloneFlow(flow)
+		return
+	}
+
 	if flow.Verdict != v1.Verdict_FORWARDED {
 		return
 	}
@@ -175,3 +185,35 @@ func (f *ForwardMetrics) update(fl *v1.Flow, labels []string) {
 		f.forwardMetric.WithLabelValues(labels...).Add(float64(utils.PacketSize(fl) + utils.PreviouslyObservedBytes(fl)))
 	}
 }
+
+func (f *ForwardMetrics) processStandaloneFlow(fl *v1.Flow) {
+	// Ingress values
+	ingressLbls := []string{
+		ingress,
+		fl.GetIP().GetSource(),
+		fl.GetSource().GetNamespace(),
+		fl.GetSource().GetPodName(),
+		"",
+		"",
+	}
+	// Egress values
+	egressLbls := []string{
+		egress,
+		fl.GetIP().GetSource(),
+		fl.GetSource().GetNamespace(),
+		fl.GetSource().GetPodName(),
+		"",
+		"",
+	}
+
+	endpointStats := GetHNSMetadata(fl).GetEndpointStats()
+
+	switch f.metricName {
+	case utils.ForwardPacketsGaugeName:
+		f.forwardMetric.WithLabelValues(ingressLbls...).Set(float64(endpointStats.GetPacketsReceived()))
+		f.forwardMetric.WithLabelValues(egressLbls...).Set(float64(endpointStats.GetPacketsSent()))
+	case utils.ForwardBytesGaugeName:
+		f.forwardMetric.WithLabelValues(ingressLbls...).Set(float64(endpointStats.GetBytesReceived()))
+		f.forwardMetric.WithLabelValues(egressLbls...).Set(float64(endpointStats.GetBytesSent()))
+	}
+}