test-integration: Add priority ordering verification for BPF links

Add integration tests to verify BPF program links are correctly ordered
by priority across XDP, TC, and TCX program types.

The new verification framework validates link ordering on each cluster
node by comparing ClusterBpfApplicationState data against actual bpfman
daemon state.

Signed-off-by: Andreas Karis <[email protected]>

Author: andreaskaris

test/integration/common.go

@@ -5,12 +5,23 @@ package integration
 
 import (
 	"bytes"
+	"fmt"
 	"regexp"
+	"slices"
 	"strconv"
 	"strings"
 	"testing"
 
+	"github.com/bpfman/bpfman-operator/apis/v1alpha1"
 	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	bpfmanNamespace      = "bpfman"
+	bpfmanContainer      = "bpfman"
+	bpfmanDaemonSelector = "name=bpfman-daemon"
 )
 
 func doKprobeCheck(t *testing.T, output *bytes.Buffer) bool {
@@ -131,3 +142,218 @@ func doProbeCommonCheck(t *testing.T, output *bytes.Buffer, str string) (bool, i
 	}
 	return false, 0
 }
+
+// clusterBpfApplicationStateSuccess returns a function that checks if the expected number of
+// ClusterBpfApplications matching the label selector have reached a successful state.
+func clusterBpfApplicationStateSuccess(t *testing.T, labelSelector string, numExpected int) func() bool {
+	return func() bool {
+		// Fetch all ClusterBpfApplications matching the label selector.
+		apps, err := bpfmanClient.BpfmanV1alpha1().ClusterBpfApplications().List(ctx, metav1.ListOptions{
+			LabelSelector: labelSelector,
+		})
+		require.NoError(t, err)
+
+		// Count how many applications have reached success state.
+		numMatches := 0
+		for _, app := range apps.Items {
+			c := meta.FindStatusCondition(app.Status.Conditions, string(v1alpha1.BpfAppStateCondSuccess))
+			if c != nil && c.Status == metav1.ConditionTrue {
+				numMatches++
+			}
+		}
+		// Return true if the number of successful applications matches expected count.
+		return numMatches == numExpected
+	}
+}
+
+// verifyClusterBpfApplicationPriority returns a function that verifies BPF program links are ordered
+// correctly according to their priority values on each node.
+func verifyClusterBpfApplicationPriority(t *testing.T, labelSelector string) func() bool {
+	return func() bool {
+		// Fetch all ClusterBpfApplications matching the label selector.
+		apps, err := bpfmanClient.BpfmanV1alpha1().ClusterBpfApplications().List(ctx, metav1.ListOptions{
+			LabelSelector: labelSelector,
+		})
+		require.NoError(t, err)
+
+		// Fetch all ClusterBpfApplicationStates to get per-node link information.
+		appStates, err := bpfmanClient.BpfmanV1alpha1().ClusterBpfApplicationStates().List(ctx, metav1.ListOptions{})
+		require.NoError(t, err)
+
+		// Build a map of node names to their associated links from ClusterBpfApplicationStates.
+		nodeLinks := map[string][]link{}
+		for _, app := range apps.Items {
+			for _, appState := range appStates.Items {
+				for _, ownerRef := range appState.OwnerReferences {
+					// Skip if this appState is not controlled by the current app.
+					if ownerRef.Controller == nil || !*ownerRef.Controller {
+						continue
+					}
+					if ownerRef.UID != app.UID {
+						continue
+					}
+					// Initialize the slice for this node if needed.
+					if nodeLinks[appState.Status.Node] == nil {
+						nodeLinks[appState.Status.Node] = []link{}
+					}
+					// Extract and append links from this appState.
+					nodeLinks[appState.Status.Node] = append(
+						nodeLinks[appState.Status.Node],
+						getClusterBpfApplicationStateLinks(t, appState)...,
+					)
+				}
+			}
+		}
+		// Verify link ordering on each node by directly querying bpfman daemon inside the pod.
+		for node, appStateLinks := range nodeLinks {
+			bpfmanLinks := []link{}
+			// Find the bpfman daemon pod running on this node.
+			pods, err := env.Cluster().Client().CoreV1().Pods(bpfmanNamespace).List(ctx, metav1.ListOptions{
+				LabelSelector: bpfmanDaemonSelector,
+				FieldSelector: fmt.Sprintf("spec.nodeName=%s", node),
+			})
+			require.NoError(t, err)
+			require.Len(t, pods.Items, 1)
+			// Query each link from bpfman and verify that bpfman get link matches the output from
+			// ClusterBpfApplicationState.
+			for _, appStateLink := range appStateLinks {
+				cmd := []string{"./bpfman", "get", "link", fmt.Sprintf("%d", appStateLink.linkId)}
+				var bpfmanOut, bpfmanErr bytes.Buffer
+				err := podExec(ctx, t, pods.Items[0], bpfmanContainer, &bpfmanOut, &bpfmanErr, cmd)
+				require.NoError(t, err)
+				t.Logf("bpfman get link output:\n%s", bpfmanOut.String())
+				// Parse the bpfman output and verify it matches.
+				bpfmanLink := parseLink(bpfmanOut.String())
+				require.True(t, linkOutputMatchesLink(t, bpfmanLink, appStateLink))
+				bpfmanLinks = append(bpfmanLinks, bpfmanLink)
+			}
+			// Verify that links are ordered correctly by priority (match priority to expected position).
+			require.True(t, verifyLinkOrder(bpfmanLinks), "position in slice should match priority", bpfmanLinks)
+		}
+		return true
+	}
+}
+
+// link represents a BPF program link with its metadata including link ID, network interface,
+// namespace path, priority, and position in the link chain.
+type link struct {
+	linkId        uint32
+	interfaceName string
+	netnsPath     string
+	priority      int32
+	position      int32
+}
+
+// parseLink parses the output from "bpfman get link" command and converts it to a link struct.
+func parseLink(out string) link {
+	l := link{}
+	lines := bytes.Split([]byte(out), []byte("\n"))
+
+	for _, line := range lines {
+		parts := bytes.SplitN(line, []byte(":"), 2)
+		if len(parts) != 2 {
+			continue
+		}
+		key := bytes.TrimSpace(parts[0])
+		value := bytes.TrimSpace(parts[1])
+
+		switch string(key) {
+		case "Link ID":
+			fmt.Sscanf(string(value), "%d", &l.linkId)
+		case "Interface":
+			l.interfaceName = string(value)
+		case "Network Namespace":
+			if string(value) != "None" {
+				l.netnsPath = string(value)
+			}
+		case "Priority":
+			fmt.Sscanf(string(value), "%d", &l.priority)
+		case "Position":
+			fmt.Sscanf(string(value), "%d", &l.position)
+		}
+	}
+
+	return l
+}
+
+// getClusterBpfApplicationStateLinks extracts link information from a ClusterBpfApplicationState
+// for XDP, TC, and TCX program types.
+func getClusterBpfApplicationStateLinks(t *testing.T, appState v1alpha1.ClusterBpfApplicationState) []link {
+	links := []link{}
+	// Iterate through all programs in the application state.
+	for _, program := range appState.Status.Programs {
+		switch program.Type {
+		case v1alpha1.ProgTypeXDP:
+			// Extract XDP program links.
+			for _, l := range program.XDP.Links {
+				require.NotNil(t, l.LinkId)
+				require.NotNil(t, l.Priority)
+				links = append(links, link{
+					linkId:        *l.LinkId,
+					interfaceName: l.InterfaceName,
+					netnsPath:     l.NetnsPath,
+					priority:      *l.Priority,
+				})
+			}
+		case v1alpha1.ProgTypeTC:
+			// Extract TC program links.
+			for _, l := range program.TC.Links {
+				require.NotNil(t, l.LinkId)
+				require.NotNil(t, l.Priority)
+				links = append(links, link{
+					linkId:        *l.LinkId,
+					interfaceName: l.InterfaceName,
+					netnsPath:     l.NetnsPath,
+					priority:      *l.Priority,
+				})
+			}
+		case v1alpha1.ProgTypeTCX:
+			// Extract TCX program links.
+			for _, l := range program.TCX.Links {
+				require.NotNil(t, l.LinkId)
+				require.NotNil(t, l.Priority)
+				links = append(links, link{
+					linkId:        *l.LinkId,
+					interfaceName: l.InterfaceName,
+					netnsPath:     l.NetnsPath,
+					priority:      *l.Priority,
+				})
+			}
+		}
+	}
+	return links
+}
+
+// linkOutputMatchesLink compares a link parsed from bpfman output with an expected link state.
+func linkOutputMatchesLink(t *testing.T, linkFromOutput, l link) bool {
+	t.Logf("Comparing output and desired link state; got:\n%+v\nwanted:\n%+v", linkFromOutput, l)
+	return l.linkId == linkFromOutput.linkId &&
+		l.interfaceName == linkFromOutput.interfaceName &&
+		l.netnsPath == linkFromOutput.netnsPath &&
+		l.priority == linkFromOutput.priority
+}
+
+// verifyLinkOrder verifies that links' positions match their priorities.
+// Side-effect: this orders `links` in place by position.
+func verifyLinkOrder(links []link) bool {
+	// Order elements by position.
+	slices.SortFunc(links, func(a, b link) int {
+		if a.position < b.position {
+			return -1
+		}
+		if a.position > b.position {
+			return 1
+		}
+		return 0
+	})
+
+	// Now, make sure that the priority of each element is >= the preceding element.
+	oldI := 0
+	for i := 1; i < len(links); i++ {
+		if links[i].priority < links[oldI].priority {
+			return false
+		}
+		oldI = i
+	}
+	return true
+}

test/integration/metrics_test.go

@@ -179,7 +179,7 @@ func testMetricsProxySelfTest(ctx context.Context, t *testing.T, pod corev1.Pod,
 	cmd := []string{"env", "TOKEN=" + token, "/metrics-proxy", "test"}
 
 	var stdout, stderr bytes.Buffer
-	err := podExec(ctx, t, pod, &stdout, &stderr, cmd)
+	err := podExec(ctx, t, pod, "", &stdout, &stderr, cmd)
 
 	// For self-test, we expect exit code 0 for success, non-zero
 	// for failure. Parse the JSON regardless of exit code to get
@@ -244,7 +244,8 @@ func testMetricsProxySelfTest(ctx context.Context, t *testing.T, pod corev1.Pod,
 	t.Logf("All self-tests passed successfully on pod %s", pod.Name)
 }
 
-func podExec(ctx context.Context, t *testing.T, pod corev1.Pod, stdout, stderr *bytes.Buffer, cmd []string) error {
+// podExec executes a command in a pod's container and captures stdout/stderr output.
+func podExec(ctx context.Context, t *testing.T, pod corev1.Pod, container string, stdout, stderr *bytes.Buffer, cmd []string) error {
 	t.Helper()
 	kubeConfig, err := config.GetConfig()
 	if err != nil {
@@ -269,6 +270,9 @@ func podExec(ctx context.Context, t *testing.T, pod corev1.Pod, stdout, stderr *
 		Stderr:  true,
 		TTY:     false,
 	}
+	if container != "" {
+		execOptions.Container = container
+	}
 
 	req.VersionedParams(execOptions, scheme.ParameterCodec)
 

test/integration/tc_test.go

@@ -5,7 +5,7 @@ package integration
 
 import (
 	"bytes"
-	"context"
+	"fmt"
 	"io"
 	"testing"
 	"time"
@@ -14,20 +14,23 @@ import (
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/ptr"
 )
 
 const (
 	tcGoCounterKustomize       = "https://github.com/bpfman/bpfman/examples/config/default/go-tc-counter/?timeout=120&ref=main"
 	tcGoCounterUserspaceNs     = "go-tc-counter"
 	tcGoCounterUserspaceDsName = "go-tc-counter-ds"
+	tcGoCounterBytecodeName    = "go-tc-counter-example"
+	tcByteCodeLabelSelector    = "app.kubernetes.io/name=tcprogram"
 )
 
 func TestTcGoCounter(t *testing.T) {
 	t.Log("deploying tc counter program")
 	require.NoError(t, clusters.KustomizeDeployForCluster(ctx, env.Cluster(), tcGoCounterKustomize))
-	addCleanup(func(context.Context) error {
+	t.Cleanup(func() {
 		cleanupLog("cleaning up tc counter program")
-		return clusters.KustomizeDeleteForCluster(ctx, env.Cluster(), tcGoCounterKustomize)
+		clusters.KustomizeDeleteForCluster(ctx, env.Cluster(), tcGoCounterKustomize)
 	})
 
 	t.Log("waiting for go tc counter userspace daemon to be available")
@@ -41,6 +44,7 @@ func TestTcGoCounter(t *testing.T) {
 
 	pods, err := env.Cluster().Client().CoreV1().Pods(tcGoCounterUserspaceNs).List(ctx, metav1.ListOptions{LabelSelector: "name=go-tc-counter"})
 	require.NoError(t, err)
+	require.Len(t, pods.Items, 1)
 	gotcCounterPod := pods.Items[0]
 
 	req := env.Cluster().Client().CoreV1().Pods(tcGoCounterUserspaceNs).GetLogs(gotcCounterPod.Name, &corev1.PodLogOptions{})
@@ -57,3 +61,73 @@ func TestTcGoCounter(t *testing.T) {
 		return doTcCheck(t, output)
 	}, 30*time.Second, time.Second)
 }
+
+func TestTcGoCounterLinkPriority(t *testing.T) {
+	priorities := []*int32{
+		nil,
+		ptr.To(int32(0)),
+		ptr.To(int32(500)),
+		ptr.To(int32(1000)),
+	}
+
+	t.Log("deploying tc counter program")
+	require.NoError(t, clusters.KustomizeDeployForCluster(ctx, env.Cluster(), tcGoCounterKustomize))
+	t.Cleanup(func() {
+		cleanupLog("cleaning up tc counter program")
+		clusters.KustomizeDeleteForCluster(ctx, env.Cluster(), tcGoCounterKustomize)
+
+		cleanupLog("cleaning up tc counter bytecode")
+		bpfmanClient.BpfmanV1alpha1().ClusterBpfApplications().DeleteCollection(ctx, metav1.DeleteOptions{},
+			metav1.ListOptions{
+				LabelSelector: tcByteCodeLabelSelector,
+			})
+	})
+
+	t.Log("creating copies of bytecode using the same link")
+	cba, err := bpfmanClient.BpfmanV1alpha1().ClusterBpfApplications().Get(ctx, tcGoCounterBytecodeName, metav1.GetOptions{})
+	require.NoError(t, err)
+	name := cba.Name
+	cba.ObjectMeta = metav1.ObjectMeta{
+		Labels: cba.Labels,
+	}
+	for i, priority := range priorities {
+		cba.Name = fmt.Sprintf("%s-%d", name, i)
+		cba.Spec.Programs[0].TC.Links[0].Priority = priority
+		_, err := bpfmanClient.BpfmanV1alpha1().ClusterBpfApplications().Create(ctx, cba, metav1.CreateOptions{})
+		require.NoError(t, err)
+	}
+	// Add priority 55 from the kustomize deployment as well.
+	priorities = append(priorities, ptr.To(int32(55)))
+
+	t.Log("waiting for bytecode to be attached successfully")
+	require.Eventually(t, clusterBpfApplicationStateSuccess(t, tcByteCodeLabelSelector, len(priorities)), 2*time.Minute, 10*time.Second)
+	require.Eventually(t, verifyClusterBpfApplicationPriority(t, tcByteCodeLabelSelector), 1*time.Minute, 10*time.Second)
+
+	t.Log("waiting for go tc counter userspace daemon to be available")
+	require.Eventually(t, func() bool {
+		daemon, err := env.Cluster().Client().AppsV1().DaemonSets(tcGoCounterUserspaceNs).Get(ctx, tcGoCounterUserspaceDsName, metav1.GetOptions{})
+		require.NoError(t, err)
+		return daemon.Status.DesiredNumberScheduled == daemon.Status.NumberAvailable
+	},
+		// Wait 5 minutes since cosign is slow, https://github.com/bpfman/bpfman/issues/1043
+		5*time.Minute, 10*time.Second)
+
+	pods, err := env.Cluster().Client().CoreV1().Pods(tcGoCounterUserspaceNs).List(ctx, metav1.ListOptions{LabelSelector: "name=go-tc-counter"})
+	require.NoError(t, err)
+	require.Len(t, pods.Items, 1)
+	goTcCounterPod := pods.Items[0]
+
+	req := env.Cluster().Client().CoreV1().Pods(tcGoCounterUserspaceNs).GetLogs(goTcCounterPod.Name, &corev1.PodLogOptions{})
+
+	require.Eventually(t, func() bool {
+		logs, err := req.Stream(ctx)
+		require.NoError(t, err)
+		defer logs.Close()
+		output := new(bytes.Buffer)
+		_, err = io.Copy(output, logs)
+		require.NoError(t, err)
+		t.Logf("counter pod log %s", output.String())
+
+		return doTcCheck(t, output)
+	}, 30*time.Second, time.Second)
+}