From 3403af6a6c89ca76599dc9a93c165692d714ec24 Mon Sep 17 00:00:00 2001 From: Ashwani Date: Tue, 11 Aug 2026 14:25:08 +0530 Subject: [PATCH] fix reconcile assignment --- pkg/floatingip/ip_controller.go | 26 ++++-- pkg/floatingip/ip_controller_test.go | 119 +++++++++++++++++++++++++-- pkg/provider/digitalocean.go | 11 +++ pkg/provider/digitalocean_test.go | 65 +++++++++++++++ 4 files changed, 208 insertions(+), 13 deletions(-) diff --git a/pkg/floatingip/ip_controller.go b/pkg/floatingip/ip_controller.go index 68d583c..2949e6c 100644 --- a/pkg/floatingip/ip_controller.go +++ b/pkg/floatingip/ip_controller.go @@ -474,14 +474,18 @@ func (i *ipController) reconcileIPStatus(ctx context.Context) { nodeProviderID: "", // reset } } - i.providerIDToIP[providerID] = ip + if providerID != "" { + i.providerIDToIP[providerID] = ip + } delete(i.providerIDToIP, expectedProviderID) - if evictedNodeName, ok := i.providerIDToNodeName[providerID]; ok { + // If the node we expected to hold this IP is still active, it lost its IP (either + // removed externally, or claimed by another node) and needs a new one. + if evictedNodeName, ok := i.providerIDToNodeName[expectedProviderID]; ok { log.WithFields(logrus.Fields{ "node": evictedNodeName, - "ip": expectedIP, - }).Info("nodes ip was claimed by other node; marking for reassignment") + "ip": ip, + }).Info("nodes ip was removed or claimed by other node; marking node for reassignment") i.assignableNodes.Add(expectedProviderID, true) } } @@ -531,9 +535,10 @@ func (i *ipController) reconcileAssignment(ctx context.Context) { ip := i.assignableIPs.Front() // If this IP was previously involved in an error we shouldn't attempt to try again before - // its retry timestamp. + // its retry timestamp. Healthy IPs also carry a nextRetry timestamp (their periodic + // status check), which must not delay assignment. status := i.ipToStatus[ip] - if !status.nextRetry.IsZero() && !status.nextRetry.After(now) { + if status.state == flipopv1alpha1.IPStateError && status.nextRetry.After(now) { retryIPs = append(retryIPs, ip) i.retry(status.nextRetry) continue @@ -544,7 +549,7 @@ func (i *ipController) reconcileAssignment(ctx context.Context) { // Similarly, if this node was involved in an error we should wait until after its retry // timestamp has elapsed. nRetry, ok := i.providerIDToRetry[providerID] - if ok && !nRetry.nextRetry.IsZero() && !nRetry.nextRetry.After(now) { + if ok && nRetry.nextRetry.After(now) { retryIPs = append(retryIPs, ip) retryProviders = append(retryProviders, providerID) i.retry(nRetry.nextRetry) @@ -597,6 +602,12 @@ func (i *ipController) reconcileAssignment(ctx context.Context) { nRetry.attempts, nRetry.nextRetry = nRetry.retrySchedule.Next(nRetry.attempts) i.providerIDToRetry[providerID] = nRetry i.retry(nRetry.nextRetry) + // The assignment failed. Roll back the speculative claim recorded above, and requeue + // both the IP and the node so assignment is retried once their backoff elapses. + status.nodeProviderID = "" + delete(i.providerIDToIP, providerID) + retryIPs = append(retryIPs, ip) + retryProviders = append(retryProviders, providerID) } i.dnsDirty = true @@ -731,7 +742,6 @@ func (i *ipController) EnableNodes(ctx context.Context, nodes ...*corev1.Node) { i.assignableIPs.Delete(ip) // Since the IP address is already assigned to the node we want to ensure the annotation reflects it as well. if i.onAnnotate != nil { - log.Warn("test") if err := i.onAnnotate(ctx, i.providerIDToNodeName[providerID], ip); err != nil { i.log.WithError(err).Error("updating Reserved IP annotation") } diff --git a/pkg/floatingip/ip_controller_test.go b/pkg/floatingip/ip_controller_test.go index c81a752..ccd9ae0 100644 --- a/pkg/floatingip/ip_controller_test.go +++ b/pkg/floatingip/ip_controller_test.go @@ -30,6 +30,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/util/uuid" + flipopv1alpha1 "github.com/digitalocean/flipop/pkg/apis/flipop/v1alpha1" "github.com/digitalocean/flipop/pkg/log" "github.com/digitalocean/flipop/pkg/provider" ) @@ -204,6 +205,21 @@ func TestIPControllerReconcileIPStatus(t *testing.T) { expectAssignableIPs: []string{}, expectAssignableNodes: []string{"mock://1"}, }, + { + name: "ip unassigned externally; node requeued", + ips: []string{"192.168.1.1"}, + responses: []ipToProviderIDRes{{ip: "192.168.1.1", providerID: ""}}, + setup: func(i *ipController) { + i.providerIDToIP["mock://1"] = "192.168.1.1" + i.ipToStatus["192.168.1.1"] = &ipStatus{ + nodeProviderID: "mock://1", + } + i.providerIDToNodeName["mock://1"] = "some-node" + }, + expectProviderIDToIP: map[string]string{}, + expectAssignableIPs: []string{"192.168.1.1"}, + expectAssignableNodes: []string{"mock://1"}, + }, { name: "provider reports ip reassigned", ips: []string{"192.168.1.1", "172.16.2.2"}, @@ -369,18 +385,111 @@ func TestIPControllerReconcileAssignment(t *testing.T) { }, }, { - name: "assignment error", + name: "assignment error", + assignableIPs: []string{"192.168.1.1"}, + assignableNodes: []string{"mock://1"}, + // The failed assignment must not leave a speculative claim behind, and both the + // IP and node must be requeued so assignment is retried after backoff. + expectProviderIDToIP: map[string]string{}, + responses: []assignIPRes{{ip: "192.168.1.1", providerID: "mock://1", err: errors.New("nope")}}, + expectIPRetry: true, // We always retry, because of assign + expectAssignableIPs: []string{"192.168.1.1"}, + expectAssignableNodes: []string{"mock://1"}, + setup: func(i *ipController) { + i.ipToStatus["192.168.1.1"] = &ipStatus{} + }, + eval: func(i *ipController) { + require.Equal(t, provider.RetrySlow, i.ipToStatus["192.168.1.1"].retrySchedule) + require.Contains(t, i.providerIDToRetry, "mock://1") + require.Empty(t, i.ipToStatus["192.168.1.1"].nodeProviderID) + }, + }, + { + name: "node in retry backoff is not attempted", + assignableIPs: []string{"192.168.1.1"}, + assignableNodes: []string{"mock://1"}, + expectProviderIDToIP: map[string]string{}, + responses: nil, // AssignIP must NOT be called. + expectIPRetry: true, + expectAssignableIPs: []string{"192.168.1.1"}, + expectAssignableNodes: []string{"mock://1"}, + setup: func(i *ipController) { + i.ipToStatus["192.168.1.1"] = &ipStatus{} + i.now = func() time.Time { return fakeNow } + i.providerIDToRetry["mock://1"] = &retry{ + attempts: 1, + nextRetry: fakeNow.Add(30 * time.Second), + retrySchedule: provider.RetryFast, + } + }, + eval: func(i *ipController) { + // The reconciler must wake when the node's backoff elapses. + require.Equal(t, fakeNow.Add(30*time.Second), i.nextRetry) + }, + }, + { + name: "node retry deadline elapsed; assignment attempted", assignableIPs: []string{"192.168.1.1"}, assignableNodes: []string{"mock://1"}, expectProviderIDToIP: map[string]string{"mock://1": "192.168.1.1"}, - responses: []assignIPRes{{ip: "192.168.1.1", providerID: "mock://1", err: errors.New("nope")}}, - expectIPRetry: true, // We always retry, because of assign + responses: []assignIPRes{{ip: "192.168.1.1", providerID: "mock://1"}}, + expectIPRetry: true, + expectAnnotateCall: &annotateCall{NodeName: "hello-world", IP: "192.168.1.1"}, setup: func(i *ipController) { i.ipToStatus["192.168.1.1"] = &ipStatus{} + i.providerIDToNodeName["mock://1"] = "hello-world" + i.now = func() time.Time { return fakeNow } + i.providerIDToRetry["mock://1"] = &retry{ + attempts: 1, + nextRetry: fakeNow.Add(-time.Second), + retrySchedule: provider.RetryFast, + } }, eval: func(i *ipController) { - require.Equal(t, provider.RetrySlow, i.ipToStatus["192.168.1.1"].retrySchedule) - require.Contains(t, i.providerIDToRetry, "mock://1") + require.NotContains(t, i.providerIDToRetry, "mock://1") + }, + }, + { + name: "ip in error backoff is not attempted", + assignableIPs: []string{"192.168.1.1"}, + assignableNodes: []string{"mock://1"}, + expectProviderIDToIP: map[string]string{}, + responses: nil, // AssignIP must NOT be called. + expectIPRetry: true, + expectAssignableIPs: []string{"192.168.1.1"}, + expectAssignableNodes: []string{"mock://1"}, + setup: func(i *ipController) { + i.now = func() time.Time { return fakeNow } + i.ipToStatus["192.168.1.1"] = &ipStatus{ + state: flipopv1alpha1.IPStateError, + retry: retry{ + attempts: 1, + nextRetry: fakeNow.Add(30 * time.Second), + retrySchedule: provider.RetryFast, + }, + } + }, + }, + { + name: "healthy ip with future status check is assigned immediately", + assignableIPs: []string{"192.168.1.1"}, + assignableNodes: []string{"mock://1"}, + expectProviderIDToIP: map[string]string{"mock://1": "192.168.1.1"}, + responses: []assignIPRes{{ip: "192.168.1.1", providerID: "mock://1"}}, + expectIPRetry: true, + expectAnnotateCall: &annotateCall{NodeName: "hello-world", IP: "192.168.1.1"}, + setup: func(i *ipController) { + i.providerIDToNodeName["mock://1"] = "hello-world" + i.now = func() time.Time { return fakeNow } + // Healthy IPs carry a nextRetry for their periodic status check; this must + // not delay assignment. + i.ipToStatus["192.168.1.1"] = &ipStatus{ + state: flipopv1alpha1.IPStateUnassigned, + retry: retry{ + nextRetry: fakeNow.Add(5 * time.Minute), + retrySchedule: healthyRetrySchedule, + }, + } }, }, } diff --git a/pkg/provider/digitalocean.go b/pkg/provider/digitalocean.go index 4002532..dfe3869 100644 --- a/pkg/provider/digitalocean.go +++ b/pkg/provider/digitalocean.go @@ -209,6 +209,11 @@ func (do *digitalOcean) AssignIP(ctx context.Context, ip, providerID string) (er return ErrNodeInUse } } + if isPendingEventError(err) { + // The droplet has another event in-flight (common during node rotation + // or provisioning). These resolve quickly; retry on the fast schedule. + return NewRetryError(err, RetryFast) + } } } return err @@ -290,6 +295,12 @@ func (do *digitalOcean) NodeToIP(ctx context.Context, providerID string) (_ stri return "", nil } +// isPendingEventError returns true if the error indicates the droplet already has an event +// in-flight (API status 422, "Droplet already has a pending event"). +func isPendingEventError(err error) bool { + return err != nil && strings.Contains(strings.ToLower(err.Error()), "pending event") +} + // asyncStatus tries to abstract the asynchronous nature of DO's floating IP updates. func (do *digitalOcean) asyncStatus(ctx context.Context, ip string) error { do.lock.Lock() diff --git a/pkg/provider/digitalocean_test.go b/pkg/provider/digitalocean_test.go index f689e64..d09ad38 100644 --- a/pkg/provider/digitalocean_test.go +++ b/pkg/provider/digitalocean_test.go @@ -19,6 +19,8 @@ package provider import ( "context" "net" + "net/http" + "net/url" "testing" "github.com/digitalocean/godo" @@ -31,6 +33,69 @@ import ( "github.com/stretchr/testify/require" ) +func TestDigitalOceanAssignIP422Classification(t *testing.T) { + tcs := []struct { + name string + message string + expectSchedule RetrySchedule + }{ + { + // A droplet with an event in-flight (e.g. during node rotation) rejects + // assignment with a 422. This resolves quickly and must be retried fast. + name: "pending event retries fast", + message: "Droplet already has a pending event.", + expectSchedule: RetryFast, + }, + { + name: "other 422 retries slow", + message: "cannot assign a reserved IP to a droplet in a different region", + expectSchedule: RetrySlow, + }, + } + for _, tc := range tcs { + tc := tc + t.Run(tc.name, func(t *testing.T) { + ctx := context.Background() + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + httpRes := &http.Response{ + StatusCode: http.StatusUnprocessableEntity, + Request: &http.Request{ + Method: http.MethodPost, + URL: &url.URL{}, + }, + } + assignErr := &godo.ErrorResponse{ + Response: httpRes, + Message: tc.message, + } + + ipActionsService := mock_godo.NewMockFloatingIPActionsService(ctrl) + ipActionsService.EXPECT(). + Assign(gomock.Any(), "10.0.0.1", 12345). + Return(nil, &godo.Response{Response: httpRes}, assignErr) + + // AssignIP consults the IP's current assignment before classifying the 422. + // Report it unassigned so classification falls through to the error itself. + ipsService := mock_godo.NewMockFloatingIPsService(ctrl) + ipsService.EXPECT(). + Get(gomock.Any(), "10.0.0.1"). + Return(&godo.FloatingIP{}, &godo.Response{}, nil) + + do := &digitalOcean{ + ipsService: ipsService, + ipActionsService: ipActionsService, + floatingIPActions: make(map[string]*doAction), + log: log.NewTestLogger(t), + } + err := do.AssignIP(ctx, "10.0.0.1", "digitalocean://12345") + require.Error(t, err) + require.Equal(t, tc.expectSchedule, ErrorToRetrySchedule(err)) + }) + } +} + func TestDigitalOceanEnsureDNSARecordSet(t *testing.T) { tcs := []struct { name string