Review comments

Author: carreter

cmd/agent/app/options/options.go

@@ -27,7 +27,6 @@ import (
 	"github.com/google/uuid"
 	"github.com/spf13/pflag"
 	"google.golang.org/grpc"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/klog/v2"
 
 	"sigs.k8s.io/apiserver-network-proxy/pkg/agent"
@@ -82,9 +81,9 @@ type GrpcProxyAgentOptions struct {
 	SyncForever    bool
 	XfrChannelSize int
 
-	// Providing a label selector enables updating the server count by counting the
-	// number of valid leases matching the selector.
-	ServerLeaseSelector string
+	// Enables updating the server count by counting the number of valid leases
+	// matching the selector.
+	CountServerLeases bool
 	// Path to kubeconfig (used by kubernetes client for lease listing)
 	KubeconfigPath string
 }
@@ -129,7 +128,7 @@ func (o *GrpcProxyAgentOptions) Flags() *pflag.FlagSet {
 	flags.BoolVar(&o.WarnOnChannelLimit, "warn-on-channel-limit", o.WarnOnChannelLimit, "Turns on a warning if the system is going to push to a full channel. The check involves an unsafe read.")
 	flags.BoolVar(&o.SyncForever, "sync-forever", o.SyncForever, "If true, the agent continues syncing, in order to support server count changes.")
 	flags.IntVar(&o.XfrChannelSize, "xfr-channel-size", 150, "Set the size of the channel for transferring data between the agent and the proxy server.")
-	flags.StringVar(&o.ServerLeaseSelector, "server-lease-selector", o.ServerLeaseSelector, "Providing a label selector enables updating the server count by counting the number of valid leases matching the selector.")
+	flags.BoolVar(&o.CountServerLeases, "count-server-leases", o.CountServerLeases, "Enables lease counting system to determine the number of proxy servers to connect to.")
 	flags.StringVar(&o.KubeconfigPath, "kubeconfig", o.KubeconfigPath, "Path to the kubeconfig file")
 	return flags
 }
@@ -207,11 +206,6 @@ func (o *GrpcProxyAgentOptions) Validate() error {
 	if err := validateAgentIdentifiers(o.AgentIdentifiers); err != nil {
 		return fmt.Errorf("agent address is invalid: %v", err)
 	}
-	if o.ServerLeaseSelector != "" {
-		if _, err := labels.Parse(o.ServerLeaseSelector); err != nil {
-			return fmt.Errorf("invalid server count lease selector: %w", err)
-		}
-	}
 	if o.KubeconfigPath != "" {
 		if _, err := os.Stat(o.KubeconfigPath); os.IsNotExist(err) {
 			return fmt.Errorf("error checking KubeconfigPath %q, got %v", o.KubeconfigPath, err)
@@ -263,7 +257,7 @@ func NewGrpcProxyAgentOptions() *GrpcProxyAgentOptions {
 		WarnOnChannelLimit:        false,
 		SyncForever:               false,
 		XfrChannelSize:            150,
-		ServerLeaseSelector:       "",
+		CountServerLeases:         false,
 		KubeconfigPath:            "",
 	}
 	return &o

cmd/agent/app/server.go

@@ -42,12 +42,16 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/clock"
 	"sigs.k8s.io/apiserver-network-proxy/cmd/agent/app/options"
 	"sigs.k8s.io/apiserver-network-proxy/pkg/agent"
 	"sigs.k8s.io/apiserver-network-proxy/pkg/util"
 )
 
-const ReadHeaderTimeout = 60 * time.Second
+const (
+	ReadHeaderTimeout = 60 * time.Second
+	LeaseNamespace    = "kube-system"
+)
 
 func NewAgentCommand(a *Agent, o *options.GrpcProxyAgentOptions) *cobra.Command {
 	cmd := &cobra.Command{
@@ -136,7 +140,7 @@ func (a *Agent) runProxyConnection(o *options.GrpcProxyAgentOptions, drainCh, st
 	}
 	cc := o.ClientSetConfig(dialOptions...)
 
-	if o.ServerLeaseSelector != "" {
+	if o.CountServerLeases {
 		var k8sClient *kubernetes.Clientset
 		config, err := clientcmd.BuildConfigFromFlags("", o.KubeconfigPath)
 		if err != nil {
@@ -146,18 +150,13 @@ func (a *Agent) runProxyConnection(o *options.GrpcProxyAgentOptions, drainCh, st
 		if err != nil {
 			return nil, fmt.Errorf("failed to create kubernetes clientset: %v", err)
 		}
-		serverLeaseSelector, err := labels.Parse(o.ServerLeaseSelector)
-		if err != nil {
-			return nil, fmt.Errorf("invalid server count lease selector: %w", err)
-		}
+		serverLeaseSelector, _ := labels.Parse("k8s-app=konnectivity-server")
 		serverLeaseCounter := agent.NewServerLeaseCounter(
+			clock.RealClock{},
 			k8sClient,
 			serverLeaseSelector,
+			LeaseNamespace,
 		)
-		if err != nil {
-			klog.Errorf("failed to create server lease counter: %v", err)
-			return nil, fmt.Errorf("failed to create server lease counter: %w", err)
-		}
 		cc.ServerLeaseCounter = serverLeaseCounter
 	}
 

cmd/server/app/options/options.go

@@ -104,11 +104,7 @@ type ProxyRunOptions struct {
 	XfrChannelSize int
 
 	// Lease controller configuration
-	LeaseDuration        time.Duration
-	LeaseRenewalInterval time.Duration
-	LeaseGCInterval      time.Duration
-	LeaseLabelSelector   string
-	LeaseNamespace       string
+	EnableLeaseController bool
 }
 
 func (o *ProxyRunOptions) Flags() *pflag.FlagSet {
@@ -145,11 +141,7 @@ func (o *ProxyRunOptions) Flags() *pflag.FlagSet {
 	flags.StringVar(&o.ProxyStrategies, "proxy-strategies", o.ProxyStrategies, "The list of proxy strategies used by the server to pick an agent/tunnel, available strategies are: default, destHost, defaultRoute.")
 	flags.StringSliceVar(&o.CipherSuites, "cipher-suites", o.CipherSuites, "The comma separated list of allowed cipher suites. Has no effect on TLS1.3. Empty means allow default list.")
 	flags.IntVar(&o.XfrChannelSize, "xfr-channel-size", o.XfrChannelSize, "The size of the two KNP server channels used in server for transferring data. One channel is for data coming from the Kubernetes API Server, and the other one is for data coming from the KNP agent.")
-	flags.DurationVar(&o.LeaseDuration, "lease-duration", o.LeaseDuration, "The duration of the KNP server lease. Lease system off by default")
-	flags.DurationVar(&o.LeaseRenewalInterval, "lease-renewal-interval", o.LeaseRenewalInterval, "The interval between KNP server lease renewal calls. Lease system off by default")
-	flags.DurationVar(&o.LeaseGCInterval, "lease-gc-interval", o.LeaseGCInterval, "The interval between KNP server garbage collection calls. Lease system off by default")
-	flags.StringVar(&o.LeaseLabelSelector, "lease-label-selector", o.LeaseLabelSelector, "The label selector for KNP server leases. Lease system off by default")
-	flags.StringVar(&o.LeaseNamespace, "lease-namespace", o.LeaseNamespace, "The namespace to which KNP server leases will be published. Lease system off by default")
+	flags.BoolVar(&o.EnableLeaseController, "enable-lease-controller", o.EnableLeaseController, "Enable lease controller to publish and garbage collect proxy server leases.")
 	flags.Bool("warn-on-channel-limit", true, "This behavior is now thread safe and always on. This flag will be removed in a future release.")
 	flags.MarkDeprecated("warn-on-channel-limit", "This behavior is now thread safe and always on. This flag will be removed in a future release.")
 
@@ -304,25 +296,6 @@ func (o *ProxyRunOptions) Validate() error {
 		}
 	}
 
-	// Validate leasing parameters. All must be empty or have a value.
-	if o.IsLeaseCountingEnabled() {
-		if o.LeaseLabelSelector == "" {
-			return fmt.Errorf("LeaseLabelSelector cannot be empty when leasing system is enabled")
-		}
-		if o.LeaseNamespace == "" {
-			return fmt.Errorf("LeaseNamespace cannot be empty when leasing system is enabled")
-		}
-		if o.LeaseDuration <= 0 {
-			return fmt.Errorf("LeaseDuration cannot be zero or negative when leasing system is enabled")
-		}
-		if o.LeaseGCInterval <= 0 {
-			return fmt.Errorf("LeaseGCInterval cannot be zero or negative when leasing system is enabled")
-		}
-		if o.LeaseRenewalInterval <= 0 {
-			return fmt.Errorf("LeaseRenewalInterval cannot be zero or negative when leasing system is enabled")
-		}
-	}
-
 	// validate the proxy strategies
 	if len(o.ProxyStrategies) == 0 {
 		return fmt.Errorf("ProxyStrategies cannot be empty")
@@ -381,11 +354,7 @@ func NewProxyRunOptions() *ProxyRunOptions {
 		ProxyStrategies:           "default",
 		CipherSuites:              make([]string, 0),
 		XfrChannelSize:            10,
-		LeaseGCInterval:           0,
-		LeaseDuration:             0,
-		LeaseRenewalInterval:      0,
-		LeaseLabelSelector:        "",
-		LeaseNamespace:            "",
+		EnableLeaseController:     false,
 	}
 	return &o
 }
@@ -398,7 +367,3 @@ func defaultServerID() string {
 	}
 	return uuid.New().String()
 }
-
-func (o *ProxyRunOptions) IsLeaseCountingEnabled() bool {
-	return o.LeaseLabelSelector != "" || o.LeaseDuration != 0 || o.LeaseGCInterval != 0 || o.LeaseRenewalInterval != 0 || o.LeaseNamespace != ""
-}

cmd/server/app/server.go

@@ -39,7 +39,6 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/keepalive"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
@@ -53,7 +52,13 @@ import (
 
 var udsListenerLock sync.Mutex
 
-const ReadHeaderTimeout = 60 * time.Second
+const (
+	ReadHeaderTimeout    = 60 * time.Second
+	LeaseDuration        = 30 * time.Second
+	LeaseRenewalInterval = 15 * time.Second
+	LeaseGCInterval      = 15 * time.Second
+	LeaseNamespace       = "kube-system"
+)
 
 func NewProxyCommand(p *Proxy, o *options.ProxyRunOptions) *cobra.Command {
 	cmd := &cobra.Command{
@@ -143,23 +148,28 @@ func (p *Proxy) Run(o *options.ProxyRunOptions, stopCh <-chan struct{}) error {
 		defer frontendStop()
 	}
 
-	if o.IsLeaseCountingEnabled() {
-		leaseLabels, err := labels.ConvertSelectorToLabelsMap(o.LeaseLabelSelector)
-		if err != nil {
-			return fmt.Errorf("could not parse lease label selector (%q): %w", o.LeaseLabelSelector, err)
-		}
-		leaseController := leases.NewController(k8sClient, o.ServerID, int32(o.LeaseDuration.Seconds()), o.LeaseRenewalInterval, o.LeaseGCInterval, fmt.Sprintf("konnectivity-proxy-server-%v", o.ServerID), o.LeaseNamespace, leaseLabels)
-		klog.V(1).Infoln("Starting lease acquisition and garbage collection controller.")
-		leaseController.Run(ctx)
-	}
-
 	klog.V(1).Infoln("Starting agent server for tunnel connections.")
 	err = p.runAgentServer(o, p.server)
 	if err != nil {
 		return fmt.Errorf("failed to run the agent server: %v", err)
 	}
 	defer p.agentServer.Stop()
 
+	if o.EnableLeaseController {
+		leaseController := leases.NewController(
+			k8sClient,
+			o.ServerID,
+			int32(LeaseDuration.Seconds()),
+			LeaseRenewalInterval,
+			LeaseGCInterval,
+			fmt.Sprintf("konnectivity-proxy-server-%v", o.ServerID),
+			LeaseNamespace,
+			map[string]string{"k8s-app": "konnectivity-server"},
+		)
+		klog.V(1).Infoln("Starting lease acquisition and garbage collection controller.")
+		leaseController.Run(ctx)
+	}
+
 	klog.V(1).Infoln("Starting admin server for debug connections.")
 	err = p.runAdminServer(o, p.server)
 	if err != nil {

pkg/agent/clientset.go

@@ -39,8 +39,9 @@ type ClientSet struct {
 	agentID string // ID of this agent
 	address string // proxy server address. Assuming HA proxy server
 
-	serverCounter           *ServerLeaseCounter // counts number of proxy server leases
+	leaseCounter            *ServerLeaseCounter // counts number of proxy server leases
 	lastReceivedServerCount int                 // last server count received from a proxy server
+	lastServerCount         int                 // last server count value from either lease system or proxy server, former takes priority
 
 	// unless it is an HA server. Initialized when the ClientSet creates
 	// the first client. When syncForever is set, it will be the most recently seen.
@@ -165,7 +166,7 @@ func (cc *ClientSetConfig) NewAgentClientSet(drainCh, stopCh <-chan struct{}) *C
 		drainCh:                 drainCh,
 		xfrChannelSize:          cc.XfrChannelSize,
 		stopCh:                  stopCh,
-		serverCounter:           cc.ServerLeaseCounter,
+		leaseCounter:            cc.ServerLeaseCounter,
 	}
 }
 
@@ -189,10 +190,9 @@ func (cs *ClientSet) sync(ctx context.Context) {
 	backoff := cs.resetBackoff()
 	var duration time.Duration
 	for {
-		if err := cs.connectOnce(ctx); err != nil {
+		if serverCount, err := cs.connectOnce(ctx); err != nil {
 			if dse, ok := err.(*DuplicateServerError); ok {
 				clientsCount := cs.ClientsCount()
-				serverCount := cs.ServerCount(ctx)
 				klog.V(4).InfoS("duplicate server", "serverID", dse.ServerID, "serverCount", serverCount, "clientsCount", clientsCount)
 				if serverCount != 0 && clientsCount >= serverCount {
 					duration = backoff.Step()
@@ -219,33 +219,38 @@ func (cs *ClientSet) sync(ctx context.Context) {
 
 func (cs *ClientSet) ServerCount(ctx context.Context) int {
 	var serverCount int
-	if cs.serverCounter != nil {
-		serverCount = cs.serverCounter.Count(ctx)
+	if cs.leaseCounter != nil {
+		serverCount = cs.leaseCounter.Count(ctx)
 		if serverCount == 0 {
 			klog.Warningf("server lease counter could not find any leases")
 		}
 	} else {
 		serverCount = cs.lastReceivedServerCount
 	}
 
+	if serverCount != cs.lastServerCount {
+		klog.Warningf("change detected in proxy server count (was: %d, now: %d)", cs.lastServerCount, serverCount)
+		cs.lastServerCount = serverCount
+	}
+
 	metrics.Metrics.SetServerCount(serverCount)
 	return serverCount
 }
 
-func (cs *ClientSet) connectOnce(ctx context.Context) error {
+func (cs *ClientSet) connectOnce(ctx context.Context) (int, error) {
 	serverCount := cs.ServerCount(ctx)
 
 	if !cs.syncForever && serverCount != 0 && cs.ClientsCount() >= serverCount {
-		return nil
+		return serverCount, nil
 	}
 	c, receivedServerCount, err := cs.newAgentClient()
 	if err != nil {
-		return err
+		return serverCount, err
 	}
 	cs.lastReceivedServerCount = receivedServerCount
 	if err := cs.AddClient(c.serverID, c); err != nil {
 		c.Close()
-		return err
+		return serverCount, err
 	}
 	klog.V(2).InfoS("sync added client connecting to proxy server", "serverID", c.serverID)
 
@@ -255,7 +260,7 @@ func (cs *ClientSet) connectOnce(ctx context.Context) error {
 		"serverID", c.serverID,
 	)
 	go runpprof.Do(context.Background(), labels, func(context.Context) { c.Serve() })
-	return nil
+	return serverCount, nil
 }
 
 func (cs *ClientSet) Serve() {