Repository: kubernetes/cloud-provider-aws
Pull request #149 (2020-10-28 22:19:12)

tags: initial implementation of tags
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Please label this pull request according to what type of issue you are addressing, especially if this is a release targeted pull request. For reference on required PR/issue labels, read here:
https://git.k8s.io/community/contributors/devel/sig-release/release.md#issue-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://git.k8s.io/community/contributors/devel/sig-testing/testing.md
4. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
5. Follow the instructions for writing a release note: https://git.k8s.io/community/contributors/guide/release-notes.md
6. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->
first pass of tags implementation

**What type of PR is this?**
/kind feature

**What this PR does / why we need it**:
We have two formats of the cluster tags in the legacy implementation:
```
// 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"
```

As @nckturner [mentioned before](https://github.com/kubernetes/cloud-provider-aws/pull/131#discussion_r501278537), the current format ("`kubernetes.io/cluster/<clusterID>": <ownership>`") causes issues for customers trying to use tags to organize billing. 

After discussion, we'd like to see tags with formats:
```
1) kubernetes.io/cluster/ = <clusterID>
2) kubernetes.io/cluster/<clusterID> = ""
```
the second format is only used for shared resources e.g. subnets

and once we have other workarounds, will make corresponding updates.
**Which issue(s) this PR fixes**:
<!-- 
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Part of #125

**Special notes for your reviewer**:

**Does this PR introduce a user-facing change?**:
<!--  
If no, just write "NONE".
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
2. 
-->
```release-note
NONE
```


k8s-ci-robot: @nicolehanjing: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-aws-check | ab502a664ffce75437bc985fa5cdc264540eed67 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-check/1321577286893309952) | `/test pull-cloud-provider-aws-check`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes&repo=cloud-provider-aws&pr=149). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anicolehanjing). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes/cloud-provider-aws/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes/cloud-provider-aws/pull/149#" title="Author self-approved">nicolehanjing</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **nckturner** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @nckturner` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes%2Fcloud-provider-aws).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/kubernetes/cloud-provider-aws/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["nckturner"]} -->
k8s-ci-robot: @nicolehanjing: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-aws-check | 1d9e7605125999101668d4610e4093734af621aa | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-check/1322227329300172800) | `/test pull-cloud-provider-aws-check`
pull-cloud-provider-aws-test | 37e4b3cb2ccf0ed3864871478e86e84001a212c3 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-test/1322276168900022272) | `/test pull-cloud-provider-aws-test`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes&repo=cloud-provider-aws&pr=149). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anicolehanjing). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes/cloud-provider-aws/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @nicolehanjing: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-aws-test | 37e4b3cb2ccf0ed3864871478e86e84001a212c3 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-test/1322276168900022272) | `/test pull-cloud-provider-aws-test`
pull-cloud-provider-aws-check | 37e4b3cb2ccf0ed3864871478e86e84001a212c3 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-check/1322276168879050752) | `/test pull-cloud-provider-aws-check`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes&repo=cloud-provider-aws&pr=149). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anicolehanjing). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes/cloud-provider-aws/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @nicolehanjing: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-aws-test | 37e4b3cb2ccf0ed3864871478e86e84001a212c3 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-test/1322276168900022272) | `/test pull-cloud-provider-aws-test`
pull-cloud-provider-aws-check | eecb5ddbdcdddc01d870854f494854b1aa1bd9a7 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-check/1322281473893470208) | `/test pull-cloud-provider-aws-check`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes&repo=cloud-provider-aws&pr=149). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anicolehanjing). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes/cloud-provider-aws/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @nicolehanjing: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-aws-check | eecb5ddbdcdddc01d870854f494854b1aa1bd9a7 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-check/1322281473893470208) | `/test pull-cloud-provider-aws-check`
pull-cloud-provider-aws-test | eecb5ddbdcdddc01d870854f494854b1aa1bd9a7 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-test/1322281473906053120) | `/test pull-cloud-provider-aws-test`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes&repo=cloud-provider-aws&pr=149). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anicolehanjing). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes/cloud-provider-aws/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @nicolehanjing: The following tests **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-aws-test | eecb5ddbdcdddc01d870854f494854b1aa1bd9a7 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-test/1322281473906053120) | `/test pull-cloud-provider-aws-test`
pull-cloud-provider-aws-check | c3c5010d9587efae6d36e313cffd38d25003f075 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-check/1323403578110382080) | `/test pull-cloud-provider-aws-check`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes&repo=cloud-provider-aws&pr=149). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anicolehanjing). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes/cloud-provider-aws/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
k8s-ci-robot: @nicolehanjing: The following test **failed**, say `/retest` to rerun all failed tests:

Test name | Commit | Details | Rerun command
--- | --- | --- | ---
pull-cloud-provider-aws-check | 0f2fe6d1f33bd5a1cb1e3955e533519f6d38f09a | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/149/pull-cloud-provider-aws-check/1326686057395130368) | `/test pull-cloud-provider-aws-check`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes&repo=cloud-provider-aws&pr=149). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Anicolehanjing). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes/cloud-provider-aws/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
nckturner: So just to summarize our conversation from this morning, we need to keep the `kubernetes.io/cluster/<clusterID> = shared|owned` for discovery, but we will allow passing in `additionalTags` (or something similar) via the config to aid in billing and other tag use cases, like where users automatically delete all resources from an account that aren't tagged a certain way.  It also makes sense to use "tag-on-create" where possible because we believe it is better supported (although I am verifying that). [There have been some changes to the APIs](https://aws.amazon.com/about-aws/whats-new/2018/07/tag-on-create-for-amazon-ec2-and-amazon-ebs-is-now-available-in-the-aws-govcloud-region/) since V1 was written, so we should take advantage of those. 
nckturner: So actually as of now I think the resource tagging API is fully supported in all regions, so I don't think we need to prefer tag-on-create for that reason.  Tag-on-create might be nice for other reasons, like being eaiser to implement and fewer total number of api calls.
nicolehanjing: > So just to summarize our conversation from this morning, we need to keep the `kubernetes.io/cluster/<clusterID> = shared|owned` for discovery, but we will allow passing in `additionalTags` (or something similar) via the config to aid in billing and other tag use cases, like where users automatically delete all resources from an account that aren't tagged a certain way. It also makes sense to use "tag-on-create" where possible because we believe it is better supported (although I am verifying that). [There have been some changes to the APIs](https://aws.amazon.com/about-aws/whats-new/2018/07/tag-on-create-for-amazon-ec2-and-amazon-ebs-is-now-available-in-the-aws-govcloud-region/) since V1 was written, so we should take advantage of those.

Thanks @nckturner for the summary
for my current implementation, I only used the first format which is "kubernetes.io/cluster/ = <clusterID>", but as we are keeping `kubernetes.io/cluster/<clusterID> = shared|owned`, I will need to keep both of them in this initial implementation, right? 
Regarding "tag-on-create", if I'm understanding it correctly, so once we have the implementation of tags, I think I need to add tags to resources, like LB, subnets, etc. while creating the resource.
nicolehanjing: @andrewsykim updated! PTAL :) 

Patch:
diff --git a/cmd/aws-cloud-controller-manager/main.go b/cmd/aws-cloud-controller-manager/main.go
index a559199..1981304 100644
--- a/cmd/aws-cloud-controller-manager/main.go
+++ b/cmd/aws-cloud-controller-manager/main.go
@@ -117,10 +117,8 @@ func main() {
 				os.Exit(1)
 			}
 
-			cloudConfigFile := c.ComponentConfig.KubeCloudShared.CloudProvider.CloudConfigFile
-
 			// initialize cloud provider with the cloud provider name and config file provided
-			cloud, err := cloudprovider.InitCloudProvider(cloudProvider, cloudConfigFile)
+			cloud, err := cloudprovider.InitCloudProvider(cloudProvider, "")
 			if err != nil {
 				klog.Fatalf("Cloud provider could not be initialized: %v", err)
 			}
diff --git a/go.mod b/go.mod
index 2c31d90..a1b9bc8 100644
--- a/go.mod
+++ b/go.mod
@@ -17,4 +17,5 @@ require (
 	k8s.io/klog/v2 v2.4.0
 	k8s.io/legacy-cloud-providers v0.20.0
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920
+	sigs.k8s.io/yaml v1.2.0
 )
diff --git a/pkg/providers/v2/cloud.go b/pkg/providers/v2/cloud.go
index a07524b..63aacb6 100644
--- a/pkg/providers/v2/cloud.go
+++ b/pkg/providers/v2/cloud.go
@@ -19,8 +19,10 @@ limitations under the License.
 package v2
 
 import (
+	"errors"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"regexp"
 
 	"github.com/aws/aws-sdk-go/aws"
@@ -30,12 +32,26 @@ import (
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/ec2"
 
+	"k8s.io/apimachinery/pkg/util/validation/field"
 	cloudprovider "k8s.io/cloud-provider"
+	awsconfigv1alpha1 "k8s.io/cloud-provider-aws/pkg/apis/config/v1alpha1"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/yaml"
 )
 
 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)
+		}
+
+		errs := validateAWSCloudConfig(cfg)
+		if len(errs) > 0 {
+			return nil, fmt.Errorf("failed to validate AWS cloud config: %v", errs.ToAggregate())
+		}
+
+		return newCloud(cfg)
 	})
 }
 
@@ -53,6 +69,14 @@ type cloud struct {
 	region    string
 	ec2       EC2
 	metadata  EC2Metadata
+	tags      awsTagging
+}
+
+// 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 +85,48 @@ type EC2Metadata interface {
 	GetMetadata(path string) (string, error)
 }
 
+func readAWSCloudConfig(config io.Reader) (*awsconfigv1alpha1.AWSCloudConfig, error) {
+	if config == nil {
+		return nil, errors.New("no AWS cloud provider config file given")
+	}
+
+	// read the config file
+	data, err := ioutil.ReadAll(config)
+	if err != nil {
+		return nil, fmt.Errorf("unable to read cloud configuration from %q [%v]", config, err)
+	}
+
+	var cfg awsconfigv1alpha1.AWSCloudConfig
+	err = yaml.Unmarshal(data, &cfg)
+	if err != nil {
+		// we got an error where the decode wasn't related to a missing type
+		return nil, err
+	}
+
+	return &cfg, nil
+}
+
+// validateAWSCloudConfig validates AWSCloudConfig
+// clusterName is required
+func validateAWSCloudConfig(config *awsconfigv1alpha1.AWSCloudConfig) field.ErrorList {
+	allErrs := field.ErrorList{}
+
+	if config.Kind != "AWSCloudConfig" {
+		allErrs = append(allErrs, field.Invalid(field.NewPath("kind"), "invalid kind for cloud config: %q", config.Kind))
+	}
+
+	if config.APIVersion != awsconfigv1alpha1.SchemeGroupVersion.String() {
+		allErrs = append(allErrs, field.Invalid(field.NewPath("apiVersion"), "invalid apiVersion for cloud config: %q", config.APIVersion))
+	}
+
+	fieldPath := field.NewPath("config")
+	if len(config.Config.ClusterName) == 0 {
+		allErrs = append(allErrs, field.Required(fieldPath.Child("clusterName"), "cluster name cannot be empty"))
+	}
+
+	return allErrs
+}
+
 func getAvailabilityZone(metadata EC2Metadata) (string, error) {
 	return metadata.GetMetadata("placement/availability-zone")
 }
@@ -82,7 +148,7 @@ func azToRegion(az string) (string, error) {
 }
 
 // newCloud creates a new instance of AWSCloud.
-func newCloud() (cloudprovider.Interface, error) {
+func newCloud(cfg *awsconfigv1alpha1.AWSCloudConfig) (cloudprovider.Interface, error) {
 	sess, err := session.NewSession(&aws.Config{})
 	if err != nil {
 		return nil, fmt.Errorf("unable to initialize AWS session: %v", err)
@@ -112,11 +178,6 @@ func newCloud() (cloudprovider.Interface, error) {
 		return nil, err
 	}
 
-	instances, err := newInstances(az, creds)
-	if err != nil {
-		return nil, err
-	}
-
 	ec2Sess, err := session.NewSession(&aws.Config{
 		Region:      aws.String(region),
 		Credentials: creds,
@@ -130,15 +191,29 @@ func newCloud() (cloudprovider.Interface, error) {
 		return nil, fmt.Errorf("error creating AWS ec2 client: %q", err)
 	}
 
-	awsCloud := &cloud{
+	var tags awsTagging
+	if cfg.Config.ClusterName != "" {
+		tags, err = newAWSTags(cfg.Config.ClusterName)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		klog.Warning("misconfigured cluster: no clusterName")
+	}
+
+	instances, err := newInstances(az, creds, tags)
+	if err != nil {
+		return nil, err
+	}
+
+	return &cloud{
 		creds:     creds,
 		instances: instances,
 		region:    region,
 		metadata:  metadataClient,
 		ec2:       ec2Service,
-	}
-
-	return awsCloud, nil
+		tags:      tags,
+	}, nil
 }
 
 // Initialize passes a Kubernetes clientBuilder interface to the cloud provider
@@ -177,7 +252,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.tags.clusterName()) > 0
 }
 
 // InstancesV2 is an implementation for instances and should only be implemented by external cloud providers.
diff --git a/pkg/providers/v2/instances.go b/pkg/providers/v2/instances.go
index 879fd1f..800cf41 100644
--- a/pkg/providers/v2/instances.go
+++ b/pkg/providers/v2/instances.go
@@ -34,7 +34,7 @@ import (
 )
 
 // newInstances returns an implementation of cloudprovider.InstancesV2
-func newInstances(az string, creds *credentials.Credentials) (cloudprovider.InstancesV2, error) {
+func newInstances(az string, creds *credentials.Credentials, tags awsTagging) (cloudprovider.InstancesV2, error) {
 	region, err := azToRegion(az)
 	if err != nil {
 		return nil, err
@@ -56,19 +56,16 @@ func newInstances(az string, creds *credentials.Credentials) (cloudprovider.Inst
 		availabilityZone: az,
 		ec2:              ec2Service,
 		region:           region,
+		tags:             tags,
 	}, 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
 	ec2              EC2
 	region           string
+	tags             awsTagging
 }
 
 // InstanceExists indicates whether a given node exists according to the cloud provider
@@ -160,6 +157,13 @@ func (i *instances) getInstance(ctx context.Context, node *v1.Node) (*ec2.Instan
 		klog.V(4).Infof("looking for node by provider ID %v", node.Spec.ProviderID)
 	}
 
+	// TODO: add additional tags from users
+	tags := i.tags.buildTags(map[string]string{}, ResourceLifecycleOwned)
+	for tagKey, tagValue := range tags {
+		request.Filters = append(request.Filters,
+			newEc2Filter("tag:"+tagKey, tagValue))
+	}
+
 	instances := []*ec2.Instance{}
 	var nextToken *string
 	for {
diff --git a/pkg/providers/v2/mocks/mock_ec2.go b/pkg/providers/v2/mocks/mock_ec2.go
index 9818a8f..83f42cc 100644
--- a/pkg/providers/v2/mocks/mock_ec2.go
+++ b/pkg/providers/v2/mocks/mock_ec2.go
@@ -1,18 +1,5 @@
-/*
-Copyright 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.
-*/
-
 // Code generated by MockGen. DO NOT EDIT.
-// Source: instances.go
+// Source: cloud.go
 
 // Package mocks is a generated GoMock package.
 package mocks
@@ -60,3 +47,56 @@ func (mr *MockEC2MockRecorder) DescribeInstances(request interface{}) *gomock.Ca
 	mr.mock.ctrl.T.Helper()
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DescribeInstances", reflect.TypeOf((*MockEC2)(nil).DescribeInstances), request)
 }
+
+// CreateTags mocks base method
+func (m *MockEC2) CreateTags(arg0 *ec2.CreateTagsInput) (*ec2.CreateTagsOutput, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "CreateTags", arg0)
+	ret0, _ := ret[0].(*ec2.CreateTagsOutput)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// CreateTags indicates an expected call of CreateTags
+func (mr *MockEC2MockRecorder) CreateTags(arg0 interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateTags", reflect.TypeOf((*MockEC2)(nil).CreateTags), arg0)
+}
+
+// MockEC2Metadata is a mock of EC2Metadata interface
+type MockEC2Metadata struct {
+	ctrl     *gomock.Controller
+	recorder *MockEC2MetadataMockRecorder
+}
+
+// MockEC2MetadataMockRecorder is the mock recorder for MockEC2Metadata
+type MockEC2MetadataMockRecorder struct {
+	mock *MockEC2Metadata
+}
+
+// NewMockEC2Metadata creates a new mock instance
+func NewMockEC2Metadata(ctrl *gomock.Controller) *MockEC2Metadata {
+	mock := &MockEC2Metadata{ctrl: ctrl}
+	mock.recorder = &MockEC2MetadataMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use
+func (m *MockEC2Metadata) EXPECT() *MockEC2MetadataMockRecorder {
+	return m.recorder
+}
+
+// GetMetadata mocks base method
+func (m *MockEC2Metadata) GetMetadata(path string) (string, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetMetadata", path)
+	ret0, _ := ret[0].(string)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetMetadata indicates an expected call of GetMetadata
+func (mr *MockEC2MetadataMockRecorder) GetMetadata(path interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetMetadata", reflect.TypeOf((*MockEC2Metadata)(nil).GetMetadata), path)
+}
diff --git a/pkg/providers/v2/tags.go b/pkg/providers/v2/tags.go
new file mode 100644
index 0000000..741b76a
--- /dev/null
+++ b/pkg/providers/v2/tags.go
@@ -0,0 +1,180 @@
+/*
+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 (
+	"errors"
+	"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.
+	// tag format: kubernetes.io/cluster/<clusterID> = shared|owned
+	// The tag key = TagNameKubernetesClusterPrefix + clusterID
+	// The tag value is an ownership value
+	TagNameKubernetesClusterPrefix = "kubernetes.io/cluster/"
+
+	// 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.
+	// TODO: revisit these values
+	createTagInitialDelay = 1 * time.Second
+	createTagFactor       = 2.0
+	createTagSteps        = 9
+
+	// 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"
+)
+
+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
+}
+
+// newAWSTags is a constructor function for awsTagging
+func newAWSTags(clusterName string) (awsTagging, error) {
+	if clusterName != "" {
+		klog.Infof("AWS cloud filtering on ClusterName: %v", clusterName)
+	} else {
+		return awsTagging{}, errors.New("No ClusterName found in the config")
+	}
+
+	return awsTagging{
+		ClusterName: clusterName,
+	}, nil
+}
+
+// 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 := 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) hasClusterTag(tags []*ec2.Tag) error {
+	if len(t.ClusterName) == 0 {
+		return errors.New("cluster name is not configured (an empty value)")
+	}
+
+	for _, tag := range tags {
+		tagKey := aws.StringValue(tag.Key)
+		if tagKey == (TagNameKubernetesClusterPrefix + t.ClusterName) {
+			return nil
+		}
+	}
+
+	return errors.New("cluster tag does not exist")
+}
+
+func (t *awsTagging) buildTags(additionalTags map[string]string, lifecycle string) map[string]string {
+	tags := make(map[string]string)
+	for tagKey, tagValue := range additionalTags {
+		tags[tagKey] = tagValue
+	}
+
+	// no clusterName is a sign of misconfigured cluster, but we can't be tagging the resources with empty
+	// strings
+	// TODO: revise the logic
+	if len(t.ClusterName) == 0 {
+		return tags
+	}
+
+	// tag format: kubernetes.io/cluster/<clusterID> = shared|owned
+	tags[TagNameKubernetesClusterPrefix+t.ClusterName] = lifecycle
+
+	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 tagKey, tagValue := range tags {
+		tag := &ec2.Tag{
+			Key:   aws.String(tagKey),
+			Value: aws.String(tagValue),
+		}
+		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
+}


Test patch:
diff --git a/pkg/providers/v2/cloud_test.go b/pkg/providers/v2/cloud_test.go
index 4f94cf1..7137f7f 100644
--- a/pkg/providers/v2/cloud_test.go
+++ b/pkg/providers/v2/cloud_test.go
@@ -19,9 +19,14 @@ limitations under the License.
 package v2
 
 import (
+	"bytes"
+	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/cloud-provider-aws/pkg/apis/config/v1alpha1"
 )
 
 func TestAzToRegion(t *testing.T) {
@@ -44,3 +49,174 @@ func TestAzToRegion(t *testing.T) {
 		assert.Equal(t, testCase.region, ret)
 	}
 }
+
+func TestReadAWSCloudConfig(t *testing.T) {
+	testcases := []struct {
+		name       string
+		configData string
+		config     *v1alpha1.AWSCloudConfig
+		expectErr  bool
+	}{
+		{
+			name: "config with valid cluster name",
+			configData: `---
+kind: AWSCloudConfig
+apiVersion: config.aws.io/v1alpha1
+config:
+    clusterName: test
+    `,
+			config: &v1alpha1.AWSCloudConfig{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "AWSCloudConfig",
+					APIVersion: "config.aws.io/v1alpha1",
+				},
+				Config: v1alpha1.AWSConfig{
+					ClusterName: "test",
+				},
+			},
+		},
+		{
+			name: "config with empty cluster name",
+			configData: `---
+kind: AWSCloudConfig
+apiVersion: config.aws.io/v1alpha1
+config:
+    clusterName: ""
+    `,
+			config: &v1alpha1.AWSCloudConfig{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "AWSCloudConfig",
+					APIVersion: "config.aws.io/v1alpha1",
+				},
+				Config: v1alpha1.AWSConfig{
+					ClusterName: "",
+				},
+			},
+		},
+		{
+			name: "config with only kind and apiVersion",
+			configData: `---
+kind: AWSCloudConfig
+apiVersion: config.aws.io/v1alpha1
+    `,
+			config: &v1alpha1.AWSCloudConfig{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "AWSCloudConfig",
+					APIVersion: "config.aws.io/v1alpha1",
+				},
+				Config: v1alpha1.AWSConfig{
+					ClusterName: "",
+				},
+			},
+		},
+		{
+			name: "config with wrong Kind",
+			configData: `---
+kind: WrongCloudConfig
+apiVersion: config.aws.io/v1alpha1
+config:
+    clusterName: test
+	`,
+			config:    nil,
+			expectErr: true,
+		},
+		{
+			name: "config with wrong apiversion",
+			configData: `---
+kind: AWSCloudConfig
+apiVersion: wrong.aws.io/v1alpha1
+config:
+    clusterName: test
+	`,
+			config:    nil,
+			expectErr: true,
+		},
+	}
+
+	for _, testcase := range testcases {
+		t.Run(testcase.name, func(t *testing.T) {
+			buffer := bytes.NewBufferString(testcase.configData)
+			cloudConfig, err := readAWSCloudConfig(buffer)
+			if err != nil && !testcase.expectErr {
+				t.Fatal(err)
+			}
+
+			if err == nil && testcase.expectErr {
+				t.Error("expected error but got none")
+			}
+
+			if !reflect.DeepEqual(cloudConfig, testcase.config) {
+				t.Logf("actual cloud config: %#v", cloudConfig)
+				t.Logf("expected cloud config: %#v", testcase.config)
+				t.Error("AWS cloud config did not match")
+			}
+		})
+	}
+}
+
+func TestValidateAWSCloudConfig(t *testing.T) {
+	testcases := []struct {
+		name      string
+		config    *v1alpha1.AWSCloudConfig
+		expectErr bool
+	}{
+		{
+			name: "valid config",
+			config: &v1alpha1.AWSCloudConfig{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "AWSCloudConfig",
+					APIVersion: "config.aws.io/v1alpha1",
+				},
+				Config: v1alpha1.AWSConfig{
+					ClusterName: "test",
+				},
+			},
+		},
+		{
+			name: "empty cluster name",
+			config: &v1alpha1.AWSCloudConfig{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "AWSCloudConfig",
+					APIVersion: "config.aws.io/v1alpha1",
+				},
+				Config: v1alpha1.AWSConfig{
+					ClusterName: "",
+				},
+			},
+			expectErr: true,
+		},
+		{
+			name: "empty config",
+			config: &v1alpha1.AWSCloudConfig{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "AWSCloudConfig",
+					APIVersion: "config.aws.io/v1alpha1",
+				},
+				Config: v1alpha1.AWSConfig{},
+			},
+			expectErr: true,
+		},
+		{
+			name: "invalid config",
+			config: &v1alpha1.AWSCloudConfig{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "AWSCloudConfig",
+					APIVersion: "config.aws.io/v1alpha1",
+				},
+			},
+			expectErr: true,
+		},
+	}
+
+	for _, testcase := range testcases {
+		t.Run(testcase.name, func(t *testing.T) {
+			errs := validateAWSCloudConfig(testcase.config)
+
+			if testcase.expectErr && len(errs) == 0 {
+				t.Errorf("expected error but got none")
+			} else if !testcase.expectErr && len(errs) > 0 {
+				t.Errorf("expected no error but received errors: %v", errs.ToAggregate())
+			}
+		})
+	}
+}
diff --git a/pkg/providers/v2/tags_test.go b/pkg/providers/v2/tags_test.go
new file mode 100644
index 0000000..aaeecdb
--- /dev/null
+++ b/pkg/providers/v2/tags_test.go
@@ -0,0 +1,89 @@
+/*
+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 (
+	"testing"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/ec2"
+)
+
+func TestFindClusterName(t *testing.T) {
+	testCases := []struct {
+		Tags                map[string]string
+		ExpectedClusterName string
+		ExpectError         bool
+	}{
+		{
+			Tags: map[string]string{},
+		},
+		{
+			Tags: map[string]string{
+				TagNameKubernetesClusterPrefix + TestClusterID: "owned",
+			},
+			ExpectedClusterName: TestClusterID,
+		},
+		{
+			Tags: map[string]string{
+				TagNameKubernetesClusterPrefix + TestClusterID: "shared",
+			},
+			ExpectedClusterName: TestClusterID,
+		},
+		{
+			Tags: map[string]string{
+				TagNameKubernetesClusterPrefix + TestClusterID: "",
+			},
+			ExpectedClusterName: TestClusterID,
+		},
+		{
+			Tags: map[string]string{
+				TagNameKubernetesClusterPrefix: "",
+			},
+			ExpectedClusterName: "",
+		},
+		{
+			Tags: map[string]string{
+				TagNameKubernetesClusterPrefix + "a": "",
+				TagNameKubernetesClusterPrefix + "b": "",
+			},
+			ExpectError: true,
+		},
+	}
+	for _, testCase := range testCases {
+		var ec2Tags []*ec2.Tag
+		for k, v := range testCase.Tags {
+			ec2Tags = append(ec2Tags, &ec2.Tag{Key: aws.String(k), Value: aws.String(v)})
+		}
+		clusterName, err := findClusterName(ec2Tags)
+		if testCase.ExpectError {
+			if err == nil {
+				t.Errorf("expected error for tags %v", testCase.Tags)
+				continue
+			}
+		} else {
+			if err != nil {
+				t.Errorf("unexpected error for tags %v: %v", testCase.Tags, err)
+				continue
+			}
+
+			if testCase.ExpectedClusterName != clusterName {
+				t.Errorf("unexpected new clusterName for tags %v: %s vs %s", testCase.Tags, testCase.ExpectedClusterName, clusterName)
+				continue
+			}
+		}
+	}
+}