Add lease lease controller and e2e tests

Author: carreter

cmd/server/app/options/options.go

@@ -102,6 +102,13 @@ type ProxyRunOptions struct {
 	// see: https://pkg.go.dev/crypto/tls#Config, so in that case, this option won't have any effect.
 	CipherSuites   []string
 	XfrChannelSize int
+
+	// Lease controller configuration
+	LeaseDuration        time.Duration
+	LeaseRenewalInterval time.Duration
+	LeaseGCInterval      time.Duration
+	LeaseLabelSelector   string
+	LeaseNamespace       string
 }
 
 func (o *ProxyRunOptions) Flags() *pflag.FlagSet {
@@ -138,7 +145,11 @@ 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.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.")
 
@@ -293,6 +304,25 @@ func (o *ProxyRunOptions) Validate() error {
 		}
 	}
 
+	// Validate leasing parameters. All must be empty or have a value.
+	if o.LeaseLabelSelector != "" || o.LeaseDuration != 0 || o.LeaseGCInterval != 0 || o.LeaseRenewalInterval != 0 || o.LeaseNamespace != "" {
+		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 when leasing system is enabled")
+		}
+		if o.LeaseGCInterval == 0 {
+			return fmt.Errorf("LeaseGCInterval cannot be zero when leasing system is enabled")
+		}
+		if o.LeaseRenewalInterval == 0 {
+			return fmt.Errorf("LeaseRenewalInterval cannot be zero when leasing system is enabled")
+		}
+	}
+
 	// validate the proxy strategies
 	if len(o.ProxyStrategies) == 0 {
 		return fmt.Errorf("ProxyStrategies cannot be empty")
@@ -351,6 +381,11 @@ func NewProxyRunOptions() *ProxyRunOptions {
 		ProxyStrategies:           "default",
 		CipherSuites:              make([]string, 0),
 		XfrChannelSize:            10,
+		LeaseGCInterval:           0,
+		LeaseDuration:             0,
+		LeaseRenewalInterval:      0,
+		LeaseLabelSelector:        "",
+		LeaseNamespace:            "",
 	}
 	return &o
 }

cmd/server/app/server.go

@@ -39,13 +39,14 @@ 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"
-
 	"sigs.k8s.io/apiserver-network-proxy/cmd/server/app/options"
 	"sigs.k8s.io/apiserver-network-proxy/konnectivity-client/proto/client"
 	"sigs.k8s.io/apiserver-network-proxy/pkg/server"
+	"sigs.k8s.io/apiserver-network-proxy/pkg/server/leases"
 	"sigs.k8s.io/apiserver-network-proxy/pkg/util"
 	"sigs.k8s.io/apiserver-network-proxy/proto/agent"
 )
@@ -142,6 +143,16 @@ func (p *Proxy) Run(o *options.ProxyRunOptions, stopCh <-chan struct{}) error {
 		defer frontendStop()
 	}
 
+	if o.LeaseLabelSelector != "" || o.LeaseDuration != 0 || o.LeaseGCInterval != 0 || o.LeaseRenewalInterval != 0 {
+		leaseLabels, err := labels.ConvertSelectorToLabelsMap(o.LeaseLabelSelector)
+		if err != nil {
+			return fmt.Errorf("could not parse lease label selector: %w", 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 {

e2e/lease_count_test.go

@@ -188,3 +188,140 @@ func TestLeaseCount(t *testing.T) {
 	feature.Setup(lc.DeleteValidLease())
 	feature.Assess("agents correctly count 3 leases (3 valid, 3 expired)", assertAgentKnownServerCount(3, adminPort))
 }
+
+func TestSingleServer_SingleAgent_LeaseCount(t *testing.T) {
+	adminPort := 8093
+
+	serverDeploymentConfig := DeploymentConfig{
+		Replicas: 1,
+		Image:    *serverImage,
+		Args: []KeyValue{
+			{"log-file", "/var/log/konnectivity-server.log"},
+			{"logtostderr", "true"},
+			{"log-file-max-size", "0"},
+			{"uds-name", "/etc/kubernetes/konnectivity-server/konnectivity-server.socket"},
+			{Key: "delete-existing-uds-file"},
+			{"cluster-cert", "/etc/kubernetes/pki/apiserver.crt"},
+			{"cluster-key", "/etc/kubernetes/pki/apiserver.key"},
+			{"server-port", "8090"},
+			{"agent-port", "8091"},
+			{"health-port", "8092"},
+			{"admin-port", strconv.Itoa(adminPort)},
+			{"keepalive-time", "1h"},
+			{"mode", "grpc"},
+			{"agent-namespace", "kube-system"},
+			{"agent-service-account", "konnectivity-agent"},
+			{"kubeconfig", "/etc/kubernetes/admin.conf"},
+			{"authentication-audience", "system:konnectivity-server"},
+			{"lease-gc-interval", "10s"},
+			{"lease-renewal-interval", "5s"},
+			{"lease-label-selector", "k8s-app=konnectivity-server"},
+			{"lease-namespace", "kube-system"},
+		},
+	}
+	serverDeployment, _, err := renderTemplate("server/deployment.yaml", serverDeploymentConfig)
+	if err != nil {
+		t.Fatalf("could not render server deployment: %v", err)
+	}
+
+	agentDeploymentConfig := DeploymentConfig{
+		Replicas: 1,
+		Image:    *agentImage,
+		Args: []KeyValue{
+			{"logtostderr", "true"},
+			{"ca-cert", "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"},
+			{"proxy-server-host", "konnectivity-server.kube-system.svc.cluster.local"},
+			{"proxy-server-port", "8091"},
+			{"sync-interval", "1s"},
+			{"sync-interval-cap", "10s"},
+			{Key: "sync-forever"},
+			{"probe-interval", "1s"},
+			{"service-account-token-path", "/var/run/secrets/tokens/konnectivity-agent-token"},
+			{"server-lease-selector", "k8s-app=konnectivity-server"},
+		},
+	}
+	agentDeployment, _, err := renderTemplate("agent/deployment.yaml", agentDeploymentConfig)
+	if err != nil {
+		t.Fatalf("could not render agent deployment: %v", err)
+	}
+
+	feature := features.New("konnectivity server and agent deployment with single replica for each")
+	feature.Setup(deployAndWaitForDeployment(serverDeployment))
+	feature.Setup(deployAndWaitForDeployment(agentDeployment))
+	feature.Assess("konnectivity server has a connected client", assertServersAreConnected(1, adminPort))
+	feature.Assess("konnectivity agent is connected to a server", assertAgentsAreConnected(1, adminPort))
+}
+
+func TestMultiServer_MultiAgent_LeaseCount(t *testing.T) {
+	adminPort := 8093
+	initialAgentReplicas := 2
+	initialServerReplicas := 2
+
+	serverDeploymentConfig := DeploymentConfig{
+		Replicas: initialServerReplicas,
+		Image:    *serverImage,
+		Args: []KeyValue{
+			{"log-file", "/var/log/konnectivity-server.log"},
+			{"logtostderr", "true"},
+			{"log-file-max-size", "0"},
+			{"uds-name", "/etc/kubernetes/konnectivity-server/konnectivity-server.socket"},
+			{Key: "delete-existing-uds-file"},
+			{"cluster-cert", "/etc/kubernetes/pki/apiserver.crt"},
+			{"cluster-key", "/etc/kubernetes/pki/apiserver.key"},
+			{"server-port", "8090"},
+			{"agent-port", "8091"},
+			{"health-port", "8092"},
+			{"admin-port", strconv.Itoa(adminPort)},
+			{"keepalive-time", "1h"},
+			{"mode", *connectionMode},
+			{"agent-namespace", "kube-system"},
+			{"agent-service-account", "konnectivity-agent"},
+			{"kubeconfig", "/etc/kubernetes/admin.conf"},
+			{"authentication-audience", "system:konnectivity-server"},
+			{"server-count", strconv.Itoa(initialServerReplicas)},
+			{"lease-gc-interval", "10s"},
+			{"lease-renewal-interval", "5s"},
+			{"lease-label-selector", "k8s-app=konnectivity-server"},
+			{"lease-namespace", "kube-system"},
+		},
+	}
+	serverDeployment, _, err := renderTemplate("server/deployment.yaml", serverDeploymentConfig)
+	if err != nil {
+		t.Fatalf("could not render server deployment: %v", err)
+	}
+
+	agentDeploymentConfig := DeploymentConfig{
+		Replicas: initialAgentReplicas,
+		Image:    *agentImage,
+		Args: []KeyValue{
+			{"logtostderr", "true"},
+			{"ca-cert", "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"},
+			{"proxy-server-host", "konnectivity-server.kube-system.svc.cluster.local"},
+			{"proxy-server-port", "8091"},
+			{"sync-interval", "1s"},
+			{"sync-interval-cap", "10s"},
+			{Key: "sync-forever"},
+			{"probe-interval", "1s"},
+			{"service-account-token-path", "/var/run/secrets/tokens/konnectivity-agent-token"},
+			{"server-lease-selector", "k8s-app=konnectivity-server"},
+		},
+	}
+	agentDeployment, _, err := renderTemplate("agent/deployment.yaml", agentDeploymentConfig)
+	if err != nil {
+		t.Fatalf("could not render agent deployment: %v", err)
+	}
+
+	feature := features.New("konnectivity server and agent deployment with single replica for each")
+	feature.Setup(deployAndWaitForDeployment(serverDeployment))
+	feature.Setup(deployAndWaitForDeployment(agentDeployment))
+	feature.Assess("all servers connected to all clients", assertServersAreConnected(initialAgentReplicas, adminPort))
+	feature.Assess("all agents connected to all servers", assertAgentsAreConnected(initialServerReplicas, adminPort))
+	feature.Setup(scaleDeployment(serverDeployment, 4))
+	feature.Setup(scaleDeployment(agentDeployment, 4))
+	feature.Assess("all servers connected to all clients after scale up", assertServersAreConnected(4, adminPort))
+	feature.Assess("all agents connected to all servers after scale up", assertAgentsAreConnected(4, adminPort))
+	feature.Setup(scaleDeployment(serverDeployment, 3))
+	feature.Setup(scaleDeployment(agentDeployment, 3))
+	feature.Assess("all servers connected to all clients after scale down", assertServersAreConnected(3, adminPort))
+	feature.Assess("all agents connected to all servers after scale down", assertAgentsAreConnected(3, adminPort))
+}

e2e/main_test.go

@@ -12,6 +12,7 @@ import (
 	"text/template"
 	"time"
 
+	appsv1api "k8s.io/api/apps/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/kubernetes/scheme"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -156,14 +157,43 @@ func renderAndApplyManifests(ctx context.Context, cfg *envconf.Config) (context.
 	return ctx, nil
 }
 
-func deployAndWaitForDeployment(deployment client.Object) func(context.Context, *testing.T, *envconf.Config) context.Context {
+func deployAndWaitForDeployment(obj client.Object) func(context.Context, *testing.T, *envconf.Config) context.Context {
 	return func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
 		client := cfg.Client()
-		err := client.Resources().Create(ctx, deployment)
+		err := client.Resources().Create(ctx, obj)
 		if err != nil {
 			t.Fatalf("could not create Deployment: %v", err)
 		}
 
+		err = wait.For(
+			conditions.New(client.Resources()).DeploymentAvailable(obj.GetName(), obj.GetNamespace()),
+			wait.WithTimeout(1*time.Minute),
+			wait.WithInterval(10*time.Second),
+		)
+		if err != nil {
+			t.Fatalf("waiting for Deployment failed: %v", err)
+		}
+
+		return ctx
+	}
+}
+
+func scaleDeployment(obj client.Object, replicas int) func(context.Context, *testing.T, *envconf.Config) context.Context {
+	return func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
+		deployment, ok := obj.(*appsv1api.Deployment)
+		if !ok {
+			t.Fatalf("provided object is not a deployment")
+		}
+
+		newReplicas := int32(replicas)
+		deployment.Spec.Replicas = &newReplicas
+
+		client := cfg.Client()
+		err := client.Resources().Update(ctx, deployment)
+		if err != nil {
+			t.Fatalf("could not update Deployment replicas: %v", err)
+		}
+
 		err = wait.For(
 			conditions.New(client.Resources()).DeploymentAvailable(deployment.GetName(), deployment.GetNamespace()),
 			wait.WithTimeout(1*time.Minute),

go.mod

@@ -16,12 +16,13 @@ require (
 	golang.org/x/net v0.26.0
 	google.golang.org/grpc v1.64.0
 	google.golang.org/protobuf v1.34.2
-	k8s.io/api v0.30.2
-	k8s.io/apimachinery v0.30.2
-	k8s.io/client-go v0.30.2
+	k8s.io/api v0.30.3
+	k8s.io/apimachinery v0.30.3
+	k8s.io/client-go v0.30.3
 	k8s.io/component-base v0.30.2
+	k8s.io/component-helpers v0.30.3
 	k8s.io/klog/v2 v2.130.0
-	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.29.0
+	sigs.k8s.io/apiserver-network-proxy/konnectivity-client v0.0.0
 	sigs.k8s.io/controller-runtime v0.18.2
 	sigs.k8s.io/e2e-framework v0.4.0
 )
@@ -41,7 +42,6 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/gnostic-models v0.6.8 // indirect
-	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/imdario/mergo v0.3.15 // indirect

go.sum

@@ -180,16 +180,18 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-k8s.io/api v0.30.2 h1:+ZhRj+28QT4UOH+BKznu4CBgPWgkXO7XAvMcMl0qKvI=
-k8s.io/api v0.30.2/go.mod h1:ULg5g9JvOev2dG0u2hig4Z7tQ2hHIuS+m8MNZ+X6EmI=
+k8s.io/api v0.30.3 h1:ImHwK9DCsPA9uoU3rVh4QHAHHK5dTSv1nxJUapx8hoQ=
+k8s.io/api v0.30.3/go.mod h1:GPc8jlzoe5JG3pb0KJCSLX5oAFIW3/qNJITlDj8BH04=
 k8s.io/apiextensions-apiserver v0.30.0 h1:jcZFKMqnICJfRxTgnC4E+Hpcq8UEhT8B2lhBcQ+6uAs=
 k8s.io/apiextensions-apiserver v0.30.0/go.mod h1:N9ogQFGcrbWqAY9p2mUAL5mGxsLqwgtUce127VtRX5Y=
-k8s.io/apimachinery v0.30.2 h1:fEMcnBj6qkzzPGSVsAZtQThU62SmQ4ZymlXRC5yFSCg=
-k8s.io/apimachinery v0.30.2/go.mod h1:iexa2somDaxdnj7bha06bhb43Zpa6eWH8N8dbqVjTUc=
-k8s.io/client-go v0.30.2 h1:sBIVJdojUNPDU/jObC+18tXWcTJVcwyqS9diGdWHk50=
-k8s.io/client-go v0.30.2/go.mod h1:JglKSWULm9xlJLx4KCkfLLQ7XwtlbflV6uFFSHTMgVs=
+k8s.io/apimachinery v0.30.3 h1:q1laaWCmrszyQuSQCfNB8cFgCuDAoPszKY4ucAjDwHc=
+k8s.io/apimachinery v0.30.3/go.mod h1:iexa2somDaxdnj7bha06bhb43Zpa6eWH8N8dbqVjTUc=
+k8s.io/client-go v0.30.3 h1:bHrJu3xQZNXIi8/MoxYtZBBWQQXwy16zqJwloXXfD3k=
+k8s.io/client-go v0.30.3/go.mod h1:8d4pf8vYu665/kUbsxWAQ/JDBNWqfFeZnvFiVdmx89U=
 k8s.io/component-base v0.30.2 h1:pqGBczYoW1sno8q9ObExUqrYSKhtE5rW3y6gX88GZII=
 k8s.io/component-base v0.30.2/go.mod h1:yQLkQDrkK8J6NtP+MGJOws+/PPeEXNpwFixsUI7h/OE=
+k8s.io/component-helpers v0.30.3 h1:KPc8l0eGx9Wg2OcKc58k9ozNcVcOInAi3NGiuS2xJ/c=
+k8s.io/component-helpers v0.30.3/go.mod h1:VOQ7g3q+YbKWwKeACG2BwPv4ftaN8jXYJ5U3xpzuYAE=
 k8s.io/klog/v2 v2.130.0 h1:5nB3+3HpqKqXJIXNtJdtxcDCfaa9KL8StJgMzGJkUkM=
 k8s.io/klog/v2 v2.130.0/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE=
 k8s.io/kube-openapi v0.0.0-20240228011516-70dd3763d340 h1:BZqlfIlq5YbRMFko6/PM7FjZpUb45WallggurYhKGag=