Implement simple endpoint slice batching

Author: mborsz

api/api-rules/violation_exceptions.list

@@ -597,6 +597,7 @@ API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,D
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,EndpointControllerConfiguration,ConcurrentEndpointSyncs
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,EndpointControllerConfiguration,EndpointUpdatesBatchPeriod
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,EndpointSliceControllerConfiguration,ConcurrentServiceEndpointSyncs
+API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,EndpointSliceControllerConfiguration,EndpointUpdatesBatchPeriod
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,EndpointSliceControllerConfiguration,MaxEndpointsPerSlice
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,GarbageCollectorControllerConfiguration,ConcurrentGCSyncs
 API rule violation: names_match,k8s.io/kube-controller-manager/config/v1alpha1,GarbageCollectorControllerConfiguration,EnableGarbageCollector

cmd/kube-controller-manager/app/discovery.go

@@ -48,6 +48,7 @@ func startEndpointSliceController(ctx ControllerContext) (http.Handler, bool, er
 		ctx.InformerFactory.Discovery().V1beta1().EndpointSlices(),
 		ctx.ComponentConfig.EndpointSliceController.MaxEndpointsPerSlice,
 		ctx.ClientBuilder.ClientOrDie("endpointslice-controller"),
+		ctx.ComponentConfig.EndpointSliceController.EndpointUpdatesBatchPeriod.Duration,
 	).Run(int(ctx.ComponentConfig.EndpointSliceController.ConcurrentServiceEndpointSyncs), ctx.Stop)
 	return nil, true, nil
 }

cmd/kube-controller-manager/app/options/endpointslicecontroller.go

@@ -18,6 +18,7 @@ package options
 
 import (
 	"fmt"
+
 	"github.com/spf13/pflag"
 
 	endpointsliceconfig "k8s.io/kubernetes/pkg/controller/endpointslice/config"
@@ -43,6 +44,7 @@ func (o *EndpointSliceControllerOptions) AddFlags(fs *pflag.FlagSet) {
 
 	fs.Int32Var(&o.ConcurrentServiceEndpointSyncs, "concurrent-service-endpoint-syncs", o.ConcurrentServiceEndpointSyncs, "The number of service endpoint syncing operations that will be done concurrently. Larger number = faster endpoint slice updating, but more CPU (and network) load. Defaults to 5.")
 	fs.Int32Var(&o.MaxEndpointsPerSlice, "max-endpoints-per-slice", o.MaxEndpointsPerSlice, "The maximum number of endpoints that will be added to an EndpointSlice. More endpoints per slice will result in less endpoint slices, but larger resources. Defaults to 100.")
+	fs.DurationVar(&o.EndpointUpdatesBatchPeriod.Duration, "endpointslice-updates-batch-period", o.EndpointUpdatesBatchPeriod.Duration, "The length of endpoint slice updates batching period. Processing of pod changes will be delayed by this duration to join them with potential upcoming updates and reduce the overall number of endpoints updates. Larger number = higher endpoint programming latency, but lower number of endpoints revision generated")
 }
 
 // ApplyTo fills up EndpointSliceController config with options.
@@ -53,6 +55,7 @@ func (o *EndpointSliceControllerOptions) ApplyTo(cfg *endpointsliceconfig.Endpoi
 
 	cfg.ConcurrentServiceEndpointSyncs = o.ConcurrentServiceEndpointSyncs
 	cfg.MaxEndpointsPerSlice = o.MaxEndpointsPerSlice
+	cfg.EndpointUpdatesBatchPeriod = o.EndpointUpdatesBatchPeriod
 
 	return nil
 }

pkg/controller/endpointslice/config/BUILD

@@ -9,6 +9,7 @@ go_library(
     ],
     importpath = "k8s.io/kubernetes/pkg/controller/endpointslice/config",
     visibility = ["//visibility:public"],
+    deps = ["//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library"],
 )
 
 filegroup(

pkg/controller/endpointslice/config/types.go

@@ -16,6 +16,10 @@ limitations under the License.
 
 package config
 
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
 // EndpointSliceControllerConfiguration contains elements describing
 // EndpointSliceController.
 type EndpointSliceControllerConfiguration struct {
@@ -28,4 +32,11 @@ type EndpointSliceControllerConfiguration struct {
 	// added to an EndpointSlice. More endpoints per slice will result in fewer
 	// and larger endpoint slices, but larger resources.
 	MaxEndpointsPerSlice int32
+
+	// EndpointUpdatesBatchPeriod can be used to batch endpoint updates.
+	// All updates of endpoint triggered by pod change will be delayed by up to
+	// 'EndpointUpdatesBatchPeriod'. If other pods in the same endpoint change
+	// in that period, they will be batched to a single endpoint update.
+	// Default 0 value means that each pod update triggers an endpoint update.
+	EndpointUpdatesBatchPeriod metav1.Duration
 }

pkg/controller/endpointslice/config/v1alpha1/zz_generated.conversion.go

@@ -66,6 +66,7 @@ func NewController(podInformer coreinformers.PodInformer,
 	endpointSliceInformer discoveryinformers.EndpointSliceInformer,
 	maxEndpointsPerSlice int32,
 	client clientset.Interface,
+	endpointUpdatesBatchPeriod time.Duration,
 ) *Controller {
 	broadcaster := record.NewBroadcaster()
 	broadcaster.StartLogging(klog.Infof)
@@ -129,6 +130,7 @@ func NewController(podInformer coreinformers.PodInformer,
 	c.eventBroadcaster = broadcaster
 	c.eventRecorder = recorder
 
+	c.endpointUpdatesBatchPeriod = endpointUpdatesBatchPeriod
 	c.serviceSelectorCache = endpointutil.NewServiceSelectorCache()
 
 	return c
@@ -194,6 +196,10 @@ type Controller struct {
 	// process the queue of service and pod changes
 	workerLoopPeriod time.Duration
 
+	// endpointUpdatesBatchPeriod is an artificial delay added to all service syncs triggered by pod changes.
+	// This can be used to reduce overall number of all endpoint slice updates.
+	endpointUpdatesBatchPeriod time.Duration
+
 	// serviceSelectorCache is a cache of service selectors to avoid high CPU consumption caused by frequent calls
 	// to AsSelectorPreValidated (see #73527)
 	serviceSelectorCache *endpointutil.ServiceSelectorCache
@@ -414,14 +420,14 @@ func (c *Controller) addPod(obj interface{}) {
 		return
 	}
 	for key := range services {
-		c.queue.Add(key)
+		c.queue.AddAfter(key, c.endpointUpdatesBatchPeriod)
 	}
 }
 
 func (c *Controller) updatePod(old, cur interface{}) {
 	services := endpointutil.GetServicesToUpdateOnPodChange(c.serviceLister, c.serviceSelectorCache, old, cur, podEndpointChanged)
 	for key := range services {
-		c.queue.Add(key)
+		c.queue.AddAfter(key, c.endpointUpdatesBatchPeriod)
 	}
 }