add PD support (kubernetes-sigs#2137)

Author: M00nF1sh

go.mod

@@ -3,7 +3,7 @@ module sigs.k8s.io/aws-load-balancer-controller
 go 1.16
 
 require (
-	github.com/aws/aws-sdk-go v1.38.67
+	github.com/aws/aws-sdk-go v1.40.7
 	github.com/go-logr/logr v0.4.0
 	github.com/golang/mock v1.6.0
 	github.com/google/go-cmp v0.5.6

go.sum

@@ -99,15 +99,13 @@ func main() {
 
 	podInfoRepo := k8s.NewDefaultPodInfoRepo(clientSet.CoreV1().RESTClient(), rtOpts.Namespace, ctrl.Log)
 	finalizerManager := k8s.NewDefaultFinalizerManager(mgr.GetClient(), ctrl.Log)
-	podENIResolver := networking.NewDefaultPodENIInfoResolver(cloud.EC2(), cloud.VpcID(), ctrl.Log)
-	nodeENIResolver := networking.NewDefaultNodeENIInfoResolver(cloud.EC2(), ctrl.Log)
 	sgManager := networking.NewDefaultSecurityGroupManager(cloud.EC2(), ctrl.Log)
 	sgReconciler := networking.NewDefaultSecurityGroupReconciler(sgManager, ctrl.Log)
 	azInfoProvider := networking.NewDefaultAZInfoProvider(cloud.EC2(), ctrl.Log.WithName("az-info-provider"))
 	subnetResolver := networking.NewDefaultSubnetsResolver(azInfoProvider, cloud.EC2(), cloud.VpcID(), controllerCFG.ClusterName, ctrl.Log.WithName("subnets-resolver"))
 	vpcResolver := networking.NewDefaultVPCResolver(cloud.EC2(), cloud.VpcID(), ctrl.Log.WithName("vpc-resolver"))
-	tgbResManager := targetgroupbinding.NewDefaultResourceManager(mgr.GetClient(), cloud.ELBV2(),
-		podInfoRepo, podENIResolver, nodeENIResolver, sgManager, sgReconciler, cloud.VpcID(), controllerCFG.ClusterName, mgr.GetEventRecorderFor("targetGroupBinding"), ctrl.Log)
+	tgbResManager := targetgroupbinding.NewDefaultResourceManager(mgr.GetClient(), cloud.ELBV2(), cloud.EC2(),
+		podInfoRepo, sgManager, sgReconciler, cloud.VpcID(), controllerCFG.ClusterName, mgr.GetEventRecorderFor("targetGroupBinding"), ctrl.Log)
 	ingGroupReconciler := ingress.NewGroupReconciler(cloud, mgr.GetClient(), mgr.GetEventRecorderFor("ingress"),
 		finalizerManager, sgManager, sgReconciler, subnetResolver,
 		controllerCFG, ctrl.Log.WithName("controllers").WithName("ingress"))

main.go

@@ -21,6 +21,7 @@ type PodInfo struct {
 	ContainerPorts []corev1.ContainerPort
 	ReadinessGates []corev1.PodReadinessGate
 	Conditions     []corev1.PodCondition
+	NodeName       string
 	PodIP          string
 
 	ENIInfos []PodENIInfo
@@ -101,6 +102,7 @@ func buildPodInfo(pod *corev1.Pod) PodInfo {
 		ContainerPorts: containerPorts,
 		ReadinessGates: pod.Spec.ReadinessGates,
 		Conditions:     pod.Status.Conditions,
+		NodeName:       pod.Spec.NodeName,
 		PodIP:          pod.Status.PodIP,
 
 		ENIInfos: podENIInfos,

pkg/aws/services/ec2_mocks.go

@@ -327,6 +327,7 @@ func Test_buildPodInfo(t *testing.T) {
 						UID:       "pod-uuid",
 					},
 					Spec: corev1.PodSpec{
+						NodeName: "ip-192-168-13-198.us-west-2.compute.internal",
 						Containers: []corev1.Container{
 							{
 								Ports: []corev1.ContainerPort{
@@ -408,7 +409,8 @@ func Test_buildPodInfo(t *testing.T) {
 						Status: corev1.ConditionTrue,
 					},
 				},
-				PodIP: "192.168.1.1",
+				NodeName: "ip-192-168-13-198.us-west-2.compute.internal",
+				PodIP:    "192.168.1.1",
 			},
 		},
 		{
@@ -424,6 +426,7 @@ func Test_buildPodInfo(t *testing.T) {
 						},
 					},
 					Spec: corev1.PodSpec{
+						NodeName: "ip-192-168-13-198.us-west-2.compute.internal",
 						Containers: []corev1.Container{
 							{
 								Ports: []corev1.ContainerPort{
@@ -505,7 +508,8 @@ func Test_buildPodInfo(t *testing.T) {
 						Status: corev1.ConditionTrue,
 					},
 				},
-				PodIP: "192.168.1.1",
+				NodeName: "ip-192-168-13-198.us-west-2.compute.internal",
+				PodIP:    "192.168.1.1",
 				ENIInfos: []PodENIInfo{
 					{
 						ENIID:     "eni-06a712e1622fda4a0",

pkg/k8s/pod_info.go

@@ -9,8 +9,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/cache"
-	"k8s.io/apimachinery/pkg/util/sets"
-	"sigs.k8s.io/aws-load-balancer-controller/pkg/aws/services"
 	"sigs.k8s.io/aws-load-balancer-controller/pkg/k8s"
 	"sync"
 	"time"
@@ -26,9 +24,9 @@ type NodeENIInfoResolver interface {
 }
 
 // NewDefaultNodeENIInfoResolver constructs new defaultNodeENIInfoResolver.
-func NewDefaultNodeENIInfoResolver(ec2Client services.EC2, logger logr.Logger) *defaultNodeENIInfoResolver {
+func NewDefaultNodeENIInfoResolver(nodeInfoProvider NodeInfoProvider, logger logr.Logger) *defaultNodeENIInfoResolver {
 	return &defaultNodeENIInfoResolver{
-		ec2Client:             ec2Client,
+		nodeInfoProvider:      nodeInfoProvider,
 		logger:                logger,
 		nodeENIInfoCache:      cache.NewExpiring(),
 		nodeENIInfoCacheMutex: sync.RWMutex{},
@@ -40,8 +38,8 @@ var _ NodeENIInfoResolver = &defaultNodeENIInfoResolver{}
 
 // default implementation for NodeENIInfoResolver.
 type defaultNodeENIInfoResolver struct {
-	// ec2 client
-	ec2Client services.EC2
+	// nodeInfoProvider
+	nodeInfoProvider NodeInfoProvider
 	// logger
 	logger logr.Logger
 
@@ -53,11 +51,11 @@ type defaultNodeENIInfoResolver struct {
 func (r *defaultNodeENIInfoResolver) Resolve(ctx context.Context, nodes []*corev1.Node) (map[types.NamespacedName]ENIInfo, error) {
 	eniInfoByNodeKey := r.fetchENIInfosFromCache(nodes)
 	nodesWithoutENIInfo := computeNodesWithoutENIInfo(nodes, eniInfoByNodeKey)
-	eniInfoByNodeKeyViaLookup, err := r.resolveViaInstanceID(ctx, nodesWithoutENIInfo)
-	if err != nil {
-		return nil, err
-	}
-	if len(eniInfoByNodeKeyViaLookup) > 0 {
+	if len(nodesWithoutENIInfo) > 0 {
+		eniInfoByNodeKeyViaLookup, err := r.resolveViaInstanceID(ctx, nodesWithoutENIInfo)
+		if err != nil {
+			return nil, err
+		}
 		r.saveENIInfosToCache(nodesWithoutENIInfo, eniInfoByNodeKeyViaLookup)
 		for nodeKey, eniInfo := range eniInfoByNodeKeyViaLookup {
 			eniInfoByNodeKey[nodeKey] = eniInfo
@@ -113,39 +111,18 @@ func (r *defaultNodeENIInfoResolver) saveENIInfosToCache(nodes []*corev1.Node, e
 }
 
 func (r *defaultNodeENIInfoResolver) resolveViaInstanceID(ctx context.Context, nodes []*corev1.Node) (map[types.NamespacedName]ENIInfo, error) {
-	nodeKeysByInstanceID := make(map[string][]types.NamespacedName)
-	for _, node := range nodes {
-		instanceID, err := k8s.ExtractNodeInstanceID(node)
-		if err != nil {
-			return nil, err
-		}
-		nodeKey := k8s.NamespacedName(node)
-		nodeKeysByInstanceID[instanceID] = append(nodeKeysByInstanceID[instanceID], nodeKey)
-	}
-	if len(nodeKeysByInstanceID) == 0 {
-		return nil, nil
-	}
-
-	instanceIDs := sets.StringKeySet(nodeKeysByInstanceID).List()
-	req := &ec2sdk.DescribeInstancesInput{
-		InstanceIds: awssdk.StringSlice(instanceIDs),
-	}
-	instances, err := r.ec2Client.DescribeInstancesAsList(ctx, req)
+	nodeInstanceByNodeKey, err := r.nodeInfoProvider.FetchNodeInstances(ctx, nodes)
 	if err != nil {
 		return nil, err
 	}
-
-	eniInfoByNodeKey := make(map[types.NamespacedName]ENIInfo)
-	for _, instance := range instances {
-		instanceID := awssdk.StringValue(instance.InstanceId)
-		primaryENI, err := findInstancePrimaryENI(instance.NetworkInterfaces)
+	eniInfoByNodeKey := make(map[types.NamespacedName]ENIInfo, len(nodeInstanceByNodeKey))
+	for nodeKey, nodeInstance := range nodeInstanceByNodeKey {
+		primaryENI, err := findInstancePrimaryENI(nodeInstance.NetworkInterfaces)
 		if err != nil {
 			return nil, err
 		}
 		eniInfo := buildENIInfoViaInstanceENI(primaryENI)
-		for _, nodeKey := range nodeKeysByInstanceID[instanceID] {
-			eniInfoByNodeKey[nodeKey] = eniInfo
-		}
+		eniInfoByNodeKey[nodeKey] = eniInfo
 	}
 	return eniInfoByNodeKey, nil
 }