Skip to content

Commit 3beba2b

Browse files
mselim00mselim00
committed
feat: add tests to validate EFA
1 parent 941030d commit 3beba2b

8 files changed

Lines changed: 700 additions & 0 deletions

File tree

‎.github/workflows/ci.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,11 @@ jobs:
2222
steps:
2323
- uses: actions/checkout@v3
2424
- run: docker build --build-arg=KUBERNETES_MINOR_VERSION=latest --file Dockerfile .
25+
build-image-efa:
26+
runs-on: ubuntu-latest
27+
steps:
28+
- uses: actions/checkout@v3
29+
- run: docker build --file test/images/efa/Dockerfile .
2530
build-image-neuronx:
2631
runs-on: ubuntu-latest
2732
steps:

‎internal/e2e/ec2.go‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
package e2e
2+
3+
import (
4+
"context"
5+
"fmt"
6+
7+
"github.com/aws/aws-k8s-tester/internal/awssdk"
8+
"github.com/aws/aws-sdk-go-v2/service/ec2"
9+
ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
10+
)
11+
12+
type EC2Client interface {
13+
DescribeInstanceType(instanceType string) (ec2types.InstanceTypeInfo, error)
14+
}
15+
16+
type ec2Client struct {
17+
client *ec2.Client
18+
}
19+
20+
func NewEC2Client() *ec2Client {
21+
return &ec2Client{
22+
client: ec2.NewFromConfig(awssdk.NewConfig()),
23+
}
24+
}
25+
26+
func (c *ec2Client) DescribeInstanceType(instanceType string) (ec2types.InstanceTypeInfo, error) {
27+
describeResponse, err := c.client.DescribeInstanceTypes(context.TODO(), &ec2.DescribeInstanceTypesInput{
28+
InstanceTypes: []ec2types.InstanceType{ec2types.InstanceType(instanceType)},
29+
})
30+
if err != nil {
31+
return ec2types.InstanceTypeInfo{}, fmt.Errorf("failed to describe instance type: %s: %v", instanceType, err)
32+
} else {
33+
return describeResponse.InstanceTypes[0], nil
34+
}
35+
}

‎test/cases/efa/commons.go‎

Lines changed: 111 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
//go:build e2e
2+
3+
package efa
4+
5+
import (
6+
"context"
7+
_ "embed"
8+
"fmt"
9+
"log"
10+
11+
"github.com/aws/aws-k8s-tester/internal/e2e"
12+
"github.com/aws/aws-sdk-go-v2/aws"
13+
corev1 "k8s.io/api/core/v1"
14+
v1 "k8s.io/api/core/v1"
15+
"k8s.io/client-go/kubernetes"
16+
"sigs.k8s.io/e2e-framework/pkg/env"
17+
"sigs.k8s.io/e2e-framework/pkg/envconf"
18+
19+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
20+
)
21+
22+
var (
23+
testenv env.Environment
24+
ec2Client e2e.EC2Client
25+
26+
testImage *string
27+
28+
pingPongSize *string
29+
pingPongIters *int
30+
pingPongDeadlineSeconds *int
31+
32+
nodeType *string
33+
expectedEFADeviceCount *int
34+
35+
verbose *bool
36+
)
37+
38+
const (
39+
EFA_RESOURCE_NAME = "vpc.amazonaws.com/efa"
40+
TEST_NAMESPACE_NAME = "efa-tests"
41+
)
42+
43+
func getEfaCapacity(node corev1.Node) int {
44+
capacity, ok := node.Status.Capacity[v1.ResourceName(EFA_RESOURCE_NAME)]
45+
if !ok {
46+
return 0
47+
}
48+
return int(capacity.Value())
49+
}
50+
51+
func getEfaNodes(ctx context.Context, config *envconf.Config) ([]corev1.Node, error) {
52+
var efaNodes []corev1.Node
53+
clientset, err := kubernetes.NewForConfig(config.Client().RESTConfig())
54+
if err != nil {
55+
return []corev1.Node{}, fmt.Errorf("failed to create Kubernetes client: %w", err)
56+
}
57+
58+
nodes, err := clientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
59+
if err != nil {
60+
return []corev1.Node{}, fmt.Errorf("failed to list nodes: %w", err)
61+
}
62+
63+
if len(nodes.Items) == 0 {
64+
return []corev1.Node{}, fmt.Errorf("no nodes found in the cluster")
65+
}
66+
67+
for _, node := range nodes.Items {
68+
instanceType := node.Labels["node.kubernetes.io/instance-type"]
69+
70+
if aws.ToString(nodeType) != "" && instanceType != aws.ToString(nodeType) {
71+
log.Printf("[INFO] Skipping node %s, node is not of target type %s", node.Name, aws.ToString(nodeType))
72+
continue
73+
}
74+
75+
numEfaDevices, err := e2e.GetNonZeroResourceCapacity(&node, EFA_RESOURCE_NAME)
76+
if err != nil {
77+
log.Printf("[INFO] Skipping node %s: %v", node.Name, err)
78+
continue
79+
}
80+
81+
allocatable, ok := node.Status.Allocatable[v1.ResourceName(EFA_RESOURCE_NAME)]
82+
if !ok {
83+
log.Printf("[INFO] Skipping node %s, node does not have EFA allocatable", node.Name)
84+
continue
85+
} else if int(allocatable.Value()) != numEfaDevices {
86+
log.Printf("[INFO] Skipping node %s, node is busy (%d of %d EFA device(s) allocatable)", node.Name, allocatable.Value(), numEfaDevices)
87+
continue
88+
}
89+
90+
expectedDeviceCount := aws.ToInt(expectedEFADeviceCount)
91+
if expectedDeviceCount == 0 {
92+
instanceInfo, err := ec2Client.DescribeInstanceType(instanceType)
93+
if err != nil {
94+
return []corev1.Node{}, err
95+
}
96+
expectedDeviceCount = int(aws.ToInt32(instanceInfo.NetworkInfo.EfaInfo.MaximumEfaInterfaces))
97+
}
98+
99+
if expectedDeviceCount != numEfaDevices {
100+
return []corev1.Node{}, fmt.Errorf("unexpected EFA device capacity on node %s: expected %d, got %d", node.Name, expectedDeviceCount, numEfaDevices)
101+
}
102+
103+
efaNodes = append(efaNodes, node)
104+
}
105+
106+
if len(efaNodes) == 0 {
107+
return []corev1.Node{}, fmt.Errorf("no nodes with EFA capacity found in the cluster")
108+
}
109+
110+
return efaNodes, nil
111+
}

‎test/cases/efa/main_test.go‎

Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
//go:build e2e
2+
3+
package efa
4+
5+
import (
6+
"context"
7+
_ "embed"
8+
"flag"
9+
"fmt"
10+
"log"
11+
"os"
12+
"os/signal"
13+
"testing"
14+
"time"
15+
16+
"github.com/aws/aws-k8s-tester/internal/e2e"
17+
"github.com/aws/aws-k8s-tester/test/manifests"
18+
appsv1 "k8s.io/api/apps/v1"
19+
corev1 "k8s.io/api/core/v1"
20+
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
21+
"sigs.k8s.io/e2e-framework/klient/wait"
22+
"sigs.k8s.io/e2e-framework/pkg/env"
23+
"sigs.k8s.io/e2e-framework/pkg/envconf"
24+
)
25+
26+
func getTestNamespace() *corev1.Namespace {
27+
return &corev1.Namespace{
28+
ObjectMeta: metav1.ObjectMeta{
29+
Name: TEST_NAMESPACE_NAME,
30+
},
31+
}
32+
}
33+
34+
func deployTestNamespace(ctx context.Context, config *envconf.Config) (context.Context, error) {
35+
36+
return ctx, nil
37+
}
38+
39+
func deployEFAPlugin(ctx context.Context, config *envconf.Config) (context.Context, error) {
40+
err := e2e.ApplyManifests(config.Client().RESTConfig(), manifests.EfaDevicePluginManifest)
41+
if err != nil {
42+
return ctx, err
43+
}
44+
45+
ds := appsv1.DaemonSet{
46+
ObjectMeta: metav1.ObjectMeta{Name: "aws-efa-k8s-device-plugin-daemonset", Namespace: "kube-system"},
47+
}
48+
err = wait.For(e2e.NewConditionExtension(config.Client().Resources()).DaemonSetReady(&ds),
49+
wait.WithContext(ctx))
50+
if err != nil {
51+
return ctx, fmt.Errorf("failed to deploy efa-device-plugin: %v", err)
52+
}
53+
54+
return ctx, nil
55+
}
56+
57+
func TestMain(m *testing.M) {
58+
testImage = flag.String("testImage", "", "container image to use for tests")
59+
pingPongSize = flag.String("pingPongSize", "all", "sizes to use for ping pong")
60+
pingPongIters = flag.Int("pingPongIters", 10000, "sizes to use for ping pong")
61+
pingPongDeadlineSeconds = flag.Int("pingPongDeadlineSeconds", 120, "maximum run time for a ping pong attempt")
62+
nodeType = flag.String("nodeType", "", "instance type to target for tests")
63+
expectedEFADeviceCount = flag.Int("expectedEFADeviceCount", 0, "expected number of efa devices for the target nodes")
64+
verbose = flag.Bool("verbose", true, "use verbose mode for tests")
65+
66+
cfg, err := envconf.NewFromFlags()
67+
if err != nil {
68+
log.Fatalf("failed to initialize test environment: %v", err)
69+
}
70+
71+
if *testImage == "" {
72+
log.Fatal("--testImage must be set, use https://github.com/aws/aws-k8s-tester/blob/main/test/efa/Dockerfile to build the image")
73+
}
74+
75+
ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
76+
timedCtx, _ := context.WithTimeout(ctx, 55*time.Minute)
77+
defer cancel()
78+
79+
testenv = env.NewWithConfig(cfg)
80+
testenv = testenv.WithContext(timedCtx)
81+
82+
ec2Client = e2e.NewEC2Client()
83+
84+
testenv.Setup(
85+
func(ctx context.Context, config *envconf.Config) (context.Context, error) {
86+
err := e2e.ApplyManifests(config.Client().RESTConfig(), manifests.EfaDevicePluginManifest)
87+
if err != nil {
88+
return ctx, err
89+
}
90+
efaDS := appsv1.DaemonSet{
91+
ObjectMeta: metav1.ObjectMeta{Name: "aws-efa-k8s-device-plugin-daemonset", Namespace: "kube-system"},
92+
}
93+
err = wait.For(e2e.NewConditionExtension(config.Client().Resources()).DaemonSetReady(&efaDS),
94+
wait.WithContext(ctx),
95+
wait.WithTimeout(5*time.Minute),
96+
)
97+
if err != nil {
98+
return ctx, err
99+
}
100+
101+
select {
102+
case <-ctx.Done():
103+
// Cooldown to let device plugin update node object with resources
104+
case <-time.After(15 * time.Second):
105+
}
106+
107+
return ctx, cfg.Client().Resources().Create(ctx, getTestNamespace())
108+
},
109+
)
110+
111+
testenv.Finish(
112+
func(ctx context.Context, config *envconf.Config) (context.Context, error) {
113+
cfg.Client().Resources().Delete(context.TODO(), getTestNamespace())
114+
err := e2e.DeleteManifests(cfg.Client().RESTConfig(), manifests.EfaDevicePluginManifest)
115+
if err != nil {
116+
return ctx, err
117+
}
118+
return ctx, nil
119+
},
120+
)
121+
122+
os.Exit(testenv.Run(m))
123+
}

0 commit comments

Comments
 (0)