Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
## unreleased

* Gate LB target admission on NodeReady for first admission while preserving the upstream stable node set for already-admitted nodes (@skumarc-do)

## v0.1.71 (beta) - September 17, 2026

* Update Go to 1.27 (@pyadagiri-do)
Expand Down
127 changes: 96 additions & 31 deletions cloud-controller-manager/do/loadbalancers.go
Original file line number Diff line number Diff line change
Expand Up @@ -240,19 +240,22 @@ func (l *loadBalancers) EnsureLoadBalancer(ctx context.Context, clusterName stri
return nil, err
}

var lbRequest *godo.LoadBalancerRequest
lbRequest, err = l.buildLoadBalancerRequest(ctx, service, nodes)
lb, lbErr := l.retrieveAndAnnotateLoadBalancer(ctx, service)
if lbErr != nil && !errors.Is(lbErr, errLBNotFound) {
return nil, lbErr
}
admittedIDs := admittedDropletIDSet(lb)

lbRequest, pendingNotReady, err := l.buildLoadBalancerRequest(ctx, service, nodes, admittedIDs)
if err != nil {
// %w keeps ErrNoEligibleBackends inspectable with errors.Is.
// %w keeps ErrNoEligibleBackends (errors.Is) and *api.RetryError (errors.As) inspectable.
return nil, fmt.Errorf("failed to build load-balancer request: %w", err)
}

var lb *godo.LoadBalancer
lb, err = l.retrieveAndAnnotateLoadBalancer(ctx, service)
switch err {
switch lbErr {
case nil:
// LB existing
lb, err = l.updateLoadBalancer(ctx, lb, service, nodes)
lb, pendingNotReady, err = l.updateLoadBalancer(ctx, lb, service, nodes)
if err != nil {
return nil, err
}
Expand All @@ -271,10 +274,6 @@ func (l *loadBalancers) EnsureLoadBalancer(ctx context.Context, clusterName stri

updateServiceAnnotation(service, annDOLoadBalancerID, lb.ID)
updateServiceAnnotation(service, annDOType, lb.Type)

default:
// unrecoverable LB retrieval error
return nil, err
}

if lb.Status == lbStatusNew {
Expand All @@ -285,6 +284,10 @@ func (l *loadBalancers) EnsureLoadBalancer(ctx context.Context, clusterName stri
return nil, fmt.Errorf("load-balancer has unexpected status %q", lb.Status)
}

// Backends were written with the currently admissible nodes, retry so nodes
// held back for IP init (pending) or NodeReady (pendingNotReady) are admitted
// once eligible, since upstream does not re-sync on those transitions.
pending = append(pending, pendingNotReady...)
if err := l.retryIfNodesPendingInit(service, pending); err != nil {
return nil, err
}
Expand Down Expand Up @@ -369,40 +372,42 @@ func (l *loadBalancers) recordUpdatedLetsEncryptCert(ctx context.Context, servic
return nil
}

func (l *loadBalancers) updateLoadBalancer(ctx context.Context, lb *godo.LoadBalancer, service *v1.Service, nodes []*v1.Node) (*godo.LoadBalancer, error) {
func (l *loadBalancers) updateLoadBalancer(ctx context.Context, lb *godo.LoadBalancer, service *v1.Service, nodes []*v1.Node) (*godo.LoadBalancer, []*v1.Node, error) {
admittedIDs := admittedDropletIDSet(lb)

// call buildLoadBalancerRequest for its error checking; we have to call it
// again just before actually updating the loadbalancer in case
// checkAndUpdateLBAndServiceCerts modifies the service
_, err := l.buildLoadBalancerRequest(ctx, service, nodes)
_, _, err := l.buildLoadBalancerRequest(ctx, service, nodes, admittedIDs)
if err != nil {
return nil, fmt.Errorf("failed to build load-balancer request: %w", err)
return nil, nil, fmt.Errorf("failed to build load-balancer request: %w", err)
}

lbCertID := getCertificateIDFromLB(lb)
serviceCertID, err := findCertificateID(ctx, service, l.resources.gclient)
if err != nil {
return nil, err
return nil, nil, err
}

err = l.recordUpdatedLetsEncryptCert(ctx, service, lbCertID, serviceCertID)
if err != nil {
return nil, err
return nil, nil, err
}

lbRequest, err := l.buildLoadBalancerRequest(ctx, service, nodes)
lbRequest, pendingNotReady, err := l.buildLoadBalancerRequest(ctx, service, nodes, admittedIDs)
if err != nil {
return nil, fmt.Errorf("failed to build load-balancer request (post-certificate update): %w", err)
return nil, nil, fmt.Errorf("failed to build load-balancer request (post-certificate update): %w", err)
}

lbID := lb.ID
lb, _, err = l.resources.gclient.LoadBalancers.Update(ctx, lb.ID, lbRequest)
if err != nil {
logLBInfo("UPDATE", lbRequest, 2)
return nil, fmt.Errorf("failed to update load-balancer with ID %s: %s", lbID, err)
return nil, nil, fmt.Errorf("failed to update load-balancer with ID %s: %s", lbID, err)
}
logLBInfo("UPDATE", lbRequest, 2)

return lb, nil
return lb, pendingNotReady, nil
}

// UpdateLoadBalancer updates the load balancer for service to balance across
Expand Down Expand Up @@ -434,11 +439,12 @@ func (l *loadBalancers) UpdateLoadBalancer(ctx context.Context, clusterName stri
return err
}

_, err = l.updateLoadBalancer(ctx, lb, service, nodes)
_, pendingNotReady, err := l.updateLoadBalancer(ctx, lb, service, nodes)
if err != nil {
return err
}

pending = append(pending, pendingNotReady...)
return l.retryIfNodesPendingInit(service, pending)
}

Expand Down Expand Up @@ -692,6 +698,44 @@ func nodeIntendsPublicNetwork(node *v1.Node) bool {
return !strings.EqualFold(node.Labels[nodeNetworkTypeLabel], nodeNetworkTypePrivate)
}

// isNodeReady reports whether the node currently has a NodeReady condition with Status=True.
func isNodeReady(node *v1.Node) bool {
for _, c := range node.Status.Conditions {
if c.Type == v1.NodeReady {
return c.Status == v1.ConditionTrue
}
}
return false
}

// admittedDropletIDSet returns a lookup set of the LB's current backend
// droplet IDs, or nil when lb is nil or has no admitted droplets.
func admittedDropletIDSet(lb *godo.LoadBalancer) map[int]bool {
if lb == nil || len(lb.DropletIDs) == 0 {
return nil
}
set := make(map[int]bool, len(lb.DropletIDs))
for _, id := range lb.DropletIDs {
set[id] = true
}
return set
}

func filterByReadiness(nodes []*v1.Node, admittedIDs map[int]bool) (admit, pending []*v1.Node) {
for _, node := range nodes {
if id, err := dropletIDFromProviderID(node.Spec.ProviderID); err == nil && admittedIDs[id] {
Comment thread
Copilot marked this conversation as resolved.
admit = append(admit, node)
continue
}
if isNodeReady(node) {
admit = append(admit, node)
continue
}
pending = append(pending, node)
}
return admit, pending
}

// pendingInitNodes splits nodes with no usable public address by declared
// network intent: those still expected to gain one (retry) and private-only
// nodes that never will (no retry). Both groups are already excluded from the
Expand Down Expand Up @@ -758,7 +802,7 @@ func (l *loadBalancers) retryIfNodesPendingInit(service *v1.Service, pending []*
return nil
}
msg := fmt.Sprintf(
"service %s/%s: updated backends with currently lb-ready nodes; retrying for %d still initializing: %s",
"service %s/%s: updated backends with the currently admissible node set; retrying for %d node(s) still pending (initializing or not yet NodeReady): %s",
service.Namespace, service.Name, len(pending), formatNodeNames(pending, 5),
)
klog.Info(msg)
Expand Down Expand Up @@ -1063,14 +1107,14 @@ func (l *loadBalancers) emitServiceEvent(service *v1.Service, eventType, reason,

// buildLoadBalancerRequest returns a *godo.LoadBalancerRequest to balance
// requests for service across nodes.
func (l *loadBalancers) buildLoadBalancerRequest(ctx context.Context, service *v1.Service, nodes []*v1.Node) (*godo.LoadBalancerRequest, error) {
func (l *loadBalancers) buildLoadBalancerRequest(ctx context.Context, service *v1.Service, nodes []*v1.Node, admittedIDs map[int]bool) (*godo.LoadBalancerRequest, []*v1.Node, error) {
lbNetwork, err := getNetwork(service)
if err != nil {
return nil, err
return nil, nil, err
}
lbType, err := getType(service, l.defaultLBType)
if err != nil {
return nil, err
return nil, nil, err
}
isInternalLB := lbNetwork == godo.LoadBalancerNetworkTypeInternal

Expand All @@ -1080,7 +1124,7 @@ func (l *loadBalancers) buildLoadBalancerRequest(ctx context.Context, service *v
lbType == godo.LoadBalancerTypeRegionalNetwork && lbNetwork == godo.LoadBalancerNetworkTypeExternal {
err := fmt.Errorf("%w: %s", ErrNoEligibleBackends, errNoEligibleBackendsMsg)
l.emitServiceEvent(service, v1.EventTypeWarning, "LoadBalancerConfigError", err.Error())
return nil, err
return nil, nil, err
}

if nodeState.filteredCount > 0 {
Expand All @@ -1099,17 +1143,38 @@ func (l *loadBalancers) buildLoadBalancerRequest(ctx context.Context, service *v
}
}

req, err := buildLoadBalancerRequestWithNodeState(ctx, service, l.resources.gclient, l.defaultLBType, nodeState)
// Already-admitted droplets stay in the target set regardless of transient
// NodeReady flips; new nodes must be Ready to be first admitted. Nodes that
// are not yet Ready are held back and converged via api.RetryError below.
Comment thread
skumarc-do marked this conversation as resolved.
admit, pendingNotReady := filterByReadiness(nodeState.lbReadyNodes, admittedIDs)
if len(pendingNotReady) > 0 {
klog.V(2).Infof("Service %s/%s: holding back %d node(s) pending NodeReady=True on first admission: %s",
service.Namespace, service.Name, len(pendingNotReady), formatNodeNames(pendingNotReady, 5))
Comment thread
skumarc-do marked this conversation as resolved.
}

if len(admit) == 0 && len(pendingNotReady) > 0 {
Comment thread
Copilot marked this conversation as resolved.
msg := fmt.Sprintf(
"service %s/%s: %d lb-ready-by-IP node(s) are not yet NodeReady=True and none is already admitted, retrying instead of writing an empty backend set: %s",
service.Namespace, service.Name, len(pendingNotReady), formatNodeNames(pendingNotReady, 5),
)
klog.Info(msg)
return nil, nil, api.NewRetryError(msg, nodesNotYetLBReadyRetryAfter)
}

// Derive the request (including NetworkStack) from the admitted nodes only, so
// a held-back node cannot shape config it will not back.
admitState := filterAndClassifyNodes(admit, lbType, lbNetwork)
req, err := buildLoadBalancerRequestWithNodeState(ctx, service, l.resources.gclient, l.defaultLBType, admitState)
if err != nil {
if errors.Is(err, ErrNetworkStackConfig) {
l.emitServiceEvent(service, v1.EventTypeWarning, "LoadBalancerConfigError", err.Error())
}
return nil, err
return nil, nil, err
}

dropletIDs, err := l.nodesToDropletIDs(ctx, nodeState.lbReadyNodes)
dropletIDs, err := l.nodesToDropletIDs(ctx, admit)
if err != nil {
return nil, err
return nil, nil, err
}
req.DropletIDs = dropletIDs

Expand All @@ -1121,7 +1186,7 @@ func (l *loadBalancers) buildLoadBalancerRequest(ctx context.Context, service *v

req.Region = l.region
req.VPCUUID = l.resources.clusterVPCID
return req, nil
return req, pendingNotReady, nil
}

// buildHealthChecks returns a godo.HealthCheck for service.
Expand Down
Loading
Loading