Merge pull request #936 from 08volt/annotations

Add Resource Annotations to L4 LB Service

Author: k8s-ci-robot

cmd/cloud-controller-manager/main.go

@@ -67,6 +67,11 @@ var enableDiscretePortForwarding bool
 // LoadBalancerClass
 var enableRBSDefaultForL4NetLB bool
 
+// enableL4LBAnnotations is bound to a command-line flag. It enables
+// the controller to write annotations related to the provisioned resources
+// for L4 Load Balancers services
+var enableL4LBAnnotations bool
+
 func main() {
 	rand.Seed(time.Now().UnixNano())
 
@@ -85,6 +90,7 @@ func main() {
 	cloudProviderFS.BoolVar(&enableMultiProject, "enable-multi-project", false, "Enables project selection from Node providerID for GCE API calls. CAUTION: Only enable if Node providerID is configured by a trusted source.")
 	cloudProviderFS.BoolVar(&enableDiscretePortForwarding, "enable-discrete-port-forwarding", false, "Enables forwarding of individual ports instead of port ranges for GCE external load balancers.")
 	cloudProviderFS.BoolVar(&enableRBSDefaultForL4NetLB, "enable-rbs-default-l4-netlb", false, "Enables RBS defaulting for GCE L4 NetLB")
+	cloudProviderFS.BoolVar(&enableL4LBAnnotations, "enable-l4-lb-annotations", false, "Enables Annotations for GCE L4 LB Services")
 
 	// add new controllers and initializers
 	nodeIpamController := nodeIPAMController{}
@@ -165,12 +171,22 @@ func cloudInitializer(config *config.CompletedConfig) cloudprovider.Interface {
 	if enableRBSDefaultForL4NetLB {
 		gceCloud, ok := (cloud).(*gce.Cloud)
 		if !ok {
-			// Fail-fast: If enableRBSDefaultForGCEL4NetLB is set, the cloud
+			// Fail-fast: If enableRBSDefaultForL4NetLB is set, the cloud
 			// provider MUST be GCE.
 			klog.Fatalf("enable-rbs-default-l4-netlb requires GCE cloud provider, but got %T", cloud)
 		}
 		gceCloud.SetEnableRBSDefaultForL4NetLB(true)
 	}
 
+	if enableL4LBAnnotations {
+		gceCloud, ok := (cloud).(*gce.Cloud)
+		if !ok {
+			// Fail-fast: If enableL4LBAnnotations is set, the cloud
+			// provider MUST be GCE.
+			klog.Fatalf("enable-l4-lb-annotations requires GCE cloud provider, but got %T", cloud)
+		}
+		gceCloud.SetEnableL4LBAnnotations(true)
+	}
+
 	return cloud
 }

providers/gce/BUILD

@@ -68,6 +68,7 @@ go_library(
         "//vendor/k8s.io/apimachinery/pkg/types",
         "//vendor/k8s.io/apimachinery/pkg/util/errors",
         "//vendor/k8s.io/apimachinery/pkg/util/sets",
+        "//vendor/k8s.io/apimachinery/pkg/util/strategicpatch",
         "//vendor/k8s.io/apimachinery/pkg/util/wait",
         "//vendor/k8s.io/apimachinery/pkg/watch",
         "//vendor/k8s.io/client-go/applyconfigurations/core/v1:core",

providers/gce/gce.go

@@ -209,6 +209,9 @@ type Cloud struct {
 
 	// enableRBSDefaultForL4NetLB disable Service controller from picking up services by default
 	enableRBSDefaultForL4NetLB bool
+
+	// enableL4LBAnnotations enable annotations related to provisioned resources in GCE
+	enableL4LBAnnotations bool
 }
 
 // ConfigGlobal is the in memory representation of the gce.conf config data
@@ -870,6 +873,10 @@ func (g *Cloud) SetEnableRBSDefaultForL4NetLB(enabled bool) {
 	g.enableRBSDefaultForL4NetLB = enabled
 }
 
+func (g *Cloud) SetEnableL4LBAnnotations(enabled bool) {
+	g.enableL4LBAnnotations = enabled
+}
+
 // getProjectsBasePath returns the compute API endpoint with the `projects/` element.
 // The suffix must be added when generating compute resource urls.
 func getProjectsBasePath(basePath string) string {

providers/gce/gce_annotations.go

@@ -89,8 +89,28 @@ const (
 
 	// RBSEnabled is an annotation to indicate the Service is opt-in for RBS
 	RBSEnabled = "enabled"
+
+	// serviceStatusPrefix is the prefix used in annotations used to record
+	// debug information in the Service annotations. This is applicable to L4 LB services.
+	serviceStatusPrefix = "networking.gke.io"
+
+	backendServiceResource = "backend-service"
+	targetPoolResource     = "target-pool"
+
+	// backendServiceKey is the annotation key used by l4 controller to record
+	// GCP Backend service name.
+	backendServiceKey = serviceStatusPrefix + "/" + backendServiceResource
+
+	// targetPoolKey is the annotation key used by l4 controller to record
+	// GCP Target pool name.
+	targetPoolKey = serviceStatusPrefix + "/" + targetPoolResource
 )
 
+var l4ResourceAnnotationKeys = []string{
+	backendServiceKey,
+	targetPoolKey,
+}
+
 // GetLoadBalancerAnnotationType returns the type of GCP load balancer which should be assembled.
 func GetLoadBalancerAnnotationType(service *v1.Service) LoadBalancerType {
 	var lbType LoadBalancerType
@@ -176,3 +196,16 @@ func GetLoadBalancerAnnotationSubnet(service *v1.Service) string {
 	}
 	return ""
 }
+
+// mergeMap merges the update map into the dst map.
+// Keys in dst are overwritten by values from update.
+// If a value in the update map is an empty string, the key is removed from dst.
+func mergeMap(dst, update map[string]string) {
+	for k, v := range update {
+		if v == "" {
+			delete(dst, k)
+		} else {
+			dst[k] = v
+		}
+	}
+}

providers/gce/gce_annotations_test.go

@@ -23,7 +23,7 @@ import (
 	"testing"
 
 	"github.com/GoogleCloudPlatform/k8s-cloud-provider/pkg/cloud"
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/stretchr/testify/assert"
@@ -72,3 +72,57 @@ func TestServiceNetworkTierAnnotationKey(t *testing.T) {
 		})
 	}
 }
+
+func TestMergeMap(t *testing.T) {
+	for _, tc := range []struct {
+		desc           string
+		existing       map[string]string
+		updates        map[string]string
+		expectedResult map[string]string
+	}{
+		{
+			desc:     "new annotations should be added",
+			existing: map[string]string{"key1": "val1"},
+			updates:  map[string]string{"key2": "val2"},
+			expectedResult: map[string]string{
+				"key1": "val1",
+				"key2": "val2",
+			},
+		},
+		{
+			desc:     "existing annotations should be overwritten",
+			existing: map[string]string{"key1": "val1"},
+			updates:  map[string]string{"key1": "val2"},
+			expectedResult: map[string]string{
+				"key1": "val2",
+			},
+		},
+		{
+			desc:     "empty value in updates should delete the key",
+			existing: map[string]string{"key1": "val1", "key2": "val2"},
+			updates:  map[string]string{"key1": ""},
+			expectedResult: map[string]string{
+				"key2": "val2",
+			},
+		},
+		{
+			desc:     "mixed updates and deletions",
+			existing: map[string]string{"key1": "val1", "key2": "val2"},
+			updates:  map[string]string{"key1": "new-val", "key2": ""},
+			expectedResult: map[string]string{
+				"key1": "new-val",
+			},
+		},
+		{
+			desc:           "empty updates should result in no change",
+			existing:       map[string]string{"key1": "val1"},
+			updates:        map[string]string{},
+			expectedResult: map[string]string{"key1": "val1"},
+		},
+	} {
+		t.Run(tc.desc, func(t *testing.T) {
+			mergeMap(tc.existing, tc.updates)
+			assert.Equal(t, tc.expectedResult, tc.existing)
+		})
+	}
+}

providers/gce/gce_loadbalancer.go

@@ -21,13 +21,17 @@ package gce
 
 import (
 	"context"
+	"encoding/json"
 	"flag"
 	"fmt"
+	"reflect"
 	"sort"
 	"strings"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/strategicpatch"
 	corev1apply "k8s.io/client-go/applyconfigurations/core/v1"
 	metav1apply "k8s.io/client-go/applyconfigurations/meta/v1"
 	"k8s.io/klog/v2"
@@ -42,6 +46,23 @@ type cidrs struct {
 	isSet bool
 }
 
+type lbSyncResult struct {
+	status      *v1.LoadBalancerStatus
+	annotations map[string]string
+}
+
+func newLBSyncResult() *lbSyncResult {
+	annotations := make(map[string]string, len(l4ResourceAnnotationKeys))
+
+	for _, key := range l4ResourceAnnotationKeys {
+		annotations[key] = "" // Initialize to empty string to indicate deletion by default if not set later
+	}
+
+	return &lbSyncResult{
+		annotations: annotations,
+	}
+}
+
 var (
 	l4LbSrcRngsFlag cidrs
 	l7lbSrcRngsFlag cidrs
@@ -197,19 +218,63 @@ func (g *Cloud) EnsureLoadBalancer(ctx context.Context, clusterName string, svc
 		}
 	}
 
-	var status *v1.LoadBalancerStatus
+	var syncResult *lbSyncResult
 	switch desiredScheme {
 	case cloud.SchemeInternal:
-		status, err = g.ensureInternalLoadBalancer(clusterName, clusterID, svc, existingFwdRule, nodes)
+		syncResult, err = g.ensureInternalLoadBalancer(clusterName, clusterID, svc, existingFwdRule, nodes)
 	default:
-		status, err = g.ensureExternalLoadBalancer(clusterName, clusterID, svc, existingFwdRule, nodes)
+		syncResult, err = g.ensureExternalLoadBalancer(clusterName, clusterID, svc, existingFwdRule, nodes)
 	}
 	if err != nil {
 		klog.Errorf("Failed to EnsureLoadBalancer(%s, %s, %s, %s, %s), err: %v", clusterName, svc.Namespace, svc.Name, loadBalancerName, g.region, err)
-		return status, err
+		return syncResult.status, err
 	}
+
+	if g.enableL4LBAnnotations {
+		if err = g.updateL4ResourcesAnnotations(ctx, svc, syncResult.annotations); err != nil {
+			return syncResult.status, fmt.Errorf("failed to set resource annotations, err: %w", err)
+		}
+	}
+
 	klog.V(4).Infof("EnsureLoadBalancer(%s, %s, %s, %s, %s): done ensuring loadbalancer.", clusterName, svc.Namespace, svc.Name, loadBalancerName, g.region)
-	return status, err
+	return syncResult.status, err
+}
+
+func (g *Cloud) updateL4ResourcesAnnotations(ctx context.Context, svc *v1.Service, newL4LBAnnotations map[string]string) error {
+	newObjectMetadata, shouldUpdate := computeNewAnnotationsIfNeeded(svc, newL4LBAnnotations)
+	if !shouldUpdate {
+		return nil
+	}
+	newSvc := svc.DeepCopy()
+	newSvc.ObjectMeta = *newObjectMetadata
+
+	patchBytes, err := servicePatchBytes(svc, newSvc)
+	if err != nil {
+		return err
+	}
+
+	_, err = g.client.CoreV1().Services(svc.Namespace).Patch(ctx, svc.Name, types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{}, "status")
+
+	return err
+}
+
+func servicePatchBytes(oldSvc, newSvc *v1.Service) ([]byte, error) {
+	oldData, err := json.Marshal(oldSvc)
+	if err != nil {
+		return nil, fmt.Errorf("failed to Marshal oldData for svc %s/%s: %v", oldSvc.Namespace, oldSvc.Name, err)
+	}
+
+	newData, err := json.Marshal(newSvc)
+	if err != nil {
+		return nil, fmt.Errorf("failed to Marshal newData for svc %s/%s: %v", newSvc.Namespace, newSvc.Name, err)
+	}
+
+	patchBytes, err := strategicpatch.CreateTwoWayMergePatch(oldData, newData, v1.Service{})
+	if err != nil {
+		return nil, fmt.Errorf("failed to CreateTwoWayMergePatch for svc %s/%s: %v", oldSvc.Namespace, oldSvc.Name, err)
+	}
+	return patchBytes, nil
+
 }
 
 // UpdateLoadBalancer is an implementation of LoadBalancer.UpdateLoadBalancer.
@@ -326,3 +391,15 @@ func hasLoadBalancerPortsError(service *v1.Service) bool {
 	}
 	return false
 }
+
+// computeNewAnnotationsIfNeeded checks if new annotations should be added to service.
+// If needed creates new service meta object.
+// This function is used by L4 LB controllers.
+func computeNewAnnotationsIfNeeded(svc *v1.Service, newAnnotations map[string]string) (*metav1.ObjectMeta, bool) {
+	newObjectMeta := svc.ObjectMeta.DeepCopy()
+	mergeMap(newObjectMeta.Annotations, newAnnotations)
+	if reflect.DeepEqual(svc.Annotations, newObjectMeta.Annotations) {
+		return nil, false
+	}
+	return newObjectMeta, true
+}

providers/gce/gce_loadbalancer_external.go

@@ -55,7 +55,9 @@ const (
 // Due to an interesting series of design decisions, this handles both creating
 // new load balancers and updating existing load balancers, recognizing when
 // each is needed.
-func (g *Cloud) ensureExternalLoadBalancer(clusterName string, clusterID string, apiService *v1.Service, existingFwdRule *compute.ForwardingRule, nodes []*v1.Node) (*v1.LoadBalancerStatus, error) {
+func (g *Cloud) ensureExternalLoadBalancer(clusterName string, clusterID string, apiService *v1.Service, existingFwdRule *compute.ForwardingRule, nodes []*v1.Node) (*lbSyncResult, error) {
+	syncResult := newLBSyncResult()
+
 	// Process services with LoadBalancerClass "networking.gke.io/l4-regional-external-legacy" used for this controller.
 	// LoadBalancerClass can't be updated so we know this controller should process the NetLB.
 	// Skip service handling if it uses Regional Backend Services and handled by other controllers
@@ -282,6 +284,7 @@ func (g *Cloud) ensureExternalLoadBalancer(clusterName string, clusterID string,
 	if err := g.ensureTargetPoolAndHealthCheck(tpExists, tpNeedsRecreation, apiService, loadBalancerName, clusterID, ipAddressToUse, hosts, hcToCreate, hcToDelete); err != nil {
 		return nil, err
 	}
+	syncResult.annotations[targetPoolKey] = loadBalancerName
 
 	if tpNeedsRecreation || fwdRuleNeedsUpdate {
 		klog.Infof("ensureExternalLoadBalancer(%s): Creating forwarding rule, IP %s (tier: %s).", lbRefStr, ipAddressToUse, netTier)
@@ -299,7 +302,8 @@ func (g *Cloud) ensureExternalLoadBalancer(clusterName string, clusterID string,
 	status := &v1.LoadBalancerStatus{}
 	status.Ingress = []v1.LoadBalancerIngress{{IP: ipAddressToUse}}
 
-	return status, nil
+	syncResult.status = status
+	return syncResult, nil
 }
 
 // updateExternalLoadBalancer is the external implementation of LoadBalancer.UpdateLoadBalancer.