add prefix status in pool and different retry time when prefix not avaialble (#268)

Author: Jie Chen

mocks/amazon-vcp-resource-controller-k8s/pkg/pool/mock_pool.go

@@ -29,10 +29,11 @@ import (
 )
 
 const (
-	RequeueAfterWhenWPEmpty         = time.Millisecond * 600
-	RequeueAfterWhenResourceCooling = time.Second * 20
-	ReasonResourceAllocationFailed  = "ResourceAllocationFailed"
-	ReasonResourceAllocated         = "ResourceAllocated"
+	RequeueAfterWhenPrefixNotAvailable = time.Minute * 2
+	RequeueAfterWhenWPEmpty            = time.Millisecond * 600
+	RequeueAfterWhenResourceCooling    = time.Second * 20
+	ReasonResourceAllocationFailed     = "ResourceAllocationFailed"
+	ReasonResourceAllocated            = "ResourceAllocated"
 )
 
 type warmResourceHandler struct {
@@ -100,6 +101,12 @@ func (w *warmResourceHandler) HandleCreate(_ int, pod *v1.Pod) (ctrl.Result, err
 				return ctrl.Result{}, nil
 			}
 			return ctrl.Result{}, err
+		case pool.ErrInsufficientCidrBlocks:
+			log.V(1).Info("prefix is not available in subnet, will retry")
+			w.APIWrapper.K8sAPI.BroadcastEvent(pod, ReasonResourceAllocationFailed,
+				fmt.Sprintf("Warm pool for resource %s is currently empty because the specified subnet does not have enough "+
+					"free cidr blocks, will retry in %s", w.resourceName, RequeueAfterWhenPrefixNotAvailable), v1.EventTypeWarning)
+			return ctrl.Result{Requeue: true, RequeueAfter: RequeueAfterWhenPrefixNotAvailable}, nil
 		default:
 			return ctrl.Result{}, err
 		}

pkg/handler/warm.go

@@ -29,6 +29,7 @@ var (
 	ErrResourceAreBeingCooledDown = fmt.Errorf("cannot assign resource now, resources are being cooled down")
 	ErrResourcesAreBeingCreated   = fmt.Errorf("cannot assign resource now, resources are being created")
 	ErrWarmPoolEmpty              = fmt.Errorf("warm pool is empty")
+	ErrInsufficientCidrBlocks     = fmt.Errorf("InsufficientCidrBlocks: The specified subnet does not have enough free cidr blocks to satisfy the request")
 	ErrResourceAlreadyAssigned    = fmt.Errorf("resource is already assigned to the requestor")
 	ErrResourceDoesntExist        = fmt.Errorf("requested resource doesn't exist in used pool")
 	ErrIncorrectResourceOwner     = fmt.Errorf("resource doesn't belong to the requestor")
@@ -42,7 +43,7 @@ type Pool interface {
 	AssignResource(requesterID string) (resourceID string, shouldReconcile bool, err error)
 	FreeResource(requesterID string, resourceID string) (shouldReconcile bool, err error)
 	GetAssignedResource(requesterID string) (resourceID string, ownsResource bool)
-	UpdatePool(job *worker.WarmPoolJob, didSucceed bool) (shouldReconcile bool)
+	UpdatePool(job *worker.WarmPoolJob, didSucceed bool, prefixAvailable bool) (shouldReconcile bool)
 	ReSync(resources []string)
 	ReconcilePool() *worker.WarmPoolJob
 	ProcessCoolDownQueue() bool
@@ -77,6 +78,8 @@ type pool struct {
 	reSyncRequired bool
 	// isPDPool indicates whether the pool is for prefix IP provider or secondary IP provider
 	isPDPool bool
+	// prefixAvailable indicates whether subnet has any prefix available
+	prefixAvailable bool
 }
 
 // Resource represents a secondary IPv4 address or a prefix-deconstructed IPv4 address, uniquely identified by GroupID and ResourceID
@@ -245,9 +248,14 @@ func (p *pool) AssignResource(requesterID string) (resourceID string, shouldReco
 		return "", false, ErrResourcesAreBeingCreated
 	}
 
-	// Caller can retry in 600 ms [Average time to create and attach a new ENI] or less
-	// Different from above check because here we want to perform reconciliation
 	if len(p.warmResources) == 0 {
+		// If prefix is not available in subnet, caller can retry in 2 min [Action required from user to change subnet]
+		if p.isPDPool && !p.prefixAvailable {
+			return "", false, ErrInsufficientCidrBlocks
+		}
+
+		// Caller can retry in 600 ms [Average time to assign a new secondary IP or prefix] or less
+		// Different from above check because here we want to perform reconciliation
 		return "", true, ErrWarmPoolEmpty
 	}
 
@@ -309,7 +317,7 @@ func (p *pool) FreeResource(requesterID string, resourceID string) (shouldReconc
 }
 
 // UpdatePool updates the warm pool with the result of the asynchronous job executed by the provider
-func (p *pool) UpdatePool(job *worker.WarmPoolJob, didSucceed bool) (shouldReconcile bool) {
+func (p *pool) UpdatePool(job *worker.WarmPoolJob, didSucceed bool, prefixAvailable bool) (shouldReconcile bool) {
 	p.lock.Lock()
 	defer p.lock.Unlock()
 
@@ -322,6 +330,13 @@ func (p *pool) UpdatePool(job *worker.WarmPoolJob, didSucceed bool) (shouldRecon
 		log.Error(fmt.Errorf("warm pool job failed: %v", job), "operation failed")
 	}
 
+	if p.isPDPool {
+		p.prefixAvailable = prefixAvailable
+		if !p.prefixAvailable {
+			log.Error(fmt.Errorf("warm pool job failed: %v", job), "prefix is not available in subnet")
+		}
+	}
+
 	if job.Resources != nil && len(job.Resources) > 0 {
 		// Add the resources to the warm pool
 		for _, resourceGroup := range job.Resources {

pkg/pool/pool.go

@@ -221,7 +221,7 @@ func TestPool_UpdatePool_OperationCreate_Succeed(t *testing.T) {
 		Operations:    worker.OperationCreate,
 		Resources:     createdResources,
 		ResourceCount: 2,
-	}, true)
+	}, true, false)
 
 	warmResources := make(map[string][]Resource)
 	warmResources[res3] = []Resource{{GroupID: res3, ResourceID: res3}}
@@ -239,7 +239,7 @@ func TestPool_UpdatePool_OperationCreate_Failed(t *testing.T) {
 		Operations:    worker.OperationCreate,
 		Resources:     nil,
 		ResourceCount: 2,
-	}, false)
+	}, false, false)
 
 	assert.True(t, shouldReconcile)
 	assert.True(t, warmPool.reSyncRequired)
@@ -253,7 +253,7 @@ func TestPool_UpdatePool_OperationDelete_Succeed(t *testing.T) {
 		Operations:    worker.OperationDeleted,
 		Resources:     nil,
 		ResourceCount: 1,
-	}, true)
+	}, true, false)
 
 	// Assert resources are not added back to the warm pool
 	assert.Zero(t, len(warmPool.warmResources))
@@ -268,7 +268,7 @@ func TestPool_UpdatePool_OperationDelete_Failed(t *testing.T) {
 		Operations:    worker.OperationDeleted,
 		Resources:     failedResources,
 		ResourceCount: 2,
-	}, false)
+	}, false, false)
 
 	failed := make(map[string][]Resource)
 	failed[res3] = []Resource{{GroupID: res3, ResourceID: res3}}

pkg/pool/pool_test.go

@@ -361,7 +361,7 @@ func (p *ipv4Provider) DeletePrivateIPv4AndUpdatePool(job *worker.WarmPoolJob) {
 // updatePoolAndReconcileIfRequired updates the resource pool and reconcile again and submit a new job if required
 func (p *ipv4Provider) updatePoolAndReconcileIfRequired(resourcePool pool.Pool, job *worker.WarmPoolJob, didSucceed bool) {
 	// Update the pool to add the created/failed resource to the warm pool and decrement the pending count
-	shouldReconcile := resourcePool.UpdatePool(job, didSucceed)
+	shouldReconcile := resourcePool.UpdatePool(job, didSucceed, false)
 
 	if shouldReconcile {
 		job := resourcePool.ReconcilePool()

pkg/provider/ip/provider.go

@@ -130,7 +130,7 @@ func TestIpv4Provider_updatePoolAndReconcileIfRequired_NoFurtherReconcile(t *tes
 
 	job := &worker.WarmPoolJob{Operations: worker.OperationCreate}
 
-	mockPool.EXPECT().UpdatePool(job, true).Return(false)
+	mockPool.EXPECT().UpdatePool(job, true, false).Return(false)
 
 	provider.updatePoolAndReconcileIfRequired(mockPool, job, true)
 }
@@ -147,7 +147,7 @@ func TestIpv4Provider_updatePoolAndReconcileIfRequired_ReconcileRequired(t *test
 
 	job := &worker.WarmPoolJob{Operations: worker.OperationCreate}
 
-	mockPool.EXPECT().UpdatePool(job, true).Return(true)
+	mockPool.EXPECT().UpdatePool(job, true, false).Return(true)
 	mockPool.EXPECT().ReconcilePool().Return(job)
 	mockWorker.EXPECT().SubmitJob(job)
 
@@ -179,7 +179,7 @@ func TestIpv4Provider_DeletePrivateIPv4AndUpdatePool(t *testing.T) {
 		Resources:     []string{},
 		ResourceCount: 2,
 		NodeName:      nodeName,
-	}, true).Return(false)
+	}, true, false).Return(false)
 
 	ipv4Provider.DeletePrivateIPv4AndUpdatePool(deleteJob)
 }
@@ -210,7 +210,7 @@ func TestIpv4Provider_DeletePrivateIPv4AndUpdatePool_SomeResourceFail(t *testing
 		Resources:     failedResources,
 		ResourceCount: 2,
 		NodeName:      nodeName,
-	}, true).Return(false)
+	}, true, false).Return(false)
 
 	ipv4Provider.DeletePrivateIPv4AndUpdatePool(&deleteJob)
 }
@@ -240,7 +240,7 @@ func TestIPv4Provider_CreatePrivateIPv4AndUpdatePool(t *testing.T) {
 		Resources:     createdResources,
 		ResourceCount: 2,
 		NodeName:      nodeName,
-	}, true).Return(false)
+	}, true, false).Return(false)
 
 	ipv4Provider.CreatePrivateIPv4AndUpdatePool(createJob)
 }
@@ -270,7 +270,7 @@ func TestIPv4Provider_CreatePrivateIPv4AndUpdatePool_Fail(t *testing.T) {
 		Resources:     createdResources,
 		ResourceCount: 2,
 		NodeName:      nodeName,
-	}, false).Return(false)
+	}, false, false).Return(false)
 
 	ipv4Provider.CreatePrivateIPv4AndUpdatePool(createJob)
 }

pkg/provider/ip/provider_test.go

@@ -290,21 +290,34 @@ func (p *ipv4PrefixProvider) CreateIPv4PrefixAndUpdatePool(job *worker.WarmPoolJ
 		p.log.Error(utils.ErrNotFound, utils.ErrMsgProviderAndPoolNotFound, "node name", job.NodeName)
 		return
 	}
-	didSucceed := true
+
+	// For successful jobs or non-retryable errors, do not re-sync or reconcile the pool.
+	notRetry := true
+	// If subnet has sufficient cidr blocks, prefixAvailable is true, otherwise false.
+	prefixAvailable := true
+
 	resources, err := instanceResource.eniManager.CreateIPV4Resource(job.ResourceCount, config.ResourceTypeIPv4Prefix, p.apiWrapper.EC2API,
 		p.log)
 
 	if err != nil {
 		p.log.Error(err, "failed to create all/some of the IPv4 prefixes", "created resources", resources)
-		didSucceed = false
+
+		// For retryable errors, set notRetry as false to re-sync and reconcile the pool.
+		if utils.ShouldRetryOnError(err) {
+			notRetry = false
+		}
+
 		//TODO: This adds a dependency on EC2 API error. Refactor later.
 		if strings.HasPrefix(err.Error(), utils.InsufficientCidrBlocksReason) {
+			// Prefix not available in the subnet, set status and pass it to pool. Note this is a non-retryable error.
+			prefixAvailable = false
+			// Send node event to inform user of insufficient CIDR blocks error
 			utils.SendNodeEvent(p.apiWrapper.K8sAPI, job.NodeName, utils.InsufficientCidrBlocksReason,
 				utils.ErrInsufficientCidrBlocks.Error(), v1.EventTypeWarning, p.log)
 		}
 	}
 	job.Resources = resources
-	p.updatePoolAndReconcileIfRequired(instanceResource.resourcePool, job, didSucceed)
+	p.updatePoolAndReconcileIfRequired(instanceResource.resourcePool, job, notRetry, prefixAvailable)
 }
 
 // DeleteIPv4PrefixAndUpdatePool executes the Delete IPv4 Prefix workflow for the list of prefixes provided in the warm pool job
@@ -324,7 +337,7 @@ func (p *ipv4PrefixProvider) DeleteIPv4PrefixAndUpdatePool(job *worker.WarmPoolJ
 		didSucceed = false
 	}
 	job.Resources = failedResources
-	p.updatePoolAndReconcileIfRequired(instanceResource.resourcePool, job, didSucceed)
+	p.updatePoolAndReconcileIfRequired(instanceResource.resourcePool, job, didSucceed, true)
 }
 
 func (p *ipv4PrefixProvider) ReSyncPool(job *worker.WarmPoolJob) {
@@ -363,9 +376,10 @@ func (p *ipv4PrefixProvider) ProcessDeleteQueue(job *worker.WarmPoolJob) (ctrl.R
 }
 
 // updatePoolAndReconcileIfRequired updates the resource pool and reconcile again and submit a new job if required
-func (p *ipv4PrefixProvider) updatePoolAndReconcileIfRequired(resourcePool pool.Pool, job *worker.WarmPoolJob, didSucceed bool) {
+func (p *ipv4PrefixProvider) updatePoolAndReconcileIfRequired(resourcePool pool.Pool, job *worker.WarmPoolJob, didSucceed bool,
+	prefixAvailable bool) {
 	// Update the pool to add the created/failed resource to the warm pool and decrement the pending count
-	shouldReconcile := resourcePool.UpdatePool(job, didSucceed)
+	shouldReconcile := resourcePool.UpdatePool(job, didSucceed, prefixAvailable)
 
 	if shouldReconcile {
 		job := resourcePool.ReconcilePool()

pkg/provider/prefix/provider.go

@@ -135,9 +135,9 @@ func TestIpv4PrefixProvider_updatePoolAndReconcileIfRequired_NoFurtherReconcile(
 
 	job := &worker.WarmPoolJob{Operations: worker.OperationCreate}
 
-	mockPool.EXPECT().UpdatePool(job, true).Return(false)
+	mockPool.EXPECT().UpdatePool(job, true, true).Return(false)
 
-	provider.updatePoolAndReconcileIfRequired(mockPool, job, true)
+	provider.updatePoolAndReconcileIfRequired(mockPool, job, true, true)
 }
 
 // TestIpv4Provider_updatePoolAndReconcileIfRequired_ReconcileRequired tests pool is updated and reconciliation is
@@ -152,11 +152,11 @@ func TestIpv4Provider_updatePoolAndReconcileIfRequired_ReconcileRequired(t *test
 
 	job := &worker.WarmPoolJob{Operations: worker.OperationCreate}
 
-	mockPool.EXPECT().UpdatePool(job, true).Return(true)
+	mockPool.EXPECT().UpdatePool(job, true, true).Return(true)
 	mockPool.EXPECT().ReconcilePool().Return(job)
 	mockWorker.EXPECT().SubmitJob(job)
 
-	provider.updatePoolAndReconcileIfRequired(mockPool, job, true)
+	provider.updatePoolAndReconcileIfRequired(mockPool, job, true, true)
 }
 
 // TestIpv4PrefixProvider_DeleteIPv4PrefixAndUpdatePool tests job with empty resources is passed back if some resource
@@ -184,7 +184,7 @@ func TestIpv4PrefixProvider_DeleteIPv4PrefixAndUpdatePool(t *testing.T) {
 		Resources:     []string{},
 		ResourceCount: 2,
 		NodeName:      nodeName,
-	}, true).Return(false)
+	}, true, true).Return(false)
 
 	provider.DeleteIPv4PrefixAndUpdatePool(deleteJob)
 }
@@ -215,7 +215,7 @@ func TestIpv4PrefixProvider_DeletePrivateIPv4AndUpdatePool_SomeResourceFail(t *t
 		Resources:     failedResources,
 		ResourceCount: 2,
 		NodeName:      nodeName,
-	}, true).Return(false)
+	}, true, true).Return(false)
 
 	prefixProvider.DeleteIPv4PrefixAndUpdatePool(&deleteJob)
 }
@@ -245,7 +245,7 @@ func TestIPv4PrefixProvider_CreateIPv4PrefixAndUpdatePool(t *testing.T) {
 		Resources:     createdResources,
 		ResourceCount: 2,
 		NodeName:      nodeName,
-	}, true).Return(false)
+	}, true, true).Return(false)
 
 	prefixProvider.CreateIPv4PrefixAndUpdatePool(createJob)
 }
@@ -256,7 +256,9 @@ func TestIPv4PrefixProvider_CreateIPv4PrefixAndUpdatePool_Fail(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()
 
-	prefixProvider := getMockIPv4PrefixProvider()
+	mockWorker := mock_worker.NewMockWorker(ctrl)
+	prefixProvider := ipv4PrefixProvider{instanceProviderAndPool: map[string]*ResourceProviderAndPool{}, workerPool: mockWorker,
+		log: zap.New(zap.UseDevMode(true)).WithName("prefix provider")}
 	mockPool := mock_pool.NewMockPool(ctrl)
 	mockManager := mock_eni.NewMockENIManager(ctrl)
 	prefixProvider.putInstanceProviderAndPool(nodeName, mockPool, mockManager, nodeCapacity, true)
@@ -276,7 +278,11 @@ func TestIPv4PrefixProvider_CreateIPv4PrefixAndUpdatePool_Fail(t *testing.T) {
 		Resources:     createdResources,
 		ResourceCount: 3,
 		NodeName:      nodeName,
-	}, false).Return(false)
+	}, false, true).Return(true)
+
+	job := &worker.WarmPoolJob{Operations: worker.OperationCreate}
+	mockPool.EXPECT().ReconcilePool().Return(job)
+	mockWorker.EXPECT().SubmitJob(job)
 
 	prefixProvider.CreateIPv4PrefixAndUpdatePool(createJob)
 }
@@ -307,12 +313,15 @@ func TestIPv4PrefixProvider_CreateIPv4PrefixAndUpdatePool_Fail_InsufficientCidrB
 
 	mockManager.EXPECT().CreateIPV4Resource(3, config.ResourceTypeIPv4Prefix, nil, gomock.Any()).Return(createdResources,
 		fmt.Errorf("InsufficientCidrBlocks: The specified subnet does not have enough free cidr blocks to satisfy the request. Status"))
+
+	// Since InsufficientCidrBlocks error is not retryable, didSucceed was not set as false to not re-sync or reconcile the pool.
+	// Also set prefixAvailable as false to notify pool
 	mockPool.EXPECT().UpdatePool(&worker.WarmPoolJob{
 		Operations:    worker.OperationCreate,
 		Resources:     createdResources,
 		ResourceCount: 3,
 		NodeName:      nodeName,
-	}, false).Return(false)
+	}, true, false).Return(false)
 
 	mockK8sWrapper.EXPECT().GetNode(nodeName).Return(node, nil).Times(1)
 	mockK8sWrapper.EXPECT().BroadcastEvent(node, utils.InsufficientCidrBlocksReason, utils.ErrInsufficientCidrBlocks.Error(), v1.EventTypeWarning).Times(1)

pkg/provider/prefix/provider_test.go

@@ -1,9 +1,23 @@
 package utils
 
-import "errors"
+import (
+	"errors"
+	"strings"
+)
 
 var (
 	ErrNotFound                   = errors.New("resource was not found")
 	ErrInsufficientCidrBlocks     = errors.New("InsufficientCidrBlocks: The specified subnet does not have enough free cidr blocks to satisfy the request")
 	ErrMsgProviderAndPoolNotFound = "cannot find the instance provider and pool from the cache"
+	NotRetryErrors                = []string{InsufficientCidrBlocksReason}
 )
+
+// ShouldRetryOnError returns true if the error is retryable, else returns false
+func ShouldRetryOnError(err error) bool {
+	for _, e := range NotRetryErrors {
+		if strings.HasPrefix(err.Error(), e) {
+			return false
+		}
+	}
+	return true
+}