zones: initial implementation of Zones interface

pkg/providers/v2/cloud.go

@@ -53,6 +53,7 @@ type cloud struct {
 	region    string
 	ec2       EC2
 	metadata  EC2Metadata
+	zones     cloudprovider.Zones
 }
 
 // EC2Metadata is an abstraction over the AWS metadata service.
@@ -117,6 +118,11 @@ func newCloud() (cloudprovider.Interface, error) {
 		return nil, err
 	}
 
+	zones, err := newZones(az, creds)
+	if err != nil {
+		return nil, err
+	}
+
 	ec2Sess, err := session.NewSession(&aws.Config{
 		Region:      aws.String(region),
 		Credentials: creds,
@@ -136,6 +142,7 @@ func newCloud() (cloudprovider.Interface, error) {
 		region:    region,
 		metadata:  metadataClient,
 		ec2:       ec2Service,
+		zones:     zones,
 	}
 
 	return awsCloud, nil
@@ -167,7 +174,7 @@ func (c *cloud) Instances() (cloudprovider.Instances, bool) {
 
 // Zones returns an implementation of Zones for Amazon Web Services.
 func (c *cloud) Zones() (cloudprovider.Zones, bool) {
-	return nil, false
+	return c.zones, true
 }
 
 // Routes returns an implementation of Routes for Amazon Web Services.

pkg/providers/v2/instances.go

@@ -30,6 +30,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/ec2"
 
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
 	cloudprovider "k8s.io/cloud-provider"
 )
 
@@ -138,32 +139,79 @@ func (i *instances) InstanceMetadata(ctx context.Context, node *v1.Node) (*cloud
 // getInstance returns the instance if the instance with the given node info still exists.
 // If false an error will be returned, the instance will be immediately deleted by the cloud controller manager.
 func (i *instances) getInstance(ctx context.Context, node *v1.Node) (*ec2.Instance, error) {
-	var request *ec2.DescribeInstancesInput
 	if node.Spec.ProviderID == "" {
-		// get Instance by private DNS name
-		request = &ec2.DescribeInstancesInput{
-			Filters: []*ec2.Filter{
-				newEc2Filter("private-dns-name", node.Name),
-			},
-		}
+		// get Instance by node name
 		klog.V(4).Infof("looking for node by private DNS name %v", node.Name)
-	} else {
-		// get Instance by provider ID
-		instanceID, err := parseInstanceIDFromProviderID(node.Spec.ProviderID)
+		return getInstanceByPrivateDNSName(ctx, types.NodeName(node.Name), i.ec2)
+	}
+
+	// get Instance by provider ID
+	klog.V(4).Infof("looking for node by provider ID %v", node.Spec.ProviderID)
+	return getInstanceByProviderID(ctx, node.Spec.ProviderID, i.ec2)
+}
+
+// getInstanceByPrivateDNSName returns the instance if the instance with the given nodeName exists.
+// If false an error will be returned, the instance will be immediately deleted by the cloud controller manager.
+func getInstanceByPrivateDNSName(ctx context.Context, nodeName types.NodeName, ec2Client EC2) (*ec2.Instance, error) {
+	request := &ec2.DescribeInstancesInput{
+		Filters: []*ec2.Filter{
+			newEc2Filter("private-dns-name", string(nodeName)),
+		},
+	}
+	klog.V(4).Infof("looking for node by private DNS name %v", nodeName)
+
+	instances := []*ec2.Instance{}
+	var nextToken *string
+	for {
+		response, err := ec2Client.DescribeInstances(request)
 		if err != nil {
-			return nil, err
+			return nil, fmt.Errorf("error describing ec2 instances: %v", err)
+		}
+
+		for _, reservation := range response.Reservations {
+			instances = append(instances, reservation.Instances...)
 		}
 
-		request = &ec2.DescribeInstancesInput{
-			InstanceIds: []*string{aws.String(instanceID)},
+		nextToken = response.NextToken
+		if aws.StringValue(nextToken) == "" {
+			break
 		}
-		klog.V(4).Infof("looking for node by provider ID %v", node.Spec.ProviderID)
+		request.NextToken = nextToken
+	}
+
+	if len(instances) == 0 {
+		return nil, cloudprovider.InstanceNotFound
+	}
+
+	if len(instances) > 1 {
+		return nil, fmt.Errorf("getInstance: multiple instances found")
+	}
+
+	state := instances[0].State.Name
+	if *state == ec2.InstanceStateNameTerminated {
+		return nil, fmt.Errorf("instance %v is terminated", instances[0].InstanceId)
+	}
+
+	return instances[0], nil
+}
+
+// getInstanceByProviderID returns the instance if the instance with the given providerID exists.
+// If false an error will be returned, the instance will be immediately deleted by the cloud controller manager.
+func getInstanceByProviderID(ctx context.Context, providerID string, ec2Client EC2) (*ec2.Instance, error) {
+	instanceID, err := parseInstanceIDFromProviderID(providerID)
+	if err != nil {
+		return nil, err
+	}
+
+	request := &ec2.DescribeInstancesInput{
+		InstanceIds: []*string{aws.String(instanceID)},
 	}
+	klog.V(4).Infof("looking for node by provider ID %v", providerID)
 
 	instances := []*ec2.Instance{}
 	var nextToken *string
 	for {
-		response, err := i.ec2.DescribeInstances(request)
+		response, err := ec2Client.DescribeInstances(request)
 		if err != nil {
 			return nil, fmt.Errorf("error describing ec2 instances: %v", err)
 		}

pkg/providers/v2/zones.go

@@ -0,0 +1,112 @@
+/*
+Copyright 2020 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+// Package v2 is an out-of-tree only implementation of the AWS cloud provider.
+// It is not compatible with v1 and should only be used on new clusters.
+package v2
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"k8s.io/apimachinery/pkg/types"
+	cloudprovider "k8s.io/cloud-provider"
+)
+
+// newZones returns an implementation of cloudprovider.Zones
+// TODO:
+// We should add zones / region support via InstancesV2 since kubernetes/kubernetes#93569 was merged in v1.20, where zone/region is just added to InstanceMetadata and implemented as part of InstancesV2
+func newZones(az string, creds *credentials.Credentials) (cloudprovider.Zones, error) {
+	region, err := azToRegion(az)
+	if err != nil {
+		return nil, err
+	}
+
+	awsConfig := &aws.Config{
+		Region:      aws.String(region),
+		Credentials: creds,
+	}
+	awsConfig = awsConfig.WithCredentialsChainVerboseErrors(true)
+
+	sess, err := session.NewSession(awsConfig)
+	if err != nil {
+		return nil, fmt.Errorf("error creating new session: %v", err)
+	}
+	ec2Service := ec2.New(sess)
+
+	return &zones{
+		availabilityZone: az,
+		ec2:              ec2Service,
+		region:           region,
+	}, nil
+}
+
+// zones is an implementation of cloudprovider.Zones
+type zones struct {
+	availabilityZone string
+	ec2              EC2
+	region           string
+}
+
+// GetZone returns the Zone containing the current failure zone and locality region that the program is running in
+func (z *zones) GetZone(ctx context.Context) (cloudprovider.Zone, error) {
+	return cloudprovider.Zone{
+		FailureDomain: z.availabilityZone,
+		Region:        z.region,
+	}, nil
+}
+
+// GetZoneByProviderID returns the Zone containing the current zone and locality region of the node specified by providerID
+func (z *zones) GetZoneByProviderID(ctx context.Context, providerID string) (cloudprovider.Zone, error) {
+	instance, err := getInstanceByProviderID(ctx, providerID, z.ec2)
+	if err != nil {
+		return cloudprovider.Zone{}, err
+	}
+
+	az := instance.Placement.AvailabilityZone
+	regionName, err := azToRegion(*az)
+	if err != nil {
+		return cloudprovider.Zone{}, err
+	}
+
+	return cloudprovider.Zone{
+		FailureDomain: *az,
+		Region:        regionName,
+	}, nil
+}
+
+// GetZoneByNodeName returns the Zone containing the current zone and locality region of the node specified by node name
+func (z *zones) GetZoneByNodeName(ctx context.Context, nodeName types.NodeName) (cloudprovider.Zone, error) {
+	instance, err := getInstanceByPrivateDNSName(ctx, nodeName, z.ec2)
+	if err != nil {
+		return cloudprovider.Zone{}, err
+	}
+
+	az := instance.Placement.AvailabilityZone
+	regionName, err := azToRegion(*az)
+	if err != nil {
+		return cloudprovider.Zone{}, err
+	}
+
+	return cloudprovider.Zone{
+		FailureDomain: *az,
+		Region:        regionName,
+	}, nil
+}