[Cherry Pick from master to 1.22] [FIXES] [Tagging Controller] Fix issues in tagging controller

pkg/controllers/tagging/metrics.go

@@ -27,6 +27,7 @@ var (
 			Name:           "cloudprovider_aws_tagging_controller_work_item_duration_seconds",
 			Help:           "workitem latency of workitem being in the queue and time it takes to process",
 			StabilityLevel: metrics.ALPHA,
+			Buckets:        metrics.ExponentialBuckets(0.5, 1.5, 20),
 		},
 		[]string{"latency_type"})
 

pkg/controllers/tagging/tagging_controller.go

@@ -14,6 +14,7 @@ limitations under the License.
 package tagging
 
 import (
+	"crypto/md5"
 	"fmt"
 	v1 "k8s.io/api/core/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -25,7 +26,10 @@ import (
 	cloudprovider "k8s.io/cloud-provider"
 	opt "k8s.io/cloud-provider-aws/pkg/controllers/options"
 	awsv1 "k8s.io/cloud-provider-aws/pkg/providers/v1"
+	nodehelpers "k8s.io/cloud-provider/node/helpers"
 	"k8s.io/klog/v2"
+	"sort"
+	"strings"
 	"time"
 )
 
@@ -37,7 +41,13 @@ type workItem struct {
 	enqueueTime    time.Time
 }
 
+func (w workItem) String() string {
+	return fmt.Sprintf("[Node: %s, RequeuingCount: %d, EnqueueTime: %s]", w.node.GetName(), w.requeuingCount, w.enqueueTime)
+}
+
 const (
+	taggingControllerLabelKey = "k8s.io/cloud-provider-aws"
+
 	maxRequeuingCount = 9
 
 	// The label for depicting total number of errors a work item encounter and succeed
@@ -105,9 +115,27 @@ func NewTaggingController(
 	// Use shared informer to listen to add/update/delete of nodes. Note that any nodes
 	// that exist before tagging controller starts will show up in the update method
 	tc.nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc:    func(obj interface{}) { tc.enqueueNode(obj, tc.tagNodesResources) },
-		UpdateFunc: func(oldObj, newObj interface{}) { tc.enqueueNode(newObj, tc.tagNodesResources) },
-		DeleteFunc: func(obj interface{}) { tc.enqueueNode(obj, tc.untagNodeResources) },
+		AddFunc: func(obj interface{}) {
+			node := obj.(*v1.Node)
+			tc.enqueueNode(node, tc.tagNodesResources)
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			node := newObj.(*v1.Node)
+			// Check if tagging is required by inspecting the labels. This check here prevents us from putting a tagged node into the
+			// work queue. We check this again before tagging the node to make sure that between when a node was put in the work queue
+			// and when it gets tagged, there might be another event which put the same item in the work queue
+			// (since the node won't have the labels yet) and hence prevents us from making an unnecessary EC2 call.
+			if !tc.isTaggingRequired(node) {
+				klog.Infof("Skip putting node %s in work queue since it was already tagged earlier.", node.GetName())
+				return
+			}
+
+			tc.enqueueNode(node, tc.tagNodesResources)
+		},
+		DeleteFunc: func(obj interface{}) {
+			node := obj.(*v1.Node)
+			tc.enqueueNode(node, tc.untagNodeResources)
+		},
 	})
 
 	return tc, nil
@@ -147,28 +175,36 @@ func (tc *Controller) process() bool {
 		return false
 	}
 
-	klog.Infof("Starting to process %v", obj)
+	klog.Infof("Starting to process %s", obj)
 
 	err := func(obj interface{}) error {
 		defer tc.workqueue.Done(obj)
 
 		workItem, ok := obj.(*workItem)
 		if !ok {
 			tc.workqueue.Forget(obj)
-			err := fmt.Errorf("expected workItem in workqueue but got %#v", obj)
+			err := fmt.Errorf("expected workItem in workqueue but got %s", obj)
 			utilruntime.HandleError(err)
 			return nil
 		}
 
 		timeTaken := time.Since(workItem.enqueueTime).Seconds()
 		recordWorkItemLatencyMetrics(workItemDequeuingTimeWorkItemMetric, timeTaken)
+		klog.Infof("Dequeuing latency %s", timeTaken)
 
 		instanceID, err := awsv1.KubernetesInstanceID(workItem.node.Spec.ProviderID).MapToAWSInstanceID()
 		if err != nil {
 			err = fmt.Errorf("Error in getting instanceID for node %s, error: %v", workItem.node.GetName(), err)
 			utilruntime.HandleError(err)
 			return nil
 		}
+		klog.Infof("Instance ID of work item %s is %s", workItem, instanceID)
+
+		if awsv1.IsFargateNode(string(instanceID)) {
+			klog.Infof("Skip processing the node %s since it is a Fargate node", instanceID)
+			tc.workqueue.Forget(obj)
+			return nil
+		}
 
 		err = workItem.action(workItem.node)
 
@@ -182,20 +218,21 @@ func (tc *Controller) process() bool {
 				return fmt.Errorf("error processing work item '%v': %s, requeuing count %d", workItem, err.Error(), workItem.requeuingCount)
 			}
 
-			klog.Errorf("error processing work item '%v': %s, requeuing count exceeded", workItem, err.Error())
+			klog.Errorf("error processing work item %s: %s, requeuing count exceeded", workItem, err.Error())
 			recordWorkItemErrorMetrics(errorsAfterRetriesExhaustedWorkItemErrorMetric, string(instanceID))
 		} else {
-			klog.Infof("Finished processing %v", workItem)
+			klog.Infof("Finished processing %s", workItem)
 			timeTaken = time.Since(workItem.enqueueTime).Seconds()
 			recordWorkItemLatencyMetrics(workItemProcessingTimeWorkItemMetric, timeTaken)
+			klog.Infof("Processing latency %s", timeTaken)
 		}
 
 		tc.workqueue.Forget(obj)
 		return nil
 	}(obj)
 
 	if err != nil {
-		klog.Errorf("Error occurred while processing %v", obj)
+		klog.Errorf("Error occurred while processing %s", obj)
 		utilruntime.HandleError(err)
 	}
 
@@ -221,16 +258,28 @@ func (tc *Controller) tagNodesResources(node *v1.Node) error {
 // tagEc2Instances applies the provided tags to each EC2 instance in
 // the cluster.
 func (tc *Controller) tagEc2Instance(node *v1.Node) error {
+	if !tc.isTaggingRequired(node) {
+		klog.Infof("Skip tagging node %s since it was already tagged earlier.", node.GetName())
+		return nil
+	}
+
 	instanceID, _ := awsv1.KubernetesInstanceID(node.Spec.ProviderID).MapToAWSInstanceID()
 
 	err := tc.cloud.TagResource(string(instanceID), tc.tags)
 
 	if err != nil {
-		klog.Errorf("Error in tagging EC2 instance for node %s, error: %v", node.GetName(), err)
+		klog.Errorf("Error in tagging EC2 instance %s for node %s, error: %v", instanceID, node.GetName(), err)
 		return err
 	}
 
-	klog.Infof("Successfully tagged %s with %v", instanceID, tc.tags)
+	labels := map[string]string{taggingControllerLabelKey: tc.getChecksumOfTags()}
+	klog.Infof("Successfully tagged %s with %v. Labeling the nodes with tagging controller labels now.", instanceID, tc.tags)
+	if !nodehelpers.AddOrUpdateLabelsOnNode(tc.kubeClient, labels, node) {
+		klog.Errorf("Couldn't apply labels %s to node %s.", labels, node.GetName())
+		return fmt.Errorf("couldn't apply labels %s to node %s", labels, node.GetName())
+	}
+
+	klog.Infof("Successfully labeled node %s with %v.", node.GetName(), labels)
 
 	return nil
 }
@@ -259,7 +308,7 @@ func (tc *Controller) untagEc2Instance(node *v1.Node) error {
 	err := tc.cloud.UntagResource(string(instanceID), tc.tags)
 
 	if err != nil {
-		klog.Errorf("Error in untagging EC2 instance for node %s, error: %v", node.GetName(), err)
+		klog.Errorf("Error in untagging EC2 instance %s for node %s, error: %v", instanceID, node.GetName(), err)
 		return err
 	}
 
@@ -270,8 +319,7 @@ func (tc *Controller) untagEc2Instance(node *v1.Node) error {
 
 // enqueueNode takes in the object and an
 // action for the object for a workitem and enqueue to the workqueue
-func (tc *Controller) enqueueNode(obj interface{}, action func(node *v1.Node) error) {
-	node := obj.(*v1.Node)
+func (tc *Controller) enqueueNode(node *v1.Node, action func(node *v1.Node) error) {
 	item := &workItem{
 		node:           node,
 		action:         action,
@@ -281,3 +329,24 @@ func (tc *Controller) enqueueNode(obj interface{}, action func(node *v1.Node) er
 	tc.workqueue.Add(item)
 	klog.Infof("Added %s to the workqueue", item)
 }
+
+func (tc *Controller) isTaggingRequired(node *v1.Node) bool {
+	if node.Labels == nil {
+		return true
+	}
+
+	if labelValue, ok := node.Labels[taggingControllerLabelKey]; !ok || labelValue != tc.getChecksumOfTags() {
+		return true
+	}
+
+	return false
+}
+
+func (tc *Controller) getChecksumOfTags() string {
+	tags := []string{}
+	for key, value := range tc.tags {
+		tags = append(tags, key+"="+value)
+	}
+	sort.Strings(tags)
+	return fmt.Sprintf("%x", md5.Sum([]byte(strings.Join(tags, ","))))
+}

pkg/controllers/tagging/tagging_controller_test.go

@@ -69,6 +69,54 @@ func Test_NodesJoiningAndLeaving(t *testing.T) {
 			toBeTagged:       true,
 			expectedMessages: []string{"Successfully tagged i-0001"},
 		},
+		{
+			name: "node0 joins the cluster and was tagged earlier with different tags.",
+			currNode: &v1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "node0",
+					CreationTimestamp: metav1.Date(2012, 1, 1, 0, 0, 0, 0, time.UTC),
+					Labels: map[string]string{
+						taggingControllerLabelKey: "9767c4972ba72e87ab553bad2afde741", // MD5 for key1=value1
+					},
+				},
+				Spec: v1.NodeSpec{
+					ProviderID: "i-0001",
+				},
+			},
+			toBeTagged:       true,
+			expectedMessages: []string{"Successfully tagged i-0001"},
+		},
+		{
+			name: "node0 joins the cluster but isn't tagged because it was already tagged earlier.",
+			currNode: &v1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "node0",
+					CreationTimestamp: metav1.Date(2012, 1, 1, 0, 0, 0, 0, time.UTC),
+					Labels: map[string]string{
+						taggingControllerLabelKey: "c812faa65d1d5e5aefa6b069b3da39df", // MD5 for key1=value1,key2=value2
+					},
+				},
+				Spec: v1.NodeSpec{
+					ProviderID: "i-0001",
+				},
+			},
+			toBeTagged:       true,
+			expectedMessages: []string{"Skip tagging node node0 since it was already tagged earlier."},
+		},
+		{
+			name: "fargate node joins the cluster.",
+			currNode: &v1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:              "fargatenode0",
+					CreationTimestamp: metav1.Date(2012, 1, 1, 0, 0, 0, 0, time.UTC),
+				},
+				Spec: v1.NodeSpec{
+					ProviderID: "aws:///us-west-2a/2ea696a557-9e55466d21eb4f83a99a9aa396bbd134/fargate-ip-10-0-55-27.us-west-2.compute.internal",
+				},
+			},
+			toBeTagged:       true,
+			expectedMessages: []string{"Skip processing the node fargate-ip-10-0-55-27.us-west-2.compute.internal since it is a Fargate node"},
+		},
 		{
 			name: "node0 leaves the cluster, failed to untag.",
 			currNode: &v1.Node{
@@ -81,7 +129,7 @@ func Test_NodesJoiningAndLeaving(t *testing.T) {
 				},
 			},
 			toBeTagged:       false,
-			expectedMessages: []string{"Error in untagging EC2 instance for node node0"},
+			expectedMessages: []string{"Error in untagging EC2 instance i-error for node node0"},
 		},
 		{
 			name: "node0 leaves the cluster.",
@@ -124,7 +172,7 @@ func Test_NodesJoiningAndLeaving(t *testing.T) {
 				kubeClient:        clientset,
 				cloud:             fakeAws,
 				nodeMonitorPeriod: 1 * time.Second,
-				tags:              map[string]string{"key": "value"},
+				tags:              map[string]string{"key2": "value2", "key1": "value1"},
 				resources:         []string{"instance"},
 				workqueue:         workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Tagging"),
 			}

pkg/providers/v1/aws.go

@@ -1836,7 +1836,7 @@ func (c *Cloud) NodeAddressesByProviderID(ctx context.Context, providerID string
 		return nil, err
 	}
 
-	if isFargateNode(string(instanceID)) {
+	if IsFargateNode(string(instanceID)) {
 		eni, err := c.describeNetworkInterfaces(string(instanceID))
 		if eni == nil || err != nil {
 			return nil, err
@@ -1879,7 +1879,7 @@ func (c *Cloud) InstanceExistsByProviderID(ctx context.Context, providerID strin
 		return false, err
 	}
 
-	if isFargateNode(string(instanceID)) {
+	if IsFargateNode(string(instanceID)) {
 		eni, err := c.describeNetworkInterfaces(string(instanceID))
 		return eni != nil, err
 	}
@@ -1919,7 +1919,7 @@ func (c *Cloud) InstanceShutdownByProviderID(ctx context.Context, providerID str
 		return false, err
 	}
 
-	if isFargateNode(string(instanceID)) {
+	if IsFargateNode(string(instanceID)) {
 		eni, err := c.describeNetworkInterfaces(string(instanceID))
 		return eni != nil, err
 	}
@@ -1980,7 +1980,7 @@ func (c *Cloud) InstanceTypeByProviderID(ctx context.Context, providerID string)
 		return "", err
 	}
 
-	if isFargateNode(string(instanceID)) {
+	if IsFargateNode(string(instanceID)) {
 		return "", nil
 	}
 
@@ -2089,7 +2089,7 @@ func (c *Cloud) GetZoneByProviderID(ctx context.Context, providerID string) (clo
 		return cloudprovider.Zone{}, err
 	}
 
-	if isFargateNode(string(instanceID)) {
+	if IsFargateNode(string(instanceID)) {
 		eni, err := c.describeNetworkInterfaces(string(instanceID))
 		if eni == nil || err != nil {
 			return cloudprovider.Zone{}, err
@@ -5248,8 +5248,8 @@ func (c *Cloud) getFullInstance(nodeName types.NodeName) (*awsInstance, *ec2.Ins
 	return awsInstance, instance, err
 }
 
-// isFargateNode returns true if given node runs on Fargate compute
-func isFargateNode(nodeName string) bool {
+// IsFargateNode returns true if given node runs on Fargate compute
+func IsFargateNode(nodeName string) bool {
 	return strings.HasPrefix(nodeName, fargateNodeNamePrefix)
 }
 

pkg/providers/v1/instances.go

@@ -78,7 +78,7 @@ func (name KubernetesInstanceID) MapToAWSInstanceID() (InstanceID, error) {
 
 	// We sanity check the resulting volume; the two known formats are
 	// i-12345678 and i-12345678abcdef01
-	if awsID == "" || !(awsInstanceRegMatch.MatchString(awsID) || isFargateNode(awsID)) {
+	if awsID == "" || !(awsInstanceRegMatch.MatchString(awsID) || IsFargateNode(awsID)) {
 		return "", fmt.Errorf("Invalid format for AWS instance (%s)", name)
 	}