feat(hubble-test): Adding Hubble Metrics test scenarios

Makefile

@@ -601,6 +601,8 @@ simplify-dashboards:
 run-perf-test:
 	go test -v ./test/e2e/retina_perf_test.go -timeout 2h -tags=perf -count=1  -args -image-tag=${TAG} -image-registry=${IMAGE_REGISTRY} -image-namespace=${IMAGE_NAMESPACE}
 
+run-e2e-test:
+	go test -v ./test/e2e/ -timeout 1h -tags=e2e -count=1  -args -image-tag=${TAG} -image-registry=${IMAGE_REGISTRY} -image-namespace=${IMAGE_NAMESPACE}
 .PHONY: update-hubble
 update-hubble:
 	@echo "Checking for Hubble updates..."

test/e2e/common/common.go

@@ -17,7 +17,6 @@ import (
 )
 
 const (
-	RetinaPort int = 10093
 	// netObsRGtag is used to tag resources created by this test suite
 	NetObsRGtag            = "-e2e-netobs-"
 	KubeSystemNamespace    = "kube-system"
@@ -51,6 +50,9 @@ var (
 	RetinaAdvancedProfilePath = func(rootDir string) string {
 		return filepath.Join(rootDir, "test", "profiles", "advanced", "values.yaml")
 	}
+	HubbleChartPath = func(rootDir string) string {
+		return filepath.Join(rootDir, "deploy", "hubble", "manifests", "controller", "helm", "retina")
+	}
 	KubeConfigFilePath = func(rootDir string) string {
 		return filepath.Join(rootDir, "test", "e2e", "test.pem")
 	}

test/e2e/common/validate-metric.go

@@ -0,0 +1,52 @@
+//nolint:revive // package name "common" is used across the E2E test suite
+package common
+
+import (
+	"errors"
+	"fmt"
+	"log"
+
+	prom "github.com/microsoft/retina/test/e2e/framework/prometheus"
+)
+
+var ErrMetricFound = errors.New("unexpected metric found")
+
+type ValidateMetric struct {
+	ForwardedPort string
+	MetricName    string
+	ValidMetrics  []map[string]string
+	ExpectMetric  bool
+	PartialMatch  bool // If true, only the specified labels need to match (metric can have additional labels)
+}
+
+func (v *ValidateMetric) Run() error {
+	promAddress := fmt.Sprintf("http://localhost:%s/metrics", v.ForwardedPort)
+
+	for _, validMetric := range v.ValidMetrics {
+		err := prom.CheckMetric(promAddress, v.MetricName, validMetric, v.PartialMatch)
+		if err != nil {
+			// If we expect the metric not to be found, return nil if it's not found.
+			if !v.ExpectMetric && errors.Is(err, prom.ErrNoMetricFound) {
+				log.Printf("metric %s not found, as expected\n", v.MetricName)
+				return nil
+			}
+			return fmt.Errorf("failed to verify prometheus metrics: %w", err)
+		}
+
+		// if we expect the metric not to be found, return an error if it is found
+		if !v.ExpectMetric {
+			return fmt.Errorf("did not expect to find metric %s matching %+v: %w", v.MetricName, validMetric, ErrMetricFound)
+		}
+
+		log.Printf("found metric %s matching %+v\n", v.MetricName, validMetric)
+	}
+	return nil
+}
+
+func (v *ValidateMetric) Prevalidate() error {
+	return nil
+}
+
+func (v *ValidateMetric) Stop() error {
+	return nil
+}

test/e2e/framework/constants/hubble.go

@@ -0,0 +1,31 @@
+package constants
+
+const (
+	// Metrics Port
+	HubbleMetricsPort = "9965"
+
+	// MetricsName
+	HubbleDNSQueryMetricName    = "hubble_dns_queries_total"
+	HubbleDNSResponseMetricName = "hubble_dns_responses_total"
+	HubbleFlowMetricName        = "hubble_flows_processed_total"
+	HubbleDropMetricName        = "hubble_drop_total"
+	HubbleTCPFlagsMetricName    = "hubble_tcp_flags_total"
+
+	// Labels
+	HubbleDestinationLabel = "destination"
+	HubbleSourceLabel      = "source"
+	HubbleIPsRetunedLabel  = "ips_returned"
+	HubbleQTypesLabel      = "qtypes"
+	HubbleRCodeLabel       = "rcode"
+	HubbleQueryLabel       = "query"
+
+	HubbleProtocolLabel = "protocol"
+	HubbleReasonLabel   = "reason"
+
+	HubbleSubtypeLabel = "subtype"
+	HubbleTypeLabel    = "type"
+	HubbleVerdictLabel = "verdict"
+
+	HubbleFamilyLabel = "family"
+	HubbleFlagLabel   = "flag"
+)

test/e2e/framework/constants/networking.go

@@ -0,0 +1,15 @@
+package constants
+
+const (
+	MetricsEndpoint = "metrics"
+
+	TCP             = "TCP"
+	UDP             = "UDP"
+	IPV4            = "IPv4"
+	IPTableRuleDrop = "IPTABLE_RULE_DROP"
+	SYN             = "SYN"
+	SYNACK          = "SYN-ACK"
+	ACK             = "ACK"
+	FIN             = "FIN"
+	RST             = "RST"
+)

test/e2e/framework/constants/retina.go

@@ -0,0 +1,17 @@
+package constants
+
+const (
+	// Metrics Port
+	RetinaMetricsPort = "10093"
+
+	// MetricsName
+	RetinaDropMetricName    = "networkobservability_drop_count"
+	RetinaForwardMetricName = "networkobservability_forward_count"
+
+	// Labels
+	RetinaSourceLabel      = "source"
+	RetinaDestinationLabel = "destination"
+	RetinaProtocolLabel    = "protocol"
+	RetinaReasonLabel      = "reason"
+	RetinaDirectionLabel   = "direction"
+)

test/e2e/framework/kubernetes/create-agnhost-statefulset.go

@@ -18,7 +18,6 @@ var ErrLabelMissingFromPod = fmt.Errorf("label missing from pod")
 
 const (
 	AgnhostHTTPPort  = 80
-	AgnhostReplicas  = 1
 	AgnhostArchAmd64 = "amd64"
 	AgnhostArchArm64 = "arm64"
 )
@@ -29,6 +28,7 @@ type CreateAgnhostStatefulSet struct {
 	ScheduleOnSameNode bool
 	KubeConfigFilePath string
 	AgnhostArch        string
+	AgnhostReplicas    *int
 }
 
 func (c *CreateAgnhostStatefulSet) Run() error {
@@ -50,7 +50,13 @@ func (c *CreateAgnhostStatefulSet) Run() error {
 		c.AgnhostArch = AgnhostArchAmd64
 	}
 
-	agnhostStatefulSet := c.getAgnhostDeployment(c.AgnhostArch)
+	// set default replicas to 1
+	replicas := 1
+	if c.AgnhostReplicas != nil {
+		replicas = *c.AgnhostReplicas
+	}
+
+	agnhostStatefulSet := c.getAgnhostDeployment(c.AgnhostArch, replicas)
 
 	err = CreateResource(ctx, agnhostStatefulSet, clientset)
 	if err != nil {
@@ -79,8 +85,8 @@ func (c *CreateAgnhostStatefulSet) Stop() error {
 	return nil
 }
 
-func (c *CreateAgnhostStatefulSet) getAgnhostDeployment(arch string) *appsv1.StatefulSet {
-	reps := int32(AgnhostReplicas)
+func (c *CreateAgnhostStatefulSet) getAgnhostDeployment(arch string, replicas int) *appsv1.StatefulSet {
+	reps := int32(replicas)
 
 	var affinity *v1.Affinity
 	if c.ScheduleOnSameNode {

test/e2e/framework/kubernetes/create-network-policy.go

@@ -36,8 +36,8 @@ func (c *CreateDenyAllNetworkPolicy) Run() error {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	agnhostStatefulSet := getNetworkPolicy(c.NetworkPolicyNamespace, c.DenyAllLabelSelector)
-	err = CreateResource(ctx, agnhostStatefulSet, clientset)
+	networkPolicy := getNetworkPolicy(c.NetworkPolicyNamespace, c.DenyAllLabelSelector)
+	err = CreateResource(ctx, networkPolicy, clientset)
 	if err != nil {
 		return fmt.Errorf("error creating simple deny-all network policy: %w", err)
 	}
@@ -96,8 +96,8 @@ func (d *DeleteDenyAllNetworkPolicy) Run() error {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	agnhostStatefulSet := getNetworkPolicy(d.NetworkPolicyNamespace, d.DenyAllLabelSelector)
-	err = DeleteResource(ctx, agnhostStatefulSet, clientset)
+	networkPolicy := getNetworkPolicy(d.NetworkPolicyNamespace, d.DenyAllLabelSelector)
+	err = DeleteResource(ctx, networkPolicy, clientset)
 	if err != nil {
 		return fmt.Errorf("error creating simple deny-all network policy: %w", err)
 	}

test/e2e/framework/kubernetes/install-hubble-helm.go

@@ -23,15 +23,15 @@ const (
 	HubbleRelayApp  = "hubble-relay"
 )
 
-type ValidateHubbleStep struct {
+type InstallHubbleHelmChart struct {
 	Namespace          string
 	ReleaseName        string
 	KubeConfigFilePath string
 	ChartPath          string
 	TagEnv             string
 }
 
-func (v *ValidateHubbleStep) Run() error {
+func (v *InstallHubbleHelmChart) Run() error {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeoutSeconds*time.Second)
 	defer cancel()
 
@@ -146,10 +146,10 @@ func (v *ValidateHubbleStep) Run() error {
 	return nil
 }
 
-func (v *ValidateHubbleStep) Prevalidate() error {
+func (v *InstallHubbleHelmChart) Prevalidate() error {
 	return nil
 }
 
-func (v *ValidateHubbleStep) Stop() error {
+func (v *InstallHubbleHelmChart) Stop() error {
 	return nil
 }

test/e2e/framework/prometheus/prometheus.go

@@ -22,13 +22,16 @@ var (
 	defaultRetryAttempts = 60
 )
 
-func CheckMetric(promAddress, metricName string, validMetric map[string]string) error {
+func CheckMetric(promAddress, metricName string, validMetric map[string]string, partial ...bool) error {
 	defaultRetrier := retry.Retrier{Attempts: defaultRetryAttempts, Delay: defaultRetryDelay}
 
 	ctx := context.Background()
 	pctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
+	// Default partial to false if not provided
+	usePartial := len(partial) > 0 && partial[0]
+
 	metrics := map[string]*promclient.MetricFamily{}
 	scrapeMetricsFn := func() error {
 		log.Printf("checking for metrics on %s", promAddress)
@@ -42,7 +45,11 @@ func CheckMetric(promAddress, metricName string, validMetric map[string]string)
 
 		// loop through each metric to check for a match,
 		// if none is found then log and return an error which will trigger a retry
-		err = verifyValidMetricPresent(metricName, metrics, validMetric)
+		if usePartial {
+			err = verifyValidMetricPresentPartial(metricName, metrics, validMetric)
+		} else {
+			err = verifyValidMetricPresent(metricName, metrics, validMetric)
+		}
 		if err != nil {
 			log.Printf("failed to find metric matching %s: %+v\n", metricName, validMetric)
 			return ErrNoMetricFound
@@ -99,6 +106,43 @@ func verifyValidMetricPresent(metricName string, data map[string]*promclient.Met
 	return fmt.Errorf("failed to find metric matching: %+v: %w", validMetric, ErrNoMetricFound)
 }
 
+// verifyValidMetricPresentPartial checks if a metric exists with labels that contain
+// all the key-value pairs in validMetric (partial matching - the metric can have additional labels)
+func verifyValidMetricPresentPartial(metricName string, data map[string]*promclient.MetricFamily, validMetric map[string]string) error {
+	for _, metric := range data {
+		if metric.GetName() == metricName {
+			for _, metric := range metric.GetMetric() {
+
+				// get all labels and values on the metric
+				metricLabels := map[string]string{}
+				for _, label := range metric.GetLabel() {
+					metricLabels[label.GetName()] = label.GetValue()
+				}
+
+				// if valid metric is empty, then we just need to make sure the metric and value is present
+				if len(validMetric) == 0 && len(metricLabels) > 0 {
+					return nil
+				}
+
+				// Check if all key-value pairs in validMetric exist in metricLabels
+				allMatch := true
+				for key, value := range validMetric {
+					if metricLabels[key] != value {
+						allMatch = false
+						break
+					}
+				}
+
+				if allMatch {
+					return nil
+				}
+			}
+		}
+	}
+
+	return fmt.Errorf("failed to find metric matching: %+v: %w", validMetric, ErrNoMetricFound)
+}
+
 func getAllPrometheusMetricsFromURL(url string) (map[string]*promclient.MetricFamily, error) {
 	client := http.Client{}
 	resp, err := client.Get(url) //nolint

test/e2e/jobs/jobs.go

@@ -9,10 +9,14 @@ import (
 	"github.com/microsoft/retina/test/e2e/framework/generic"
 	"github.com/microsoft/retina/test/e2e/framework/kubernetes"
 	"github.com/microsoft/retina/test/e2e/framework/types"
-	"github.com/microsoft/retina/test/e2e/hubble"
 	"github.com/microsoft/retina/test/e2e/scenarios/capture"
 	"github.com/microsoft/retina/test/e2e/scenarios/dns"
 	"github.com/microsoft/retina/test/e2e/scenarios/drop"
+	hubble_dns "github.com/microsoft/retina/test/e2e/scenarios/hubble/dns"
+	hubble_drop "github.com/microsoft/retina/test/e2e/scenarios/hubble/drop"
+	hubble_flow "github.com/microsoft/retina/test/e2e/scenarios/hubble/flow"
+	hubble_service "github.com/microsoft/retina/test/e2e/scenarios/hubble/service"
+	hubble_tcp "github.com/microsoft/retina/test/e2e/scenarios/hubble/tcp"
 	"github.com/microsoft/retina/test/e2e/scenarios/latency"
 	tcp "github.com/microsoft/retina/test/e2e/scenarios/tcp"
 	"github.com/microsoft/retina/test/e2e/scenarios/windows"
@@ -273,20 +277,36 @@ func UpgradeAndTestRetinaAdvancedMetrics(kubeConfigFilePath, chartPath, valuesFi
 	return job
 }
 
-func ValidateHubble(kubeConfigFilePath, chartPath string, testPodNamespace string) *types.Job {
+func InstallAndTestHubbleMetrics(kubeConfigFilePath, chartPath string) *types.Job {
 	job := types.NewJob("Validate Hubble")
 
-	job.AddStep(&kubernetes.ValidateHubbleStep{
+	job.AddStep(&kubernetes.InstallHubbleHelmChart{
 		Namespace:          common.KubeSystemNamespace,
 		ReleaseName:        "retina",
 		KubeConfigFilePath: kubeConfigFilePath,
 		ChartPath:          chartPath,
 		TagEnv:             generic.DefaultTagEnv,
 	}, nil)
 
-	job.AddScenario(hubble.ValidateHubbleRelayService())
+	hubbleScenarios := []*types.Scenario{
+		hubble_service.ValidateHubbleRelayService(),
+		hubble_service.ValidateHubbleUIService(kubeConfigFilePath),
+	}
 
-	job.AddScenario(hubble.ValidateHubbleUIService(kubeConfigFilePath))
+	for _, arch := range common.Architectures {
+		hubbleScenarios = append(hubbleScenarios,
+			hubble_dns.ValidateDNSMetric(arch),
+			hubble_flow.ValidatePodToPodIntraNodeHubbleFlowMetric(arch),
+			hubble_flow.ValidatePodToPodInterNodeHubbleFlowMetric(arch),
+			hubble_flow.ValidatePodToWorldHubbleFlowMetric(arch),
+			hubble_drop.ValidateDropMetric(arch),
+			hubble_tcp.ValidateTCPMetric(arch),
+		)
+	}
+
+	for _, scenario := range hubbleScenarios {
+		job.AddScenario(scenario)
+	}
 
 	job.AddStep(&kubernetes.EnsureStableComponent{
 		PodNamespace:           common.KubeSystemNamespace,

test/e2e/retina_e2e_test.go

@@ -26,8 +26,6 @@ func TestE2ERetina(t *testing.T) {
 	// Get to root of the repo by going up two directories
 	rootDir := filepath.Dir(filepath.Dir(cwd))
 
-	hubblechartPath := filepath.Join(rootDir, "deploy", "hubble", "manifests", "controller", "helm", "retina")
-
 	err = jobs.LoadGenericFlags().Run()
 	require.NoError(t, err, "failed to load generic flags")
 
@@ -54,14 +52,9 @@ func TestE2ERetina(t *testing.T) {
 	)
 	advanceMetricsE2E.Run(ctx)
 
-	// Install and test Hubble basic metrics
-	validatehubble := types.NewRunner(t,
-		jobs.ValidateHubble(
-			common.KubeConfigFilePath(rootDir),
-			hubblechartPath,
-			common.TestPodNamespace),
-	)
-	validatehubble.Run(ctx)
+	// Install and test Hubble  metrics
+	hubbleMetricsE2E := types.NewRunner(t, jobs.InstallAndTestHubbleMetrics(common.KubeConfigFilePath(rootDir), common.HubbleChartPath(rootDir)))
+	hubbleMetricsE2E.Run(ctx)
 
 	// Install Retina basic and test captures
 	captureE2E := types.NewRunner(t,