Retry LB sync while nodes are still cloud-provider initializing

cloud-controller-manager/do/loadbalancers.go

@@ -70,6 +70,18 @@ var (
 
 	errNoEligibleBackendsMsg = "REGIONAL_NETWORK EXTERNAL load balancers require at least one node with a public IPv4 address; no eligible backends found. For private-only nodes, set service.beta.kubernetes.io/do-loadbalancer-type to \"REGIONAL\" or service.beta.kubernetes.io/do-loadbalancer-network to \"INTERNAL\"."
 
+	// nodesNotYetLBReadyRetryAfter is the base retry delay while nodes are still
+	// cloud-provider-initializing. Kept short so scale-up races heal quickly.
+	nodesNotYetLBReadyRetryAfter = 5 * time.Second
+
+	// nodeInitTaintedGrace bounds how long a node carrying the cloud-provider
+	// uninitialized taint is treated as still initializing rather than stuck.
+	nodeInitTaintedGrace = 10 * time.Minute
+
+	// nodeInitAddressGrace covers the gap after the taint is removed but before
+	// Status.Addresses is patched; those are two separate writes.
+	nodeInitAddressGrace = 60 * time.Second
+
 	// Sticky sessions types.
 	stickySessionsTypeNone    = "none"
 	stickySessionsTypeCookies = "cookies"
@@ -106,6 +118,8 @@ type loadBalancers struct {
 	lbActiveTimeout   int
 	lbActiveCheckTick int
 	defaultLBType     string
+	// now overrides the clock in tests; nil means time.Now.
+	now func() time.Time
 }
 
 type servicePatcher struct {
@@ -125,14 +139,27 @@ func newServicePatcher(kclient kubernetes.Interface, base *v1.Service) servicePa
 // Patch will submit a patch request for the Service unless the updated service
 // reference contains the same set of annotations as the base copied during
 // servicePatcher initialization.
+//
+// A successful patch returns err untouched: utilerrors.Aggregate does not
+// implement errors.As, so wrapping would hide api.RetryError from upstream.
 func (sp *servicePatcher) Patch(ctx context.Context, err error) error {
 	if reflect.DeepEqual(sp.base.Annotations, sp.updated.Annotations) {
 		return err
 	}
 	perr := patchService(ctx, sp.kclient, sp.base, sp.updated)
+	if perr == nil {
+		return err
+	}
 	return utilerrors.NewAggregate([]error{err, perr})
 }
 
+func (l *loadBalancers) clock() time.Time {
+	if l.now != nil {
+		return l.now()
+	}
+	return time.Now()
+}
+
 // newLoadbalancers returns a cloudprovider.LoadBalancer whose concrete type is a *loadbalancer.
 func newLoadBalancers(resources *resources, region string, defaultLBType string) cloudprovider.LoadBalancer {
 	return &loadBalancers{
@@ -215,10 +242,16 @@ func (l *loadBalancers) EnsureLoadBalancer(ctx context.Context, clusterName stri
 	patcher := newServicePatcher(l.resources.kclient, service)
 	defer func() { err = patcher.Patch(ctx, err) }()
 
+	pending, err := l.prepareNodesForLBSync(service, nodes)
+	if err != nil {
+		return nil, err
+	}
+
 	var lbRequest *godo.LoadBalancerRequest
 	lbRequest, err = l.buildLoadBalancerRequest(ctx, service, nodes)
 	if err != nil {
-		return nil, fmt.Errorf("failed to build load-balancer request: %s", err)
+		// %w keeps ErrNoEligibleBackends inspectable with errors.Is.
+		return nil, fmt.Errorf("failed to build load-balancer request: %w", err)
 	}
 
 	var lb *godo.LoadBalancer
@@ -256,6 +289,10 @@ func (l *loadBalancers) EnsureLoadBalancer(ctx context.Context, clusterName stri
 		return nil, fmt.Errorf("load-balancer has unexpected status %q", lb.Status)
 	}
 
+	if err := l.retryIfNodesPendingInit(service, pending); err != nil {
+		return nil, err
+	}
+
 	// If a LB hostname annotation is specified, return with it instead of the IP.
 	hostname := getHostname(service)
 	if hostname != "" {
@@ -342,7 +379,7 @@ func (l *loadBalancers) updateLoadBalancer(ctx context.Context, lb *godo.LoadBal
 	// checkAndUpdateLBAndServiceCerts modifies the service
 	_, err := l.buildLoadBalancerRequest(ctx, service, nodes)
 	if err != nil {
-		return nil, fmt.Errorf("failed to build load-balancer request: %s", err)
+		return nil, fmt.Errorf("failed to build load-balancer request: %w", err)
 	}
 
 	lbCertID := getCertificateIDFromLB(lb)
@@ -358,7 +395,7 @@ func (l *loadBalancers) updateLoadBalancer(ctx context.Context, lb *godo.LoadBal
 
 	lbRequest, err := l.buildLoadBalancerRequest(ctx, service, nodes)
 	if err != nil {
-		return nil, fmt.Errorf("failed to build load-balancer request (post-certificate update): %s", err)
+		return nil, fmt.Errorf("failed to build load-balancer request (post-certificate update): %w", err)
 	}
 
 	lbID := lb.ID
@@ -395,8 +432,18 @@ func (l *loadBalancers) UpdateLoadBalancer(ctx context.Context, clusterName stri
 		return err
 	}
 
+	// Classify after annotating so the real LB type is used, not the default.
+	pending, err := l.prepareNodesForLBSync(service, nodes)
+	if err != nil {
+		return err
+	}
+
 	_, err = l.updateLoadBalancer(ctx, lb, service, nodes)
-	return err
+	if err != nil {
+		return err
+	}
+
+	return l.retryIfNodesPendingInit(service, pending)
 }
 
 // EnsureLoadBalancerDeleted deletes the specified loadbalancer if it exists.
@@ -540,6 +587,10 @@ type nodeState struct {
 	lbReadyNodes          []*v1.Node
 	filteredCount         int
 	publicNetUnreadyCount int
+	// publicNetUnreadyNodes are nodes classified as publicNetUnready (no usable
+	// external/fallback IP for this LB type/network). Used to detect transient
+	// cloud-provider init races vs permanently ineligible nodes.
+	publicNetUnreadyNodes []*v1.Node
 	singleStackV4Nodes    []*v1.Node
 	singleStackV6Nodes    []*v1.Node
 	dualStackNodes        []*v1.Node
@@ -594,10 +645,11 @@ func filterAndClassifyNodes(nodes []*v1.Node, lbType, lbNetwork string) *nodeSta
 	isInternalLB := lbNetwork == godo.LoadBalancerNetworkTypeInternal
 
 	state := &nodeState{
-		lbReadyNodes:       make([]*v1.Node, 0, len(nodes)),
-		singleStackV4Nodes: make([]*v1.Node, 0),
-		singleStackV6Nodes: make([]*v1.Node, 0),
-		dualStackNodes:     make([]*v1.Node, 0),
+		lbReadyNodes:          make([]*v1.Node, 0, len(nodes)),
+		publicNetUnreadyNodes: make([]*v1.Node, 0),
+		singleStackV4Nodes:    make([]*v1.Node, 0),
+		singleStackV6Nodes:    make([]*v1.Node, 0),
+		dualStackNodes:        make([]*v1.Node, 0),
 	}
 
 	for _, node := range nodes {
@@ -612,6 +664,7 @@ func filterAndClassifyNodes(nodes []*v1.Node, lbType, lbNetwork string) *nodeSta
 			klog.V(4).Infof("Node %s filtered: no external IP addresses (required for REGIONAL_NETWORK EXTERNAL load balancer)", node.Name)
 			state.filteredCount++
 			state.publicNetUnreadyCount++
+			state.publicNetUnreadyNodes = append(state.publicNetUnreadyNodes, node)
 			continue
 		}
 
@@ -635,6 +688,122 @@ func filterAndClassifyNodes(nodes []*v1.Node, lbType, lbNetwork string) *nodeSta
 	return state
 }
 
+// isCloudProviderUninitialized reports whether the node still carries the
+// external cloud-provider uninitialized taint, meaning a missing ExternalIP is
+// transient rather than a permanent private-only configuration.
+func isCloudProviderUninitialized(node *v1.Node) bool {
+	for _, taint := range node.Spec.Taints {
+		if taint.Key == api.TaintExternalCloudProvider {
+			return true
+		}
+	}
+	return false
+}
+
+// pendingInitNodes splits publicNetUnready nodes into those that may still be
+// completing cloud-provider initialization and those past their grace period.
+//
+// Expired nodes are permanently ineligible: a stuck or genuinely private-only
+// node must not keep the retry loop alive forever.
+func pendingInitNodes(state *nodeState, now time.Time) (pending, expired []*v1.Node) {
+	for _, node := range state.publicNetUnreadyNodes {
+		age := now.Sub(node.CreationTimestamp.Time)
+		grace := nodeInitAddressGrace
+		if isCloudProviderUninitialized(node) {
+			grace = nodeInitTaintedGrace
+		}
+		if age < grace {
+			pending = append(pending, node)
+		} else {
+			expired = append(expired, node)
+		}
+	}
+	return pending, expired
+}
+
+// retryAfterFor slows the retry cadence as nodes stay unready, bounding the
+// load balancer writes a stuck node can cause. The youngest pending node sets
+// the pace so one stuck node cannot throttle a node that is merely racing.
+func retryAfterFor(pending []*v1.Node, now time.Time) time.Duration {
+	if len(pending) == 0 {
+		return nodesNotYetLBReadyRetryAfter
+	}
+
+	youngest := now.Sub(pending[0].CreationTimestamp.Time)
+	for _, node := range pending[1:] {
+		if age := now.Sub(node.CreationTimestamp.Time); age < youngest {
+			youngest = age
+		}
+	}
+
+	switch {
+	case youngest < 30*time.Second:
+		return nodesNotYetLBReadyRetryAfter
+	case youngest < 2*time.Minute:
+		return 15 * time.Second
+	default:
+		return 60 * time.Second
+	}
+}
+
+// classifyServiceNodes resolves LB type/network and classifies the given nodes.
+func (l *loadBalancers) classifyServiceNodes(service *v1.Service, nodes []*v1.Node) (*nodeState, error) {
+	lbNetwork, err := getNetwork(service)
+	if err != nil {
+		return nil, err
+	}
+	lbType, err := getType(service, l.defaultLBType)
+	if err != nil {
+		return nil, err
+	}
+	return filterAndClassifyNodes(nodes, lbType, lbNetwork), nil
+}
+
+// prepareNodesForLBSync classifies nodes and returns those that may still be
+// initializing, so the caller can apply the lb-ready ones and retry for the
+// rest. It fails with a RetryError when no node is lb-ready yet.
+func (l *loadBalancers) prepareNodesForLBSync(service *v1.Service, nodes []*v1.Node) (pending []*v1.Node, err error) {
+	nodeState, err := l.classifyServiceNodes(service, nodes)
+	if err != nil {
+		return nil, err
+	}
+
+	pending, expired := pendingInitNodes(nodeState, l.clock())
+	if len(expired) > 0 {
+		klog.Infof(
+			"service %s/%s: permanently omitting %d node(s) with no usable address after init grace period: %s",
+			service.Namespace, service.Name, len(expired), formatNodeNames(expired, 5),
+		)
+	}
+
+	// Never write an empty backend set while nodes are still coming up.
+	if len(nodeState.lbReadyNodes) == 0 && len(pending) > 0 {
+		msg := fmt.Sprintf(
+			"service %s/%s: %d node(s) still initializing and not yet lb-ready (e.g. missing ExternalIP); retrying instead of updating backends with an empty node set: %s",
+			service.Namespace, service.Name, len(pending), formatNodeNames(pending, 5),
+		)
+		klog.Info(msg)
+		return nil, api.NewRetryError(msg, retryAfterFor(pending, l.clock()))
+	}
+
+	return pending, nil
+}
+
+// retryIfNodesPendingInit asks upstream to sync again after backends were
+// updated with a partial node set. This retry is the only convergence path:
+// address population does not re-trigger the service controller.
+func (l *loadBalancers) retryIfNodesPendingInit(service *v1.Service, pending []*v1.Node) error {
+	if len(pending) == 0 {
+		return nil
+	}
+	msg := fmt.Sprintf(
+		"service %s/%s: updated backends with currently lb-ready nodes; retrying for %d still initializing: %s",
+		service.Namespace, service.Name, len(pending), formatNodeNames(pending, 5),
+	)
+	klog.Info(msg)
+	return api.NewRetryError(msg, retryAfterFor(pending, l.clock()))
+}
+
 // classifyNode determines node's IP stack classification.
 // When allowInternalFallback is true, a private IPv4 InternalIP is treated as IPv4-capable.
 func classifyNode(node *v1.Node, allowInternalFallback bool) nodeClassification {