WIP

Signed-off-by: Martynas Pumputis <[email protected]>

connectivity/check/action.go

@@ -758,10 +758,7 @@ func (a *Action) followFlows(ctx context.Context, ready chan bool) error {
 	// All tests are initiated from the source Pod, so filtering traffic
 	// originating from and destined to the Pod should capture what we need.
 	pod := a.Source()
-	filter := []*flow.FlowFilter{
-		{SourcePod: []string{pod.Name()}},
-		{DestinationPod: []string{pod.Name()}},
-	}
+	filter := pod.FlowFilter()
 
 	// Initiate long-poll against Hubble Relay.
 	b, err := hubbleClient.GetFlows(ctx, &observer.GetFlowsRequest{

connectivity/check/context.go

@@ -67,8 +67,9 @@ type ConnectivityTest struct {
 
 	lastFlowTimestamps map[string]time.Time
 
-	nodes              map[string]*corev1.Node
-	nodesWithoutCilium []string
+	nodes                 map[string]*corev1.Node
+	nodesWithoutCilium    []string
+	nodesWithoutCiliumMap map[string]struct{}
 
 	manifests      map[string]string
 	helmYAMLValues string

connectivity/check/deployment.go

@@ -1160,10 +1160,13 @@ func (ct *ConnectivityTest) validateDeployment(ctx context.Context) error {
 	}
 
 	for _, pod := range hostNetNSPods.Items {
-		ct.hostNetNSPodsByNode[pod.Spec.NodeName] = Pod{
+		_, ok := ct.nodesWithoutCiliumMap[pod.Spec.NodeName]
+		p := Pod{
 			K8sClient: ct.client,
 			Pod:       pod.DeepCopy(),
+			Outside:   ok,
 		}
+		ct.hostNetNSPodsByNode[pod.Spec.NodeName] = p
 	}
 
 	var logOnce sync.Once

connectivity/check/features.go

@@ -271,10 +271,12 @@ func (ct *ConnectivityTest) extractFeaturesFromNodes(ctx context.Context, client
 	}
 
 	nodes := []string{}
+	ct.nodesWithoutCiliumMap = make(map[string]struct{})
 	for _, node := range nodeList.Items {
 		node := node
 		if !canNodeRunCilium(&node) {
 			nodes = append(nodes, node.ObjectMeta.Name)
+			ct.nodesWithoutCiliumMap[node.ObjectMeta.Name] = struct{}{}
 		}
 	}
 

connectivity/check/peer.go

@@ -8,6 +8,8 @@ import (
 	"net/url"
 	"strconv"
 
+	"github.com/cilium/cilium/api/v1/flow"
+	"github.com/cilium/cilium/pkg/identity"
 	ciliumv2 "github.com/cilium/cilium/pkg/k8s/apis/cilium.io/v2"
 	corev1 "k8s.io/api/core/v1"
 
@@ -40,6 +42,8 @@ type TestPeer interface {
 
 	// Labels returns copy of peer labels
 	Labels() map[string]string
+
+	FlowFilter() []*flow.FlowFilter
 }
 
 // Pod is a Kubernetes Pod acting as a peer in a connectivity test.
@@ -60,6 +64,9 @@ type Pod struct {
 
 	// Port the Pods is listening on for connectivity tests.
 	port uint32
+
+	// The pod is running on a node which doesn't run Cilium
+	Outside bool
 }
 
 func (p Pod) String() string {
@@ -112,6 +119,19 @@ func (p Pod) Labels() map[string]string {
 	return newMap
 }
 
+func (p Pod) FlowFilter() []*flow.FlowFilter {
+	if !p.Outside {
+		return []*flow.FlowFilter{
+			{SourcePod: []string{p.Name()}},
+			{DestinationPod: []string{p.Name()}},
+		}
+	}
+	return []*flow.FlowFilter{
+		{SourceIdentity: []uint32{uint32(identity.ReservedIdentityWorld)}},
+		{DestinationIdentity: []uint32{uint32(identity.ReservedIdentityWorld)}},
+	}
+}
+
 // Service is a service acting as a peer in a connectivity test.
 // It implements interface TestPeer.
 type Service struct {
@@ -161,6 +181,10 @@ func (s Service) Labels() map[string]string {
 	return newMap
 }
 
+func (s Service) FlowFilter() []*flow.FlowFilter {
+	return nil
+}
+
 // ExternalWorkload is an external workload acting as a peer in a
 // connectivity test. It implements interface TestPeer.
 type ExternalWorkload struct {
@@ -208,6 +232,10 @@ func (e ExternalWorkload) Labels() map[string]string {
 	return newMap
 }
 
+func (e ExternalWorkload) FlowFilter() []*flow.FlowFilter {
+	return nil
+}
+
 // ICMPEndpoint returns a new ICMP endpoint.
 func ICMPEndpoint(name, host string) TestPeer {
 	return icmpEndpoint{
@@ -260,6 +288,10 @@ func (ie icmpEndpoint) Labels() map[string]string {
 	return make(map[string]string)
 }
 
+func (ie icmpEndpoint) FlowFilter() []*flow.FlowFilter {
+	return nil
+}
+
 // HTTPEndpoint returns a new endpoint with the given name and raw URL.
 // Panics if rawurl cannot be parsed.
 func HTTPEndpoint(name, rawurl string) TestPeer {
@@ -344,3 +376,7 @@ func (he httpEndpoint) Labels() map[string]string {
 	}
 	return newMap
 }
+
+func (he httpEndpoint) FlowFilter() []*flow.FlowFilter {
+	return nil
+}

connectivity/suite.go

@@ -200,42 +200,6 @@ func Run(ctx context.Context, ct *check.ConnectivityTest) error {
 		return ct.Run(ctx)
 	}
 
-	// Datapath Conformance Tests
-	if ct.Params().Datapath {
-		ct.NewTest("north-south-loadbalancing").
-			WithFeatureRequirements(check.RequireFeatureEnabled(check.FeatureNodeWithoutCilium)).
-			WithScenarios(
-				tests.OutsideToNodePort(),
-			)
-		ct.NewTest("north-south-loadbalancing-with-l7-policy").
-			WithFeatureRequirements(check.RequireFeatureEnabled(check.FeatureNodeWithoutCilium)).
-			WithCiliumPolicy(echoIngressL7HTTPFromAnywherePolicyYAML).
-			WithScenarios(
-				tests.OutsideToNodePort(),
-			)
-		ct.NewTest("pod-to-pod-encryption").
-			WithFeatureRequirements(check.RequireFeatureEnabled(check.FeatureEncryptionPod)).
-			WithScenarios(
-				tests.PodToPodEncryption(),
-			)
-		ct.NewTest("node-to-node-encryption").
-			WithFeatureRequirements(check.RequireFeatureEnabled(check.FeatureEncryptionPod),
-				check.RequireFeatureEnabled(check.FeatureEncryptionNode)).
-			WithScenarios(
-				tests.NodeToNodeEncryption(),
-			)
-
-		ct.NewTest("egress-gateway").
-			WithCiliumEgressGatewayPolicy(egressGatewayPolicyYAML).
-			WithFeatureRequirements(check.RequireFeatureEnabled(check.FeatureEgressGateway),
-				check.RequireFeatureEnabled(check.FeatureNodeWithoutCilium)).
-			WithScenarios(
-				tests.EgressGateway(),
-			)
-
-		return ct.Run(ctx)
-	}
-
 	// Run all tests without any policies in place.
 	noPoliciesScenarios := []check.Scenario{
 		tests.PodToPod(),
@@ -704,6 +668,37 @@ func Run(ctx context.Context, ct *check.ConnectivityTest) error {
 		WithFeatureRequirements(check.RequireFeatureEnabled(check.FeatureHealthChecking)).
 		WithScenarios(tests.CiliumHealth())
 
+	ct.NewTest("north-south-loadbalancing").
+		WithFeatureRequirements(check.RequireFeatureEnabled(check.FeatureNodeWithoutCilium)).
+		WithScenarios(
+			tests.OutsideToNodePort(),
+		)
+	ct.NewTest("north-south-loadbalancing-with-l7-policy").
+		WithFeatureRequirements(check.RequireFeatureEnabled(check.FeatureNodeWithoutCilium)).
+		WithCiliumPolicy(echoIngressL7HTTPFromAnywherePolicyYAML).
+		WithScenarios(
+			tests.OutsideToNodePort(),
+		)
+	ct.NewTest("pod-to-pod-encryption").
+		WithFeatureRequirements(check.RequireFeatureEnabled(check.FeatureEncryptionPod)).
+		WithScenarios(
+			tests.PodToPodEncryption(),
+		)
+	ct.NewTest("node-to-node-encryption").
+		WithFeatureRequirements(check.RequireFeatureEnabled(check.FeatureEncryptionPod),
+			check.RequireFeatureEnabled(check.FeatureEncryptionNode)).
+		WithScenarios(
+			tests.NodeToNodeEncryption(),
+		)
+
+	ct.NewTest("egress-gateway").
+		WithCiliumEgressGatewayPolicy(egressGatewayPolicyYAML).
+		WithFeatureRequirements(check.RequireFeatureEnabled(check.FeatureEgressGateway),
+			check.RequireFeatureEnabled(check.FeatureNodeWithoutCilium)).
+		WithScenarios(
+			tests.EgressGateway(),
+		)
+
 	// The following tests have DNS redirect policies. They should be executed last.
 
 	// Test L7 HTTP introspection using an ingress policy on echo pods.