From e441f11fa244e669e32e5b59b950e02293774f26 Mon Sep 17 00:00:00 2001 From: Evan Vetere Date: Fri, 2 Oct 2026 12:46:51 -0400 Subject: [PATCH] fix: Keep the leader lease renewing under API load Lease renewals shared each manager's API connection and client-side rate limiter with every other request. When the API server throttled or the connection slowed, renewals queued behind controller traffic, the renew deadline passed, and the replica lost its lease and restarted. This was seen in production on 2 Oct for milo, where lease GET and PUT timed out awaiting headers during throttling. Key changes: - Build a dedicated client config for leader election in both the VPC controller and the fabric identity controller: a copy of the base config with its own dialer, no shared rate limiter, and QPS 5, burst 10 - Release the lease when the manager stops, so a rollout or drain hands over at once instead of waiting out the lease. The manager is the only thing either process runs and each exits when it returns, so no controller can act after the lease is released - Add tests showing the base config is untouched, and that leader election ignores a saturated main limiter and opens its own connection --- cmd/fabric-identity-controller/main.go | 16 +- cmd/main.go | 16 +- internal/leaderelection/leaderelection.go | 42 +++++ .../leaderelection/leaderelection_test.go | 145 ++++++++++++++++++ 4 files changed, 207 insertions(+), 12 deletions(-) create mode 100644 internal/leaderelection/leaderelection.go create mode 100644 internal/leaderelection/leaderelection_test.go diff --git a/cmd/fabric-identity-controller/main.go b/cmd/fabric-identity-controller/main.go index 11828cb..5479564 100644 --- a/cmd/fabric-identity-controller/main.go +++ b/cmd/fabric-identity-controller/main.go @@ -40,6 +40,7 @@ import ( cloudv1alpha1 "go.datum.net/cloud/api/v1alpha1" "go.datum.net/cloud/internal/controller" "go.datum.net/cloud/internal/ipam" + "go.datum.net/cloud/internal/leaderelection" networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha" ) @@ -112,12 +113,15 @@ func main() { // leadership only as available as that plane: a hub hiccup would churn // leadership and restart the controller. An unreachable hub has to cost // this component its work. It does not have to cost it its identity. - mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{ - Scheme: scheme, - Metrics: metricsserver.Options{BindAddress: metricsAddr}, - HealthProbeBindAddress: probeAddr, - LeaderElection: enableLeaderElection, - LeaderElectionID: "fabric-identity-controller.cloud.datumapis.com", + restConfig := ctrl.GetConfigOrDie() + mgr, err := ctrl.NewManager(restConfig, ctrl.Options{ + Scheme: scheme, + Metrics: metricsserver.Options{BindAddress: metricsAddr}, + HealthProbeBindAddress: probeAddr, + LeaderElection: enableLeaderElection, + LeaderElectionID: "fabric-identity-controller.cloud.datumapis.com", + LeaderElectionConfig: leaderelection.RestConfig(restConfig), + LeaderElectionReleaseOnCancel: true, }) if err != nil { setupLog.Error(err, "unable to start manager") diff --git a/cmd/main.go b/cmd/main.go index f499299..ea42da3 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -39,6 +39,7 @@ import ( cloudv1alpha1 "go.datum.net/cloud/api/v1alpha1" "go.datum.net/cloud/internal/controller" "go.datum.net/cloud/internal/galactic" + "go.datum.net/cloud/internal/leaderelection" datumwebhook "go.datum.net/cloud/internal/webhook" computev1alpha "go.datum.net/compute/api/v1alpha" networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha" @@ -83,12 +84,15 @@ func main() { os.Exit(1) } - mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{ - Scheme: scheme, - Metrics: metricsserver.Options{BindAddress: metricsAddr}, - HealthProbeBindAddress: probeAddr, - LeaderElection: enableLeaderElection, - LeaderElectionID: "vpc-controller.cloud.datumapis.com", + restConfig := ctrl.GetConfigOrDie() + mgr, err := ctrl.NewManager(restConfig, ctrl.Options{ + Scheme: scheme, + Metrics: metricsserver.Options{BindAddress: metricsAddr}, + HealthProbeBindAddress: probeAddr, + LeaderElection: enableLeaderElection, + LeaderElectionID: "vpc-controller.cloud.datumapis.com", + LeaderElectionConfig: leaderelection.RestConfig(restConfig), + LeaderElectionReleaseOnCancel: true, WebhookServer: webhook.NewServer(webhook.Options{ Port: webhookPort, CertDir: webhookCertDir, diff --git a/internal/leaderelection/leaderelection.go b/internal/leaderelection/leaderelection.go new file mode 100644 index 0000000..f4a1163 --- /dev/null +++ b/internal/leaderelection/leaderelection.go @@ -0,0 +1,42 @@ +/* +Copyright © 2026 Datum Technology, Inc. All rights reserved. + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as +published by the Free Software Foundation, either version 3 of the +License, or (at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . +*/ + +package leaderelection + +import ( + "net" + "time" + + "k8s.io/client-go/rest" +) + +const ( + QPS = 5 + Burst = 10 +) + +func RestConfig(base *rest.Config) *rest.Config { + cfg := rest.CopyConfig(base) + cfg.RateLimiter = nil + cfg.QPS = QPS + cfg.Burst = Burst + cfg.Dial = (&net.Dialer{ + Timeout: 30 * time.Second, + KeepAlive: 30 * time.Second, + }).DialContext + return cfg +} diff --git a/internal/leaderelection/leaderelection_test.go b/internal/leaderelection/leaderelection_test.go new file mode 100644 index 0000000..b1ca341 --- /dev/null +++ b/internal/leaderelection/leaderelection_test.go @@ -0,0 +1,145 @@ +/* +Copyright © 2026 Datum Technology, Inc. All rights reserved. + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as +published by the Free Software Foundation, either version 3 of the +License, or (at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . +*/ + +package leaderelection + +import ( + "context" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "k8s.io/apimachinery/pkg/runtime/schema" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/rest" + "k8s.io/client-go/util/flowcontrol" +) + +type remoteAddrRecorder struct { + mu sync.Mutex + addrs []string +} + +func (r *remoteAddrRecorder) ServeHTTP(w http.ResponseWriter, req *http.Request) { + r.mu.Lock() + r.addrs = append(r.addrs, req.RemoteAddr) + r.mu.Unlock() + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{}`)) +} + +func (r *remoteAddrRecorder) last() string { + r.mu.Lock() + defer r.mu.Unlock() + return r.addrs[len(r.addrs)-1] +} + +func restClientFor(t *testing.T, cfg *rest.Config) *rest.RESTClient { + t.Helper() + cfg = rest.CopyConfig(cfg) + cfg.GroupVersion = &schema.GroupVersion{Group: "coordination.k8s.io", Version: "v1"} + cfg.APIPath = "/apis" + cfg.NegotiatedSerializer = clientgoscheme.Codecs.WithoutConversion() + c, err := rest.RESTClientFor(cfg) + if err != nil { + t.Fatalf("building REST client: %v", err) + } + return c +} + +func getLease(ctx context.Context, c *rest.RESTClient) error { + return c.Get().Namespace("default").Resource("leases").Name("lease").Do(ctx).Error() +} + +func TestLeaderElectionRestConfigDoesNotMutateBase(t *testing.T) { + limiter := flowcontrol.NewTokenBucketRateLimiter(1, 1) + base := &rest.Config{Host: "https://example.invalid", QPS: 50, Burst: 100, RateLimiter: limiter} + + cfg := RestConfig(base) + + if cfg.RateLimiter != nil { + t.Errorf("leader election config reuses the base rate limiter") + } + if cfg.QPS != QPS || cfg.Burst != Burst { + t.Errorf("got QPS %v Burst %d, want %v and %d", cfg.QPS, cfg.Burst, QPS, Burst) + } + if cfg.Dial == nil { + t.Errorf("leader election config has no dialer of its own") + } + if base.RateLimiter != limiter || base.QPS != 50 || base.Burst != 100 || base.Dial != nil { + t.Errorf("base config was mutated: %+v", base) + } +} + +func TestLeaderElectionIgnoresSaturatedMainLimiter(t *testing.T) { + srv := httptest.NewServer(&remoteAddrRecorder{}) + defer srv.Close() + + base := &rest.Config{ + Host: srv.URL, + RateLimiter: flowcontrol.NewTokenBucketRateLimiter(0.001, 1), + } + mainClient := restClientFor(t, base) + + if err := getLease(context.Background(), mainClient); err != nil { + t.Fatalf("first request on main client: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + if err := getLease(ctx, mainClient); err == nil { + t.Fatalf("expected the main client to be throttled") + } + + leaderClient := restClientFor(t, RestConfig(base)) + ctx, cancel = context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if err := getLease(ctx, leaderClient); err != nil { + t.Fatalf("leader election request was throttled by the main limiter: %v", err) + } +} + +func TestLeaderElectionUsesItsOwnConnection(t *testing.T) { + rec := &remoteAddrRecorder{} + srv := httptest.NewServer(rec) + defer srv.Close() + + base := &rest.Config{Host: srv.URL, QPS: -1} + mainClient := restClientFor(t, base) + ctx := context.Background() + + if err := getLease(ctx, mainClient); err != nil { + t.Fatalf("main client request: %v", err) + } + mainAddr := rec.last() + + if err := getLease(ctx, restClientFor(t, base)); err != nil { + t.Fatalf("second main client request: %v", err) + } + if rec.last() != mainAddr { + t.Fatalf("clients built from the same config did not share a connection, so this test cannot tell connections apart") + } + + if err := getLease(ctx, restClientFor(t, RestConfig(base))); err != nil { + t.Fatalf("leader election request: %v", err) + } + if rec.last() == mainAddr { + t.Errorf("leader election reused the main client's connection %s", mainAddr) + } +}