From ca5f7ec2c72c4401b2dc26abd8db6ebc9ad8ff57 Mon Sep 17 00:00:00 2001 From: Viktor Voytovych Date: Thu, 15 Oct 2026 20:05:10 +0300 Subject: [PATCH 1/3] fix: Serve DNSRecordSet webhooks on every replica The replicator handed the Milo project provider to the multicluster manager as it is, so the provider's Start ran on the elected leader only. A replica that is not the leader engaged no project, and its mutating webhook read each record set's DNSZone from the deployment cluster, found none, and admitted the record set without its display annotations. Hide Start from that wiring and start the provider on every replica, as network-services-operator does. Gate readiness on the webhook server serving and on the projects found at start being registered, so a new replica takes no admission request it cannot answer. --- cmd/main.go | 168 +++++++++++++------- cmd/main_test.go | 17 +-- cmd/nonleader_test.go | 347 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 464 insertions(+), 68 deletions(-) create mode 100644 cmd/nonleader_test.go diff --git a/cmd/main.go b/cmd/main.go index 3658110a..50afe5ec 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -8,6 +8,7 @@ import ( "errors" "flag" "fmt" + "net/http" "os" "time" @@ -22,6 +23,7 @@ import ( "k8s.io/apimachinery/pkg/runtime/serializer" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientgoscheme "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/rest" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/cluster" @@ -44,6 +46,8 @@ import ( // +kubebuilder:scaffold:imports ) +const leaderElectionID = "1813fe7c.datum.cloud" + var ( scheme = runtime.NewScheme() setupLog = ctrl.Log.WithName("setup") @@ -234,7 +238,7 @@ func main() { LeaseDuration: &leaderElectionLeaseDuration, RenewDeadline: &leaderElectionRenewDeadline, RetryPeriod: &leaderElectionRetryPeriod, - LeaderElectionID: "1813fe7c.datum.cloud", + LeaderElectionID: leaderElectionID, }) if err != nil { setupLog.Error(err, "unable to start manager") @@ -314,23 +318,16 @@ func main() { os.Exit(1) } - // Initialize cluster discovery provider (single or milo) - runnables, provider, err := initializeClusterDiscovery(serverConfig, deploymentCluster, scheme) - if err != nil { - setupLog.Error(err, "unable to initialize cluster discovery") - os.Exit(1) - } - // Multicluster manager (serves mutating webhook for display annotations) webhookServer := webhook.NewServer(webhookServerOptions) webhookServer = dnswebhook.NewClusterAwareWebhookServer(webhookServer) - mcmgr, err := mcmanager.New(cfg, provider, ctrl.Options{ + runnables, mcmgr, err := newReplicatorManager(serverConfig, cfg, deploymentCluster, ctrl.Options{ Scheme: scheme, Metrics: metricsServerOptions, WebhookServer: webhookServer, HealthProbeBindAddress: probeAddr, LeaderElection: enableLeaderElection, - LeaderElectionID: "1813fe7c.datum.cloud", + LeaderElectionID: leaderElectionID, }) if err != nil { setupLog.Error(err, "unable to start multicluster manager") @@ -356,17 +353,8 @@ func main() { os.Exit(1) } - if err := dnswebhook.SetupDNSRecordSetWebhook(mcmgr); err != nil { - setupLog.Error(err, "unable to create webhook", "webhook", "DNSRecordSet") - os.Exit(1) - } - - if err := mcmgr.AddHealthzCheck("healthz", healthz.Ping); err != nil { - setupLog.Error(err, "unable to set up health check") - os.Exit(1) - } - if err := mcmgr.AddReadyzCheck("readyz", healthz.Ping); err != nil { - setupLog.Error(err, "unable to set up ready check") + if err := setupReplicatorWebhook(mcmgr); err != nil { + setupLog.Error(err, "unable to set up the DNSRecordSet webhook and its checks") os.Exit(1) } @@ -383,6 +371,41 @@ func main() { } } +// newReplicatorManager builds the replicator's cluster discovery and the +// multicluster manager that discovers clusters through it. +func newReplicatorManager( + serverConfig config.DNSOperator, + cfg *rest.Config, + deploymentCluster cluster.Cluster, + opts ctrl.Options, +) ([]manager.Runnable, mcmanager.Manager, error) { + runnables, provider, setup, err := initializeClusterDiscovery(serverConfig, deploymentCluster, opts.Scheme) + if err != nil { + return nil, nil, fmt.Errorf("unable to initialize cluster discovery: %w", err) + } + mcmgr, err := mcmanager.New(cfg, provider, opts) + if err != nil { + return nil, nil, err + } + if err := setup(mcmgr); err != nil { + return nil, nil, err + } + return runnables, mcmgr, nil +} + +// setupReplicatorWebhook registers the DNSRecordSet webhooks, the +// replicator's health check, and a readiness check on its webhook server. +func setupReplicatorWebhook(mcmgr mcmanager.Manager) error { + if err := dnswebhook.SetupDNSRecordSetWebhook(mcmgr); err != nil { + return fmt.Errorf("unable to create the DNSRecordSet webhook: %w", err) + } + if err := mcmgr.AddHealthzCheck("healthz", healthz.Ping); err != nil { + return fmt.Errorf("unable to set up health check: %w", err) + } + // The mutating webhook fails closed. + return mcmgr.AddReadyzCheck("webhook", mcmgr.GetWebhookServer().StartedChecker()) +} + // runReplicator starts the pre-created runnables, the downstream cluster, and // the multicluster manager, and blocks until ctx is cancelled or one of them // returns an error. @@ -418,8 +441,9 @@ func initializeClusterDiscovery( serverConfig config.DNSOperator, deploymentCluster cluster.Cluster, scheme *runtime.Scheme, -) (runnables []manager.Runnable, provider multicluster.Provider, err error) { +) (runnables []manager.Runnable, provider multicluster.Provider, setup func(mcmanager.Manager) error, err error) { runnables = append(runnables, deploymentCluster) + setup = func(mcmanager.Manager) error { return nil } switch serverConfig.Discovery.Mode { case multiclusterproviders.ProviderSingle: // The single provider implements multicluster.ProviderRunnable and @@ -428,40 +452,20 @@ func initializeClusterDiscovery( provider = mcsingle.New("single", deploymentCluster) case multiclusterproviders.ProviderMilo: - discoveryRestConfig, err := serverConfig.Discovery.DiscoveryRestConfig() + discoveryManager, miloProvider, err := newMiloProvider(serverConfig.Discovery, scheme) if err != nil { - return nil, nil, fmt.Errorf("unable to get discovery rest config: %w", err) + return nil, nil, nil, err } - - projectRestConfig, err := serverConfig.Discovery.ProjectRestConfig() - if err != nil { - return nil, nil, fmt.Errorf("unable to get project rest config: %w", err) + // Handed to the multicluster manager as it is, the provider's Start + // runs on the elected leader only, and the webhooks of every other + // replica could reach no project. + provider = milomulticluster.WithoutAutoStart(miloProvider) + setup = func(mgr mcmanager.Manager) error { + if err := milomulticluster.EngageAlways(mgr, miloProvider); err != nil { + return fmt.Errorf("unable to start the project provider on every replica: %w", err) + } + return mgr.AddReadyzCheck("projects", projectsRegistered(miloProvider)) } - - discoveryManager, err := manager.New(discoveryRestConfig, manager.Options{ - Client: client.Options{ - Cache: &client.CacheOptions{ - Unstructured: true, - }, - }, - }) - if err != nil { - return nil, nil, fmt.Errorf("unable to set up overall controller manager: %w", err) - } - - provider, err = milomulticluster.New(discoveryManager, milomulticluster.Options{ - ClusterOptions: []cluster.Option{ - func(o *cluster.Options) { - o.Scheme = scheme - }, - }, - InternalServiceDiscovery: serverConfig.Discovery.InternalServiceDiscovery, - ProjectRestConfig: projectRestConfig, - }) - if err != nil { - return nil, nil, fmt.Errorf("unable to create datum project provider: %w", err) - } - runnables = append(runnables, discoveryManager) // case providers.ProviderKind: @@ -474,13 +478,67 @@ func initializeClusterDiscovery( // }) default: - return nil, nil, fmt.Errorf( + return nil, nil, nil, fmt.Errorf( "unsupported cluster discovery mode %s", serverConfig.Discovery.Mode, ) } - return runnables, provider, nil + return runnables, provider, setup, nil +} + +// newMiloProvider builds the manager that discovers projects and the Milo +// provider that runs its discovery on it. +func newMiloProvider( + discovery config.DiscoveryConfig, + scheme *runtime.Scheme, +) (manager.Manager, *milomulticluster.Provider, error) { + discoveryRestConfig, err := discovery.DiscoveryRestConfig() + if err != nil { + return nil, nil, fmt.Errorf("unable to get discovery rest config: %w", err) + } + + projectRestConfig, err := discovery.ProjectRestConfig() + if err != nil { + return nil, nil, fmt.Errorf("unable to get project rest config: %w", err) + } + + discoveryManager, err := manager.New(discoveryRestConfig, manager.Options{ + Client: client.Options{ + Cache: &client.CacheOptions{ + Unstructured: true, + }, + }, + }) + if err != nil { + return nil, nil, fmt.Errorf("unable to set up overall controller manager: %w", err) + } + + provider, err := milomulticluster.New(discoveryManager, milomulticluster.Options{ + ClusterOptions: []cluster.Option{ + func(o *cluster.Options) { + o.Scheme = scheme + }, + }, + InternalServiceDiscovery: discovery.InternalServiceDiscovery, + ProjectRestConfig: projectRestConfig, + }) + if err != nil { + return nil, nil, fmt.Errorf("unable to create datum project provider: %w", err) + } + return discoveryManager, provider, nil +} + +// projectsRegistered reports ready once the provider has registered every +// project it found at start, so a new replica does not answer admission +// requests for a project it cannot reach yet. +func projectsRegistered(provider interface{ HasSynced() bool }) healthz.Checker { + return func(*http.Request) error { + if !provider.HasSynced() { + return errors.New("projects found at start are not all registered yet") + } + return nil + } } func ignoreCanceled(err error) error { diff --git a/cmd/main_test.go b/cmd/main_test.go index 1b9c2107..ea19c3a4 100644 --- a/cmd/main_test.go +++ b/cmd/main_test.go @@ -18,7 +18,6 @@ import ( metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" "sigs.k8s.io/controller-runtime/pkg/reconcile" mcbuilder "sigs.k8s.io/multicluster-runtime/pkg/builder" - mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager" mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile" multiclusterproviders "go.miloapis.com/milo/pkg/multicluster-runtime" @@ -27,8 +26,8 @@ import ( ) // TestReplicatorProviderLifecycle exercises the replicator startup path from -// cmd/main.go against envtest: initializeClusterDiscovery builds the single -// cluster discovery provider, and runReplicator starts everything. +// cmd/main.go against envtest: newReplicatorManager builds the single cluster +// discovery provider and the manager, and runReplicator starts everything. // // It asserts that the multicluster manager engages the cluster on its own - // there is no explicit provider start call anymore - by requiring that a @@ -66,21 +65,13 @@ func TestReplicatorProviderLifecycle(t *testing.T) { serverConfig := config.DNSOperator{ Discovery: config.DiscoveryConfig{Mode: multiclusterproviders.ProviderSingle}, } - runnables, provider, err := initializeClusterDiscovery(serverConfig, deploymentCluster, scheme) - if err != nil { - t.Fatalf("initializeClusterDiscovery: %v", err) - } - if provider == nil { - t.Fatal("initializeClusterDiscovery returned a nil provider") - } - - mcmgr, err := mcmanager.New(cfg, provider, ctrl.Options{ + runnables, mcmgr, err := newReplicatorManager(serverConfig, cfg, deploymentCluster, ctrl.Options{ Scheme: scheme, Metrics: metricsserver.Options{BindAddress: "0"}, HealthProbeBindAddress: "0", }) if err != nil { - t.Fatalf("creating multicluster manager: %v", err) + t.Fatalf("newReplicatorManager: %v", err) } // A controller registered through the multicluster builder only starts diff --git a/cmd/nonleader_test.go b/cmd/nonleader_test.go new file mode 100644 index 00000000..2db9a47e --- /dev/null +++ b/cmd/nonleader_test.go @@ -0,0 +1,347 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package main + +import ( + "context" + "io" + "net" + "net/http" + "net/http/httptest" + "net/http/httputil" + "os/exec" + "path/filepath" + "regexp" + "strings" + "testing" + "time" + + coordinationv1 "k8s.io/api/coordination/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" + clientcmdapi "k8s.io/client-go/tools/clientcmd/api" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/cluster" + "sigs.k8s.io/controller-runtime/pkg/envtest" + metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" + "sigs.k8s.io/controller-runtime/pkg/webhook" + mccontext "sigs.k8s.io/multicluster-runtime/pkg/context" + mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager" + + multiclusterproviders "go.miloapis.com/milo/pkg/multicluster-runtime" + + dnsv1alpha1 "go.miloapis.com/dns-operator/api/v1alpha1" + "go.miloapis.com/dns-operator/internal/config" + "go.miloapis.com/dns-operator/internal/display" + dnswebhook "go.miloapis.com/dns-operator/internal/webhook" +) + +const ( + testProject = "p1" + testZone = "zone" + testNamespace = "default" + testDomain = "example.com" +) + +// projectControlPlanePath is the path Milo serves each project's control plane +// under; the provider appends it to the project kubeconfig's host. +var projectControlPlanePath = regexp.MustCompile(`^/apis/resourcemanager\.miloapis\.com/v1alpha1/projects/[^/]+/control-plane`) + +// TestNonLeaderReplica runs the replicator as a replica that never wins the +// lease, with the Milo provider production uses and a webhook server that +// never serves. One replica serves both subtests: the provider registers a +// process-wide controller name, so a second provider in this process would +// not start. +func TestNonLeaderReplica(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + probe := freeAddress(t) + mcmgr := startNonLeaderReplica(ctx, t, ctrl.Options{ + HealthProbeBindAddress: probe, + WebhookServer: neverServing{webhook.NewServer(webhook.Options{})}, + }, setupReplicatorWebhook) + + // The webhook must read the record set's DNSZone from the project's + // control plane and stamp the display annotations. + t.Run("its webhook reads the project's zone", func(t *testing.T) { + // The same construction SetupDNSRecordSetWebhook registers. + mutator := &dnswebhook.DNSRecordSetMutator{Manager: mcmgr, Client: mcmgr.GetLocalManager().GetClient()} + if !annotatedWithin(ctx, t, mutator, 60*time.Second) { + t.Fatal("a replica that is not the leader admitted a DNSRecordSet without its display annotations: " + + "it never engaged the project, so its webhook read the DNSZone from the deployment cluster") + } + }) + + // The webhook fails closed, so a replica must not take traffic before its + // webhook server serves, nor before it has registered the projects it + // found at start. So /readyz must fail on the webhook while the projects + // check passes. + t.Run("it is ready only when its webhook serves", func(t *testing.T) { + status, body := readyzOnceProjectsRegistered(t, probe, 60*time.Second) + if status == http.StatusOK || !strings.Contains(body, "[-]webhook failed") { + t.Errorf("readyz = %d, want a failure with the webhook check failing, while the webhook server does not serve:\n%s", + status, body) + } + if !strings.Contains(body, "[+]projects ok") { + t.Errorf("readyz never reported the projects registered:\n%s", body) + } + }) + requireNotLeader(t, mcmgr) +} + +// readyzOnceProjectsRegistered reads the replica's verbose readiness until it +// reports the projects registered, or until timeout, and returns the last read. +func readyzOnceProjectsRegistered(t *testing.T, address string, timeout time.Duration) (int, string) { + t.Helper() + status, body := readyz(t, address) + for deadline := time.Now().Add(timeout); time.Now().Before(deadline); time.Sleep(time.Second) { + if status, body = readyz(t, address); strings.Contains(body, "[+]projects ok") { + break + } + } + return status, body +} + +// startNonLeaderReplica builds the replicator the way main does, with the Milo +// provider production uses, as a replica that cannot win the lease, and +// starts it. Two API servers stand in for production's two: the deployment +// cluster, which holds the Projects, the Lease and downstream DNSZones, and a +// project control plane, which holds the project's DNSZone. A proxy in front +// of the second strips the project path, as Milo does. So a webhook that never +// engaged the project falls back to the deployment cluster, finds no zone +// there, and admits the record set without annotations. +func startNonLeaderReplica( + ctx context.Context, t *testing.T, opts ctrl.Options, setups ...func(mcmanager.Manager) error, +) mcmanager.Manager { + t.Helper() + dnsCRDs := filepath.Join("..", "config", "crd", "bases") + deploymentCfg := startEnvtest(t, dnsCRDs, miloProjectCRDs(t)) + projectCfg := startEnvtest(t, dnsCRDs) + + deploymentClient := newClient(t, deploymentCfg) + holdLease(ctx, t, deploymentClient) + createReadyProject(ctx, t, deploymentClient) + createZone(ctx, t, newClient(t, projectCfg)) + + serverConfig := config.DNSOperator{Discovery: config.DiscoveryConfig{ + Mode: multiclusterproviders.ProviderMilo, + DiscoveryKubeconfigPath: writeKubeconfig(t, deploymentCfg), + ProjectKubeconfigPath: writeKubeconfig(t, &rest.Config{Host: projectControlPlaneProxy(t, projectCfg)}), + }} + opts.Scheme = scheme + opts.Metrics = metricsserver.Options{BindAddress: "0"} + opts.LeaderElection, opts.LeaderElectionID, opts.LeaderElectionNamespace = true, leaderElectionID, testNamespace + runnables, mcmgr, err := newReplicatorManager(serverConfig, deploymentCfg, newCluster(t, deploymentCfg), opts) + if err != nil { + t.Fatalf("newReplicatorManager: %v", err) + } + for _, setup := range setups { + if err := setup(mcmgr); err != nil { + t.Fatalf("setting up the replica: %v", err) + } + } + go func() { _ = runReplicator(ctx, runnables, newCluster(t, deploymentCfg), mcmgr) }() + return mcmgr +} + +// requireNotLeader fails the test if the replica won the lease another holder +// keeps, since then it did not run a replica that is not the leader. +func requireNotLeader(t *testing.T, mcmgr mcmanager.Manager) { + t.Helper() + select { + case <-mcmgr.Elected(): + t.Fatal("the replica won the lease another holder keeps, so this test did not run a non-leader") + default: + } +} + +// neverServing is a webhook server that registers handlers and never serves. +type neverServing struct{ webhook.Server } + +func (neverServing) Start(ctx context.Context) error { + <-ctx.Done() + return nil +} + +// readyz reads the replica's verbose readiness from its health probe address. +func readyz(t *testing.T, address string) (int, string) { + t.Helper() + resp, err := http.Get("http://" + address + "/readyz?verbose") + if err != nil { + return 0, err.Error() + } + defer func() { _ = resp.Body.Close() }() + body, _ := io.ReadAll(resp.Body) + return resp.StatusCode, string(body) +} + +// freeAddress returns a loopback address with a port nothing listens on. +func freeAddress(t *testing.T) string { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("finding a free port: %v", err) + } + defer func() { _ = l.Close() }() + return l.Addr().String() +} + +// annotatedWithin admits a DNSRecordSet in the test project until the webhook +// stamps its display annotations, and reports whether it did before timeout. +func annotatedWithin(ctx context.Context, t *testing.T, m *dnswebhook.DNSRecordSetMutator, timeout time.Duration) bool { + t.Helper() + projectCtx := mccontext.WithCluster(ctx, testProject) + for deadline := time.Now().Add(timeout); time.Now().Before(deadline); time.Sleep(time.Second) { + rs := &dnsv1alpha1.DNSRecordSet{ObjectMeta: metav1.ObjectMeta{Name: "www", Namespace: testNamespace}} + rs.Spec.DNSZoneRef.Name = testZone + rs.Spec.Records = []dnsv1alpha1.RecordEntry{{Name: "www"}} + if err := m.Default(projectCtx, rs); err != nil { + t.Logf("webhook: %v", err) + continue + } + if rs.Annotations[display.AnnotationDisplayName] == "www."+testDomain { + return true + } + } + return false +} + +// startEnvtest starts an API server with the CRDs in dirs and stops it when +// the test ends. +func startEnvtest(t *testing.T, dirs ...string) *rest.Config { + t.Helper() + env := &envtest.Environment{CRDDirectoryPaths: dirs, ErrorIfCRDPathMissing: true} + if dir := firstEnvTestBinaryDir(t); dir != "" { + env.BinaryAssetsDirectory = dir + } + cfg, err := env.Start() + if err != nil { + t.Fatalf("starting envtest: %v (run 'make setup-envtest' or set KUBEBUILDER_ASSETS)", err) + } + t.Cleanup(func() { + if err := env.Stop(); err != nil { + t.Errorf("stopping envtest: %v", err) + } + }) + return cfg +} + +// miloProjectCRDs returns the directory of Milo's resourcemanager CRDs, the +// Project among them, in the module this repository pins. +func miloProjectCRDs(t *testing.T) string { + t.Helper() + out, err := exec.Command("go", "list", "-m", "-f", "{{.Dir}}", "go.miloapis.com/milo").Output() + if err != nil { + t.Fatalf("locating the milo module: %v", err) + } + return filepath.Join(strings.TrimSpace(string(out)), "config", "crd", "bases", "resourcemanager") +} + +// projectControlPlaneProxy serves the project control plane at Milo's project +// path, forwarding to cfg's API server, and returns its URL. +func projectControlPlaneProxy(t *testing.T, cfg *rest.Config) string { + t.Helper() + transport, err := rest.TransportFor(cfg) + if err != nil { + t.Fatalf("project transport: %v", err) + } + target, err := http.NewRequest(http.MethodGet, cfg.Host, nil) + if err != nil { + t.Fatalf("project host: %v", err) + } + proxy := httputil.NewSingleHostReverseProxy(target.URL) + proxy.Transport = transport + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + r.URL.Path = projectControlPlanePath.ReplaceAllString(r.URL.Path, "") + r.URL.RawPath = "" + proxy.ServeHTTP(w, r) + })) + t.Cleanup(server.Close) + return server.URL +} + +// writeKubeconfig writes cfg as a kubeconfig file and returns its path. +func writeKubeconfig(t *testing.T, cfg *rest.Config) string { + t.Helper() + kubeconfig := clientcmdapi.NewConfig() + kubeconfig.Clusters["c"] = &clientcmdapi.Cluster{Server: cfg.Host, CertificateAuthorityData: cfg.CAData} + kubeconfig.AuthInfos["u"] = &clientcmdapi.AuthInfo{ + ClientCertificateData: cfg.CertData, ClientKeyData: cfg.KeyData, Token: cfg.BearerToken, + } + kubeconfig.Contexts["c"] = &clientcmdapi.Context{Cluster: "c", AuthInfo: "u"} + kubeconfig.CurrentContext = "c" + path := filepath.Join(t.TempDir(), "kubeconfig") + if err := clientcmd.WriteToFile(*kubeconfig, path); err != nil { + t.Fatalf("writing kubeconfig: %v", err) + } + return path +} + +// holdLease records another replica as the holder of the replicator's lease +// for an hour, so the replica under test cannot win it. +func holdLease(ctx context.Context, t *testing.T, c client.Client) { + t.Helper() + holder, seconds, now := "another-replica", int32(3600), metav1.NewMicroTime(time.Now()) + lease := &coordinationv1.Lease{ + ObjectMeta: metav1.ObjectMeta{Name: leaderElectionID, Namespace: testNamespace}, + Spec: coordinationv1.LeaseSpec{ + HolderIdentity: &holder, LeaseDurationSeconds: &seconds, AcquireTime: &now, RenewTime: &now, + }, + } + if err := c.Create(ctx, lease); err != nil { + t.Fatalf("creating the held lease: %v", err) + } +} + +// createReadyProject creates the test Project in the deployment cluster and +// marks it Ready, which is what makes the Milo provider register it. +func createReadyProject(ctx context.Context, t *testing.T, c client.Client) { + t.Helper() + project := &unstructured.Unstructured{} + project.SetAPIVersion("resourcemanager.miloapis.com/v1alpha1") + project.SetKind("Project") + project.SetName(testProject) + project.Object["spec"] = map[string]any{"ownerRef": map[string]any{"kind": "Organization", "name": "org"}} + if err := c.Create(ctx, project); err != nil { + t.Fatalf("creating the project: %v", err) + } + project.Object["status"] = map[string]any{"conditions": []any{map[string]any{ + "type": "Ready", "status": "True", "reason": "Ready", "message": "ready", + "lastTransitionTime": time.Now().UTC().Format(time.RFC3339), + }}} + if err := c.Status().Update(ctx, project); err != nil { + t.Fatalf("marking the project ready: %v", err) + } +} + +// createZone creates the record set's DNSZone in the project control plane. +func createZone(ctx context.Context, t *testing.T, c client.Client) { + t.Helper() + zone := &dnsv1alpha1.DNSZone{ObjectMeta: metav1.ObjectMeta{Name: testZone, Namespace: testNamespace}} + zone.Spec.DomainName = testDomain + zone.Spec.DNSZoneClassName = "test" + if err := c.Create(ctx, zone); err != nil { + t.Fatalf("creating the zone: %v", err) + } +} + +func newClient(t *testing.T, cfg *rest.Config) client.Client { + t.Helper() + c, err := client.New(cfg, client.Options{Scheme: scheme}) + if err != nil { + t.Fatalf("creating client: %v", err) + } + return c +} + +func newCluster(t *testing.T, cfg *rest.Config) cluster.Cluster { + t.Helper() + cl, err := cluster.New(cfg, func(o *cluster.Options) { o.Scheme = scheme }) + if err != nil { + t.Fatalf("creating cluster: %v", err) + } + return cl +} From 41a95297f3332fae3a12e9ce9e70d325fa16e148 Mon Sep 17 00:00:00 2001 From: Viktor Voytovych Date: Fri, 16 Oct 2026 11:52:39 +0300 Subject: [PATCH 2/3] test: Run the replicator as production does One replica, leader election off, the Milo provider: the replicator reconciles the project's DNSZone and the webhook annotates a record set. The same test passes on the previous wiring, so it guards production's configuration. --- cmd/nonleader_test.go | 122 ++++++++++++++++++++++++++++++++++-------- 1 file changed, 99 insertions(+), 23 deletions(-) diff --git a/cmd/nonleader_test.go b/cmd/nonleader_test.go index 2db9a47e..31335def 100644 --- a/cmd/nonleader_test.go +++ b/cmd/nonleader_test.go @@ -9,9 +9,11 @@ import ( "net/http" "net/http/httptest" "net/http/httputil" + "os" "os/exec" "path/filepath" "regexp" + "slices" "strings" "testing" "time" @@ -35,6 +37,7 @@ import ( dnsv1alpha1 "go.miloapis.com/dns-operator/api/v1alpha1" "go.miloapis.com/dns-operator/internal/config" + "go.miloapis.com/dns-operator/internal/controller" "go.miloapis.com/dns-operator/internal/display" dnswebhook "go.miloapis.com/dns-operator/internal/webhook" ) @@ -59,7 +62,8 @@ func TestNonLeaderReplica(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() probe := freeAddress(t) - mcmgr := startNonLeaderReplica(ctx, t, ctrl.Options{ + p := startPlanes(t) + mcmgr := startReplica(ctx, t, p, true, newCluster(t, p.deployment), ctrl.Options{ HealthProbeBindAddress: probe, WebhookServer: neverServing{webhook.NewServer(webhook.Options{})}, }, setupReplicatorWebhook) @@ -105,36 +109,97 @@ func readyzOnceProjectsRegistered(t *testing.T, address string, timeout time.Dur return status, body } -// startNonLeaderReplica builds the replicator the way main does, with the Milo -// provider production uses, as a replica that cannot win the lease, and -// starts it. Two API servers stand in for production's two: the deployment -// cluster, which holds the Projects, the Lease and downstream DNSZones, and a -// project control plane, which holds the project's DNSZone. A proxy in front -// of the second strips the project path, as Milo does. So a webhook that never +const ownProcessEnv = "DNS_OPERATOR_TEST_OWN_PROCESS" + +func TestSingleReplicaWithoutLeaderElection(t *testing.T) { + // The Milo provider registers a process-wide controller name, and + // TestNonLeaderReplica starts one, so this test runs in a process of its own. + if os.Getenv(ownProcessEnv) != t.Name() { + runInOwnProcess(t) + return + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + p := startPlanes(t) + downstream := newCluster(t, p.deployment) + zoneReplicator := func(mcmgr mcmanager.Manager) error { + return (&controller.DNSZoneReplicator{DownstreamClient: downstream.GetClient()}).SetupWithManager(mcmgr, downstream) + } + mcmgr := startReplica(ctx, t, p, false, downstream, ctrl.Options{ + HealthProbeBindAddress: freeAddress(t), + WebhookServer: neverServing{webhook.NewServer(webhook.Options{})}, + }, zoneReplicator, setupReplicatorWebhook) + + if !finalizedWithin(ctx, t, newClient(t, p.project), 60*time.Second) { + t.Fatal("the replicator never reconciled the project's DNSZone") + } + mutator := &dnswebhook.DNSRecordSetMutator{Manager: mcmgr, Client: mcmgr.GetLocalManager().GetClient()} + if !annotatedWithin(ctx, t, mutator, 60*time.Second) { + t.Fatal("the webhook admitted a DNSRecordSet without its display annotations") + } +} + +func runInOwnProcess(t *testing.T) { + t.Helper() + cmd := exec.Command(os.Args[0], "-test.run=^"+t.Name()+"$", "-test.v") + cmd.Env = append(os.Environ(), ownProcessEnv+"="+t.Name()) + out, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("%s in its own process: %v\n%s", t.Name(), err, out) + } +} + +func finalizedWithin(ctx context.Context, t *testing.T, c client.Client, timeout time.Duration) bool { + t.Helper() + for deadline := time.Now().Add(timeout); time.Now().Before(deadline); time.Sleep(time.Second) { + var zone dnsv1alpha1.DNSZone + err := c.Get(ctx, client.ObjectKey{Namespace: testNamespace, Name: testZone}, &zone) + if err == nil && slices.Contains(zone.Finalizers, "dns.networking.miloapis.com/finalize-dnszone") { + return true + } + } + return false +} + +// planes stand in for production's two API servers: the deployment cluster, +// which holds the Projects, the Lease and downstream DNSZones, and a project +// control plane, which holds the project's DNSZone. A proxy in front of the +// second strips the project path, as Milo does. So a webhook that never // engaged the project falls back to the deployment cluster, finds no zone // there, and admits the record set without annotations. -func startNonLeaderReplica( - ctx context.Context, t *testing.T, opts ctrl.Options, setups ...func(mcmanager.Manager) error, -) mcmanager.Manager { +type planes struct{ deployment, project *rest.Config } + +func startPlanes(t *testing.T) planes { t.Helper() dnsCRDs := filepath.Join("..", "config", "crd", "bases") - deploymentCfg := startEnvtest(t, dnsCRDs, miloProjectCRDs(t)) - projectCfg := startEnvtest(t, dnsCRDs) + return planes{deployment: startEnvtest(t, dnsCRDs, miloProjectCRDs(t)), project: startEnvtest(t, dnsCRDs, nsoDomainCRD(t))} +} - deploymentClient := newClient(t, deploymentCfg) - holdLease(ctx, t, deploymentClient) +// startReplica builds the replicator the way main does, with the Milo +// provider production uses, and starts it. A contended replica runs with +// leader election on while another replica holds the lease, so it cannot win +// it; otherwise leader election is off, as production runs it. +func startReplica( + ctx context.Context, t *testing.T, p planes, contended bool, downstream cluster.Cluster, + opts ctrl.Options, setups ...func(mcmanager.Manager) error, +) mcmanager.Manager { + t.Helper() + deploymentClient := newClient(t, p.deployment) + if contended { + holdLease(ctx, t, deploymentClient) + } createReadyProject(ctx, t, deploymentClient) - createZone(ctx, t, newClient(t, projectCfg)) + createZone(ctx, t, newClient(t, p.project)) serverConfig := config.DNSOperator{Discovery: config.DiscoveryConfig{ Mode: multiclusterproviders.ProviderMilo, - DiscoveryKubeconfigPath: writeKubeconfig(t, deploymentCfg), - ProjectKubeconfigPath: writeKubeconfig(t, &rest.Config{Host: projectControlPlaneProxy(t, projectCfg)}), + DiscoveryKubeconfigPath: writeKubeconfig(t, p.deployment), + ProjectKubeconfigPath: writeKubeconfig(t, &rest.Config{Host: projectControlPlaneProxy(t, p.project)}), }} opts.Scheme = scheme opts.Metrics = metricsserver.Options{BindAddress: "0"} - opts.LeaderElection, opts.LeaderElectionID, opts.LeaderElectionNamespace = true, leaderElectionID, testNamespace - runnables, mcmgr, err := newReplicatorManager(serverConfig, deploymentCfg, newCluster(t, deploymentCfg), opts) + opts.LeaderElection, opts.LeaderElectionID, opts.LeaderElectionNamespace = contended, leaderElectionID, testNamespace + runnables, mcmgr, err := newReplicatorManager(serverConfig, p.deployment, newCluster(t, p.deployment), opts) if err != nil { t.Fatalf("newReplicatorManager: %v", err) } @@ -143,7 +208,7 @@ func startNonLeaderReplica( t.Fatalf("setting up the replica: %v", err) } } - go func() { _ = runReplicator(ctx, runnables, newCluster(t, deploymentCfg), mcmgr) }() + go func() { _ = runReplicator(ctx, runnables, downstream, mcmgr) }() return mcmgr } @@ -233,11 +298,22 @@ func startEnvtest(t *testing.T, dirs ...string) *rest.Config { // Project among them, in the module this repository pins. func miloProjectCRDs(t *testing.T) string { t.Helper() - out, err := exec.Command("go", "list", "-m", "-f", "{{.Dir}}", "go.miloapis.com/milo").Output() + return filepath.Join(moduleDir(t, "go.miloapis.com/milo"), "config", "crd", "bases", "resourcemanager") +} + +func nsoDomainCRD(t *testing.T) string { + t.Helper() + return filepath.Join(moduleDir(t, "go.datum.net/network-services-operator"), + "config", "crd", "bases", "networking.datumapis.com_domains.yaml") +} + +func moduleDir(t *testing.T, module string) string { + t.Helper() + out, err := exec.Command("go", "list", "-m", "-f", "{{.Dir}}", module).Output() if err != nil { - t.Fatalf("locating the milo module: %v", err) + t.Fatalf("locating the %s module: %v", module, err) } - return filepath.Join(strings.TrimSpace(string(out)), "config", "crd", "bases", "resourcemanager") + return strings.TrimSpace(string(out)) } // projectControlPlaneProxy serves the project control plane at Milo's project From ad50d7b402d51f4f530ca6750da7d04d59636d72 Mon Sep 17 00:00:00 2001 From: Viktor Voytovych Date: Fri, 16 Oct 2026 12:38:13 +0300 Subject: [PATCH 3/3] style: Wrap two test lines over the lint limit The lll linter caps a line at 120 characters. Two lines in the non-leader test were longer, so the Lint workflow failed on this branch. --- cmd/nonleader_test.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/cmd/nonleader_test.go b/cmd/nonleader_test.go index 31335def..9636cb92 100644 --- a/cmd/nonleader_test.go +++ b/cmd/nonleader_test.go @@ -51,7 +51,8 @@ const ( // projectControlPlanePath is the path Milo serves each project's control plane // under; the provider appends it to the project kubeconfig's host. -var projectControlPlanePath = regexp.MustCompile(`^/apis/resourcemanager\.miloapis\.com/v1alpha1/projects/[^/]+/control-plane`) +var projectControlPlanePath = regexp.MustCompile( + `^/apis/resourcemanager\.miloapis\.com/v1alpha1/projects/[^/]+/control-plane`) // TestNonLeaderReplica runs the replicator as a replica that never wins the // lease, with the Milo provider production uses and a webhook server that @@ -172,7 +173,10 @@ type planes struct{ deployment, project *rest.Config } func startPlanes(t *testing.T) planes { t.Helper() dnsCRDs := filepath.Join("..", "config", "crd", "bases") - return planes{deployment: startEnvtest(t, dnsCRDs, miloProjectCRDs(t)), project: startEnvtest(t, dnsCRDs, nsoDomainCRD(t))} + return planes{ + deployment: startEnvtest(t, dnsCRDs, miloProjectCRDs(t)), + project: startEnvtest(t, dnsCRDs, nsoDomainCRD(t)), + } } // startReplica builds the replicator the way main does, with the Milo