test: improve custom networking integration tests (#3668)

* test: improve custom networking integration tests with cross-node enforcement, ClusterIP connectivity, and ASG-driven instance recycling

* allowing cluster sg ingress rule to allow custom networking sgid

* making set asg more resilient, and adding error for sg creation call

---------

Co-authored-by: Hao Zhou <haouc@users.noreply.github.com>

Author: yash97

test/framework/resources/aws/services/autoscaling.go

@@ -24,6 +24,8 @@ import (
 
 type AutoScaling interface {
 	DescribeAutoScalingGroup(ctx context.Context, autoScalingGroupName string) ([]types.AutoScalingGroup, error)
+	GetASGForInstance(ctx context.Context, instanceID string) (string, error)
+	SetDesiredCapacity(ctx context.Context, asgName string, desired int32) error
 }
 
 // Directly using the client to interact with the service instead of an interface.
@@ -51,3 +53,43 @@ func (d defaultAutoScaling) DescribeAutoScalingGroup(ctx context.Context, autoSc
 
 	return asg.AutoScalingGroups, nil
 }
+
+func (d defaultAutoScaling) GetASGForInstance(ctx context.Context, instanceID string) (string, error) {
+	out, err := d.client.DescribeAutoScalingInstances(ctx, &autoscaling.DescribeAutoScalingInstancesInput{
+		InstanceIds: []string{instanceID},
+	})
+	if err != nil {
+		return "", err
+	}
+	if len(out.AutoScalingInstances) == 0 || out.AutoScalingInstances[0].AutoScalingGroupName == nil {
+		return "", fmt.Errorf("instance %s is not part of any ASG", instanceID)
+	}
+	return *out.AutoScalingInstances[0].AutoScalingGroupName, nil
+}
+
+func (d defaultAutoScaling) SetDesiredCapacity(ctx context.Context, asgName string, desired int32) error {
+	descOut, err := d.client.DescribeAutoScalingGroups(ctx, &autoscaling.DescribeAutoScalingGroupsInput{
+		AutoScalingGroupNames: []string{asgName},
+	})
+	if err != nil {
+		return err
+	}
+	if len(descOut.AutoScalingGroups) == 0 {
+		return fmt.Errorf("ASG %s not found", asgName)
+	}
+	if aws.ToInt32(descOut.AutoScalingGroups[0].MaxSize) < desired || aws.ToInt32(descOut.AutoScalingGroups[0].MinSize) > desired {
+		if _, err := d.client.UpdateAutoScalingGroup(ctx, &autoscaling.UpdateAutoScalingGroupInput{
+			AutoScalingGroupName: aws.String(asgName),
+			MaxSize:              aws.Int32(max(desired, aws.ToInt32(descOut.AutoScalingGroups[0].MaxSize))),
+			MinSize:              aws.Int32(min(desired, aws.ToInt32(descOut.AutoScalingGroups[0].MinSize))),
+		}); err != nil {
+			return fmt.Errorf("raise MaxSize: %v", err)
+		}
+	}
+	_, err = d.client.SetDesiredCapacity(ctx, &autoscaling.SetDesiredCapacityInput{
+		AutoScalingGroupName: aws.String(asgName),
+		DesiredCapacity:      aws.Int32(desired),
+		HonorCooldown:        aws.Bool(false),
+	})
+	return err
+}

test/framework/resources/aws/utils/nodegroup.go

@@ -29,6 +29,9 @@ import (
 	"github.com/aws/amazon-vpc-cni-k8s/test/framework/utils"
 	"github.com/aws/aws-sdk-go-v2/aws"
 	cloudformationtypes "github.com/aws/aws-sdk-go-v2/service/cloudformation/types"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
 const (
@@ -366,18 +369,64 @@ func TerminateInstances(f *framework.Framework) error {
 	if err != nil {
 		return fmt.Errorf("failed to get list of nodes created: %v", err)
 	}
+	if len(nodeList.Items) == 0 {
+		return nil
+	}
 
 	var instanceIDs []string
 	for _, node := range nodeList.Items {
 		instanceIDs = append(instanceIDs, k8sUtils.GetInstanceIDFromNode(node))
 	}
+	expected := int32(len(instanceIDs))
 
-	err = f.CloudServices.EC2().TerminateInstance(context.TODO(), instanceIDs)
+	// Find the ASG owning the first instance. Assumes all nodes belong to the same ASG.
+	asgName, err := f.CloudServices.AutoScaling().GetASGForInstance(context.TODO(), instanceIDs[0])
 	if err != nil {
-		return fmt.Errorf("failed to terminate instances: %v", err)
+		return fmt.Errorf("failed to find ASG for instance %s: %v", instanceIDs[0], err)
 	}
 
-	// Wait for instances to be replaced
-	time.Sleep(time.Minute * 8)
-	return nil
+	// Scale the ASG to 0 so it terminates all instances through its own lifecycle
+	if err := f.CloudServices.AutoScaling().SetDesiredCapacity(context.TODO(), asgName, 0); err != nil {
+		return fmt.Errorf("failed to set desired capacity to 0 on %s: %v", asgName, err)
+	}
+
+	// Wait until ASG has actually finished terminating its instances before scaling
+	if err := wait.PollUntilContextTimeout(context.TODO(), 10*time.Second, 10*time.Minute, true, func(ctx context.Context) (bool, error) {
+		asgs, derr := f.CloudServices.AutoScaling().DescribeAutoScalingGroup(ctx, asgName)
+		if derr != nil || len(asgs) == 0 {
+			return false, nil
+		}
+		return len(asgs[0].Instances) == 0, nil
+	}); err != nil {
+		return fmt.Errorf("timed out waiting for ASG %s to scale to 0: %v", asgName, err)
+	}
+
+	// Force-delete stale Node objects so the scheduler/aws-node don't wait on wedged kubelets.
+	zero := int64(0)
+	opts := &client.DeleteOptions{GracePeriodSeconds: &zero, PropagationPolicy: func() *metav1.DeletionPropagation {
+		p := metav1.DeletePropagationBackground
+		return &p
+	}()}
+	for i := range nodeList.Items {
+		_ = f.K8sResourceManagers.NodeManager().DeleteNode(&nodeList.Items[i], opts)
+	}
+
+	if err := f.CloudServices.AutoScaling().SetDesiredCapacity(context.TODO(), asgName, expected); err != nil {
+		return fmt.Errorf("failed to set desired capacity back to %d on %s: %v", expected, asgName, err)
+	}
+
+	// Wait until ASG reports `expected` instances InService.
+	return wait.PollUntilContextTimeout(context.TODO(), 10*time.Second, 8*time.Minute, true, func(ctx context.Context) (bool, error) {
+		asgs, derr := f.CloudServices.AutoScaling().DescribeAutoScalingGroup(ctx, asgName)
+		if derr != nil || len(asgs) == 0 {
+			return false, nil
+		}
+		inService := int32(0)
+		for _, inst := range asgs[0].Instances {
+			if inst.LifecycleState == "InService" {
+				inService++
+			}
+		}
+		return inService >= expected, nil
+	})
 }

test/framework/resources/k8s/resources/node.go

@@ -19,6 +19,7 @@ import (
 	"github.com/aws/amazon-vpc-cni-k8s/test/framework/utils"
 
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -27,6 +28,7 @@ type NodeManager interface {
 	GetNodes(nodeLabelKey string, nodeLabelVal string) (v1.NodeList, error)
 	GetAllNodes() (v1.NodeList, error)
 	UpdateNode(oldNode *v1.Node, newNode *v1.Node) error
+	DeleteNode(node *v1.Node, opts ...client.DeleteOption) error
 	WaitTillNodesReady(nodeLabelKey string, nodeLabelVal string, asgSize int) error
 }
 
@@ -73,6 +75,14 @@ func (d *defaultNodeManager) UpdateNode(oldNode *v1.Node, newNode *v1.Node) erro
 	return d.k8sClient.Patch(context.Background(), newNode, client.MergeFrom(oldNode))
 }
 
+func (d *defaultNodeManager) DeleteNode(node *v1.Node, opts ...client.DeleteOption) error {
+	err := d.k8sClient.Delete(context.Background(), node, opts...)
+	if errors.IsNotFound(err) {
+		return nil
+	}
+	return err
+}
+
 func (d *defaultNodeManager) WaitTillNodesReady(nodeLabelKey string, nodeLabelVal string, asgSize int) error {
 	return wait.PollImmediateUntil(utils.PollIntervalLong, func() (done bool, err error) {
 		nodeList, err := d.GetNodes(nodeLabelKey, nodeLabelVal)

test/framework/utils/const.go

@@ -35,4 +35,5 @@ const (
 	PollIntervalLong   = time.Second * 20
 
 	DefaultDeploymentReadyTimeout = time.Second * 300
+	ShortDeploymentReadyTimeout   = time.Second * 120
 )

test/integration/custom-networking/custom_networking_suite_test.go

@@ -57,6 +57,7 @@ var (
 	corednsSGOpenPort            = 53
 	primaryENISGID               string
 	primaryENISGList             []string
+	clusterSGID                  string
 	// List of ENIConfig per Availability Zone
 	eniConfigList        []*v1alpha1.ENIConfig
 	eniConfigBuilderList []*manifest.ENIConfigBuilder
@@ -87,6 +88,16 @@ var _ = BeforeSuite(func() {
 	Expect(err).ToNot(HaveOccurred())
 	customNetworkingSGID = *createSecurityGroupOutput.GroupId
 
+	By("Getting Cluster Security Group ID")
+	clusterRes, err := f.CloudServices.EKS().DescribeCluster(context.TODO(), f.Options.ClusterName)
+	Expect(err).NotTo(HaveOccurred())
+	clusterSGID = *(clusterRes.Cluster.ResourcesVpcConfig.ClusterSecurityGroupId)
+
+	By("allowing custom networking SG in cluster SG")
+	err = f.CloudServices.EC2().AuthorizeSecurityGroupIngress(context.TODO(), clusterSGID, "-1",
+		-1, -1, customNetworkingSGID, true)
+	Expect(err).ToNot(HaveOccurred())
+
 	By("authorizing egress and ingress on security group for single port")
 	f.CloudServices.EC2().AuthorizeSecurityGroupEgress(context.TODO(), customNetworkingSGID, "TCP",
 		customNetworkingSGOpenPort, customNetworkingSGOpenPort, "0.0.0.0/0")
@@ -218,6 +229,10 @@ var _ = AfterSuite(func() {
 			-1, -1, customNetworkingSGID, true)
 	}
 
+	By("removing custom networking SG from cluster SG")
+	_ = f.CloudServices.EC2().RevokeSecurityGroupIngress(context.TODO(), clusterSGID, "-1",
+		-1, -1, customNetworkingSGID, true)
+
 	By("deleting security group")
 	errs.Append(f.CloudServices.EC2().DeleteSecurityGroup(context.TODO(), customNetworkingSGID))
 

test/integration/custom-networking/custom_networking_test.go

@@ -14,19 +14,22 @@
 package custom_networking
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"strconv"
 	"time"
 
 	awsUtils "github.com/aws/amazon-vpc-cni-k8s/test/framework/resources/aws/utils"
 	"github.com/aws/amazon-vpc-cni-k8s/test/framework/resources/k8s/manifest"
+	k8sUtils "github.com/aws/amazon-vpc-cni-k8s/test/framework/resources/k8s/utils"
 	"github.com/aws/amazon-vpc-cni-k8s/test/framework/utils"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	v1 "k8s.io/api/apps/v1"
 	coreV1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 )
 
 var _ = Describe("Custom Networking Test", func() {
@@ -51,7 +54,6 @@ var _ = Describe("Custom Networking Test", func() {
 				Command([]string{"nc"}).
 				Args([]string{"-k", "-l", strconv.Itoa(port)}).
 				Build()
-
 			deploymentBuilder := manifest.NewBusyBoxDeploymentBuilder(f.Options.TestImageRegistry).
 				Container(container).
 				Replicas(replicaCount).
@@ -90,7 +92,29 @@ var _ = Describe("Custom Networking Test", func() {
 					Parallelism(1).
 					Build()
 
+				// Force client Job onto a DIFFERENT node than THIS target pod so that
+				// traffic actually traverses the ENI and AWS SG enforcement applies
+				// (same-node pod-to-pod bypasses ENI-level SG evaluation in the Linux bridge).
+				// We only exclude the one target pod's node, not all target nodes, so the
+				// Job can still schedule on 2-node clusters when targets span both nodes.
+				testJob.Spec.Template.Spec.Affinity = &coreV1.Affinity{
+					NodeAffinity: &coreV1.NodeAffinity{
+						RequiredDuringSchedulingIgnoredDuringExecution: &coreV1.NodeSelector{
+							NodeSelectorTerms: []coreV1.NodeSelectorTerm{{
+								MatchExpressions: []coreV1.NodeSelectorRequirement{{
+									Key:      "kubernetes.io/hostname",
+									Operator: coreV1.NodeSelectorOpNotIn,
+									Values:   []string{pod.Spec.NodeName},
+								}},
+							}},
+						},
+					},
+				}
+
 				_, err := f.K8sResourceManagers.JobManager().CreateAndWaitTillJobCompleted(testJob)
+				logJobPodDiag(testJob.Name)
+				logTargetENIDiag(pod)
+
 				if shouldConnect {
 					By("verifying connection to pod succeeds on port " + strconv.Itoa(port))
 					Expect(err).ToNot(HaveOccurred())
@@ -116,6 +140,7 @@ var _ = Describe("Custom Networking Test", func() {
 				shouldConnect = true
 			})
 			It("should connect", func() {})
+
 		})
 
 		Context("when connecting to unreachable port", func() {
@@ -128,6 +153,49 @@ var _ = Describe("Custom Networking Test", func() {
 		})
 	})
 
+	Context("when a custom-networking pod connects to an external endpoint on an egress-allowed port", func() {
+		It("should succeed reaching 1.1.1.1:53", func() {
+			testJob := manifest.NewDefaultJobBuilder().
+				Container(manifest.NewNetCatAlpineContainer(f.Options.TestImageRegistry).
+					Command([]string{"nc"}).
+					Args([]string{"-z", "-v", "-w5", "1.1.1.1", "53"}).
+					Build()).
+				Name("external-reachability").
+				Parallelism(1).
+				Build()
+			_, err := f.K8sResourceManagers.JobManager().CreateAndWaitTillJobCompleted(testJob)
+			logJobPodDiag(testJob.Name)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(f.K8sResourceManagers.JobManager().DeleteAndWaitTillJobIsDeleted(testJob)).ToNot(HaveOccurred())
+		})
+	})
+
+	Context("when a custom-networking pod connects to the Kubernetes API ClusterIP", func() {
+		It("should reach the API server via ClusterIP", func() {
+			// Use the well-known kubernetes.default ClusterIP service which exists on every cluster.
+			// Resolve the ClusterIP dynamically to avoid hardcoding.
+			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+			defer cancel()
+
+			kubeSvc, err := f.K8sResourceManagers.ServiceManager().GetService(ctx, "default", "kubernetes")
+			Expect(err).ToNot(HaveOccurred())
+			clusterIP := kubeSvc.Spec.ClusterIP
+
+			testJob := manifest.NewDefaultJobBuilder().
+				Container(manifest.NewNetCatAlpineContainer(f.Options.TestImageRegistry).
+					Command([]string{"nc"}).
+					Args([]string{"-z", "-v", "-w5", clusterIP, "443"}).
+					Build()).
+				Name("k8s-api-clusterip").
+				Parallelism(1).
+				Build()
+			_, err = f.K8sResourceManagers.JobManager().CreateAndWaitTillJobCompleted(testJob)
+			logJobPodDiag(testJob.Name)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(f.K8sResourceManagers.JobManager().DeleteAndWaitTillJobIsDeleted(testJob)).ToNot(HaveOccurred())
+		})
+	})
+
 	Context("when creating deployment on nodes that do not have ENIConfig", func() {
 		JustBeforeEach(func() {
 			By("deleting ENIConfig for all availability zones")
@@ -158,7 +226,7 @@ var _ = Describe("Custom Networking Test", func() {
 
 			By("verifying deployment should not succeed")
 			deployment, err = f.K8sResourceManagers.DeploymentManager().
-				CreateAndWaitTillDeploymentIsReady(deployment, utils.DefaultDeploymentReadyTimeout)
+				CreateAndWaitTillDeploymentIsReady(deployment, utils.ShortDeploymentReadyTimeout)
 			Expect(err).To(HaveOccurred())
 
 			By("deleting the failed deployment")
@@ -208,3 +276,46 @@ var _ = Describe("Custom Networking Test", func() {
 		})
 	})
 })
+
+// logJobPodDiag prints pod name, node, IP, phase, and container logs for every
+// pod belonging to the given Job. Useful for post-mortem when a Job fails.
+func logJobPodDiag(jobName string) {
+	pods, err := f.K8sResourceManagers.PodManager().GetPodsWithLabelSelector("job-name", jobName)
+	if err != nil {
+		return
+	}
+	for _, p := range pods.Items {
+		logs, _ := f.K8sResourceManagers.PodManager().PodLogs(p.Namespace, p.Name)
+		fmt.Fprintf(GinkgoWriter, "[DIAG] pod=%s node=%s ip=%s phase=%s\n%s\n",
+			p.Name, p.Spec.NodeName, p.Status.PodIP, p.Status.Phase, logs)
+	}
+}
+
+// logTargetENIDiag looks up the EC2 ENI that owns the target pod's IP and
+// prints its ENI ID, subnet, and security groups.
+func logTargetENIDiag(pod coreV1.Pod) {
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
+	defer cancel()
+
+	node := &coreV1.Node{}
+	if err := f.K8sClient.Get(ctx, types.NamespacedName{Name: pod.Spec.NodeName}, node); err != nil {
+		return
+	}
+	instance, err := f.CloudServices.EC2().DescribeInstance(ctx, k8sUtils.GetInstanceIDFromNode(*node))
+	if err != nil {
+		return
+	}
+	for _, nic := range instance.NetworkInterfaces {
+		for _, pip := range nic.PrivateIpAddresses {
+			if pip.PrivateIpAddress != nil && *pip.PrivateIpAddress == pod.Status.PodIP {
+				var sgs []string
+				for _, g := range nic.Groups {
+					sgs = append(sgs, *g.GroupId)
+				}
+				fmt.Fprintf(GinkgoWriter, "[DIAG] target ENI=%s subnet=%s SGs=%v\n",
+					*nic.NetworkInterfaceId, *nic.SubnetId, sgs)
+				return
+			}
+		}
+	}
+}