added prefix provider and support for PD configuration params (#223)

* added support for loading PD configuration from config map

* added prefix provider to manage ipv4 prefixes as a new resource

* updated pool to set PD configs to default if not valid

* added logger to log failures when parsing PD configs

controllers/core/configmap_controller.go

@@ -21,12 +21,14 @@ import (
 	"github.com/aws/amazon-vpc-resource-controller-k8s/pkg/config"
 	"github.com/aws/amazon-vpc-resource-controller-k8s/pkg/k8s"
 	"github.com/aws/amazon-vpc-resource-controller-k8s/pkg/node/manager"
+
 	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 )
 
 // ConfigMapReconciler reconciles a ConfigMap object
@@ -39,6 +41,9 @@ type ConfigMapReconciler struct {
 	Condition                         condition.Conditions
 	curWinIPAMEnabledCond             bool
 	curWinPrefixDelegationEnabledCond bool
+	curWinPDWarmIPTarget              int
+	curWinPDMinIPTarget               int
+	curWinPDWarmPrefixTarget          int
 }
 
 //+kubebuilder:rbac:groups=core,resources=configmaps,namespace=kube-system,resourceNames=amazon-vpc-cni,verbs=get;list;watch
@@ -65,7 +70,7 @@ func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		}
 	}
 
-	// Check if the flag value has changed
+	// Check if the Windows IPAM flag has changed
 	newWinIPAMEnabledCond := r.Condition.IsWindowsIPAMEnabled()
 
 	var isIPAMFlagUpdated bool
@@ -87,8 +92,21 @@ func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		isPrefixFlagUpdated = true
 	}
 
+	// Check if configurations for Windows prefix delegation have changed
+	var isPDConfigUpdated bool
+	warmIPTarget, minIPTarget, warmPrefixTarget := config.ParseWinPDTargets(r.Log, configmap)
+	if r.curWinPDWarmIPTarget != warmIPTarget || r.curWinPDMinIPTarget != minIPTarget || r.curWinPDWarmPrefixTarget != warmPrefixTarget {
+		r.curWinPDWarmIPTarget = warmIPTarget
+		r.curWinPDMinIPTarget = minIPTarget
+		r.curWinPDWarmPrefixTarget = warmPrefixTarget
+		logger.Info("updated PD configs from configmap", config.WarmIPTarget, r.curWinPDWarmIPTarget,
+			config.MinimumIPTarget, r.curWinPDMinIPTarget, config.WarmPrefixTarget, r.curWinPDWarmPrefixTarget)
+
+		isPDConfigUpdated = true
+	}
+
 	// Flag is updated, update all nodes
-	if isIPAMFlagUpdated || isPrefixFlagUpdated {
+	if isIPAMFlagUpdated || isPrefixFlagUpdated || isPDConfigUpdated {
 		err := UpdateNodesOnConfigMapChanges(r.K8sAPI, r.NodeManager)
 		if err != nil {
 			// Error in updating nodes
@@ -102,8 +120,11 @@ func (r *ConfigMapReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 
 // SetupWithManager sets up the controller with the Manager.
 func (r *ConfigMapReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	// Explicitly set MaxConcurrentReconciles to 1 to ensure concurrent reconciliation NOT supported for config map controller.
+	// Don't change to more than 1 unless the struct is guarded against concurrency issues.
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&corev1.ConfigMap{}).
+		WithOptions(controller.Options{MaxConcurrentReconciles: 1}).
 		Complete(r)
 }
 

controllers/core/pod_controller.go

@@ -201,18 +201,21 @@ func (r *PodReconciler) SetupWithManager(ctx context.Context, manager ctrl.Manag
 // updateResourceName updates resource name according to pod event and which IP allocation mode is enabled
 func (r *PodReconciler) updateResourceName(isDeletionEvent bool, pod *v1.Pod) string {
 	resourceName := config.ResourceNameIPAddress
+
 	// Pod deletion must use the handler that assigned the IP resource regardless which IP allocation mode is active
 	if isDeletionEvent {
 		resourceID, present := pod.Annotations[resourceName]
 		if !present {
 			r.Log.Error(fmt.Errorf("resource ID was not found in annotation"),
-				"resource", resourceName, "is delete event", isDeletionEvent)
+				"failed to find resource ID", "resource", resourceName, "is delete event", isDeletionEvent)
 			return resourceName
 		}
 		// Check prefix provider to see if it's a prefix deconstructed IP
 		prefixProvider, found := r.ResourceManager.GetResourceProvider(config.ResourceNameIPAddressFromPrefix)
 		if !found {
-			r.Log.Error(fmt.Errorf("resource provider was not found"), "resource", resourceName)
+			// If prefix provider not found, log the error and continue to use the secondary IP provider
+			r.Log.Error(fmt.Errorf("resource provider was not found"), "failed to find resource provider",
+				"resource", resourceName)
 			return resourceName
 		}
 		resourcePool, ok := prefixProvider.GetPool(pod.Spec.NodeName)
@@ -223,7 +226,7 @@ func (r *PodReconciler) updateResourceName(isDeletionEvent bool, pod *v1.Pod) st
 	} else {
 		// Pod creation should use the currently active resource handler
 		if r.Condition.IsWindowsPrefixDelegationEnabled() {
-			// If prefix delegation is enabled, update resource name so that prefix handler will be used
+			// If prefix delegation is enabled, update resource name so that prefix IP handler will be used
 			resourceName = config.ResourceNameIPAddressFromPrefix
 		}
 	}

controllers/core/pod_controller_test.go

@@ -22,9 +22,11 @@ import (
 	mock_k8s "github.com/aws/amazon-vpc-resource-controller-k8s/mocks/amazon-vcp-resource-controller-k8s/pkg/k8s"
 	mock_node "github.com/aws/amazon-vpc-resource-controller-k8s/mocks/amazon-vcp-resource-controller-k8s/pkg/node"
 	mock_manager "github.com/aws/amazon-vpc-resource-controller-k8s/mocks/amazon-vcp-resource-controller-k8s/pkg/node/manager"
+	mock_pool "github.com/aws/amazon-vpc-resource-controller-k8s/mocks/amazon-vcp-resource-controller-k8s/pkg/pool"
+	mock_provider "github.com/aws/amazon-vpc-resource-controller-k8s/mocks/amazon-vcp-resource-controller-k8s/pkg/provider"
 	mock_resource "github.com/aws/amazon-vpc-resource-controller-k8s/mocks/amazon-vcp-resource-controller-k8s/pkg/resource"
+	"github.com/aws/amazon-vpc-resource-controller-k8s/pkg/config"
 	"github.com/aws/amazon-vpc-resource-controller-k8s/pkg/k8s/pod"
-
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
@@ -53,8 +55,9 @@ var (
 	mockPod = &v1.Pod{
 		TypeMeta: metav1.TypeMeta{},
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      mockPodName,
-			Namespace: mockPodNS,
+			Name:        mockPodName,
+			Namespace:   mockPodNS,
+			Annotations: map[string]string{config.ResourceNameIPAddress: "192.168.10.0/32"},
 		},
 		Spec: v1.PodSpec{
 			NodeName: mockNodeName,
@@ -93,6 +96,7 @@ type Mock struct {
 	MockNode            *mock_node.MockNode
 	PodReconciler       *PodReconciler
 	MockHandler         *mock_handler.MockHandler
+	MockProvider        *mock_provider.MockResourceProvider
 }
 
 func NewMock(ctrl *gomock.Controller, mockPod *v1.Pod) Mock {
@@ -101,7 +105,6 @@ func NewMock(ctrl *gomock.Controller, mockPod *v1.Pod) Mock {
 	mockResourceManager := mock_resource.NewMockResourceManager(ctrl)
 	mockNode := mock_node.NewMockNode(ctrl)
 	mockK8sWrapper := mock_k8s.NewMockK8sWrapper(ctrl)
-
 	converter := pod.PodConverter{}
 	mockIndexer := cache.NewIndexer(converter.Indexer, pod.NodeNameIndexer())
 	mockIndexer.Add(mockPod)
@@ -308,3 +311,38 @@ func TestPodReconcile_Reconcile_PodDeletedManagedNodeDeletedFromCluster(t *testi
 	assert.NoError(t, err)
 	assert.Equal(t, result, controllerruntime.Result{})
 }
+
+// TestUpdateResourceName_IsDeleteEvent_PrefixProvider tests for pod deletion events, prefix handler is used when ip is managed by prefix
+// ip pool
+func TestUpdateResourceName_IsDeleteEvent_PrefixProvider(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mock := NewMock(ctrl, mockPod)
+	mockPool := mock_pool.NewMockPool(ctrl)
+	mockProvider := mock_provider.NewMockResourceProvider(ctrl)
+	mock.MockResourceManager.EXPECT().GetResourceProvider(config.ResourceNameIPAddressFromPrefix).Return(mockProvider, true)
+	mockProvider.EXPECT().GetPool(mockPod.Spec.NodeName).Return(mockPool, true)
+	mockPool.EXPECT().IsManagedResource(mockPod.Annotations[config.ResourceNameIPAddress]).Return(true)
+	resourceName := mock.PodReconciler.updateResourceName(true, mockPod)
+
+	assert.Equal(t, config.ResourceNameIPAddressFromPrefix, resourceName)
+}
+
+// TestUpdateResourceName_IsDeleteEvent_SecondaryIPProvider tests for pod deletion events, secondary ip handler is used when ip is managed
+// by secondary ip pool
+func TestUpdateResourceName_IsDeleteEvent_SecondaryIPProvider(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	mock := NewMock(ctrl, mockPod)
+	mockPool := mock_pool.NewMockPool(ctrl)
+	mockProvider := mock_provider.NewMockResourceProvider(ctrl)
+	mock.MockResourceManager.EXPECT().GetResourceProvider(config.ResourceNameIPAddressFromPrefix).Return(mockProvider, true)
+	mockProvider.EXPECT().GetPool(mockPod.Spec.NodeName).Return(mockPool, true)
+	mockPool.EXPECT().IsManagedResource(mockPod.Annotations[config.ResourceNameIPAddress]).Return(false)
+	resourceName := mock.PodReconciler.updateResourceName(true, mockPod)
+
+	// since resource ip is not managed by prefix ip pool, resource name remains unchanged
+	assert.Equal(t, config.ResourceNameIPAddress, resourceName)
+}

main.go

@@ -278,8 +278,7 @@ func main() {
 	// hasPodDataStoreSynced is set to true when the custom controller has synced
 	controllerConditions := condition.NewControllerConditions(
 		ctrl.Log.WithName("controller conditions"), k8sApi)
-	//supportedResources := []string{config.ResourceNamePodENI, config.ResourceNameIPAddress, config.ResourceNameIPAddressFromPrefix}
-	supportedResources := []string{config.ResourceNamePodENI, config.ResourceNameIPAddress}
+	supportedResources := []string{config.ResourceNamePodENI, config.ResourceNameIPAddress, config.ResourceNameIPAddressFromPrefix}
 	resourceManager, err := resource.NewResourceManager(ctx, supportedResources, apiWrapper, controllerConditions)
 	if err != nil {
 		ctrl.Log.Error(err, "failed to init resources", "resources", supportedResources)

pkg/config/loader.go

@@ -13,6 +13,13 @@
 
 package config
 
+import (
+	"strconv"
+
+	"github.com/go-logr/logr"
+	v1 "k8s.io/api/core/v1"
+)
+
 const (
 	// TODO: Should we always do this max retry no matter why it fails
 	// such deleted pods will also be retried 5 times, which could be an issue for large pods loads and high churning rate.
@@ -59,6 +66,67 @@ func LoadResourceConfig() map[string]ResourceConfig {
 	return getDefaultResourceConfig()
 }
 
+func LoadResourceConfigFromConfigMap(log logr.Logger, vpcCniConfigMap *v1.ConfigMap) map[string]ResourceConfig {
+	resourceConfig := getDefaultResourceConfig()
+
+	warmIPTarget, minIPTarget, warmPrefixTarget := ParseWinPDTargets(log, vpcCniConfigMap)
+
+	// If no PD configuration is set in configMap or none is valid, return default resource config
+	if warmIPTarget == 0 && minIPTarget == 0 && warmPrefixTarget == 0 {
+		return resourceConfig
+	}
+
+	resourceConfig[ResourceNameIPAddressFromPrefix].WarmPoolConfig.WarmIPTarget = warmIPTarget
+	resourceConfig[ResourceNameIPAddressFromPrefix].WarmPoolConfig.MinIPTarget = minIPTarget
+	resourceConfig[ResourceNameIPAddressFromPrefix].WarmPoolConfig.WarmPrefixTarget = warmPrefixTarget
+
+	return resourceConfig
+}
+
+// ParseWinPDTargets parses config map for Windows prefix delegation configurations set by users
+func ParseWinPDTargets(log logr.Logger, vpcCniConfigMap *v1.ConfigMap) (warmIPTarget int, minIPTarget int, warmPrefixTarget int) {
+	warmIPTarget, minIPTarget, warmPrefixTarget = 0, 0, 0
+
+	if vpcCniConfigMap.Data == nil {
+		return warmIPTarget, minIPTarget, warmPrefixTarget
+	}
+
+	warmIPTargetStr, foundWarmIP := vpcCniConfigMap.Data[WarmIPTarget]
+	minIPTargetStr, foundMinIP := vpcCniConfigMap.Data[MinimumIPTarget]
+	warmPrefixTargetStr, foundWarmPrefix := vpcCniConfigMap.Data[WarmPrefixTarget]
+
+	// If no configuration is found, return 0
+	if !foundWarmIP && !foundMinIP && !foundWarmPrefix {
+		return warmIPTarget, minIPTarget, warmPrefixTarget
+	}
+
+	if foundWarmIP {
+		warmIPTargetInt, err := strconv.Atoi(warmIPTargetStr)
+		if err != nil {
+			log.Error(err, "failed to parse warm ip target", "warm ip target", warmIPTargetStr)
+		} else {
+			warmIPTarget = warmIPTargetInt
+		}
+	}
+	if foundMinIP {
+		minIPTargetInt, err := strconv.Atoi(minIPTargetStr)
+		if err != nil {
+			log.Error(err, "failed to parse minimum ip target", "minimum ip target", minIPTargetStr)
+		} else {
+			minIPTarget = minIPTargetInt
+		}
+	}
+	if foundWarmPrefix {
+		warmPrefixTargetInt, err := strconv.Atoi(warmPrefixTargetStr)
+		if err != nil {
+			log.Error(err, "failed to parse warm prefix target", "warm prefix target", warmPrefixTargetStr)
+		} else {
+			warmPrefixTarget = warmPrefixTargetInt
+		}
+	}
+	return warmIPTarget, minIPTarget, warmPrefixTarget
+}
+
 // getDefaultResourceConfig returns the default Resource Configuration.
 func getDefaultResourceConfig() map[string]ResourceConfig {
 
@@ -87,5 +155,22 @@ func getDefaultResourceConfig() map[string]ResourceConfig {
 	}
 	config[ResourceNameIPAddress] = ipV4Config
 
+	// Create default configuration for prefix-deconstructed IPv4 resource pool
+	prefixIPv4WarmPoolConfig := WarmPoolConfig{
+		DesiredSize:      IPv4PDDefaultWPSize,
+		MaxDeviation:     IPv4PDDefaultMaxDev,
+		ReservedSize:     IPv4PDDefaultResSize,
+		WarmIPTarget:     IPv4PDDefaultWarmIPTargetSize,
+		MinIPTarget:      IPv4PDDefaultMinIPTargetSize,
+		WarmPrefixTarget: IPv4PDDefaultWarmPrefixTargetSize,
+	}
+	prefixIPv4Config := ResourceConfig{
+		Name:           ResourceNameIPAddressFromPrefix,
+		WorkerCount:    IPv4PDDefaultWorker,
+		SupportedOS:    map[string]bool{OSWindows: true, OSLinux: false},
+		WarmPoolConfig: &prefixIPv4WarmPoolConfig,
+	}
+	config[ResourceNameIPAddressFromPrefix] = prefixIPv4Config
+
 	return config
 }