Merge pull request #41695 from justinsb/shared_tag

pkg/cloudprovider/providers/aws/BUILD

@@ -21,6 +21,7 @@ go_library(
         "regions.go",
         "retry_handler.go",
         "sets_ippermissions.go",
+        "tags.go",
         "volumes.go",
     ],
     tags = ["automanaged"],
@@ -56,6 +57,7 @@ go_test(
         "device_allocator_test.go",
         "regions_test.go",
         "retry_handler_test.go",
+        "tags_test.go",
     ],
     library = ":go_default_library",
     tags = ["automanaged"],

pkg/cloudprovider/providers/aws/aws_loadbalancer.go

@@ -55,9 +55,14 @@ func (c *Cloud) ensureLoadBalancer(namespacedName types.NamespacedName, loadBala
 
 		createRequest.SecurityGroups = stringPointerArray(securityGroupIDs)
 
-		createRequest.Tags = []*elb.Tag{
-			{Key: aws.String(TagNameKubernetesCluster), Value: aws.String(c.getClusterName())},
-			{Key: aws.String(TagNameKubernetesService), Value: aws.String(namespacedName.String())},
+		tags := c.tagging.buildTags(ResourceLifecycleOwned, map[string]string{
+			TagNameKubernetesService: namespacedName.String(),
+		})
+
+		for k, v := range tags {
+			createRequest.Tags = append(createRequest.Tags, &elb.Tag{
+				Key: aws.String(k), Value: aws.String(v),
+			})
 		}
 
 		glog.Infof("Creating load balancer for %v with name: %s", namespacedName, loadBalancerName)

pkg/cloudprovider/providers/aws/aws_routes.go

@@ -29,14 +29,20 @@ func (c *Cloud) findRouteTable(clusterName string) (*ec2.RouteTable, error) {
 	// This should be unnecessary (we already filter on TagNameKubernetesCluster,
 	// and something is broken if cluster name doesn't match, but anyway...
 	// TODO: All clouds should be cluster-aware by default
-	filters := []*ec2.Filter{newEc2Filter("tag:"+TagNameKubernetesCluster, clusterName)}
-	request := &ec2.DescribeRouteTablesInput{Filters: c.addFilters(filters)}
+	request := &ec2.DescribeRouteTablesInput{Filters: c.tagging.addFilters(nil)}
 
-	tables, err := c.ec2.DescribeRouteTables(request)
+	response, err := c.ec2.DescribeRouteTables(request)
 	if err != nil {
 		return nil, err
 	}
 
+	var tables []*ec2.RouteTable
+	for _, table := range response {
+		if c.tagging.hasClusterTag(table.Tags) {
+			tables = append(tables, table)
+		}
+	}
+
 	if len(tables) == 0 {
 		return nil, fmt.Errorf("unable to find route table for AWS cluster: %s", clusterName)
 	}

pkg/cloudprovider/providers/aws/aws_test.go

@@ -146,7 +146,7 @@ func NewFakeAWSServices() *FakeAWSServices {
 	s.instances = []*ec2.Instance{selfInstance}
 
 	var tag ec2.Tag
-	tag.Key = aws.String(TagNameKubernetesCluster)
+	tag.Key = aws.String(TagNameKubernetesClusterLegacy)
 	tag.Value = aws.String(TestClusterId)
 	selfInstance.Tags = []*ec2.Tag{&tag}
 
@@ -177,24 +177,6 @@ func (s *FakeAWSServices) Metadata() (EC2Metadata, error) {
 	return s.metadata, nil
 }
 
-func TestFilterTags(t *testing.T) {
-	awsServices := NewFakeAWSServices()
-	c, err := newAWSCloud(strings.NewReader("[global]"), awsServices)
-	if err != nil {
-		t.Errorf("Error building aws cloud: %v", err)
-		return
-	}
-
-	if len(c.filterTags) != 1 {
-		t.Errorf("unexpected filter tags: %v", c.filterTags)
-		return
-	}
-
-	if c.filterTags[TagNameKubernetesCluster] != TestClusterId {
-		t.Errorf("unexpected filter tags: %v", c.filterTags)
-	}
-}
-
 func TestNewAWSCloud(t *testing.T) {
 	tests := []struct {
 		name string
@@ -279,6 +261,15 @@ func instanceMatchesFilter(instance *ec2.Instance, filter *ec2.Filter) bool {
 		return contains(filter.Values, *instance.State.Name)
 	}
 
+	if name == "tag-key" {
+		for _, instanceTag := range instance.Tags {
+			if contains(filter.Values, aws.StringValue(instanceTag.Key)) {
+				return true
+			}
+		}
+		return false
+	}
+
 	if strings.HasPrefix(name, "tag:") {
 		tagName := name[4:]
 		for _, instanceTag := range instance.Tags {
@@ -286,7 +277,9 @@ func instanceMatchesFilter(instance *ec2.Instance, filter *ec2.Filter) bool {
 				return true
 			}
 		}
+		return false
 	}
+
 	panic("Unknown filter name: " + name)
 }
 
@@ -974,13 +967,14 @@ func TestIpPermissionExistsHandlesMultipleGroupIdsWithUserIds(t *testing.T) {
 		t.Errorf("Should have not been considered equal since first is not in the second array of groups")
 	}
 }
+
 func TestFindInstanceByNodeNameExcludesTerminatedInstances(t *testing.T) {
 	awsServices := NewFakeAWSServices()
 
 	nodeName := types.NodeName("my-dns.internal")
 
 	var tag ec2.Tag
-	tag.Key = aws.String(TagNameKubernetesCluster)
+	tag.Key = aws.String(TagNameKubernetesClusterLegacy)
 	tag.Value = aws.String(TestClusterId)
 	tags := []*ec2.Tag{&tag}
 
@@ -1024,8 +1018,8 @@ func TestFindInstancesByNodeNameCached(t *testing.T) {
 	nodeNameTwo := "my-dns-two.internal"
 
 	var tag ec2.Tag
-	tag.Key = aws.String(TagNameKubernetesCluster)
-	tag.Value = aws.String(TestClusterId)
+	tag.Key = aws.String(TagNameKubernetesClusterPrefix + TestClusterId)
+	tag.Value = aws.String("")
 	tags := []*ec2.Tag{&tag}
 
 	var runningInstance ec2.Instance

pkg/cloudprovider/providers/aws/tags.go

+/*
+Copyright 2017 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 aws
+
+import (
+	"fmt"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"github.com/golang/glog"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"strings"
+)
+
+// TagNameKubernetesClusterPrefix is the tag name we use to differentiate multiple
+// logically independent clusters running in the same AZ.
+// The tag key = TagNameKubernetesClusterPrefix + clusterID
+// The tag value is an ownership value
+const TagNameKubernetesClusterPrefix = "kubernetes.io/cluster/"
+
+// TagNameKubernetesClusterLegacy is the legacy tag name we use to differentiate multiple
+// logically independent clusters running in the same AZ.  The problem with it was that it
+// did not allow shared resources.
+const TagNameKubernetesClusterLegacy = "KubernetesCluster"
+
+type ResourceLifecycle string
+
+const (
+	// ResourceLifecycleOwned is the value we use when tagging resources to indicate
+	// that the resource is considered owned and managed by the cluster,
+	// and in particular that the lifecycle is tied to the lifecycle of the cluster.
+	ResourceLifecycleOwned = "owned"
+	// 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"
+)
+
+type awsTagging struct {
+	// ClusterID 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.
+	// This gives us similar functionality to GCE projects.
+	ClusterID string
+
+	// usesLegacyTags is true if we are using the legacy TagNameKubernetesClusterLegacy tags
+	usesLegacyTags bool
+}
+
+func (t *awsTagging) init(legacyClusterID string, clusterID string) error {
+	if legacyClusterID != "" {
+		if clusterID != "" && legacyClusterID != clusterID {
+			return fmt.Errorf("ClusterID tags did not match: %q vs %q", clusterID, legacyClusterID)
+		}
+		t.usesLegacyTags = true
+		clusterID = legacyClusterID
+	}
+
+	t.ClusterID = clusterID
+
+	if clusterID != "" {
+		glog.Infof("AWS cloud filtering on ClusterID: %v", clusterID)
+	} else {
+		glog.Infof("AWS cloud - no clusterID filtering")
+	}
+
+	return nil
+}
+
+// Extracts a clusterID from the given tags, if one is present
+// If no clusterID is found, returns "", nil
+// If multiple (different) clusterIDs are found, returns an error
+func (t *awsTagging) initFromTags(tags []*ec2.Tag) error {
+	legacyClusterID, newClusterID, err := findClusterIDs(tags)
+	if err != nil {
+		return err
+	}
+
+	if legacyClusterID == "" && newClusterID == "" {
+		glog.Errorf("Tag %q nor %q not found; Kubernetes may behave unexpectedly.", TagNameKubernetesClusterLegacy, TagNameKubernetesClusterPrefix+"...")
+	}
+
+	return t.init(legacyClusterID, newClusterID)
+}
+
+// Extracts the legacy & new cluster ids from the given tags, if they are present
+// If duplicate tags are found, returns an error
+func findClusterIDs(tags []*ec2.Tag) (string, string, error) {
+	legacyClusterID := ""
+	newClusterID := ""
+
+	for _, tag := range tags {
+		tagKey := aws.StringValue(tag.Key)
+		if strings.HasPrefix(tagKey, TagNameKubernetesClusterPrefix) {
+			id := strings.TrimPrefix(tagKey, TagNameKubernetesClusterPrefix)
+			if newClusterID != "" {
+				return "", "", fmt.Errorf("Found multiple cluster tags with prefix %s (%q and %q)", TagNameKubernetesClusterPrefix, newClusterID, id)
+			}
+			newClusterID = id
+		}
+
+		if tagKey == TagNameKubernetesClusterLegacy {
+			id := aws.StringValue(tag.Value)
+			if legacyClusterID != "" {
+				return "", "", fmt.Errorf("Found multiple %s tags (%q and %q)", TagNameKubernetesClusterLegacy, legacyClusterID, id)
+			}
+			legacyClusterID = id
+		}
+	}
+
+	return legacyClusterID, newClusterID, nil
+}
+
+func (t *awsTagging) clusterTagKey() string {
+	return TagNameKubernetesClusterPrefix + t.ClusterID
+}
+
+func (t *awsTagging) hasClusterTag(tags []*ec2.Tag) bool {
+	clusterTagKey := t.clusterTagKey()
+	for _, tag := range tags {
+		tagKey := aws.StringValue(tag.Key)
+		// For 1.6, we continue to recognize the legacy tags, for the 1.5 -> 1.6 upgrade
+		if tagKey == TagNameKubernetesClusterLegacy {
+			return aws.StringValue(tag.Value) == t.ClusterID
+		}
+
+		if tagKey == clusterTagKey {
+			return true
+		}
+	}
+	return false
+}
+
+// Ensure that a resource has the correct tags
+// If it has no tags, we assume that this was a problem caused by an error in between creation and tagging,
+// and we add the tags.  If it has a different cluster's tags, that is an error.
+func (c *awsTagging) readRepairClusterTags(client EC2, resourceID string, lifecycle ResourceLifecycle, additionalTags map[string]string, observedTags []*ec2.Tag) error {
+	actualTagMap := make(map[string]string)
+	for _, tag := range observedTags {
+		actualTagMap[aws.StringValue(tag.Key)] = aws.StringValue(tag.Value)
+	}
+
+	expectedTags := c.buildTags(lifecycle, additionalTags)
+
+	addTags := make(map[string]string)
+	for k, expected := range expectedTags {
+		actual := actualTagMap[k]
+		if actual == expected {
+			continue
+		}
+		if actual == "" {
+			glog.Warningf("Resource %q was missing expected cluster tag %q.  Will add (with value %q)", resourceID, k, expected)
+			addTags[k] = expected
+		} else {
+			return fmt.Errorf("resource %q has tag belonging to another cluster: %q=%q (expected %q)", resourceID, k, actual, expected)
+		}
+	}
+
+	if err := c.createTags(client, resourceID, lifecycle, additionalTags); err != nil {
+		return fmt.Errorf("error adding missing tags to resource %q: %v", resourceID, err)
+	}
+
+	return nil
+}
+
+// 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(client EC2, resourceID string, lifecycle ResourceLifecycle, additionalTags map[string]string) error {
+	tags := t.buildTags(lifecycle, additionalTags)
+
+	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{}
+	request.Resources = []*string{&resourceID}
+	request.Tags = awsTags
+
+	var lastErr error
+	err := wait.ExponentialBackoff(backoff, func() (bool, error) {
+		_, err := client.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
+		glog.V(2).Infof("Failed to create tags; will retry.  Error was %v", err)
+		lastErr = err
+		return false, nil
+	})
+	if err == wait.ErrWaitTimeout {
+		// return real CreateTags error instead of timeout
+		err = lastErr
+	}
+	return err
+}
+
+// Add additional filters, to match on our tags
+// This lets us run multiple k8s clusters in a single EC2 AZ
+func (t *awsTagging) addFilters(filters []*ec2.Filter) []*ec2.Filter {
+	// For 1.6, we always recognize the legacy tag, for the 1.5 -> 1.6 upgrade
+	// There are no "or" filters by key, so we look for both the legacy and new key, and then we have to post-filter
+	f := newEc2Filter("tag-key", TagNameKubernetesClusterLegacy, t.clusterTagKey())
+	filters = append(filters, f)
+
+	if len(filters) == 0 {
+		// We can't pass a zero-length Filters to AWS (it's an error)
+		// So if we end up with no filters; just return nil
+		return nil
+	}
+
+	return filters
+}
+
+func (t *awsTagging) buildTags(lifecycle ResourceLifecycle, additionalTags map[string]string) map[string]string {
+	tags := make(map[string]string)
+	for k, v := range additionalTags {
+		tags[k] = v
+	}
+	// We only create legacy tags if we are using legacy tags, i.e. if we have seen a legacy tag on our instance
+	if t.usesLegacyTags {
+		tags[TagNameKubernetesClusterLegacy] = t.ClusterID
+	}
+	tags[t.clusterTagKey()] = string(lifecycle)
+
+	return tags
+}

pkg/cloudprovider/providers/aws/tags_test.go

+/*
+Copyright 2014 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 aws
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+	"strings"
+	"testing"
+)
+
+func TestFilterTags(t *testing.T) {
+	awsServices := NewFakeAWSServices()
+	c, err := newAWSCloud(strings.NewReader("[global]"), awsServices)
+	if err != nil {
+		t.Errorf("Error building aws cloud: %v", err)
+		return
+	}
+
+	if c.tagging.ClusterID != TestClusterId {
+		t.Errorf("unexpected ClusterID: %v", c.tagging.ClusterID)
+	}
+}
+
+func TestFindClusterID(t *testing.T) {
+	grid := []struct {
+		Tags           map[string]string
+		ExpectedNew    string
+		ExpectedLegacy string
+		ExpectError    bool
+	}{
+		{
+			Tags: map[string]string{},
+		},
+		{
+			Tags: map[string]string{
+				TagNameKubernetesClusterLegacy: "a",
+			},
+			ExpectedLegacy: "a",
+		},
+		{
+			Tags: map[string]string{
+				TagNameKubernetesClusterPrefix + "a": "owned",
+			},
+			ExpectedNew: "a",
+		},
+		{
+			Tags: map[string]string{
+				TagNameKubernetesClusterPrefix + "a": "",
+			},
+			ExpectedNew: "a",
+		},
+		{
+			Tags: map[string]string{
+				TagNameKubernetesClusterLegacy:       "a",
+				TagNameKubernetesClusterPrefix + "a": "",
+			},
+			ExpectedLegacy: "a",
+			ExpectedNew:    "a",
+		},
+		{
+			Tags: map[string]string{
+				TagNameKubernetesClusterPrefix + "a": "",
+				TagNameKubernetesClusterPrefix + "b": "",
+			},
+			ExpectError: true,
+		},
+	}
+	for _, g := range grid {
+		var ec2Tags []*ec2.Tag
+		for k, v := range g.Tags {
+			ec2Tags = append(ec2Tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
+		}
+		actualLegacy, actualNew, err := findClusterIDs(ec2Tags)
+		if g.ExpectError {
+			if err == nil {
+				t.Errorf("expected error for tags %v", g.Tags)
+				continue
+			}
+		} else {
+			if err != nil {
+				t.Errorf("unexpected error for tags %v: %v", g.Tags, err)
+				continue
+			}
+
+			if g.ExpectedNew != actualNew {
+				t.Errorf("unexpected new clusterid for tags %v: %s vs %s", g.Tags, g.ExpectedNew, actualNew)
+				continue
+			}
+
+			if g.ExpectedLegacy != actualLegacy {
+				t.Errorf("unexpected new clusterid for tags %v: %s vs %s", g.Tags, g.ExpectedLegacy, actualLegacy)
+				continue
+			}
+		}
+	}
+}