tags: initial implementation of tags

Author: nicolehanjing

Diff for: pkg/providers/v2/cloud.go

@@ -29,13 +29,19 @@ import (
 	"github.com/aws/aws-sdk-go/aws/ec2metadata"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/ec2"
+	"gopkg.in/gcfg.v1"
 
 	cloudprovider "k8s.io/cloud-provider"
 )
 
 func init() {
 	cloudprovider.RegisterCloudProvider(ProviderName, func(config io.Reader) (cloudprovider.Interface, error) {
-		return newCloud()
+		cfg, err := readAWSCloudConfig(config)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read AWS cloud provider config file: %v", err)
+		}
+
+		return newCloud(*cfg)
 	})
 }
 
@@ -53,6 +59,22 @@ type cloud struct {
 	region    string
 	ec2       EC2
 	metadata  EC2Metadata
+	tagging   awsTagging
+}
+
+// AWSCloudConfig wraps the settings for the AWS cloud provider.
+type AWSCloudConfig struct {
+	Config struct {
+		// ClusterName is the cluster name we'll use to identify our cluster resources
+		ClusterName string
+	}
+}
+
+// EC2 is an interface defining only the methods we call from the AWS EC2 SDK.
+type EC2 interface {
+	DescribeInstances(request *ec2.DescribeInstancesInput) (*ec2.DescribeInstancesOutput, error)
+
+	CreateTags(*ec2.CreateTagsInput) (*ec2.CreateTagsOutput, error)
 }
 
 // EC2Metadata is an abstraction over the AWS metadata service.
@@ -61,6 +83,20 @@ type EC2Metadata interface {
 	GetMetadata(path string) (string, error)
 }
 
+func readAWSCloudConfig(config io.Reader) (*AWSCloudConfig, error) {
+	if config == nil {
+		return nil, fmt.Errorf("no AWS cloud provider config file given")
+	}
+
+	var cfg AWSCloudConfig
+	err := gcfg.ReadInto(&cfg, config)
+	if err != nil {
+		return nil, err
+	}
+
+	return &cfg, nil
+}
+
 func getAvailabilityZone(metadata EC2Metadata) (string, error) {
 	return metadata.GetMetadata("placement/availability-zone")
 }
@@ -82,7 +118,7 @@ func azToRegion(az string) (string, error) {
 }
 
 // newCloud creates a new instance of AWSCloud.
-func newCloud() (cloudprovider.Interface, error) {
+func newCloud(cfg AWSCloudConfig) (cloudprovider.Interface, error) {
 	sess, err := session.NewSession(&aws.Config{})
 	if err != nil {
 		return nil, fmt.Errorf("unable to initialize AWS session: %v", err)
@@ -138,6 +174,12 @@ func newCloud() (cloudprovider.Interface, error) {
 		ec2:       ec2Service,
 	}
 
+	if cfg.Config.ClusterName != "" {
+		if err := awsCloud.tagging.init(cfg.Config.ClusterName); err != nil {
+			return nil, err
+		}
+	}
+
 	return awsCloud, nil
 }
 
@@ -177,7 +219,7 @@ func (c *cloud) Routes() (cloudprovider.Routes, bool) {
 
 // HasClusterID returns true if the cluster has a clusterID
 func (c *cloud) HasClusterID() bool {
-	return false
+	return len(c.tagging.clusterName()) > 0
 }
 
 // InstancesV2 is an implementation for instances and should only be implemented by external cloud providers.

Diff for: pkg/providers/v2/instances.go

@@ -59,11 +59,6 @@ func newInstances(az string, creds *credentials.Credentials) (cloudprovider.Inst
 	}, nil
 }
 
-// EC2 is an interface defining only the methods we call from the AWS EC2 SDK.
-type EC2 interface {
-	DescribeInstances(request *ec2.DescribeInstancesInput) (*ec2.DescribeInstancesOutput, error)
-}
-
 // instances is an implementation of cloudprovider.InstancesV2
 type instances struct {
 	availabilityZone string

Diff for: pkg/providers/v2/mocks/mock_ec2.go

@@ -0,0 +1,211 @@
+/*
+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 (
+	"fmt"
+	"strings"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/klog/v2"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+)
+
+const (
+	// TagNameKubernetesClusterPrefix is the tag name we use to differentiate multiple
+	// logically independent clusters running in the same AZ.
+	// we support two tags:
+	// 1) kubernetes.io/cluster/=<clusterName>
+	// 2) for shared resoueces between clusters, kubernetes.io/cluster/<clusterName>=""
+	TagNameKubernetesClusterPrefix = "kubernetes.io/cluster/"
+
+	// ResourceLifecycleShared is the value we use when tagging resources to indicate
+	// that the resource is shared between multiple clusters, and should not be destroyed
+	// if the cluster is destroyed.
+	ResourceLifecycleShared = "shared"
+
+	// createTag* is configuration of exponential backoff for CreateTag call. We
+	// retry mainly because if we create an object, we cannot tag it until it is
+	// "fully created" (eventual consistency). Starting with 1 second, doubling
+	// it every step and taking 9 steps results in 255 second total waiting
+	// time.
+	createTagInitialDelay = 1 * time.Second
+	createTagFactor       = 2.0
+	createTagSteps        = 9
+)
+
+type awsTagging struct {
+	// ClusterName is our cluster identifier: we tag AWS resources with this value,
+	// and thus we can run two independent clusters in the same VPC or subnets.
+	ClusterName string
+
+	// resourceShared is true if the resource is shared between multiple clusters
+	resourceShared bool
+}
+
+// Extracts the cluster name from the given tags, if they are present
+// If duplicate tags are found, returns an error
+func findClusterName(tags []*ec2.Tag) (string, error) {
+	clusterName := ""
+
+	for _, tag := range tags {
+		tagKey := aws.StringValue(tag.Key)
+		if strings.HasPrefix(tagKey, TagNameKubernetesClusterPrefix) {
+			name := aws.StringValue(tag.Value)
+			if name == "" {
+				name = strings.TrimPrefix(tagKey, TagNameKubernetesClusterPrefix)
+			}
+
+			if clusterName != "" {
+				return "", fmt.Errorf("Found multiple cluster tags with prefix %s (%q and %q)", TagNameKubernetesClusterPrefix, clusterName, name)
+			}
+			clusterName = name
+		}
+	}
+
+	return clusterName, nil
+}
+
+func (t *awsTagging) init(clusterName string) error {
+	t.ClusterName = clusterName
+
+	if clusterName != "" {
+		klog.Infof("AWS cloud filtering on ClusterName: %v", clusterName)
+	} else {
+		return fmt.Errorf("AWS cloud failed to find ClusterName")
+	}
+
+	return nil
+}
+
+// Extracts a cluster name from the given tags, if one is present
+// If no clusterName is found, returns "", nil
+// If multiple (different) clusterNames are found, returns an error
+func (t *awsTagging) initFromTags(tags []*ec2.Tag) error {
+	clusterName, err := findClusterName(tags)
+	if err != nil {
+		return err
+	}
+
+	if clusterName == "" {
+		klog.Errorf("Tag %q not found; Kubernetes may behave unexpectedly.", TagNameKubernetesClusterPrefix)
+	}
+
+	return t.init(clusterName)
+}
+
+func (t *awsTagging) hasClusterTag(tags []*ec2.Tag) bool {
+	// if the clusterName is not configured -- we consider all instances.
+	if len(t.ClusterName) == 0 {
+		return true
+	}
+
+	for _, tag := range tags {
+		tagKey := aws.StringValue(tag.Key)
+		if (tagKey == TagNameKubernetesClusterPrefix) && (aws.StringValue(tag.Value) == t.ClusterName) {
+			return true
+		}
+
+		// for shared resources
+		if tagKey == (TagNameKubernetesClusterPrefix + t.ClusterName) {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (t *awsTagging) buildTags(additionalTags map[string]string, lifecycle string) map[string]string {
+	tags := make(map[string]string)
+	for k, v := range additionalTags {
+		tags[k] = v
+	}
+
+	// no clusterName is a sign of misconfigured cluster, but we can't be tagging the resources with empty
+	// strings
+	if len(t.ClusterName) == 0 {
+		return tags
+	}
+
+	// we support two tags:
+	// 1) kubernetes.io/cluster/=<clusterName>
+	// 2) for shared resoueces between clusters, kubernetes.io/cluster/<clusterName>=""
+	if lifecycle == ResourceLifecycleShared {
+		tags[TagNameKubernetesClusterPrefix+t.ClusterName] = ""
+	} else {
+		tags[TagNameKubernetesClusterPrefix] = t.ClusterName
+	}
+
+	return tags
+}
+
+// createTags calls EC2 CreateTags, but adds retry-on-failure logic
+// We retry mainly because if we create an object, we cannot tag it until it is "fully created" (eventual consistency)
+// The error code varies though (depending on what we are tagging), so we simply retry on all errors
+func (t *awsTagging) createTags(ec2Client EC2, resourceID string, lifecycle string, additionalTags map[string]string) error {
+	tags := t.buildTags(additionalTags, lifecycle)
+
+	if tags == nil || len(tags) == 0 {
+		return nil
+	}
+
+	var awsTags []*ec2.Tag
+	for k, v := range tags {
+		tag := &ec2.Tag{
+			Key:   aws.String(k),
+			Value: aws.String(v),
+		}
+		awsTags = append(awsTags, tag)
+	}
+
+	backoff := wait.Backoff{
+		Duration: createTagInitialDelay,
+		Factor:   createTagFactor,
+		Steps:    createTagSteps,
+	}
+
+	request := &ec2.CreateTagsInput{
+		Resources: []*string{&resourceID},
+		Tags:      awsTags,
+	}
+
+	var lastErr error
+	err := wait.ExponentialBackoff(backoff, func() (bool, error) {
+		_, err := ec2Client.CreateTags(request)
+		if err == nil {
+			return true, nil
+		}
+
+		// We could check that the error is retryable, but the error code changes based on what we are tagging
+		// SecurityGroup: InvalidGroup.NotFound
+		klog.V(2).Infof("Failed to create tags; will retry.  Error was %q", err)
+		lastErr = err
+		return false, nil
+	})
+	if err == wait.ErrWaitTimeout {
+		// return real CreateTags error instead of timeout
+		err = lastErr
+	}
+
+	return err
+}
+
+func (t *awsTagging) clusterName() string {
+	return t.ClusterName
+}