working implementation on both target-type IP and NodePort

Author: mmiller-sh

main.go

@@ -185,7 +185,7 @@ func main() {
 tgbResManager := targetgroupbinding.NewDefaultResourceManager(mgr.GetClient(), cloud.ELBV2(),
 podInfoRepo, networkingManager, vpcInfoProvider, multiClusterManager, lbcMetricsCollector,
 cloud.VpcID(), controllerCFG.FeatureGates.Enabled(config.EndpointsFailOpen), controllerCFG.EnableEndpointSlices,
-mgr.GetEventRecorderFor("targetGroupBinding"), ctrl.Log)
+mgr.GetEventRecorderFor("targetGroupBinding"), ctrl.Log, controllerCFG.MaxTargetsPerInstance)
 backendSGProvider := networking.NewBackendSGProvider(controllerCFG.ClusterName, controllerCFG.BackendSecurityGroup,
 cloud.VpcID(), cloud.EC2(), mgr.GetClient(), controllerCFG.DefaultTags, nlbGatewayEnabled || albGatewayEnabled, ctrl.Log.WithName("backend-sg-provider"))
 sgResolver := networking.NewDefaultSecurityGroupResolver(cloud.EC2(), cloud.VpcID())

pkg/targetgroupbinding/resource_manager.go

@@ -48,7 +48,8 @@ func NewDefaultResourceManager(k8sClient client.Client, elbv2Client services.ELB
 podInfoRepo k8s.PodInfoRepo, networkingManager networking.NetworkingManager,
 vpcInfoProvider networking.VPCInfoProvider, multiClusterManager MultiClusterManager, metricsCollector lbcmetrics.MetricCollector,
 vpcID string, failOpenEnabled bool, endpointSliceEnabled bool,
-eventRecorder record.EventRecorder, logger logr.Logger) *defaultResourceManager {
+eventRecorder record.EventRecorder, logger logr.Logger, maxTargetsPerInstance int) *defaultResourceManager {
+
 targetsManager := NewCachedTargetsManager(elbv2Client, logger)
 endpointResolver := backend.NewDefaultEndpointResolver(k8sClient, podInfoRepo, failOpenEnabled, endpointSliceEnabled, logger)
 return &defaultResourceManager{
@@ -61,6 +62,7 @@ func NewDefaultResourceManager(k8sClient client.Client, elbv2Client services.ELB
 vpcID: vpcID,
 vpcInfoProvider: vpcInfoProvider,
 podInfoRepo: podInfoRepo,
+maxTargetsPerInstance: maxTargetsPerInstance,
 multiClusterManager: multiClusterManager,
 metricsCollector: metricsCollector,
 
@@ -83,6 +85,7 @@ type defaultResourceManager struct {
 logger logr.Logger
 vpcInfoProvider networking.VPCInfoProvider
 podInfoRepo k8s.PodInfoRepo
+maxTargetsPerInstance int
 multiClusterManager MultiClusterManager
 metricsCollector lbcmetrics.MetricCollector
 vpcID string
@@ -180,6 +183,7 @@ func (m *defaultResourceManager) reconcileWithIPTargetType(ctx context.Context,
 if err != nil {
 return "", "", false, ctrlerrors.NewErrorWithMetrics(controllerName, "list_targets_error", err, m.metricsCollector)
 }
+totalTargets := len(targets)
 
 notDrainingTargets, _ := partitionTargetsByDrainingStatus(targets)
 matchedEndpointAndTargets, unmatchedEndpoints, unmatchedTargets := matchPodEndpointWithTargets(endpoints, notDrainingTargets)
@@ -240,6 +244,23 @@ func (m *defaultResourceManager) reconcileWithIPTargetType(ctx context.Context,
 return "", "", false, ctrlerrors.NewErrorWithMetrics(controllerName, "update_tracked_ip_targets_error", err, m.metricsCollector)
 }
 
+// Log that we're witholding target additions to prevent exceeding max-targets-per-instance
+var limitedUnmatchedEndpoints []backend.PodEndpoint
+
+if m.maxTargetsPerInstance > 0 && len(unmatchedEndpoints) + totalTargets > m.maxTargetsPerInstance {
+maxAdditions := m.maxTargetsPerInstance - totalTargets
+if maxAdditions > 0 {
+limitedUnmatchedEndpoints = unmatchedEndpoints[:maxAdditions]
+}
+tgbScopedLogger.Info("Limiting target additions due to max-targets-per-instance configuration",
+"currentTargets", totalTargets,
+"maxTargetsPerInstance", m.maxTargetsPerInstance,
+"proposedAdditions", len(unmatchedEndpoints),
+"numberOmitted", len(unmatchedEndpoints) - len(limitedUnmatchedEndpoints))
+
+unmatchedEndpoints = limitedUnmatchedEndpoints
+}
+
 if err := m.registerPodEndpoints(ctx, tgb, unmatchedEndpoints); err != nil {
 return "", "", false, ctrlerrors.NewErrorWithMetrics(controllerName, "register_pod_endpoint_error", err, m.metricsCollector)
 }
@@ -307,6 +328,7 @@ func (m *defaultResourceManager) reconcileWithInstanceTargetType(ctx context.Con
 if err != nil {
 return "", "", false, ctrlerrors.NewErrorWithMetrics(controllerName, "list_targets_error", err, m.metricsCollector)
 }
+totalTargets := len(targets)
 
 notDrainingTargets, _ := partitionTargetsByDrainingStatus(targets)
 
@@ -341,6 +363,22 @@ func (m *defaultResourceManager) reconcileWithInstanceTargetType(ctx context.Con
 return "", "", false, ctrlerrors.NewErrorWithMetrics(controllerName, "update_tracked_instance_targets_error", err, m.metricsCollector)
 }
 
+var limitedUnmatchedEndpoints []backend.NodePortEndpoint
+
+if m.maxTargetsPerInstance > 0 && len(unmatchedEndpoints) + totalTargets > m.maxTargetsPerInstance {
+maxAdditions := m.maxTargetsPerInstance - totalTargets
+if maxAdditions > 0 {
+limitedUnmatchedEndpoints = unmatchedEndpoints[:maxAdditions]
+}
+tgbScopedLogger.Info("Limiting target additions due to max-targets-per-instance configuration",
+"currentTargets", totalTargets,
+"maxTargetsPerInstance", m.maxTargetsPerInstance,
+"proposedAdditions", len(unmatchedEndpoints),
+"numberOmitted", len(unmatchedEndpoints) - len(limitedUnmatchedEndpoints))
+
+unmatchedEndpoints = limitedUnmatchedEndpoints
+}
+
 if err := m.registerNodePortEndpoints(ctx, tgb, unmatchedEndpoints); err != nil {
 return "", "", false, ctrlerrors.NewErrorWithMetrics(controllerName, "update_node_port_endpoints_error", err, m.metricsCollector)
 }

pkg/targetgroupbinding/targets_manager.go

@@ -18,7 +18,6 @@ const (
 defaultTargetsCacheTTL = 5 * time.Minute
 defaultRegisterTargetsChunkSize = 200
 defaultDeregisterTargetsChunkSize = 200
-maxTargetsPerInstance = 500
 )
 
 // TargetsManager is an abstraction around ELBV2's targets API.
@@ -81,16 +80,7 @@ type targetsCacheItem struct {
 
 func (m *cachedTargetsManager) RegisterTargets(ctx context.Context, tgb *elbv2api.TargetGroupBinding, targets []elbv2types.TargetDescription) error {
 tgARN := tgb.Spec.TargetGroupARN
-sampledTargets := targets
-
-m.logger.Info("Number of targets", len(targets), "registering a subset per max-targets-per-instance", maxTargetsPerInstance)
-if maxTargetsPerInstance > 0 && len(targets) > maxTargetsPerInstance {
-m.logger.Info("Max number of targets exceeded", len(targets), "registering a subset per max-targets-per-instance", maxTargetsPerInstance)
-m.logger.Info("Max number of targets exceeded", len(targets), "registering a subset per max-targets-per-instance", maxTargetsPerInstance)
-sampledTargets = targets[:maxTargetsPerInstance]
-}
-
-targetsChunks := chunkTargetDescriptions(sampledTargets, m.registerTargetsChunkSize)
+targetsChunks := chunkTargetDescriptions(targets, m.registerTargetsChunkSize)
 for _, targetsChunk := range targetsChunks {
 req := &elbv2sdk.RegisterTargetsInput{
 TargetGroupArn: aws.String(tgARN),