Repository: linode/linode-cloud-controller-manager
Pull request #168 (2024-01-14 19:30:37)

always return lbnotfound from getNodeBalancerForService
Found an edge case where I had deleted a nodebalancer but it was set on a service and wouldn't clear the finalizer and never saw the short-circuit log from here: https://github.com/linode/linode-cloud-controller-manager/blob/main/cloud/linode/loadbalancers.go#L413

the lbnotfound type check was rewriting to a new error and never returning so moving the onus on checking lbnotfound to the create path so a second load balancer is not created and the skip now works.

okokes-akamai: Can you reflect this change in tests? Fix the failing one and perhaps add some new ones.
luthermonson: > Can you reflect this change in tests? Fix the failing one and perhaps add some new ones.

tests added, fail in main but pass in my branch

Patch:
diff --git a/cloud/linode/common.go b/cloud/linode/common.go
index a56c1a6..c4562c3 100644
--- a/cloud/linode/common.go
+++ b/cloud/linode/common.go
@@ -16,12 +16,8 @@ func (e invalidProviderIDError) Error() string {
 	return fmt.Sprintf("invalid provider ID %q", e.value)
 }
 
-func isLinodeProviderID(providerID string) bool {
-	return strings.HasPrefix(providerID, providerIDPrefix)
-}
-
 func parseProviderID(providerID string) (int, error) {
-	if !isLinodeProviderID(providerID) {
+	if !strings.HasPrefix(providerID, providerIDPrefix) {
 		return 0, invalidProviderIDError{providerID}
 	}
 	id, err := strconv.Atoi(strings.TrimPrefix(providerID, providerIDPrefix))
diff --git a/cloud/linode/instances.go b/cloud/linode/instances.go
index da0da98..22e8e46 100644
--- a/cloud/linode/instances.go
+++ b/cloud/linode/instances.go
@@ -53,11 +53,12 @@ type instances struct {
 }
 
 func newInstances(client Client) *instances {
-	timeout := 15
+	var timeout int
 	if raw, ok := os.LookupEnv("LINODE_INSTANCE_CACHE_TTL"); ok {
-		if t, _ := strconv.Atoi(raw); t > 0 {
-			timeout = t
-		}
+		timeout, _ = strconv.Atoi(raw)
+	}
+	if timeout == 0 {
+		timeout = 15
 	}
 
 	return &instances{client, &nodeCache{
@@ -107,7 +108,7 @@ func (i *instances) lookupLinode(ctx context.Context, node *v1.Node) (*linodego.
 	sentry.SetTag(ctx, "provider_id", providerID)
 	sentry.SetTag(ctx, "node_name", node.Name)
 
-	if providerID != "" && isLinodeProviderID(providerID){
+	if providerID != "" {
 		id, err := parseProviderID(providerID)
 		if err != nil {
 			sentry.CaptureError(ctx, err)
diff --git a/cloud/linode/loadbalancers.go b/cloud/linode/loadbalancers.go
index d2088b6..e3346cf 100644
--- a/cloud/linode/loadbalancers.go
+++ b/cloud/linode/loadbalancers.go
@@ -70,17 +70,7 @@ func (l *loadbalancers) getNodeBalancerForService(ctx context.Context, service *
 
 	if hasIDAnn {
 		sentry.SetTag(ctx, "load_balancer_id", rawID)
-		nb, err := l.getNodeBalancerByID(ctx, service, id)
-		switch err.(type) {
-		case nil:
-			return nb, nil
-
-		case lbNotFoundError:
-			return nil, fmt.Errorf("%s annotation points to a NodeBalancer that does not exist: %s", annLinodeNodeBalancerID, err)
-
-		default:
-			return nil, err
-		}
+		return l.getNodeBalancerByID(ctx, service, id)
 	}
 	return l.getNodeBalancerByStatus(ctx, service)
 }
@@ -120,7 +110,7 @@ func (l *loadbalancers) getNodeBalancerByStatus(ctx context.Context, service *v1
 func (l *loadbalancers) cleanupOldNodeBalancer(ctx context.Context, service *v1.Service) error {
 	// unless there's an annotation, we can never get a past and current NB to differ,
 	// because they're looked up the same way
-	if _, ok := service.GetAnnotations()[annLinodeNodeBalancerID]; !ok {
+	if _, ok := getServiceAnnotation(service, annLinodeNodeBalancerID); !ok {
 		return nil
 	}
 
@@ -199,6 +189,13 @@ func (l *loadbalancers) EnsureLoadBalancer(ctx context.Context, clusterName stri
 	nb, err = l.getNodeBalancerForService(ctx, service)
 	switch err.(type) {
 	case lbNotFoundError:
+		if service.GetAnnotations()[annLinodeNodeBalancerID] != "" {
+			// a load balancer annotation has been created so a NodeBalancer is coming, error out and retry later
+			klog.Infof("NodeBalancer created but not available yet, waiting...")
+			sentry.CaptureError(ctx, err)
+			return nil, err
+		}
+
 		if nb, err = l.buildLoadBalancerRequest(ctx, clusterName, service, nodes); err != nil {
 			sentry.CaptureError(ctx, err)
 			return nil, err
@@ -474,7 +471,7 @@ func (l *loadbalancers) getNodeBalancerByID(ctx context.Context, service *v1.Ser
 
 func (l *loadbalancers) getLoadBalancerTags(_ context.Context, clusterName string, service *v1.Service) []string {
 	tags := []string{clusterName}
-	tagStr, ok := service.GetAnnotations()[annLinodeLoadBalancerTags]
+	tagStr, ok := getServiceAnnotation(service, annLinodeLoadBalancerTags)
 	if ok {
 		return append(tags, strings.Split(tagStr, ",")...)
 	}
@@ -494,7 +491,7 @@ func (l *loadbalancers) createNodeBalancer(ctx context.Context, clusterName stri
 		Tags:               tags,
 	}
 
-	fwid, ok := service.GetAnnotations()[annLinodeCloudFirewallID]
+	fwid, ok := getServiceAnnotation(service, annLinodeCloudFirewallID)
 	if ok {
 		firewallID, err := strconv.Atoi(fwid)
 		if err != nil {
@@ -525,7 +522,7 @@ func (l *loadbalancers) buildNodeBalancerConfig(ctx context.Context, service *v1
 	}
 
 	if health == linodego.CheckHTTP || health == linodego.CheckHTTPBody {
-		path := service.GetAnnotations()[annLinodeCheckPath]
+		path := service.Annotations[annLinodeCheckPath]
 		if path == "" {
 			path = "/"
 		}
@@ -533,14 +530,14 @@ func (l *loadbalancers) buildNodeBalancerConfig(ctx context.Context, service *v1
 	}
 
 	if health == linodego.CheckHTTPBody {
-		body := service.GetAnnotations()[annLinodeCheckBody]
+		body := service.Annotations[annLinodeCheckBody]
 		if body == "" {
 			return config, fmt.Errorf("for health check type http_body need body regex annotation %v", annLinodeCheckBody)
 		}
 		config.CheckBody = body
 	}
 	checkInterval := 5
-	if ci, ok := service.GetAnnotations()[annLinodeHealthCheckInterval]; ok {
+	if ci, ok := service.Annotations[annLinodeHealthCheckInterval]; ok {
 		if checkInterval, err = strconv.Atoi(ci); err != nil {
 			return config, err
 		}
@@ -548,7 +545,7 @@ func (l *loadbalancers) buildNodeBalancerConfig(ctx context.Context, service *v1
 	config.CheckInterval = checkInterval
 
 	checkTimeout := 3
-	if ct, ok := service.GetAnnotations()[annLinodeHealthCheckTimeout]; ok {
+	if ct, ok := service.Annotations[annLinodeHealthCheckTimeout]; ok {
 		if checkTimeout, err = strconv.Atoi(ct); err != nil {
 			return config, err
 		}
@@ -556,7 +553,7 @@ func (l *loadbalancers) buildNodeBalancerConfig(ctx context.Context, service *v1
 	config.CheckTimeout = checkTimeout
 
 	checkAttempts := 2
-	if ca, ok := service.GetAnnotations()[annLinodeHealthCheckAttempts]; ok {
+	if ca, ok := service.Annotations[annLinodeHealthCheckAttempts]; ok {
 		if checkAttempts, err = strconv.Atoi(ca); err != nil {
 			return config, err
 		}
@@ -564,7 +561,7 @@ func (l *loadbalancers) buildNodeBalancerConfig(ctx context.Context, service *v1
 	config.CheckAttempts = checkAttempts
 
 	checkPassive := true
-	if cp, ok := service.GetAnnotations()[annLinodeHealthCheckPassive]; ok {
+	if cp, ok := service.Annotations[annLinodeHealthCheckPassive]; ok {
 		if checkPassive, err = strconv.ParseBool(cp); err != nil {
 			return config, err
 		}
@@ -685,7 +682,7 @@ func getPortConfig(service *v1.Service, port int) (portConfig, error) {
 	protocol := portConfigAnnotation.Protocol
 	if protocol == "" {
 		protocol = "tcp"
-		if p, ok := service.GetAnnotations()[annLinodeDefaultProtocol]; ok {
+		if p, ok := service.Annotations[annLinodeDefaultProtocol]; ok {
 			protocol = p
 		}
 	}
@@ -695,7 +692,7 @@ func getPortConfig(service *v1.Service, port int) (portConfig, error) {
 	if proxyProtocol == "" {
 		proxyProtocol = string(linodego.ProxyProtocolNone)
 		for _, ann := range []string{annLinodeDefaultProxyProtocol, annLinodeProxyProtocolDeprecated} {
-			if pp, ok := service.GetAnnotations()[ann]; ok {
+			if pp, ok := service.Annotations[ann]; ok {
 				proxyProtocol = pp
 				break
 			}
@@ -722,7 +719,7 @@ func getPortConfig(service *v1.Service, port int) (portConfig, error) {
 }
 
 func getHealthCheckType(service *v1.Service) (linodego.ConfigCheck, error) {
-	hType, ok := service.GetAnnotations()[annLinodeHealthCheckType]
+	hType, ok := service.Annotations[annLinodeHealthCheckType]
 	if !ok {
 		return linodego.CheckConnection, nil
 	}
@@ -735,7 +732,7 @@ func getHealthCheckType(service *v1.Service) (linodego.ConfigCheck, error) {
 func getPortConfigAnnotation(service *v1.Service, port int) (portConfigAnnotation, error) {
 	annotation := portConfigAnnotation{}
 	annotationKey := annLinodePortConfigPrefix + strconv.Itoa(port)
-	annotationJSON, ok := service.GetAnnotations()[annotationKey]
+	annotationJSON, ok := service.Annotations[annotationKey]
 
 	if !ok {
 		return annotation, nil
@@ -789,7 +786,7 @@ func getTLSCertInfo(ctx context.Context, kubeClient kubernetes.Interface, namesp
 func getConnectionThrottle(service *v1.Service) int {
 	connThrottle := 20
 
-	if connThrottleString := service.GetAnnotations()[annLinodeThrottle]; connThrottleString != "" {
+	if connThrottleString := service.Annotations[annLinodeThrottle]; connThrottleString != "" {
 		parsed, err := strconv.Atoi(connThrottleString)
 		if err == nil {
 			if parsed < 0 {
@@ -836,8 +833,16 @@ func getServiceNn(service *v1.Service) string {
 	return fmt.Sprintf("%s/%s", service.Namespace, service.Name)
 }
 
+func getServiceAnnotation(service *v1.Service, name string) (string, bool) {
+	if service.Annotations == nil {
+		return "", false
+	}
+	val, ok := service.Annotations[name]
+	return val, ok
+}
+
 func getServiceBoolAnnotation(service *v1.Service, name string) bool {
-	value, ok := service.GetAnnotations()[name]
+	value, ok := getServiceAnnotation(service, name)
 	if !ok {
 		return false
 	}
diff --git a/cloud/linode/node_controller.go b/cloud/linode/node_controller.go
index ed53a41..5834d19 100644
--- a/cloud/linode/node_controller.go
+++ b/cloud/linode/node_controller.go
@@ -3,9 +3,6 @@ package linode
 import (
 	"context"
 	"net/http"
-	"os"
-	"strconv"
-	"sync"
 	"time"
 
 	"github.com/appscode/go/wait"
@@ -20,35 +17,21 @@ import (
 )
 
 type nodeController struct {
-	sync.RWMutex
-
 	client     Client
 	instances  *instances
 	kubeclient kubernetes.Interface
 	informer   v1informers.NodeInformer
 
-	metadataLastUpdate map[string]time.Time
-	ttl                time.Duration
-
 	queue workqueue.DelayingInterface
 }
 
 func newNodeController(kubeclient kubernetes.Interface, client Client, informer v1informers.NodeInformer) *nodeController {
-	timeout := 300
-	if raw, ok := os.LookupEnv("LINODE_METADATA_TTL"); ok {
-		if t, _ := strconv.Atoi(raw); t > 0 {
-			timeout = t
-		}
-	}
-
 	return &nodeController{
-		client:             client,
-		instances:          newInstances(client),
-		kubeclient:         kubeclient,
-		informer:           informer,
-		ttl:                time.Duration(timeout) * time.Second,
-		metadataLastUpdate: make(map[string]time.Time),
-		queue:              workqueue.NewDelayingQueue(),
+		client:     client,
+		instances:  newInstances(client),
+		kubeclient: kubeclient,
+		informer:   informer,
+		queue:      workqueue.NewDelayingQueue(),
 	}
 }
 
@@ -63,6 +46,15 @@ func (s *nodeController) Run(stopCh <-chan struct{}) {
 			klog.Infof("NodeController will handle newly created node (%s) metadata", node.Name)
 			s.queue.Add(node)
 		},
+		UpdateFunc: func(_, new interface{}) {
+			node, ok := new.(*v1.Node)
+			if !ok {
+				return
+			}
+
+			klog.Infof("NodeController will handle updated node (%s) metadata", node.Name)
+			s.queue.Add(node)
+		},
 	})
 
 	go wait.Until(s.worker, time.Second, stopCh)
@@ -89,7 +81,7 @@ func (s *nodeController) processNext() bool {
 		return true
 	}
 
-	err := s.handleNode(context.TODO(), node)
+	err := s.handleNodeAdded(context.TODO(), node)
 	switch deleteErr := err.(type) {
 	case nil:
 		break
@@ -106,27 +98,8 @@ func (s *nodeController) processNext() bool {
 	return true
 }
 
-func (s *nodeController) LastMetadataUpdate(nodeName string) time.Time {
-	s.RLock()
-	defer s.RUnlock()
-	return s.metadataLastUpdate[nodeName]
-}
-
-func (s *nodeController) SetLastMetadataUpdate(nodeName string) {
-	s.Lock()
-	defer s.Unlock()
-	s.metadataLastUpdate[nodeName] = time.Now()
-}
-
-func (s *nodeController) handleNode(ctx context.Context, node *v1.Node) error {
-	klog.Infof("NodeController handling node (%s) metadata", node.Name)
-
-	lastUpdate := s.LastMetadataUpdate(node.Name)
-
-	uuid, ok := node.Labels[annLinodeHostUUID]
-	if ok && time.Since(lastUpdate) < s.ttl {
-		return nil
-	}
+func (s *nodeController) handleNodeAdded(ctx context.Context, node *v1.Node) error {
+	klog.Infof("NodeController handling node (%s) addition", node.Name)
 
 	linode, err := s.instances.lookupLinode(ctx, node)
 	if err != nil {
@@ -134,8 +107,8 @@ func (s *nodeController) handleNode(ctx context.Context, node *v1.Node) error {
 		return err
 	}
 
-	if uuid == linode.HostUUID {
-		s.SetLastMetadataUpdate(node.Name)
+	uuid, ok := node.Labels[annLinodeHostUUID]
+	if ok && uuid == linode.HostUUID {
 		return nil
 	}
 
@@ -147,7 +120,5 @@ func (s *nodeController) handleNode(ctx context.Context, node *v1.Node) error {
 		return err
 	}
 
-	s.SetLastMetadataUpdate(node.Name)
-
 	return nil
 }


Test patch:
diff --git a/cloud/linode/instances_test.go b/cloud/linode/instances_test.go
index a93711a..878b129 100644
--- a/cloud/linode/instances_test.go
+++ b/cloud/linode/instances_test.go
@@ -233,6 +233,17 @@ func TestMalformedProviders(t *testing.T) {
 
 	client := NewMockClient(ctrl)
 
+	t.Run("fails on malformed providerID", func(t *testing.T) {
+		instances := newInstances(client)
+		providerID := "bogus://bogus"
+		node := nodeWithProviderID(providerID)
+		client.EXPECT().ListInstances(gomock.Any(), nil).Times(1).Return([]linodego.Instance{}, nil)
+
+		meta, err := instances.InstanceMetadata(ctx, node)
+		assert.ErrorIs(t, err, invalidProviderIDError{providerID})
+		assert.Nil(t, meta)
+	})
+
 	t.Run("fails on non-numeric providerID", func(t *testing.T) {
 		instances := newInstances(client)
 		providerID := providerIDPrefix + "abc"
diff --git a/cloud/linode/loadbalancers_test.go b/cloud/linode/loadbalancers_test.go
index fd20129..67ca7bb 100644
--- a/cloud/linode/loadbalancers_test.go
+++ b/cloud/linode/loadbalancers_test.go
@@ -1776,10 +1776,12 @@ func testGetNodeBalancerForServiceIDDoesNotExist(t *testing.T, client *linodego.
 		t.Fatal("expected getNodeBalancerForService to return an error")
 	}
 
-	expectedErr := fmt.Sprintf("%s annotation points to a NodeBalancer that does not exist: LoadBalancer (%s) not found for service (%s)",
-		annLinodeNodeBalancerID, bogusNodeBalancerID, getServiceNn(svc))
-
-	if err.Error() != expectedErr {
+	nbid, _ := strconv.Atoi(bogusNodeBalancerID)
+	expectedErr := lbNotFoundError{
+		serviceNn:      getServiceNn(svc),
+		nodeBalancerID: nbid,
+	}
+	if err.Error() != expectedErr.Error() {
 		t.Errorf("expected error to be '%s' but got '%s'", expectedErr, err)
 	}
 }
diff --git a/e2e/test/ccm_e2e_test.go b/e2e/test/ccm_e2e_test.go
index 7d4373a..d5ce41b 100644
--- a/e2e/test/ccm_e2e_test.go
+++ b/e2e/test/ccm_e2e_test.go
@@ -6,6 +6,9 @@ import (
 	"fmt"
 	"os/exec"
 	"strconv"
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/errors"
 
 	"github.com/linode/linodego"
 	. "github.com/onsi/ginkgo/v2"
@@ -90,6 +93,14 @@ var _ = Describe("e2e tests", func() {
 		Eventually(watcher.ResultChan()).Should(Receive(EnsuredService()))
 	}
 
+	ensureServiceWasDeleted := func() {
+		err := func() error {
+			_, err := f.LoadBalancer.GetService()
+			return err
+		}
+		Eventually(err).WithTimeout(10 * time.Second).Should(MatchError(errors.IsNotFound, "IsNotFound"))
+	}
+
 	createServiceWithSelector := func(selector map[string]string, ports []core.ServicePort, isSessionAffinityClientIP bool) {
 		Expect(f.LoadBalancer.CreateService(selector, nil, ports, isSessionAffinityClientIP)).NotTo(HaveOccurred())
 		Eventually(f.LoadBalancer.GetServiceEndpoints).Should(Not(BeEmpty()))
@@ -942,7 +953,7 @@ var _ = Describe("e2e tests", func() {
 					By("Creating new NodeBalancer")
 					nbID := createNodeBalancer()
 
-					By("Waiting for currenct NodeBalancer to be ready")
+					By("Waiting for current NodeBalancer to be ready")
 					checkNodeBalancerID(framework.TestServerResourceName, nodeBalancerID)
 
 					By("Annotating service with new NodeBalancer ID")
@@ -957,6 +968,97 @@ var _ = Describe("e2e tests", func() {
 				})
 			})
 
+			Context("Deleted Service when NodeBalancer not present", func() {
+				var (
+					pods         []string
+					labels       map[string]string
+					annotations  map[string]string
+					servicePorts []core.ServicePort
+
+					nodeBalancerID int
+				)
+
+				BeforeEach(func() {
+					pods = []string{"test-pod-1"}
+					ports := []core.ContainerPort{
+						{
+							Name:          "http-1",
+							ContainerPort: 8080,
+						},
+					}
+					servicePorts = []core.ServicePort{
+						{
+							Name:       "http-1",
+							Port:       80,
+							TargetPort: intstr.FromInt(8080),
+							Protocol:   "TCP",
+						},
+					}
+
+					labels = map[string]string{
+						"app": "test-loadbalancer-with-nodebalancer-id",
+					}
+
+					By("Creating NodeBalancer")
+					nodeBalancerID = createNodeBalancer()
+
+					annotations = map[string]string{
+						annLinodeNodeBalancerID: strconv.Itoa(nodeBalancerID),
+					}
+
+					By("Creating Pod")
+					createPodWithLabel(pods, ports, framework.TestServerImage, labels, false)
+
+					By("Creating Service")
+					createServiceWithAnnotations(labels, annotations, servicePorts, false)
+				})
+
+				AfterEach(func() {
+					By("Deleting the Pods")
+					deletePods(pods)
+
+					err := root.Recycle()
+					Expect(err).NotTo(HaveOccurred())
+				})
+
+				It("should use the specified NodeBalancer", func() {
+					By("Checking the NodeBalancerID")
+					checkNodeBalancerID(framework.TestServerResourceName, nodeBalancerID)
+				})
+
+				It("should use the newly specified NodeBalancer ID", func() {
+					By("Creating new NodeBalancer")
+					nbID := createNodeBalancer()
+
+					By("Waiting for current NodeBalancer to be ready")
+					checkNodeBalancerID(framework.TestServerResourceName, nodeBalancerID)
+
+					By("Annotating service with new NodeBalancer ID")
+					annotations[annLinodeNodeBalancerID] = strconv.Itoa(nbID)
+					updateServiceWithAnnotations(labels, annotations, servicePorts, false)
+
+					By("Checking the NodeBalancer ID")
+					checkNodeBalancerID(framework.TestServerResourceName, nbID)
+
+					By("Checking old NodeBalancer was deleted")
+					checkNodeBalancerNotExists(nodeBalancerID)
+				})
+
+				It("should delete the service with no NodeBalancer present", func() {
+					By("Deleting the NodeBalancer")
+					deleteNodeBalancer(nodeBalancerID)
+
+					By("Checking old NodeBalancer was deleted")
+					checkNodeBalancerNotExists(nodeBalancerID)
+
+					By("Deleting the Service")
+					deleteService()
+
+					By("Checking if the service was deleted")
+					ensureServiceWasDeleted()
+				})
+			})
+
 			Context("With Preserve Annotation", func() {
 				var (
 					pods           []string
diff --git a/e2e/test/framework/service.go b/e2e/test/framework/service.go
index 2f0b8aa..e1c1d8b 100644
--- a/e2e/test/framework/service.go
+++ b/e2e/test/framework/service.go
@@ -69,6 +69,10 @@ func (i *lbInvocation) GetServiceWatcher() (watch.Interface, error) {
 	return watcher, nil
 }
 
+func (i *lbInvocation) GetService() (*core.Service, error) {
+	return i.kubeClient.CoreV1().Services(i.Namespace()).Get(context.TODO(), TestServerResourceName, metav1.GetOptions{})
+}
+
 func (i *lbInvocation) CreateService(selector, annotations map[string]string, ports []core.ServicePort, isSessionAffinityClientIP bool) error {
 	return i.createOrUpdateService(selector, annotations, ports, isSessionAffinityClientIP, true)
 }