Repository: kubernetes/cloud-provider-aws
Pull request #413 (2022-06-17 21:02:42)

Add rate limiting support for Tagging Controller
<!--  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
-->

**What type of PR is this?**
> Uncomment only one, leave it on its own line:

/kind feature

**What this PR does / why we need it**: Adds rate limiting support for Tagging Controller. This helps in controlling the rate at which the controller processes items and hence control the rate at which it makes calls to services like EC2.

**Which issue(s) this PR fixes**: N/A

**Special notes for your reviewer**: N/A

**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
Adds rate limiting support for Tagging Controller. This helps in controlling the rate at which the controller processes items and hence control the rate at which it makes calls to services like EC2.
```


k8s-ci-robot: @saurav-agarwalla: This issue is currently awaiting triage.

If cloud-provider-aws contributors determine this is a relevant issue, they will accept it by applying the `triage/accepted` label and provide further guidance.

The `triage/accepted` label can be added by org members by writing `/triage accepted` in a comment.


<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.
</details>
k8s-ci-robot: Hi @saurav-agarwalla. Thanks for your PR.

I'm waiting for a [kubernetes](https://github.com/orgs/kubernetes/people) member to verify that this patch is reasonable to test. If it is, they should reply with `/ok-to-test` on its own line. Until that is done, I will not automatically test new commits in this PR, but the usual testing commands by org members will still work. Regular contributors should [join the org](https://git.k8s.io/community/community-membership.md#member) to skip this step.

Once the patch is verified, the new status will be reflected by the `ok-to-test` label.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=kubernetes%2Fcloud-provider-aws).

<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.
</details>

k8s-ci-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **andrewsykim** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @andrewsykim` 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/release-1.20/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":["andrewsykim"]} -->
nckturner: /triage accepted
/ok-to-test
k8s-ci-robot: @saurav-agarwalla: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-aws-test | 7282cf8bc95f832beb5d7341a408abe30b140702 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/413/pull-cloud-provider-aws-test/1537907410851598336) | true | `/test pull-cloud-provider-aws-test`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes&repo=cloud-provider-aws&pr=413). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaurav-agarwalla). 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: @saurav-agarwalla: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-aws-test | 7282cf8bc95f832beb5d7341a408abe30b140702 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/413/pull-cloud-provider-aws-test/1537907410851598336) | true | `/test pull-cloud-provider-aws-test`
pull-cloud-provider-aws-check | 7282cf8bc95f832beb5d7341a408abe30b140702 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/413/pull-cloud-provider-aws-check/1537907410818043904) | true | `/test pull-cloud-provider-aws-check`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes&repo=cloud-provider-aws&pr=413). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaurav-agarwalla). 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: @saurav-agarwalla: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-cloud-provider-aws-test | 7282cf8bc95f832beb5d7341a408abe30b140702 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/413/pull-cloud-provider-aws-test/1537907410851598336) | true | `/test pull-cloud-provider-aws-test`
pull-cloud-provider-aws-check | 7282cf8bc95f832beb5d7341a408abe30b140702 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/413/pull-cloud-provider-aws-check/1537907410818043904) | true | `/test pull-cloud-provider-aws-check`
pull-cloud-provider-aws-e2e | 7282cf8bc95f832beb5d7341a408abe30b140702 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/cloud-provider-aws/413/pull-cloud-provider-aws-e2e/1537907410885152768) | true | `/test pull-cloud-provider-aws-e2e`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes&repo=cloud-provider-aws&pr=413). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaurav-agarwalla). 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 -->

Patch:
diff --git a/go.mod b/go.mod
index f7ab7b0..6462c05 100644
--- a/go.mod
+++ b/go.mod
@@ -8,6 +8,7 @@ require (
 	github.com/google/go-cmp v0.5.5
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.6.1
+	golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e
 	gopkg.in/gcfg.v1 v1.2.0
 	gopkg.in/warnings.v0 v0.1.1 // indirect
 	k8s.io/api v0.20.15
diff --git a/pkg/controllers/options/tagging_controller.go b/pkg/controllers/options/tagging_controller.go
index 3c1de76..c606a85 100644
--- a/pkg/controllers/options/tagging_controller.go
+++ b/pkg/controllers/options/tagging_controller.go
@@ -21,14 +21,20 @@ import (
 // TaggingControllerOptions contains the inputs that can
 // be used in the tagging controller
 type TaggingControllerOptions struct {
-	Tags      map[string]string
-	Resources []string
+	Tags       map[string]string
+	Resources  []string
+	RateLimit  float64
+	BurstLimit int
 }
 
 // AddFlags add the additional flags for the controller
 func (o *TaggingControllerOptions) AddFlags(fs *pflag.FlagSet) {
 	fs.StringToStringVar(&o.Tags, "tags", o.Tags, "Tags to apply to AWS resources in the tagging controller, in a form of key=value.")
 	fs.StringArrayVar(&o.Resources, "resources", o.Resources, "AWS resources name to add/remove tags in the tagging controller.")
+	fs.Float64Var(&o.RateLimit, "tagging-controller-rate-limit", o.RateLimit,
+		"Steady-state rate limit (per sec) at which the controller processes items in its queue. A value of zero (default) disables rate limiting.")
+	fs.IntVar(&o.BurstLimit, "tagging-controller-burst-limit", o.BurstLimit,
+		"Burst limit at which the controller processes items in its queue. A value of zero (default) disables rate limiting.")
 }
 
 // Validate checks for errors from user input
@@ -41,6 +47,14 @@ func (o *TaggingControllerOptions) Validate() error {
 		return fmt.Errorf("--resources must not be empty")
 	}
 
+	if o.RateLimit < 0.0 {
+		return fmt.Errorf("--tagging-controller-rate-limit should not be less than zero")
+	}
+
+	if o.BurstLimit < 0 {
+		return fmt.Errorf("--tagging-controller-burst-limit should not be less than zero")
+	}
+
 	for _, r := range o.Resources {
 		for _, resource := range SupportedResources {
 			if r != resource {
diff --git a/pkg/controllers/tagging/tagging_controller.go b/pkg/controllers/tagging/tagging_controller.go
index d20a9de..b930746 100644
--- a/pkg/controllers/tagging/tagging_controller.go
+++ b/pkg/controllers/tagging/tagging_controller.go
@@ -16,6 +16,7 @@ package tagging
 import (
 	"crypto/md5"
 	"fmt"
+	"golang.org/x/time/rate"
 	v1 "k8s.io/api/core/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -83,6 +84,8 @@ type Controller struct {
 
 	// Representing the resources to tag
 	resources []string
+
+	rateLimitEnabled bool
 }
 
 // NewTaggingController creates a NewTaggingController object
@@ -92,7 +95,9 @@ func NewTaggingController(
 	cloud cloudprovider.Interface,
 	nodeMonitorPeriod time.Duration,
 	tags map[string]string,
-	resources []string) (*Controller, error) {
+	resources []string,
+	rateLimit float64,
+	burstLimit int) (*Controller, error) {
 
 	awsCloud, ok := cloud.(*awsv1.Cloud)
 	if !ok {
@@ -100,6 +105,23 @@ func NewTaggingController(
 		return nil, err
 	}
 
+	var rateLimiter workqueue.RateLimiter
+	var rateLimitEnabled bool
+	if rateLimit > 0.0 && burstLimit > 0 {
+		klog.Infof("Rate limit enabled on controller with rate %f and burst %d.", rateLimit, burstLimit)
+		// This is the workqueue.DefaultControllerRateLimiter() but in case where throttling is enabled on the controller,
+		// the rate and burst values are set to the provided values.
+		rateLimiter = workqueue.NewMaxOfRateLimiter(
+			workqueue.NewItemExponentialFailureRateLimiter(5*time.Millisecond, 1000*time.Second),
+			&workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(rateLimit), burstLimit)},
+		)
+		rateLimitEnabled = true
+	} else {
+		klog.Infof("Rate limit disabled on controller.")
+		rateLimiter = workqueue.DefaultControllerRateLimiter()
+		rateLimitEnabled = false
+	}
+
 	registerMetrics()
 	tc := &Controller{
 		nodeInformer:      nodeInformer,
@@ -107,9 +129,10 @@ func NewTaggingController(
 		cloud:             awsCloud,
 		tags:              tags,
 		resources:         resources,
-		workqueue:         workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Tagging"),
+		workqueue:         workqueue.NewNamedRateLimitingQueue(rateLimiter, "Tagging"),
 		nodesSynced:       nodeInformer.Informer().HasSynced,
 		nodeMonitorPeriod: nodeMonitorPeriod,
+		rateLimitEnabled:  rateLimitEnabled,
 	}
 
 	// Use shared informer to listen to add/update/delete of nodes. Note that any nodes
@@ -326,8 +349,14 @@ func (tc *Controller) enqueueNode(node *v1.Node, action func(node *v1.Node) erro
 		requeuingCount: 0,
 		enqueueTime:    time.Now(),
 	}
-	tc.workqueue.Add(item)
-	klog.Infof("Added %s to the workqueue", item)
+
+	if tc.rateLimitEnabled {
+		tc.workqueue.AddRateLimited(item)
+		klog.Infof("Added %s to the workqueue (rate-limited)", item)
+	} else {
+		tc.workqueue.Add(item)
+		klog.Infof("Added %s to the workqueue (without any rate-limit)", item)
+	}
 }
 
 func (tc *Controller) isTaggingRequired(node *v1.Node) bool {
diff --git a/pkg/controllers/tagging/tagging_controller_wrapper.go b/pkg/controllers/tagging/tagging_controller_wrapper.go
index 3722872..9795dd6 100644
--- a/pkg/controllers/tagging/tagging_controller_wrapper.go
+++ b/pkg/controllers/tagging/tagging_controller_wrapper.go
@@ -45,7 +45,9 @@ func (tc *ControllerWrapper) startTaggingController(ctx genericcontrollermanager
 		cloud,
 		completedConfig.ComponentConfig.KubeCloudShared.NodeMonitorPeriod.Duration,
 		tc.Options.Tags,
-		tc.Options.Resources)
+		tc.Options.Resources,
+		tc.Options.RateLimit,
+		tc.Options.BurstLimit)
 
 	if err != nil {
 		klog.Warningf("failed to start tagging controller: %s", err)


Test patch:
diff --git a/pkg/controllers/tagging/tagging_controller_test.go b/pkg/controllers/tagging/tagging_controller_test.go
index 29ff076..f33931b 100644
--- a/pkg/controllers/tagging/tagging_controller_test.go
+++ b/pkg/controllers/tagging/tagging_controller_test.go
@@ -40,6 +40,7 @@ func Test_NodesJoiningAndLeaving(t *testing.T) {
 		currNode         *v1.Node
 		toBeTagged       bool
 		expectedMessages []string
+		rateLimited      bool
 	}{
 		{
 			name: "node0 joins the cluster, but fail to tag.",
@@ -67,7 +68,22 @@ func Test_NodesJoiningAndLeaving(t *testing.T) {
 				},
 			},
 			toBeTagged:       true,
-			expectedMessages: []string{"Successfully tagged i-0001"},
+			expectedMessages: []string{"Successfully tagged i-0001", "to the workqueue (without any rate-limit)"},
+		},
+		{
+			name: "node0 joins the cluster (rate-limited).",
+			currNode: &v1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "node0",
+					CreationTimestamp: metav1.Date(2012, 1, 1, 0, 0, 0, 0, time.UTC),
+				},
+				Spec: v1.NodeSpec{
+					ProviderID: "i-0001",
+				},
+			},
+			toBeTagged:       true,
+			expectedMessages: []string{"Successfully tagged i-0001", "to the workqueue (rate-limited)"},
+			rateLimited:      true,
 		},
 		{
 			name: "node0 joins the cluster and was tagged earlier with different tags.",
@@ -175,6 +191,7 @@ func Test_NodesJoiningAndLeaving(t *testing.T) {
 				tags:              map[string]string{"key2": "value2", "key1": "value1"},
 				resources:         []string{"instance"},
 				workqueue:         workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Tagging"),
+				rateLimitEnabled:  testcase.rateLimited,
 			}
 
 			if testcase.toBeTagged {
@@ -183,6 +200,11 @@ func Test_NodesJoiningAndLeaving(t *testing.T) {
 				tc.enqueueNode(testcase.currNode, tc.untagNodeResources)
 			}
 
+			if tc.rateLimitEnabled {
+				// If rate limit is enabled, sleep for 10 ms to wait for the item to be added to the queue since the base delay is 5 ms.
+				time.Sleep(10 * time.Millisecond)
+			}
+
 			for tc.workqueue.Len() > 0 {
 				tc.process()