From 07751de942287b9ae20532e89ef1901739f3e7e1 Mon Sep 17 00:00:00 2001 From: Jiawei Huang Date: Wed, 22 Jul 2026 15:17:27 -0700 Subject: [PATCH] serval: render the non-cluster-host L7 entrypoint from NonClusterHost Render serval, the single HTTPS entrypoint for non-cluster hosts, from the existing NonClusterHost CR. spec.typhaEndpoint is deprecated: unset selects the tunnel and the nonclusterhost controller renders serval; set keeps today's direct-to-typha path and renders calico-typha-noncluster-host. Exactly one exists at a time, both owned by the same CR, so flipping the field is an ordinary reconcile. There is no Serval CR and no separate controller. Serval runs Typha in-process, sourced from the in-cluster calico-typha, so it needs a Typha client keypair (serval-typha-client-certs, common name typha-client) and no datastore RBAC. Its egress rule to calico-typha matches the Service rather than an endpoint selector, because a plain selector does not match ClusterIP-DNAT'd traffic; a service selector implies the Service's ports, so Ports must not be set alongside it. Serval carries the per-host fan-out, so the HostEndpoint-count autoscaler drives its replica count, as it did for calico-typha-noncluster-host. The in-cluster calico-typha keeps scaling on nodes alone: each serval replica is a single client of it however many hosts it serves. The HostEndpoint count now includes only the endpoints the node init labels as non-cluster hosts, so auto and in-cluster host endpoints no longer inflate it. The non-cluster-host Typha policy is deleted alongside the other v3 policies rather than with that deployment and service. A v3 resource is only reconcilable while the API server is healthy, so deleting it from the Typha component made every reconcile fail on a cluster that has no Calico API server -- including the FV environment, where it stalled unrelated components behind a degraded installation. Also refresh the imported FelixConfiguration CRDs, which had drifted from calico master (nftables flowtable offload fields). Co-Authored-By: Claude Opus 4.8 --- pkg/common/common.go | 7 + pkg/controller/csr/csr_controller.go | 5 +- .../installation/core_controller.go | 95 ++-- .../installation/core_controller_test.go | 81 +++- .../installation/typha_autoscaler.go | 320 ------------- .../logcollector/logcollector_controller.go | 38 +- .../logcollector_controller_test.go | 7 +- .../nonclusterhost_controller.go | 215 ++++++++- .../nonclusterhost_controller_test.go | 2 + .../typhaautoscaler/typha_autoscaler.go | 328 ++++++++++++++ .../typha_autoscaler_suite_test.go | 27 ++ .../typha_autoscaler_test.go | 74 +-- .../common/networkpolicy/networkpolicy.go | 4 + pkg/render/logcollector/fluentbit_test.go | 17 +- pkg/render/logcollector/logcollector.go | 14 +- pkg/render/logcollector/networkpolicy.go | 11 +- pkg/render/logcollector/pipeline.go | 4 +- .../logcollector/rendered_config_test.go | 1 + pkg/render/nonclusterhost/nonclusterhost.go | 24 +- pkg/render/serval/component.go | 424 ++++++++++++++++++ pkg/render/serval/component_test.go | 177 ++++++++ pkg/render/serval/suite_test.go | 29 ++ pkg/render/typha.go | 50 ++- pkg/render/typha_test.go | 47 ++ 24 files changed, 1518 insertions(+), 483 deletions(-) delete mode 100644 pkg/controller/installation/typha_autoscaler.go create mode 100644 pkg/controller/typhaautoscaler/typha_autoscaler.go create mode 100644 pkg/controller/typhaautoscaler/typha_autoscaler_suite_test.go rename pkg/controller/{installation => typhaautoscaler}/typha_autoscaler_test.go (79%) create mode 100644 pkg/render/serval/component.go create mode 100644 pkg/render/serval/component_test.go create mode 100644 pkg/render/serval/suite_test.go diff --git a/pkg/common/common.go b/pkg/common/common.go index 485046c415..3ac3672630 100644 --- a/pkg/common/common.go +++ b/pkg/common/common.go @@ -21,6 +21,13 @@ const ( KubeControllersDeploymentName = "calico-kube-controllers" WindowsDaemonSetName = "calico-node-windows" + // ServalName is the name of Serval's Deployment and of the Service in front of + // it. Typha needs the Service name too, so it lives here rather than in the + // serval render package, which imports render. + ServalName = "serval" + // ServalServicePortName is the name of the HTTPS port on that Service. + ServalServicePortName = "https" + // Monitor + Prometheus related const TigeraPrometheusNamespace = "tigera-prometheus" diff --git a/pkg/controller/csr/csr_controller.go b/pkg/controller/csr/csr_controller.go index 32618b5139..e16c97484f 100644 --- a/pkg/controller/csr/csr_controller.go +++ b/pkg/controller/csr/csr_controller.go @@ -204,8 +204,9 @@ func (r *reconcileCSR) Reconcile(ctx context.Context, request reconcile.Request) } needsCSRRole = monitorCR.Spec.ExternalPrometheus != nil - // Check whether the non-cluster host feature is enabled. - // Non-cluster hosts generate CSRs to establish mTLS connections with the cluster. + // Check whether the non-cluster host feature is enabled. Non-cluster hosts + // generate CSRs to establish mTLS connections with the cluster, in both the + // serval gateway and legacy modes. if !needsCSRRole { nonclusterhost, err := utils.GetNonClusterHost(ctx, r.client) if err != nil { diff --git a/pkg/controller/installation/core_controller.go b/pkg/controller/installation/core_controller.go index a766f061f4..6af801944d 100644 --- a/pkg/controller/installation/core_controller.go +++ b/pkg/controller/installation/core_controller.go @@ -57,7 +57,6 @@ import ( "sigs.k8s.io/controller-runtime/pkg/reconcile" v3 "github.com/tigera/api/pkg/apis/projectcalico/v3" - calicoclient "github.com/tigera/api/pkg/client/clientset_generated/clientset" operatorv1 "github.com/tigera/operator/api/v1" "github.com/tigera/operator/pkg/active" "github.com/tigera/operator/pkg/common" @@ -72,6 +71,7 @@ import ( "github.com/tigera/operator/pkg/controller/migration/datastoremigration" "github.com/tigera/operator/pkg/controller/options" "github.com/tigera/operator/pkg/controller/status" + "github.com/tigera/operator/pkg/controller/typhaautoscaler" "github.com/tigera/operator/pkg/controller/utils" "github.com/tigera/operator/pkg/controller/utils/imageset" "github.com/tigera/operator/pkg/ctrlruntime" @@ -340,8 +340,7 @@ func newReconciler(mgr manager.Manager, opts options.ControllerOptions) (*Reconc go nodeIndexInformer.Run(opts.ShutdownContext.Done()) // Create a Typha autoscaler. - typhaListWatch := cache.NewListWatchFromClient(opts.K8sClientset.AppsV1().RESTClient(), "deployments", "calico-system", fields.OneTermEqualSelector("metadata.name", "calico-typha")) - typhaScaler := newTyphaAutoscaler(opts.K8sClientset, nodeIndexInformer, typhaListWatch, statusManager) + typhaScaler := typhaautoscaler.New(opts.K8sClientset, nodeIndexInformer, statusManager, []string{common.TyphaDeploymentName}) r := &ReconcileInstallation{ config: mgr.GetConfig(), @@ -365,7 +364,7 @@ func newReconciler(mgr manager.Manager, opts options.ControllerOptions) (*Reconc apiDiscovery: opts.APIDiscovery, } r.status.Run(opts.ShutdownContext) - r.typhaAutoscaler.start(opts.ShutdownContext) + r.typhaAutoscaler.Start(opts.ShutdownContext) return r, nil } @@ -402,26 +401,25 @@ var _ reconcile.Reconciler = &ReconcileInstallation{} type ReconcileInstallation struct { // This client, initialized using mgr.Client() above, is a split client // that reads objects from the cache and writes to the apiserver - config *rest.Config - client client.Client - clientset *kubernetes.Clientset - scheme *runtime.Scheme - shutdownContext context.Context - watches map[runtime.Object]struct{} - autoDetectedProvider operatorv1.Provider - status status.StatusManager - typhaAutoscaler *typhaAutoscaler - typhaAutoscalerNonClusterHost *typhaAutoscaler - namespaceMigration migration.NamespaceMigration - enterpriseCRDsExist bool - migrationChecked bool - clusterDomain string - manageCRDs bool - tierWatchReady *utils.ReadyFlag - migrationWatchReady *utils.ReadyFlag - v3CRDs bool - kubernetesVersion *common.VersionInfo - apiDiscovery *discovery.APIDiscovery + config *rest.Config + client client.Client + clientset *kubernetes.Clientset + scheme *runtime.Scheme + shutdownContext context.Context + watches map[runtime.Object]struct{} + autoDetectedProvider operatorv1.Provider + status status.StatusManager + typhaAutoscaler *typhaautoscaler.Autoscaler + namespaceMigration migration.NamespaceMigration + enterpriseCRDsExist bool + migrationChecked bool + clusterDomain string + manageCRDs bool + tierWatchReady *utils.ReadyFlag + migrationWatchReady *utils.ReadyFlag + v3CRDs bool + kubernetesVersion *common.VersionInfo + apiDiscovery *discovery.APIDiscovery // newComponentHandler returns a new component handler. Useful stub for unit testing. newComponentHandler func(log logr.Logger, client client.Client, scheme *runtime.Scheme, cr metav1.Object) utils.ComponentHandler @@ -1012,19 +1010,12 @@ func (r *ReconcileInstallation) Reconcile(ctx context.Context, request reconcile if !installationMarkedForDeletion { // If the autoscalar is degraded then trigger a run and recheck the degraded status. If it is still degraded after the // the run the reset the degraded status and requeue the request. - if r.typhaAutoscaler.isDegraded() { - if err := r.typhaAutoscaler.triggerRun(); err != nil { + if r.typhaAutoscaler.IsDegraded() { + if err := r.typhaAutoscaler.TriggerRun(); err != nil { r.status.SetDegraded(operatorv1.ResourceScalingError, "Failed to scale typha", err, reqLogger) return reconcile.Result{RequeueAfter: utils.StandardRetry}, nil } } - - if r.typhaAutoscalerNonClusterHost != nil && r.typhaAutoscalerNonClusterHost.isDegraded() { - if err := r.typhaAutoscalerNonClusterHost.triggerRun(); err != nil { - r.status.SetDegraded(operatorv1.ResourceScalingError, "Failed to scale typha for noncluster hosts", err, reqLogger) - return reconcile.Result{RequeueAfter: utils.StandardRetry}, nil - } - } } // The operator supports running in a "Calico only" mode so that it doesn't need to run enterprise-specific controllers. @@ -1450,14 +1441,18 @@ func (r *ReconcileInstallation) Reconcile(ctx context.Context, request reconcile TrustedBundle: typhaNodeTLS.TrustedBundle, })) - // Check if non-cluster host feature is enabled. - var nonclusterhost *operatorv1.NonClusterHost + // Check if the non-cluster host feature is enabled. The legacy directly-exposed Typha + // deployment (calico-typha-noncluster-host) renders only when a NonClusterHost sets + // spec.typhaEndpoint. An unset endpoint selects the serval gateway (rendered by the + // nonclusterhost controller), whose in-process Typha replaces this deployment. + var legacyNonClusterHost *operatorv1.NonClusterHost if instance.Spec.Variant.IsEnterprise() { - nonclusterhost, err = utils.GetNonClusterHost(ctx, r.client) + nonclusterhost, err := utils.GetNonClusterHost(ctx, r.client) if err != nil { r.status.SetDegraded(operatorv1.ResourceReadError, "Failed to query NonClusterHost resource", err, reqLogger) return reconcile.Result{}, err - } else if nonclusterhost != nil { + } else if nonclusterhost != nil && nonclusterhost.Spec.TyphaEndpoint != "" { + legacyNonClusterHost = nonclusterhost // This is the default common name in CSR from non-cluster hosts. typhaNodeTLS.NodeNonClusterHostCommonName = render.FelixCommonName + render.TyphaNonClusterHostSuffix // Attempt to retrieve the BYO node certificates for non-cluster hosts if they are present. @@ -1473,22 +1468,6 @@ func (r *ReconcileInstallation) Reconcile(ctx context.Context, request reconcile typhaNodeTLS.NodeNonClusterHostCommonName = cn typhaNodeTLS.NodeNonClusterHostURISAN = urisan } - - if r.typhaAutoscalerNonClusterHost == nil { - calicoClient, err := calicoclient.NewForConfig(r.config) - if err != nil { - r.status.SetDegraded(operatorv1.InvalidConfigurationError, "Failed to initialize Calico client", err, reqLogger) - return reconcile.Result{}, err - } - - hepListWatch := cache.NewListWatchFromClient(calicoClient.ProjectcalicoV3().RESTClient(), "hostendpoints", corev1.NamespaceAll, fields.Everything()) - hepIndexInformer := cache.NewSharedIndexInformer(hepListWatch, &v3.HostEndpoint{}, 0, cache.Indexers{}) - go hepIndexInformer.Run(r.shutdownContext.Done()) - - typhaNonClusterHostWatch := cache.NewListWatchFromClient(r.clientset.AppsV1().RESTClient(), "deployments", "calico-system", fields.OneTermEqualSelector("metadata.name", "calico-typha"+render.TyphaNonClusterHostSuffix)) - r.typhaAutoscalerNonClusterHost = newTyphaAutoscaler(r.clientset, hepIndexInformer, typhaNonClusterHostWatch, r.status, typhaAutoscalerOptionNonclusterHost(true)) - r.typhaAutoscalerNonClusterHost.start(r.shutdownContext) - } } } @@ -1500,7 +1479,7 @@ func (r *ReconcileInstallation) Reconcile(ctx context.Context, request reconcile TLS: typhaNodeTLS, MigrateNamespaces: needsNamespaceMigration, ClusterDomain: r.clusterDomain, - NonClusterHost: nonclusterhost, + NonClusterHost: legacyNonClusterHost, FelixHealthPort: *felixConfiguration.Spec.HealthPort, } components = append(components, render.Typha(&typhaCfg)) @@ -1738,8 +1717,14 @@ func (r *ReconcileInstallation) Reconcile(ctx context.Context, request reconcile // deployment becomes unhealthy and reconciliation of non-NetworkPolicy resources in the core controller // would resolve it, we render the network policies of components last to prevent a chicken-and-egg scenario. if includeV3NetworkPolicy { - if nonclusterhost != nil { + if legacyNonClusterHost != nil { components = append(components, render.NewTyphaNonClusterHostPolicy(&typhaCfg)) + } else { + // Garbage-collect the policy when the legacy Typha is not deployed. It + // belongs here rather than alongside that deployment and service, + // because a v3 resource is only reconcilable while the API server is + // healthy. + components = append(components, render.NewDeletionPassthrough(render.TyphaNonClusterHostPolicyForDeletion())) } components = append(components, kubecontrollers.NewCalicoKubeControllersPolicy(&kubeControllersCfg, calicoSystemDefaultDenyForCalicoSystem()), diff --git a/pkg/controller/installation/core_controller_test.go b/pkg/controller/installation/core_controller_test.go index bda4f6065b..6a9e2eb9b3 100644 --- a/pkg/controller/installation/core_controller_test.go +++ b/pkg/controller/installation/core_controller_test.go @@ -36,6 +36,7 @@ import ( rbacv1 "k8s.io/api/rbac/v1" schedv1 "k8s.io/api/scheduling/v1" storagev1 "k8s.io/api/storage/v1" + kerror "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" @@ -57,6 +58,7 @@ import ( "github.com/tigera/operator/pkg/components" "github.com/tigera/operator/pkg/controller/certificatemanager" "github.com/tigera/operator/pkg/controller/status" + "github.com/tigera/operator/pkg/controller/typhaautoscaler" "github.com/tigera/operator/pkg/controller/utils" ctrlrfake "github.com/tigera/operator/pkg/ctrlruntime/client/fake" "github.com/tigera/operator/pkg/dns" @@ -195,7 +197,7 @@ var _ = Describe("Testing core-controller installation", func() { scheme: scheme, autoDetectedProvider: operator.ProviderNone, status: mockStatus, - typhaAutoscaler: newTyphaAutoscaler(cs, nodeIndexInformer, test.NewTyphaListWatch(cs), mockStatus), + typhaAutoscaler: typhaautoscaler.New(cs, nodeIndexInformer, mockStatus, []string{common.TyphaDeploymentName}), namespaceMigration: &fakeNamespaceMigration{}, enterpriseCRDsExist: true, migrationChecked: true, @@ -204,7 +206,7 @@ var _ = Describe("Testing core-controller installation", func() { newComponentHandler: utils.NewComponentHandler, } - r.typhaAutoscaler.start(ctx) + r.typhaAutoscaler.Start(ctx) certificateManager, err := certificatemanager.Create(c, nil, "", common.OperatorNamespace(), certificatemanager.AllowCACreation()) Expect(err).NotTo(HaveOccurred()) @@ -254,22 +256,26 @@ var _ = Describe("Testing core-controller installation", func() { cancel() }) + // These cover the legacy direct mode, which renders + // calico-typha-noncluster-host. spec.typhaEndpoint is what selects it; + // leaving the field unset selects the tunnel, where hosts reach Typha + // through serval and this deployment is not rendered. Context("non-cluster host tests", func() { nonClusterHostObjectMeta := metav1.ObjectMeta{Name: "tigera-secure"} BeforeEach(func() { - By("Creating a NonClusterHost CR") + By("Creating a NonClusterHost CR in the legacy direct mode") Expect(c.Create(ctx, &operator.NonClusterHost{ TypeMeta: metav1.TypeMeta{Kind: "NonClusterHost", APIVersion: "operator.tigera.io/v1"}, ObjectMeta: nonClusterHostObjectMeta, + Spec: operator.NonClusterHostSpec{ + Endpoint: "https://1.2.3.4:443", + TyphaEndpoint: "5.6.7.8:5473", + }, })).NotTo(HaveOccurred()) - - r.typhaAutoscalerNonClusterHost = newTyphaAutoscaler(cs, nodeIndexInformer, test.NewTyphaListWatch(cs), mockStatus) - r.typhaAutoscalerNonClusterHost.start(ctx) }) AfterEach(func() { - r.typhaAutoscalerNonClusterHost = nil Expect(c.Delete(ctx, &operator.NonClusterHost{ObjectMeta: nonClusterHostObjectMeta})).NotTo(HaveOccurred()) }) @@ -345,6 +351,51 @@ var _ = Describe("Testing core-controller installation", func() { }) }) + // The counterpart to the legacy direct mode above: with spec.typhaEndpoint + // unset the hosts reach Typha through serval, so the core controller must + // not deploy a Typha of its own for them. + Context("non-cluster host tests, tunnel mode", func() { + nonClusterHostObjectMeta := metav1.ObjectMeta{Name: "tigera-secure"} + + BeforeEach(func() { + By("Creating a NonClusterHost CR with no typhaEndpoint") + Expect(c.Create(ctx, &operator.NonClusterHost{ + TypeMeta: metav1.TypeMeta{Kind: "NonClusterHost", APIVersion: "operator.tigera.io/v1"}, + ObjectMeta: nonClusterHostObjectMeta, + Spec: operator.NonClusterHostSpec{Endpoint: "https://1.2.3.4:443"}, + })).NotTo(HaveOccurred()) + }) + + AfterEach(func() { + Expect(c.Delete(ctx, &operator.NonClusterHost{ObjectMeta: nonClusterHostObjectMeta})).NotTo(HaveOccurred()) + }) + + It("should not create a separate Typha deployment for non-cluster hosts", func() { + _, err := r.Reconcile(ctx, reconcile.Request{}) + Expect(err).NotTo(HaveOccurred()) + + err = c.Get(ctx, types.NamespacedName{Name: "calico-typha-noncluster-host", Namespace: common.CalicoNamespace}, &appsv1.Deployment{}) + Expect(kerror.IsNotFound(err)).To(BeTrue(), "calico-typha-noncluster-host should not be deployed in tunnel mode") + }) + + It("should still deploy the in-cluster Typha", func() { + _, err := r.Reconcile(ctx, reconcile.Request{}) + Expect(err).NotTo(HaveOccurred()) + + // The hosts are served through serval, which is an ordinary client of + // this deployment, so it must be unaffected by the mode. + typha := appsv1.Deployment{} + Expect(c.Get(ctx, types.NamespacedName{Name: common.TyphaDeploymentName, Namespace: common.CalicoNamespace}, &typha)).NotTo(HaveOccurred()) + + // Serval is a client per replica rather than per node, so Typha counts + // its endpoints when it sizes its connection limit. + Expect(typha.Spec.Template.Spec.Containers[0].Env).To(ContainElements( + corev1.EnvVar{Name: "TYPHA_K8SEXTRACLIENTSERVICENAME", Value: "serval"}, + corev1.EnvVar{Name: "TYPHA_K8SEXTRACLIENTPORTNAME", Value: "https"}, + )) + }) + }) + Context("with Goldmane installed", func() { BeforeEach(func() { // Create a Goldmane CR. @@ -824,7 +875,7 @@ var _ = Describe("Testing core-controller installation", func() { scheme: scheme, autoDetectedProvider: operator.ProviderNone, status: mockStatus, - typhaAutoscaler: newTyphaAutoscaler(cs, nodeIndexInformer, test.NewTyphaListWatch(cs), mockStatus), + typhaAutoscaler: typhaautoscaler.New(cs, nodeIndexInformer, mockStatus, []string{common.TyphaDeploymentName}), namespaceMigration: &fakeNamespaceMigration{}, enterpriseCRDsExist: true, migrationChecked: true, @@ -833,7 +884,7 @@ var _ = Describe("Testing core-controller installation", func() { migrationWatchReady: &utils.ReadyFlag{}, newComponentHandler: utils.NewComponentHandler, } - r.typhaAutoscaler.start(ctx) + r.typhaAutoscaler.Start(ctx) cr = &operator.Installation{ ObjectMeta: metav1.ObjectMeta{Name: "default"}, @@ -1046,7 +1097,7 @@ var _ = Describe("Testing core-controller installation", func() { scheme: scheme, autoDetectedProvider: operator.ProviderNone, status: mockStatus, - typhaAutoscaler: newTyphaAutoscaler(cs, nodeIndexInformer, test.NewTyphaListWatch(cs), mockStatus), + typhaAutoscaler: typhaautoscaler.New(cs, nodeIndexInformer, mockStatus, []string{common.TyphaDeploymentName}), namespaceMigration: &fakeNamespaceMigration{}, enterpriseCRDsExist: true, migrationChecked: true, @@ -1055,7 +1106,7 @@ var _ = Describe("Testing core-controller installation", func() { newComponentHandler: utils.NewComponentHandler, } - r.typhaAutoscaler.start(ctx) + r.typhaAutoscaler.Start(ctx) ca, err := tls.MakeCA("test") Expect(err).NotTo(HaveOccurred()) cert, _, _ := ca.Config.GetPEMBytes() // create a valid pem block @@ -2334,7 +2385,7 @@ var _ = Describe("Testing core-controller installation", func() { scheme: scheme, autoDetectedProvider: operator.ProviderNone, status: mockStatus, - typhaAutoscaler: newTyphaAutoscaler(cs, nodeIndexInformer, test.NewTyphaListWatch(cs), mockStatus), + typhaAutoscaler: typhaautoscaler.New(cs, nodeIndexInformer, mockStatus, []string{common.TyphaDeploymentName}), namespaceMigration: &fakeNamespaceMigration{}, enterpriseCRDsExist: true, migrationChecked: true, @@ -2343,7 +2394,7 @@ var _ = Describe("Testing core-controller installation", func() { migrationWatchReady: &utils.ReadyFlag{}, newComponentHandler: utils.NewComponentHandler, } - r.typhaAutoscaler.start(ctx) + r.typhaAutoscaler.Start(ctx) cr = &operator.Installation{ ObjectMeta: metav1.ObjectMeta{Name: "default"}, @@ -2471,7 +2522,7 @@ var _ = Describe("Testing core-controller installation", func() { scheme: scheme, autoDetectedProvider: operator.ProviderNone, status: mockStatus, - typhaAutoscaler: newTyphaAutoscaler(cs, nodeIndexInformer, test.NewTyphaListWatch(cs), mockStatus), + typhaAutoscaler: typhaautoscaler.New(cs, nodeIndexInformer, mockStatus, []string{common.TyphaDeploymentName}), namespaceMigration: &fakeNamespaceMigration{}, enterpriseCRDsExist: true, migrationChecked: true, @@ -2482,7 +2533,7 @@ var _ = Describe("Testing core-controller installation", func() { }, } - r.typhaAutoscaler.start(ctx) + r.typhaAutoscaler.Start(ctx) certificateManager, err := certificatemanager.Create(c, nil, "", common.OperatorNamespace(), certificatemanager.AllowCACreation()) Expect(err).NotTo(HaveOccurred()) diff --git a/pkg/controller/installation/typha_autoscaler.go b/pkg/controller/installation/typha_autoscaler.go deleted file mode 100644 index 21db6b0a53..0000000000 --- a/pkg/controller/installation/typha_autoscaler.go +++ /dev/null @@ -1,320 +0,0 @@ -// Copyright (c) 2020-2026 Tigera, Inc. All rights reserved. - -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package installation - -import ( - "context" - "fmt" - "time" - - appsv1 "k8s.io/api/apps/v1" - v1 "k8s.io/api/core/v1" - apierrors "k8s.io/apimachinery/pkg/api/errors" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - "k8s.io/client-go/kubernetes" - "k8s.io/client-go/tools/cache" - logf "sigs.k8s.io/controller-runtime/pkg/log" - - v3 "github.com/tigera/api/pkg/apis/projectcalico/v3" - operator "github.com/tigera/operator/api/v1" - "github.com/tigera/operator/pkg/common" - "github.com/tigera/operator/pkg/controller/status" - "github.com/tigera/operator/pkg/render" -) - -var typhaLog = logf.Log.WithName("typha_autoscaler") - -const ( - defaultTyphaAutoscalerSyncPeriod = 10 * time.Second - - hepCreatedLabelKey = "projectcalico.org/created-by" - hepCreatedLabelValue = "calico-kube-controllers" -) - -// typhaAutoscaler periodically lists the nodes and, if needed, scales the Typha deployment up/down. -// Number of replicas should be at least (1 typha for every 200 nodes) + 1 but the number of typhas -// cannot exceed the number of nodes+masters. -type typhaAutoscaler struct { - client kubernetes.Interface - syncPeriod time.Duration - statusManager status.StatusManager - triggerRunChan chan chan error - isDegradedChan chan chan bool - indexInformer cache.SharedIndexInformer - typhaInformer cache.Controller - typhaIndexer cache.Store - nonClusterHost bool - - // done is closed when the autoscaler goroutine exits, so callers can wait for shutdown. - done chan struct{} - - // Number of currently running replicas. - activeReplicas int32 -} - -type typhaAutoscalerOption func(*typhaAutoscaler) - -// typhaAutoscalerOptionPeriod is an option that sets a custom sync period for the Typha autoscaler. -func typhaAutoscalerOptionPeriod(syncPeriod time.Duration) typhaAutoscalerOption { - return func(t *typhaAutoscaler) { - t.syncPeriod = syncPeriod - } -} - -// typhaAutoScalerOptionNonclusterHost is an option that sets the Typha autoscaler to for non-cluster host. -func typhaAutoscalerOptionNonclusterHost(nonClusterHost bool) typhaAutoscalerOption { - return func(t *typhaAutoscaler) { - t.nonClusterHost = nonClusterHost - } -} - -// newTyphaAutoscaler creates a new Typha autoscaler, optionally applying any options to the default autoscaler instance. -// The default sync period is 10 seconds. -func newTyphaAutoscaler(cs kubernetes.Interface, indexInformer cache.SharedIndexInformer, typhaListWatch cache.ListerWatcher, statusManager status.StatusManager, options ...typhaAutoscalerOption) *typhaAutoscaler { - ta := &typhaAutoscaler{ - client: cs, - statusManager: statusManager, - syncPeriod: defaultTyphaAutoscalerSyncPeriod, - triggerRunChan: make(chan chan error), - isDegradedChan: make(chan chan bool), - indexInformer: indexInformer, - } - - // Configure an informer to monitor the active replicas. - typhaHandlers := cache.ResourceEventHandlerFuncs{ - AddFunc: func(obj interface{}) { - if d, ok := obj.(*appsv1.Deployment); ok { - if d.Spec.Replicas != nil { - ta.activeReplicas = *d.Spec.Replicas - } - } - }, - UpdateFunc: func(old, obj interface{}) { - if d, ok := obj.(*appsv1.Deployment); ok { - if d.Spec.Replicas != nil { - ta.activeReplicas = *d.Spec.Replicas - } - } - }, - } - ta.typhaIndexer, ta.typhaInformer = cache.NewInformerWithOptions(cache.InformerOptions{ - ListerWatcher: typhaListWatch, - ObjectType: &appsv1.Deployment{}, - ResyncPeriod: 0, - Handler: typhaHandlers, - Indexers: cache.Indexers{}, - }) - - for _, option := range options { - option(ta) - } - return ta -} - -// start starts the Typha autoscaler, updating the Typha deployment's replica count every sync period. The triggerRunChan -// can be used to trigger an auto scale run immediately, while the isDegradedChan can be used to get the degraded status -// of the last run. The triggerRun and isDegraded functions should be used instead of instead of access these channels directly. -func (t *typhaAutoscaler) start(ctx context.Context) { - t.done = make(chan struct{}) - go func() { - defer close(t.done) - degraded := false - ticker := time.NewTicker(t.syncPeriod) - defer ticker.Stop() - typhaLog.Info("Starting typha autoscaler", "syncPeriod", t.syncPeriod) - - // Start the informer. - go t.typhaInformer.Run(ctx.Done()) - // Wait for the informers to sync, bailing out if we're asked to shut down first. - for !t.indexInformer.HasSynced() || !t.typhaInformer.HasSynced() { - select { - case <-ctx.Done(): - typhaLog.Info("typha autoscaler shutting down") - return - case <-time.After(100 * time.Millisecond): - } - } - - // Don't autoscale or report degraded if the context has been cancelled - we're shutting down. - if ctx.Err() != nil { - typhaLog.Info("typha autoscaler shutting down") - return - } - - // Autoscale on start up then do it again every tick. - if err := t.autoscaleReplicas(); err != nil { - degraded = true - typhaLog.Error(err, "Failed to autoscale typha") - t.statusManager.SetDegraded(operator.ResourceScalingError, fmt.Sprintf("Failed to autoscale typha - %s", err.Error()), nil, log) - } - - for { - select { - case <-ticker.C: - if err := t.autoscaleReplicas(); err != nil { - degraded = true - typhaLog.Error(err, "Failed to autoscale typha") - - // Since this run was triggered by the ticker we need to degrade the tigera status now. - t.statusManager.SetDegraded(operator.ResourceScalingError, fmt.Sprintf("Failed to autoscale typha - %s", err.Error()), nil, log) - } else { - degraded = false - } - case errCh := <-t.triggerRunChan: - if err := t.autoscaleReplicas(); err != nil { - degraded = true - - // Return the error so the "caller" can decided what to do with the error - errCh <- err - } else { - degraded = false - } - - close(errCh) - - ticker.Stop() - ticker = time.NewTicker(t.syncPeriod) - case boolCh := <-t.isDegradedChan: - boolCh <- degraded - close(boolCh) - case <-ctx.Done(): - typhaLog.Info("typha autoscaler shutting down") - return - } - } - }() -} - -// waitForShutdown blocks until the autoscaler goroutine started by start() has exited. It is a no-op -// if the autoscaler was never started. Cancel the context passed to start() to trigger shutdown. -func (t *typhaAutoscaler) waitForShutdown() { - if t.done != nil { - <-t.done - } -} - -func (t *typhaAutoscaler) triggerRun() error { - errChan := make(chan error) - t.triggerRunChan <- errChan - - return <-errChan -} - -// isDegraded checks if the last run autoscale run failed and returns true if it did and false otherwise. -func (t *typhaAutoscaler) isDegraded() bool { - boolChan := make(chan bool) - t.isDegradedChan <- boolChan - - return <-boolChan -} - -// autoscaleReplicas calculates the number of typha pods that should be running and scales the typha deployment accordingly -func (t *typhaAutoscaler) autoscaleReplicas() error { - var expectedReplicas int - if t.nonClusterHost { - heps := t.getHostEndpointCounts() - expectedReplicas = common.GetExpectedTyphaScale(heps) - } else { - allSchedulableNodes, linuxNodes := t.getNodeCounts() - typhaLog.V(5).Info("Number of nodes to consider for typha autoscaling", "all", allSchedulableNodes, "linux", linuxNodes) - expectedReplicas = common.GetExpectedTyphaScale(allSchedulableNodes) - if linuxNodes < expectedReplicas { - return fmt.Errorf("not enough linux nodes to schedule typha pods on, require %d and have %d", expectedReplicas, linuxNodes) - } - } - - typhaLog.V(5).Info("Checking if we need to scale typha", "expectedReplicas", expectedReplicas, "currentReplicas", t.activeReplicas) - if int32(expectedReplicas) != t.activeReplicas { - err := t.updateReplicas(int32(expectedReplicas)) - if err != nil && !apierrors.IsNotFound(err) { - return fmt.Errorf("could not scale Typha deployment: %w", err) - } - } - - return nil -} - -// updateReplicas updates the Typha deployment to the expected replicas if the current replica count differs. -func (t *typhaAutoscaler) updateReplicas(expectedReplicas int32) error { - name := common.TyphaDeploymentName - if t.nonClusterHost { - name += render.TyphaNonClusterHostSuffix - } - typha, err := t.client.AppsV1().Deployments(common.CalicoNamespace).Get(context.Background(), name, metav1.GetOptions{}) - if err != nil { - return err - } - - // The replicas field defaults to 1. We need this in case spec.Replicas is nil. - var prevReplicas int32 - prevReplicas = 1 - if typha.Spec.Replicas != nil { - prevReplicas = *typha.Spec.Replicas - } - - if prevReplicas == expectedReplicas { - return nil - } - - typhaLog.Info(fmt.Sprintf("Updating typha replicas from %d to %d", prevReplicas, expectedReplicas)) - typha.Spec.Replicas = &expectedReplicas - _, err = t.client.AppsV1().Deployments(common.CalicoNamespace).Update(context.Background(), typha, metav1.UpdateOptions{}) - return err -} - -// getNodeCounts returns the number of all the schedulable nodes and the number of the schedulable linux nodes. The linux -// node count is needed because typha pods can only be scheduled on linux nodes, however, nodes of other os types (i.e. windows) -// still need to use typha. -func (t *typhaAutoscaler) getNodeCounts() (int, int) { - linuxNodes := 0 - schedulable := 0 - for _, obj := range t.indexInformer.GetIndexer().List() { - n := obj.(*v1.Node) - if n.Spec.Unschedulable { - continue - } - - if _, ok := n.Labels["kubernetes.azure.com/cluster"]; ok && n.Labels["type"] == "virtual-kubelet" { - // in AKS, there is a feature called 'virtual-nodes' which represent azure's container service as a node in the kubernetes cluster. - // virtual-nodes have many limitations, and are tainted to prevent pods from running on them. - // calico-node isn't run there as they don't support hostNetwork or host volume mounts. - // as such, we shouldn't consider virtual-nodes in the count towards how many typha pods should be run. - // furthermore, typha can't run on virtual-nodes as it is hostnetworked, so we don't want it's desired - // replica count to include it. - continue - } - - schedulable++ - if n.Labels["kubernetes.io/os"] == "linux" { - linuxNodes++ - } - } - return schedulable, linuxNodes -} - -// getHostEndpointCounts returns the number of host endpoints in the cluster that are not created by the kube-controllers. -func (t *typhaAutoscaler) getHostEndpointCounts() int { - heps := 0 - for _, obj := range t.indexInformer.GetIndexer().List() { - // Exclude auto host endpoints that are created by calico-kube-controllers. - hep := obj.(*v3.HostEndpoint) - if _, ok := hep.Labels[hepCreatedLabelKey]; ok && hep.Labels[hepCreatedLabelKey] == hepCreatedLabelValue { - continue - } - - heps++ - } - return heps -} diff --git a/pkg/controller/logcollector/logcollector_controller.go b/pkg/controller/logcollector/logcollector_controller.go index 8dae009ddc..50ff9ec711 100644 --- a/pkg/controller/logcollector/logcollector_controller.go +++ b/pkg/controller/logcollector/logcollector_controller.go @@ -184,7 +184,7 @@ func add(mgr manager.Manager, c ctrlruntime.Controller) error { } if err = c.WatchObject(&operatorv1.NonClusterHost{}, &handler.EnqueueRequestForObject{}); err != nil { - return fmt.Errorf("logcollector-controller failed to watch resource: %w", err) + return fmt.Errorf("logcollector-controller failed to watch NonClusterHost resource: %w", err) } return nil } @@ -643,28 +643,28 @@ func (r *ReconcileLogCollector) Reconcile(ctx context.Context, request reconcile return reconcile.Result{}, err } } - // Create a component handler to manage the rendered component. handler := utils.NewComponentHandler(log, r.client, r.scheme, instance) fluentBitCfg := &rlogcollector.FluentBitConfiguration{ - LogCollector: instance, - S3Credential: s3Credential, - SplkCredential: splunkCredential, - Filters: filters, - EKSConfig: eksConfig, - PullSecrets: pullSecrets, - Installation: installationSpec, - ClusterDomain: r.opts.ClusterDomain, - FluentBitKeyPair: fluentBitKeyPair, - TrustedBundle: trustedBundle, - ManagedCluster: managedCluster, - UseSyslogCertificate: useSyslogCertificate, - Tenant: tenant, - ExternalElastic: r.opts.ElasticExternal, - EKSLogForwarderKeyPair: eksLogForwarderKeyPair, - NonClusterHost: nonclusterhost, - LicenseExpired: licenseExpired, + LogCollector: instance, + S3Credential: s3Credential, + SplkCredential: splunkCredential, + Filters: filters, + EKSConfig: eksConfig, + PullSecrets: pullSecrets, + Installation: installationSpec, + ClusterDomain: r.opts.ClusterDomain, + FluentBitKeyPair: fluentBitKeyPair, + TrustedBundle: trustedBundle, + ManagedCluster: managedCluster, + UseSyslogCertificate: useSyslogCertificate, + Tenant: tenant, + ExternalElastic: r.opts.ElasticExternal, + EKSLogForwarderKeyPair: eksLogForwarderKeyPair, + NonClusterHost: nonclusterhost, + NonClusterHostLogIngestion: nonclusterhost != nil, + LicenseExpired: licenseExpired, } // Render the fluent-bit component for Linux. The same configuration drives // the shared and Windows components below; each applies its OS-specific diff --git a/pkg/controller/logcollector/logcollector_controller_test.go b/pkg/controller/logcollector/logcollector_controller_test.go index e35ae4180e..989eea4e22 100644 --- a/pkg/controller/logcollector/logcollector_controller_test.go +++ b/pkg/controller/logcollector/logcollector_controller_test.go @@ -268,9 +268,10 @@ var _ = Describe("LogCollector controller tests", func() { ObjectMeta: metav1.ObjectMeta{Name: "calico-system.allow-calico-fluent-bit", Namespace: render.LogCollectorNamespace}, } Expect(test.GetResource(c, &policy)).To(BeNil()) - // Metrics rule (2020) + non-cluster-host rule (9880). Without the fix the - // Windows render (applied last) drops the 9880 rule, leaving only one. - Expect(policy.Spec.Ingress).To(HaveLen(2)) + // Metrics rule (2020) + the two non-cluster-host rules on 9880 + // (voltron and serval sources). Without the fix the Windows render + // (applied last) drops the 9880 rules, leaving only the metrics rule. + Expect(policy.Spec.Ingress).To(HaveLen(3)) }) It("should degrade when the syslog endpoint scheme is not tcp or udp", func() { diff --git a/pkg/controller/nonclusterhost/nonclusterhost_controller.go b/pkg/controller/nonclusterhost/nonclusterhost_controller.go index dd0868ab25..098519a647 100644 --- a/pkg/controller/nonclusterhost/nonclusterhost_controller.go +++ b/pkg/controller/nonclusterhost/nonclusterhost_controller.go @@ -19,6 +19,7 @@ import ( "fmt" "net" + "github.com/go-logr/logr" "k8s.io/apimachinery/pkg/runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller" @@ -28,11 +29,21 @@ import ( "sigs.k8s.io/controller-runtime/pkg/reconcile" operatorv1 "github.com/tigera/operator/api/v1" + "github.com/tigera/operator/pkg/common" + "github.com/tigera/operator/pkg/controller/certificatemanager" + "github.com/tigera/operator/pkg/controller/k8sapi" "github.com/tigera/operator/pkg/controller/options" "github.com/tigera/operator/pkg/controller/status" + "github.com/tigera/operator/pkg/controller/typhaautoscaler" "github.com/tigera/operator/pkg/controller/utils" + "github.com/tigera/operator/pkg/controller/utils/imageset" "github.com/tigera/operator/pkg/ctrlruntime" + "github.com/tigera/operator/pkg/dns" + "github.com/tigera/operator/pkg/render" + rcertificatemanagement "github.com/tigera/operator/pkg/render/certificatemanagement" "github.com/tigera/operator/pkg/render/nonclusterhost" + "github.com/tigera/operator/pkg/render/serval" + "github.com/tigera/operator/pkg/tls/certificatemanagement" "github.com/tigera/operator/pkg/url" ) @@ -40,13 +51,20 @@ const controllerName = "nonclusterhost-controller" var log = logf.Log.WithName("controller_nonclusterhost") +// typhaNonClusterHostDeployment is the legacy directly-exposed Typha deployment, +// used when spec.typhaEndpoint is set. +var typhaNonClusterHostDeployment = common.TyphaDeploymentName + render.TyphaNonClusterHostSuffix + func Add(mgr manager.Manager, opts options.ControllerOptions) error { if !opts.EnterpriseCRDExists { return nil } // create the reconciler - reconciler := newReconciler(mgr, opts) + reconciler, err := newReconciler(mgr, opts) + if err != nil { + return err + } // create a new controller c, err := ctrlruntime.NewController(controllerName, mgr, controller.Options{Reconciler: reconciler}) @@ -57,14 +75,33 @@ func Add(mgr manager.Manager, opts options.ControllerOptions) error { return add(mgr, c) } -func newReconciler(mgr manager.Manager, opts options.ControllerOptions) reconcile.Reconciler { +func newReconciler(mgr manager.Manager, opts options.ControllerOptions) (reconcile.Reconciler, error) { + statusManager := status.New(mgr.GetClient(), "non-cluster-hosts", opts.KubernetesVersion) + + // Both modes fan out to the non-cluster hosts, so both scale with the registered-host + // (HostEndpoint) count: serval (typhaEndpoint unset), whose in-process Typha + // serves the hosts, or the legacy calico-typha-noncluster-host deployment (typhaEndpoint + // set). Exactly one exists at a time, so a single autoscaler drives both — it applies the + // count to each and skips the one that is absent. + autoscaler, err := typhaautoscaler.NewHostEndpointScaler( + mgr.GetConfig(), opts.K8sClientset, statusManager, + []string{serval.ServalDeploymentName, typhaNonClusterHostDeployment}, + opts.ShutdownContext.Done()) + if err != nil { + return nil, fmt.Errorf("failed to create the non-cluster-host autoscaler: %w", err) + } + autoscaler.Start(opts.ShutdownContext) + r := &ReconcileNonClusterHost{ - client: mgr.GetClient(), - scheme: mgr.GetScheme(), - status: status.New(mgr.GetClient(), "non-cluster-hosts", opts.KubernetesVersion), + client: mgr.GetClient(), + scheme: mgr.GetScheme(), + status: statusManager, + provider: opts.DetectedProvider, + clusterDomain: opts.ClusterDomain, + autoscaler: autoscaler, } r.status.Run(opts.ShutdownContext) - return r + return r, nil } func add(mgr manager.Manager, c ctrlruntime.Controller) error { @@ -72,15 +109,40 @@ func add(mgr manager.Manager, c ctrlruntime.Controller) error { return fmt.Errorf("%s failed to watch resource: %w", controllerName, err) } + if err := utils.AddInstallationWatch(c); err != nil { + return fmt.Errorf("%s failed to watch Installation resource: %w", controllerName, err) + } + + // Serval's serving cert and its Typha client keypair drive a re-render. + for _, secretName := range []string{serval.ServalKeyPairSecret, serval.ServalTyphaClientKeyPairSecret, certificatemanagement.CASecretName} { + if err := utils.AddSecretsWatch(c, secretName, common.OperatorNamespace()); err != nil { + return fmt.Errorf("%s failed to watch secret %s: %w", controllerName, secretName, err) + } + } + + if err := imageset.AddImageSetWatch(c); err != nil { + return fmt.Errorf("%s failed to watch ImageSet: %w", controllerName, err) + } + + if err := utils.AddPeriodicReconcile(c, utils.PeriodicReconcileTime, &handler.EnqueueRequestForObject{}); err != nil { + return fmt.Errorf("%s failed to create periodic reconcile watch: %w", controllerName, err) + } + return nil } var _ reconcile.Reconciler = &ReconcileNonClusterHost{} type ReconcileNonClusterHost struct { - client client.Client - scheme *runtime.Scheme - status status.StatusManager + client client.Client + scheme *runtime.Scheme + status status.StatusManager + provider operatorv1.Provider + clusterDomain string + + // autoscaler scales serval and the legacy calico-typha-noncluster-host + // deployment by HostEndpoint count; only one exists at a time. + autoscaler *typhaautoscaler.Autoscaler } func (r *ReconcileNonClusterHost) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) { @@ -100,28 +162,72 @@ func (r *ReconcileNonClusterHost) Reconcile(ctx context.Context, request reconci r.status.OnCRFound() defer r.status.SetMetaData(&instance.ObjectMeta) - // Validate endpoint fields - _, _, _, err = url.ParseEndpoint(instance.Spec.Endpoint) + // Re-trigger the autoscaler if degraded and requeue if it stays degraded. + if r.autoscaler != nil && r.autoscaler.IsDegraded() { + if err := r.autoscaler.TriggerRun(); err != nil { + r.status.SetDegraded(operatorv1.ResourceScalingError, "Failed to scale Typha for non-cluster hosts", err, logc) + return reconcile.Result{RequeueAfter: utils.StandardRetry}, nil + } + } + + // Validate the endpoint; it is the HTTPS front door in both modes. + _, endpointHost, _, err := url.ParseEndpoint(instance.Spec.Endpoint) if err != nil { r.status.SetDegraded(operatorv1.ResourceValidationError, "Invalid endpoint", err, logc) return reconcile.Result{}, err } - _, _, err = net.SplitHostPort(instance.Spec.TyphaEndpoint) - if err != nil { - r.status.SetDegraded(operatorv1.ResourceValidationError, "Invalid Typha endpoint", err, logc) - return reconcile.Result{}, err + // typhaEndpoint set selects the deprecated direct-to-typha (legacy) mode; unset selects + // serval. + tunnelMode := instance.Spec.TyphaEndpoint == "" + if !tunnelMode { + logc.Info("spec.typhaEndpoint is deprecated; clear it to reach Typha through the tunnel") + if _, _, err = net.SplitHostPort(instance.Spec.TyphaEndpoint); err != nil { + r.status.SetDegraded(operatorv1.ResourceValidationError, "Invalid Typha endpoint", err, logc) + return reconcile.Result{}, err + } + } + + // The host identity (ServiceAccount, token, ClusterRole) is shared by both modes. + components := []render.Component{ + nonclusterhost.NonClusterHost(&nonclusterhost.Config{NonClusterHost: instance.Spec}), } - config := &nonclusterhost.Config{ - NonClusterHost: instance.Spec, + if tunnelMode { + // Only tunnel mode needs an Installation (for images and certificates). + variant, installationSpec, err := utils.GetInstallationSpec(ctx, r.client) + if err != nil { + r.status.SetDegraded(operatorv1.ResourceReadError, "Failed to query Installation", err, logc) + return reconcile.Result{}, err + } else if installationSpec == nil { + r.status.SetDegraded(operatorv1.ResourceNotFound, "Installation not found", nil, logc) + return reconcile.Result{}, nil + } + + tunnelComponents, res, err := r.tunnelComponents(ctx, installationSpec, endpointHost, logc) + if err != nil || res != nil { + if res != nil { + return *res, err + } + return reconcile.Result{}, err + } + components = append(components, tunnelComponents...) + + if err = imageset.ApplyImageSet(ctx, r.client, variant, components...); err != nil { + r.status.SetDegraded(operatorv1.ResourceUpdateError, "Error with images from ImageSet", err, logc) + return reconcile.Result{}, err + } + } else { + // Legacy mode: tear down serval if it was previously rendered. No images to resolve. + components = append(components, serval.Serval(&serval.Configuration{Deleted: true})) } - component := nonclusterhost.NonClusterHost(config) ch := utils.NewComponentHandler(logc, r.client, r.scheme, instance) - if err = ch.CreateOrUpdateOrDelete(ctx, component, r.status); err != nil { - r.status.SetDegraded(operatorv1.ResourceUpdateError, "Error creating / updating resource", err, logc) - return reconcile.Result{}, err + for _, component := range components { + if err = ch.CreateOrUpdateOrDelete(ctx, component, r.status); err != nil { + r.status.SetDegraded(operatorv1.ResourceUpdateError, "Error creating / updating resource", err, logc) + return reconcile.Result{}, err + } } r.status.ReadyToMonitor() @@ -133,3 +239,70 @@ func (r *ReconcileNonClusterHost) Reconcile(ctx context.Context, request reconci return reconcile.Result{}, nil } + +// tunnelComponents builds the serval components (cert management + serval). It returns +// a non-nil *reconcile.Result when the caller should return early (e.g. waiting on a keypair). +func (r *ReconcileNonClusterHost) tunnelComponents(ctx context.Context, installationSpec *operatorv1.InstallationSpec, endpointHost string, logc logr.Logger) ([]render.Component, *reconcile.Result, error) { + pullSecrets, err := utils.GetInstallationPullSecrets(installationSpec, r.client) + if err != nil { + r.status.SetDegraded(operatorv1.ResourceReadError, "Error retrieving pull secrets", err, logc) + return nil, &reconcile.Result{}, err + } + + certificateManager, err := certificatemanager.Create(r.client, installationSpec, r.clusterDomain, common.OperatorNamespace()) + if err != nil { + r.status.SetDegraded(operatorv1.ResourceCreateError, "Unable to create the certificate manager", err, logc) + return nil, &reconcile.Result{}, err + } + + // The serving certificate covers the in-cluster service names plus the external endpoint + // host that non-cluster hosts connect to. + dnsNames := dns.GetServiceDNSNames(serval.ServalServiceName, serval.ServalNamespace, r.clusterDomain) + dnsNames = append(dnsNames, endpointHost) + keyPair, err := certificateManager.GetOrCreateKeyPair(r.client, serval.ServalKeyPairSecret, common.OperatorNamespace(), dnsNames) + if err != nil { + r.status.SetDegraded(operatorv1.ResourceCreateError, "Error creating TLS certificate", err, logc) + return nil, &reconcile.Result{}, err + } + + // Serval relays felix tunnels to the in-cluster calico-typha as an ordinary + // Typha client, so it needs a client keypair whose CN (typha-client) + // calico-typha accepts, signed by the cluster CA. + typhaClientKeyPair, err := certificateManager.GetOrCreateKeyPair( + r.client, serval.ServalTyphaClientKeyPairSecret, common.OperatorNamespace(), + []string{render.FelixCommonName}) + if err != nil { + r.status.SetDegraded(operatorv1.ResourceCreateError, "Error creating the Typha client certificate", err, logc) + return nil, &reconcile.Result{}, err + } + + trustedBundle, err := certificateManager.CreateNamedTrustedBundleFromSecrets(serval.ServalDeploymentName, r.client, common.OperatorNamespace(), false) + if err != nil { + r.status.SetDegraded(operatorv1.ResourceCreateError, "Unable to create the trusted bundle", err, logc) + return nil, &reconcile.Result{}, err + } + + certComponent := rcertificatemanagement.CertificateManagement(&rcertificatemanagement.Config{ + Namespace: serval.ServalNamespace, + TruthNamespace: common.OperatorNamespace(), + ServiceAccounts: []string{serval.ServalServiceAccountName}, + KeyPairOptions: []rcertificatemanagement.KeyPairOption{ + rcertificatemanagement.NewKeyPairOption(keyPair, true, true), + rcertificatemanagement.NewKeyPairOption(typhaClientKeyPair, true, true), + }, + TrustedBundle: trustedBundle, + }) + + servalComponent := serval.Serval(&serval.Configuration{ + PullSecrets: pullSecrets, + OpenShift: r.provider.IsOpenShift(), + Installation: installationSpec, + TrustedCertBundle: trustedBundle, + ServerKeyPair: keyPair, + TyphaClientKeyPair: typhaClientKeyPair, + ClusterDomain: r.clusterDomain, + K8sServiceEp: k8sapi.Endpoint, + }) + + return []render.Component{certComponent, servalComponent}, nil, nil +} diff --git a/pkg/controller/nonclusterhost/nonclusterhost_controller_test.go b/pkg/controller/nonclusterhost/nonclusterhost_controller_test.go index b15e141618..2a4199f43e 100644 --- a/pkg/controller/nonclusterhost/nonclusterhost_controller_test.go +++ b/pkg/controller/nonclusterhost/nonclusterhost_controller_test.go @@ -64,6 +64,8 @@ var _ = Describe("NonClusterHost controller tests", func() { mockStatus.On("OnCRNotFound").Return() mockStatus.On("ReadyToMonitor") mockStatus.On("SetMetaData", mock.Anything).Return() + // Legacy mode tears down any serval deployment, so the handler untracks it. + mockStatus.On("RemoveDeployments", mock.Anything).Return() r = ReconcileNonClusterHost{ client: cli, diff --git a/pkg/controller/typhaautoscaler/typha_autoscaler.go b/pkg/controller/typhaautoscaler/typha_autoscaler.go new file mode 100644 index 0000000000..5bcca04909 --- /dev/null +++ b/pkg/controller/typhaautoscaler/typha_autoscaler.go @@ -0,0 +1,328 @@ +// Copyright (c) 2020-2026 Tigera, Inc. All rights reserved. + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package typhaautoscaler scales a Typha-like deployment to track the size of +// the thing it fans out to: the cluster's node count for the in-cluster Typha, +// or the registered-host (HostEndpoint) count for the non-cluster-host Typha +// and Serval. Controllers own the deployment they scale; this +// package only adjusts its replica count. +package typhaautoscaler + +import ( + "context" + "fmt" + "time" + + v1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/fields" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/cache" + logf "sigs.k8s.io/controller-runtime/pkg/log" + + v3 "github.com/tigera/api/pkg/apis/projectcalico/v3" + calicoclient "github.com/tigera/api/pkg/client/clientset_generated/clientset" + operator "github.com/tigera/operator/api/v1" + "github.com/tigera/operator/pkg/common" + "github.com/tigera/operator/pkg/controller/status" +) + +var typhaLog = logf.Log.WithName("typha_autoscaler") + +const ( + defaultSyncPeriod = 10 * time.Second + + // nonClusterHostLabelKey/Value tag a HostEndpoint that a non-cluster host + // registered (the node init's label updater sets them). Only these count + // toward scaling: auto host endpoints and in-cluster ones are not served by + // serval. + nonClusterHostLabelKey = "hostendpoint.projectcalico.org/type" + nonClusterHostLabelValue = "nonclusterhost" +) + +// Autoscaler periodically counts the nodes (or host endpoints) and, if needed, scales its +// deployment(s) to match. Number of replicas should be at least (1 typha for every 200 nodes) + 1 +// but the number of typhas cannot exceed the number of nodes+masters. It can drive more than one +// deployment from the same count — the non-cluster-host controller scales serval and +// the legacy calico-typha-noncluster-host deployment with one instance, since exactly one of them +// exists at a time and both want the same replica count. +type Autoscaler struct { + client kubernetes.Interface + syncPeriod time.Duration + statusManager status.StatusManager + triggerRunChan chan chan error + isDegradedChan chan chan bool + indexInformer cache.SharedIndexInformer + + // deploymentNames are the deployments this autoscaler scales, in the calico-system + // namespace. A deployment that does not exist is skipped. + deploymentNames []string + // scaleByHostEndpoints selects the count that drives scaling: host endpoints when true + // (non-cluster-host Typha, Serval), schedulable nodes when false (in-cluster Typha). + scaleByHostEndpoints bool + + // done is closed when the autoscaler goroutine exits, so callers can wait for shutdown. + done chan struct{} +} + +type Option func(*Autoscaler) + +// OptionSyncPeriod sets a custom sync period for the autoscaler. +func OptionSyncPeriod(syncPeriod time.Duration) Option { + return func(t *Autoscaler) { + t.syncPeriod = syncPeriod + } +} + +// OptionScaleByHostEndpoints scales by the registered-host (HostEndpoint) count instead of +// the node count. Use it for the non-cluster-host Typha and Serval. +func OptionScaleByHostEndpoints() Option { + return func(t *Autoscaler) { + t.scaleByHostEndpoints = true + } +} + +// New creates a new autoscaler that scales the named deployments (in the calico-system +// namespace) to a single computed replica count, optionally applying any options. The default +// sync period is 10 seconds and, unless OptionScaleByHostEndpoints is given, scaling tracks the +// node count. +func New(cs kubernetes.Interface, indexInformer cache.SharedIndexInformer, statusManager status.StatusManager, deploymentNames []string, options ...Option) *Autoscaler { + ta := &Autoscaler{ + client: cs, + statusManager: statusManager, + syncPeriod: defaultSyncPeriod, + triggerRunChan: make(chan chan error), + isDegradedChan: make(chan chan bool), + indexInformer: indexInformer, + deploymentNames: deploymentNames, + } + + for _, option := range options { + option(ta) + } + return ta +} + +// NewHostEndpointScaler builds an autoscaler that scales the named deployments (in the +// calico-system namespace) by the registered-host (HostEndpoint) count. It creates the +// HostEndpoint informer from calicoConfig and starts it on stopCh. The caller must still call +// Start to begin scaling. Used by the non-cluster-host controller, which drives both +// serval and the legacy Typha deployment from one count. +func NewHostEndpointScaler(calicoConfig *rest.Config, clientset kubernetes.Interface, statusManager status.StatusManager, deploymentNames []string, stopCh <-chan struct{}) (*Autoscaler, error) { + calicoClient, err := calicoclient.NewForConfig(calicoConfig) + if err != nil { + return nil, err + } + + hepListWatch := cache.NewListWatchFromClient(calicoClient.ProjectcalicoV3().RESTClient(), "hostendpoints", metav1.NamespaceAll, fields.Everything()) + hepIndexInformer := cache.NewSharedIndexInformer(hepListWatch, &v3.HostEndpoint{}, 0, cache.Indexers{}) + go hepIndexInformer.Run(stopCh) + + return New(clientset, hepIndexInformer, statusManager, deploymentNames, OptionScaleByHostEndpoints()), nil +} + +// Start starts the autoscaler, updating the deployment's replica count every sync period. The +// triggerRunChan can be used to trigger an auto scale run immediately, while the isDegradedChan +// can be used to get the degraded status of the last run. TriggerRun and IsDegraded should be +// used instead of accessing these channels directly. +func (t *Autoscaler) Start(ctx context.Context) { + t.done = make(chan struct{}) + go func() { + defer close(t.done) + degraded := false + ticker := time.NewTicker(t.syncPeriod) + defer ticker.Stop() + typhaLog.Info("Starting typha autoscaler", "deployments", t.deploymentNames, "syncPeriod", t.syncPeriod) + + // Wait for the informer to sync, bailing out if we're asked to shut down first. + for !t.indexInformer.HasSynced() { + select { + case <-ctx.Done(): + typhaLog.Info("typha autoscaler shutting down") + return + case <-time.After(100 * time.Millisecond): + } + } + + // Don't autoscale or report degraded if the context has been cancelled - we're shutting down. + if ctx.Err() != nil { + typhaLog.Info("typha autoscaler shutting down") + return + } + + // Autoscale on start up then do it again every tick. + if err := t.autoscaleReplicas(); err != nil { + degraded = true + typhaLog.Error(err, "Failed to autoscale typha") + t.statusManager.SetDegraded(operator.ResourceScalingError, fmt.Sprintf("Failed to autoscale typha - %s", err.Error()), nil, typhaLog) + } + + for { + select { + case <-ticker.C: + if err := t.autoscaleReplicas(); err != nil { + degraded = true + typhaLog.Error(err, "Failed to autoscale typha") + + // Since this run was triggered by the ticker we need to degrade the tigera status now. + t.statusManager.SetDegraded(operator.ResourceScalingError, fmt.Sprintf("Failed to autoscale typha - %s", err.Error()), nil, typhaLog) + } else { + degraded = false + } + case errCh := <-t.triggerRunChan: + if err := t.autoscaleReplicas(); err != nil { + degraded = true + + // Return the error so the "caller" can decided what to do with the error + errCh <- err + } else { + degraded = false + } + + close(errCh) + + ticker.Stop() + ticker = time.NewTicker(t.syncPeriod) + case boolCh := <-t.isDegradedChan: + boolCh <- degraded + close(boolCh) + case <-ctx.Done(): + typhaLog.Info("typha autoscaler shutting down") + return + } + } + }() +} + +// WaitForShutdown blocks until the autoscaler goroutine started by Start() has exited. It is a +// no-op if the autoscaler was never started. Cancel the context passed to Start() to trigger +// shutdown. +func (t *Autoscaler) WaitForShutdown() { + if t.done != nil { + <-t.done + } +} + +// TriggerRun triggers an autoscale run immediately and returns any error from it. +func (t *Autoscaler) TriggerRun() error { + errChan := make(chan error) + t.triggerRunChan <- errChan + + return <-errChan +} + +// IsDegraded checks if the last autoscale run failed and returns true if it did and false otherwise. +func (t *Autoscaler) IsDegraded() bool { + boolChan := make(chan bool) + t.isDegradedChan <- boolChan + + return <-boolChan +} + +// autoscaleReplicas calculates the number of typha pods that should be running and scales the deployment accordingly +func (t *Autoscaler) autoscaleReplicas() error { + var expectedReplicas int + if t.scaleByHostEndpoints { + heps := t.getHostEndpointCounts() + expectedReplicas = common.GetExpectedTyphaScale(heps) + } else { + allSchedulableNodes, linuxNodes := t.getNodeCounts() + typhaLog.V(5).Info("Number of nodes to consider for typha autoscaling", "all", allSchedulableNodes, "linux", linuxNodes) + expectedReplicas = common.GetExpectedTyphaScale(allSchedulableNodes) + if linuxNodes < expectedReplicas { + return fmt.Errorf("not enough linux nodes to schedule typha pods on, require %d and have %d", expectedReplicas, linuxNodes) + } + } + + typhaLog.V(5).Info("Checking if we need to scale typha", "expectedReplicas", expectedReplicas, "deployments", t.deploymentNames) + for _, name := range t.deploymentNames { + // A deployment that does not exist in this mode (e.g. serval while typhaEndpoint is + // set, or vice versa) is simply skipped. + if err := t.updateReplicas(name, int32(expectedReplicas)); err != nil && !apierrors.IsNotFound(err) { + return fmt.Errorf("could not scale deployment %s: %w", name, err) + } + } + + return nil +} + +// updateReplicas updates the named deployment to the expected replicas if its current replica +// count differs. +func (t *Autoscaler) updateReplicas(name string, expectedReplicas int32) error { + typha, err := t.client.AppsV1().Deployments(common.CalicoNamespace).Get(context.Background(), name, metav1.GetOptions{}) + if err != nil { + return err + } + + // The replicas field defaults to 1. We need this in case spec.Replicas is nil. + var prevReplicas int32 + prevReplicas = 1 + if typha.Spec.Replicas != nil { + prevReplicas = *typha.Spec.Replicas + } + + if prevReplicas == expectedReplicas { + return nil + } + + typhaLog.Info(fmt.Sprintf("Updating %s replicas from %d to %d", name, prevReplicas, expectedReplicas)) + typha.Spec.Replicas = &expectedReplicas + _, err = t.client.AppsV1().Deployments(common.CalicoNamespace).Update(context.Background(), typha, metav1.UpdateOptions{}) + return err +} + +// getNodeCounts returns the number of all the schedulable nodes and the number of the schedulable linux nodes. The linux +// node count is needed because typha pods can only be scheduled on linux nodes, however, nodes of other os types (i.e. windows) +// still need to use typha. +func (t *Autoscaler) getNodeCounts() (int, int) { + linuxNodes := 0 + schedulable := 0 + for _, obj := range t.indexInformer.GetIndexer().List() { + n := obj.(*v1.Node) + if n.Spec.Unschedulable { + continue + } + + if _, ok := n.Labels["kubernetes.azure.com/cluster"]; ok && n.Labels["type"] == "virtual-kubelet" { + // in AKS, there is a feature called 'virtual-nodes' which represent azure's container service as a node in the kubernetes cluster. + // virtual-nodes have many limitations, and are tainted to prevent pods from running on them. + // calico-node isn't run there as they don't support hostNetwork or host volume mounts. + // as such, we shouldn't consider virtual-nodes in the count towards how many typha pods should be run. + // furthermore, typha can't run on virtual-nodes as it is hostnetworked, so we don't want it's desired + // replica count to include it. + continue + } + + schedulable++ + if n.Labels["kubernetes.io/os"] == "linux" { + linuxNodes++ + } + } + return schedulable, linuxNodes +} + +// getHostEndpointCounts returns the number of HostEndpoints that non-cluster hosts registered, +// identified by the label the node init sets. Auto host endpoints and in-cluster ones are excluded. +func (t *Autoscaler) getHostEndpointCounts() int { + heps := 0 + for _, obj := range t.indexInformer.GetIndexer().List() { + hep := obj.(*v3.HostEndpoint) + if hep.Labels[nonClusterHostLabelKey] == nonClusterHostLabelValue { + heps++ + } + } + return heps +} diff --git a/pkg/controller/typhaautoscaler/typha_autoscaler_suite_test.go b/pkg/controller/typhaautoscaler/typha_autoscaler_suite_test.go new file mode 100644 index 0000000000..45873c527e --- /dev/null +++ b/pkg/controller/typhaautoscaler/typha_autoscaler_suite_test.go @@ -0,0 +1,27 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package typhaautoscaler + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestTyphaAutoscaler(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Typha Autoscaler Suite") +} diff --git a/pkg/controller/installation/typha_autoscaler_test.go b/pkg/controller/typhaautoscaler/typha_autoscaler_test.go similarity index 79% rename from pkg/controller/installation/typha_autoscaler_test.go rename to pkg/controller/typhaautoscaler/typha_autoscaler_test.go index 8042593a64..af1b294764 100644 --- a/pkg/controller/installation/typha_autoscaler_test.go +++ b/pkg/controller/typhaautoscaler/typha_autoscaler_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package installation +package typhaautoscaler import ( "context" @@ -24,6 +24,7 @@ import ( "github.com/stretchr/testify/mock" operator "github.com/tigera/operator/api/v1" + "github.com/tigera/operator/pkg/common" "github.com/tigera/operator/pkg/controller/status" . "github.com/tigera/operator/test" @@ -41,9 +42,9 @@ var _ = Describe("Test typha autoscaler ", func() { var c *kfake.Clientset var ctx context.Context var cancel context.CancelFunc - var nlw, tlw cache.ListerWatcher + var nlw cache.ListerWatcher var nodeIndexInformer cache.SharedIndexInformer - var ta *typhaAutoscaler + var ta *Autoscaler BeforeEach(func() { ta = nil @@ -59,9 +60,8 @@ var _ = Describe("Test typha autoscaler ", func() { } c = kfake.NewClientset(objs...) nlw = NewNodeListWatch(c) - tlw = NewTyphaListWatch(c) - // Create the indexer and informer used by the typhaAutoscaler + // Create the indexer and informer used by the autoscaler nodeIndexInformer = cache.NewSharedIndexInformer(nlw, &corev1.Node{}, 0, cache.Indexers{}) ctx, cancel = context.WithCancel(context.Background()) @@ -77,13 +77,13 @@ var _ = Describe("Test typha autoscaler ", func() { // spec has ended, panicking a later, unrelated spec. cancel() if ta != nil { - ta.waitForShutdown() + ta.WaitForShutdown() } }) It("should initialize an autoscaler", func() { - ta = newTyphaAutoscaler(c, nodeIndexInformer, tlw, statusManager) - ta.start(ctx) + ta = New(c, nodeIndexInformer, statusManager, []string{common.TyphaDeploymentName}) + ta.Start(ctx) }) It("should get the correct number of nodes", func() { @@ -93,7 +93,7 @@ var _ = Describe("Test typha autoscaler ", func() { // Don't start the autoscaler - this test only exercises getNodeCounts(), which reads // from the nodeIndexInformer directly. Starting it would race with node creation, // since autoscaleReplicas() can fire before the informer has picked up the new nodes. - ta = newTyphaAutoscaler(c, nodeIndexInformer, tlw, statusManager) + ta = New(c, nodeIndexInformer, statusManager, []string{common.TyphaDeploymentName}) Eventually(func() error { schedulableNodes, linuxNodes := ta.getNodeCounts() @@ -144,8 +144,8 @@ var _ = Describe("Test typha autoscaler ", func() { CreateNode(c, "node2", map[string]string{"kubernetes.io/os": "linux"}, nil) // Create the autoscaler and run it - ta = newTyphaAutoscaler(c, nodeIndexInformer, tlw, statusManager, typhaAutoscalerOptionPeriod(10*time.Millisecond)) - ta.start(ctx) + ta = New(c, nodeIndexInformer, statusManager, []string{common.TyphaDeploymentName}, OptionSyncPeriod(10*time.Millisecond)) + ta.Start(ctx) // For clusters smaller than 3 nodes we only expect 1 replica. verifyTyphaReplicas(c, 1) @@ -198,8 +198,8 @@ var _ = Describe("Test typha autoscaler ", func() { }).Should(HaveLen(5)) // Create the autoscaler and run it - ta = newTyphaAutoscaler(c, nodeIndexInformer, tlw, statusManager, typhaAutoscalerOptionPeriod(10*time.Millisecond)) - ta.start(ctx) + ta = New(c, nodeIndexInformer, statusManager, []string{common.TyphaDeploymentName}, OptionSyncPeriod(10*time.Millisecond)) + ta.Start(ctx) verifyTyphaReplicas(c, 3) }) @@ -233,8 +233,8 @@ var _ = Describe("Test typha autoscaler ", func() { }).Should(HaveLen(5)) // Create the autoscaler and run it - ta = newTyphaAutoscaler(c, nodeIndexInformer, tlw, statusManager, typhaAutoscalerOptionPeriod(10*time.Millisecond)) - ta.start(ctx) + ta = New(c, nodeIndexInformer, statusManager, []string{common.TyphaDeploymentName}, OptionSyncPeriod(10*time.Millisecond)) + ta.Start(ctx) // normally we'd expect to see three replicas for five nodes, but since one node is a virtual-kubelet, // we should still only expect two @@ -273,34 +273,48 @@ var _ = Describe("Test typha autoscaler ", func() { }).Should(HaveLen(5)) // Create the autoscaler and run it - ta = newTyphaAutoscaler(c, nodeIndexInformer, tlw, statusManager, typhaAutoscalerOptionPeriod(10*time.Millisecond)) - ta.start(ctx) + ta = New(c, nodeIndexInformer, statusManager, []string{common.TyphaDeploymentName}, OptionSyncPeriod(10*time.Millisecond)) + ta.Start(ctx) // This blocks until the first run is done. - ta.isDegraded() + ta.IsDegraded() statusManager.AssertExpectations(GinkgoT()) }) - It("should not autoscale or report degraded once its context is cancelled", func() { - // statusManager has no SetDegraded expectation configured, so the mock panics if it's - // called. With zero linux nodes the startup autoscale would normally report degraded, so - // a cancelled autoscaler that still runs the startup autoscale would panic here. - cancelledCtx, cancelStart := context.WithCancel(context.Background()) - cancelStart() + It("should scale the deployment it is named for, not one derived from the counting mode", func() { + // Regression: the target deployment is an explicit constructor argument, so an + // autoscaler scales exactly that deployment (e.g. serval) rather than a + // name derived from a flag. Previously a host-endpoint autoscaler always scaled + // calico-typha-noncluster-host regardless of which deployment it was created for. + deploymentName := "serval" + var r int32 = 0 + _, err := c.AppsV1().Deployments("calico-system").Create(ctx, &appsv1.Deployment{ + TypeMeta: metav1.TypeMeta{Kind: "Deployment", APIVersion: "apps/v1"}, + ObjectMeta: metav1.ObjectMeta{Name: deploymentName, Namespace: "calico-system"}, + Spec: appsv1.DeploymentSpec{Replicas: &r}, + }, metav1.CreateOptions{}) + Expect(err).NotTo(HaveOccurred()) - ta = newTyphaAutoscaler(c, nodeIndexInformer, tlw, statusManager) - ta.start(cancelledCtx) + CreateNode(c, "node1", map[string]string{"kubernetes.io/os": "linux"}, nil) + CreateNode(c, "node2", map[string]string{"kubernetes.io/os": "linux"}, nil) + CreateNode(c, "node3", map[string]string{"kubernetes.io/os": "linux"}, nil) + + ta = New(c, nodeIndexInformer, statusManager, []string{deploymentName}, OptionSyncPeriod(10*time.Millisecond)) + ta.Start(ctx) - // The goroutine should observe the cancelled context and exit without autoscaling. - ta.waitForShutdown() - statusManager.AssertNotCalled(GinkgoT(), "SetDegraded", mock.Anything, mock.Anything, mock.Anything, mock.Anything) + // Three nodes scale to two replicas, applied to the named deployment. + verifyReplicas(c, deploymentName, 2) }) }) func verifyTyphaReplicas(c kubernetes.Interface, expectedReplicas int) { + verifyReplicas(c, "calico-typha", expectedReplicas) +} + +func verifyReplicas(c kubernetes.Interface, deploymentName string, expectedReplicas int) { EventuallyWithOffset(1, func() int32 { - typha, err := c.AppsV1().Deployments("calico-system").Get(context.Background(), "calico-typha", metav1.GetOptions{}) + typha, err := c.AppsV1().Deployments("calico-system").Get(context.Background(), deploymentName, metav1.GetOptions{}) Expect(err).NotTo(HaveOccurred()) // Just return an invalid number that will never match an expected replica count. if typha.Spec.Replicas == nil { diff --git a/pkg/render/common/networkpolicy/networkpolicy.go b/pkg/render/common/networkpolicy/networkpolicy.go index b1c18c5587..c5775c2f04 100644 --- a/pkg/render/common/networkpolicy/networkpolicy.go +++ b/pkg/render/common/networkpolicy/networkpolicy.go @@ -89,6 +89,10 @@ func CreateEntityRule(namespace string, deploymentName string, ports ...uint16) } } +// ServalSourceEntityRule matches traffic from the serval gateway, the +// non-cluster host entrypoint. +var ServalSourceEntityRule = CreateSourceEntityRule("calico-system", "serval") + // CreateSourceEntityRule creates a conventional entity rule that matches ingress traffic based on namespace and deployment name. func CreateSourceEntityRule(namespace string, deploymentName string) v3.EntityRule { return v3.EntityRule{ diff --git a/pkg/render/logcollector/fluentbit_test.go b/pkg/render/logcollector/fluentbit_test.go index bc42622f5d..d51fd8db5a 100644 --- a/pkg/render/logcollector/fluentbit_test.go +++ b/pkg/render/logcollector/fluentbit_test.go @@ -1249,6 +1249,7 @@ var _ = Describe("Tigera Secure Fluent Bit rendering tests", func() { Endpoint: "https://1.2.3.4:5678", }, } + cfg.NonClusterHostLogIngestion = true cfg.LogCollector.Spec.AdditionalStores = additionalStoreSpecAllHosts expectedResources = append(expectedResources, &corev1.Service{ObjectMeta: metav1.ObjectMeta{Name: render.FluentBitInputService, Namespace: render.LogCollectorNamespace}, TypeMeta: metav1.TypeMeta{Kind: "Service", APIVersion: "v1"}}) @@ -1340,13 +1341,15 @@ var _ = Describe("Tigera Secure Fluent Bit rendering tests", func() { Endpoint: "https://1.2.3.4:5678", }, } + cfg.NonClusterHostLogIngestion = true resourcesWithNonClusterHosts, _ := logcollector.FluentBitShared(cfg).Objects() policyWithNonClusterHosts := testutils.GetCalicoSystemPolicyFromResources(policyName, resourcesWithNonClusterHosts) - // Validate that we have a single ingress rule added for the fluent-bit service. + // Validate the ingress rules added for the fluent-bit input: one + // for voltron and one for the Serval gateway. Expect(policyWithoutNonClusterHosts.Spec.Egress).To(Equal(policyWithNonClusterHosts.Spec.Egress)) - Expect(len(policyWithoutNonClusterHosts.Spec.Ingress)).To(Equal(len(policyWithNonClusterHosts.Spec.Ingress) - 1)) - Expect(len(policyWithNonClusterHosts.Spec.Ingress)).To(Equal(2)) + Expect(len(policyWithoutNonClusterHosts.Spec.Ingress)).To(Equal(len(policyWithNonClusterHosts.Spec.Ingress) - 2)) + Expect(len(policyWithNonClusterHosts.Spec.Ingress)).To(Equal(3)) Expect(policyWithNonClusterHosts.Spec.Ingress[1]).To(Equal(v3.Rule{ Action: v3.Allow, Protocol: &networkpolicy.TCPProtocol, @@ -1358,6 +1361,14 @@ var _ = Describe("Tigera Secure Fluent Bit rendering tests", func() { Ports: networkpolicy.Ports(logcollector.FluentBitInputPort), }, })) + Expect(policyWithNonClusterHosts.Spec.Ingress[2]).To(Equal(v3.Rule{ + Action: v3.Allow, + Protocol: &networkpolicy.TCPProtocol, + Source: networkpolicy.ServalSourceEntityRule, + Destination: v3.EntityRule{ + Ports: networkpolicy.Ports(logcollector.FluentBitInputPort), + }, + })) }) }) diff --git a/pkg/render/logcollector/logcollector.go b/pkg/render/logcollector/logcollector.go index 0dd9064c1e..09690e3a29 100644 --- a/pkg/render/logcollector/logcollector.go +++ b/pkg/render/logcollector/logcollector.go @@ -169,6 +169,13 @@ type FluentBitConfiguration struct { NonClusterHost *operatorv1.NonClusterHost + // NonClusterHostLogIngestion enables the in-cluster fluent-bit HTTP input + // and the non_cluster_* pipeline that receive host logs. It is true while + // either a NonClusterHost (legacy voltron relay) or a Serval resource (the + // gateway relays to the same input) exists, so the input survives deleting + // one CR while the other is still in use. + NonClusterHostLogIngestion bool + // LicenseExpired indicates the license has expired and fluent-bit DaemonSet should be removed. LicenseExpired bool } @@ -235,11 +242,12 @@ func (c *fluentBitComponent) Objects() ([]client.Object, []client.Object) { } if c.osType == rmeta.OSTypeLinux { - if c.cfg.NonClusterHost != nil { + if c.cfg.NonClusterHostLogIngestion { objs = append(objs, c.nonClusterHostInputService()) } else { - // Clean up the input service when the NonClusterHost resource is - // removed; the rendered config drops the http input at the same time. + // Clean up the input service when neither a NonClusterHost nor a + // Serval resource exists; the rendered config drops the http input at + // the same time. toDelete = append(toDelete, c.nonClusterHostInputService()) } } diff --git a/pkg/render/logcollector/networkpolicy.go b/pkg/render/logcollector/networkpolicy.go index a4ea51b83a..27dc3f5993 100644 --- a/pkg/render/logcollector/networkpolicy.go +++ b/pkg/render/logcollector/networkpolicy.go @@ -193,7 +193,7 @@ func (c *fluentBitComponent) calicoSystemPolicy() *v3.NetworkPolicy { }, } - if c.cfg.NonClusterHost != nil { + if c.cfg.NonClusterHostLogIngestion { ingressRules = append(ingressRules, v3.Rule{ Action: v3.Allow, Protocol: &networkpolicy.TCPProtocol, @@ -202,6 +202,15 @@ func (c *fluentBitComponent) calicoSystemPolicy() *v3.NetworkPolicy { Ports: networkpolicy.Ports(FluentBitInputPort), }, }) + // The Serval gateway relays non-cluster host logs to the same input. + ingressRules = append(ingressRules, v3.Rule{ + Action: v3.Allow, + Protocol: &networkpolicy.TCPProtocol, + Source: networkpolicy.ServalSourceEntityRule, + Destination: v3.EntityRule{ + Ports: networkpolicy.Ports(FluentBitInputPort), + }, + }) } return &v3.NetworkPolicy{ diff --git a/pkg/render/logcollector/pipeline.go b/pkg/render/logcollector/pipeline.go index a179213969..4ec3137f50 100644 --- a/pkg/render/logcollector/pipeline.go +++ b/pkg/render/logcollector/pipeline.go @@ -112,7 +112,7 @@ func (c *fluentBitComponent) addInputs(cfg *fluentBitConfig) { // The ingress path for non-cluster-host log forwarding: voltron relays the // hosts' posts to this input, and in_http derives the non_cluster_* tags // from the request paths. - if c.cfg.NonClusterHost != nil && c.osType == rmeta.OSTypeLinux { + if c.cfg.NonClusterHostLogIngestion && c.osType == rmeta.OSTypeLinux { cfg.Pipeline.Inputs = append(cfg.Pipeline.Inputs, map[string]interface{}{ "name": "http", "listen": "0.0.0.0", @@ -257,7 +257,7 @@ func (c *fluentBitComponent) linseedTags() []string { } tags = append(tags, in.tag) } - if c.cfg.NonClusterHost != nil && c.osType == rmeta.OSTypeLinux { + if c.cfg.NonClusterHostLogIngestion && c.osType == rmeta.OSTypeLinux { tags = append(tags, "non_cluster_flows", "non_cluster_dns", "non_cluster_policy_activity") } return tags diff --git a/pkg/render/logcollector/rendered_config_test.go b/pkg/render/logcollector/rendered_config_test.go index 75e2b58cbb..bdc37120bc 100644 --- a/pkg/render/logcollector/rendered_config_test.go +++ b/pkg/render/logcollector/rendered_config_test.go @@ -115,6 +115,7 @@ func TestRenderedConfigGoldens(t *testing.T) { ObjectMeta: metav1.ObjectMeta{Name: "tigera-secure"}, Spec: operatorv1.NonClusterHostSpec{Endpoint: "https://1.2.3.4:5678"}, } + cfg.NonClusterHostLogIngestion = true }, }, { diff --git a/pkg/render/nonclusterhost/nonclusterhost.go b/pkg/render/nonclusterhost/nonclusterhost.go index 7e0eb19a1b..a0fdc118c5 100644 --- a/pkg/render/nonclusterhost/nonclusterhost.go +++ b/pkg/render/nonclusterhost/nonclusterhost.go @@ -45,6 +45,12 @@ type nonClusterHostComponent struct { cfg *Config } +// name is the object name for the rendered host scaffolding: the single +// tigera-noncluster-host identity that non-cluster hosts authenticate as. +func (c *nonClusterHostComponent) name() string { + return NonClusterHostObjectName +} + func (c *nonClusterHostComponent) ResolveImages(is *operatorv1.ImageSet) error { return nil } @@ -74,7 +80,7 @@ func (c *nonClusterHostComponent) serviceAccount() *corev1.ServiceAccount { APIVersion: "v1", }, ObjectMeta: metav1.ObjectMeta{ - Name: NonClusterHostObjectName, + Name: c.name(), Namespace: common.CalicoNamespace, }, } @@ -87,11 +93,11 @@ func (c *nonClusterHostComponent) tokenSecret() *corev1.Secret { APIVersion: "v1", }, ObjectMeta: metav1.ObjectMeta{ - Name: NonClusterHostObjectName, + Name: c.name(), Namespace: common.CalicoNamespace, // The annotation below will result in the auto-creation of spec.data.token. Annotations: map[string]string{ - "kubernetes.io/service-account.name": NonClusterHostObjectName, + "kubernetes.io/service-account.name": c.name(), }, }, Type: "kubernetes.io/service-account-token", @@ -186,7 +192,9 @@ func (c *nonClusterHostComponent) clusterRole() *rbacv1.ClusterRole { Verbs: []string{"create"}, }, { - // Used to read endpoint field from the NonClusterHost resource. + // Used to read the endpoint and typhaEndpoint fields from the + // NonClusterHost resource: an unset typhaEndpoint tells the host to + // reach Typha through the gateway tunnel. APIGroups: []string{"operator.tigera.io"}, Resources: []string{"nonclusterhosts"}, Verbs: []string{"get", "list", "watch"}, @@ -237,7 +245,7 @@ func (c *nonClusterHostComponent) clusterRole() *rbacv1.ClusterRole { return &rbacv1.ClusterRole{ TypeMeta: metav1.TypeMeta{Kind: "ClusterRole", APIVersion: "rbac.authorization.k8s.io/v1"}, ObjectMeta: metav1.ObjectMeta{ - Name: NonClusterHostObjectName, + Name: c.name(), }, Rules: rules, } @@ -247,17 +255,17 @@ func (c *nonClusterHostComponent) clusterRoleBinding() *rbacv1.ClusterRoleBindin return &rbacv1.ClusterRoleBinding{ TypeMeta: metav1.TypeMeta{Kind: "ClusterRoleBinding", APIVersion: "rbac.authorization.k8s.io/v1"}, ObjectMeta: metav1.ObjectMeta{ - Name: NonClusterHostObjectName, + Name: c.name(), }, RoleRef: rbacv1.RoleRef{ APIGroup: "rbac.authorization.k8s.io", Kind: "ClusterRole", - Name: NonClusterHostObjectName, + Name: c.name(), }, Subjects: []rbacv1.Subject{ { Kind: "ServiceAccount", - Name: NonClusterHostObjectName, + Name: c.name(), Namespace: common.CalicoNamespace, }, }, diff --git a/pkg/render/serval/component.go b/pkg/render/serval/component.go new file mode 100644 index 0000000000..4cc9fb4b9d --- /dev/null +++ b/pkg/render/serval/component.go @@ -0,0 +1,424 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package serval renders Serval: the single layer-7 entrypoint +// through which non-cluster hosts reach the cluster. It serves the +// kube-apiserver proxy, log ingestion, and the felix-to-typha WebSocket +// tunnel on one HTTPS endpoint. +package serval + +import ( + "fmt" + + v3 "github.com/tigera/api/pkg/apis/projectcalico/v3" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + + operatorv1 "github.com/tigera/operator/api/v1" + "github.com/tigera/operator/pkg/common" + "github.com/tigera/operator/pkg/components" + "github.com/tigera/operator/pkg/controller/k8sapi" + "github.com/tigera/operator/pkg/render" + "github.com/tigera/operator/pkg/render/common/meta" + "github.com/tigera/operator/pkg/render/common/networkpolicy" + "github.com/tigera/operator/pkg/render/common/podaffinity" + "github.com/tigera/operator/pkg/render/common/secret" + "github.com/tigera/operator/pkg/render/common/securitycontext" + "github.com/tigera/operator/pkg/render/logcollector" + "github.com/tigera/operator/pkg/tls/certificatemanagement" +) + +const ( + ServalName = common.ServalName + ServalNamespace = common.CalicoNamespace + ServalServiceAccountName = ServalName + ServalDeploymentName = ServalName + ServalServiceName = ServalName + ServalPolicyName = networkpolicy.CalicoComponentPolicyPrefix + "serval" + ServalKeyPairSecret = "serval-key-pair" + ServalTyphaClientKeyPairSecret = "serval-typha-client-certs" + + ServalServicePortName = common.ServalServicePortName + ServalServicePort = 443 + ServalTargetPort = 8449 + ServalHealthPort = 8080 +) + +func Serval(cfg *Configuration) render.Component { + return &Component{cfg: cfg} +} + +// Configuration contains all the config information needed to render the component. +type Configuration struct { + PullSecrets []*corev1.Secret + OpenShift bool + Installation *operatorv1.InstallationSpec + TrustedCertBundle certificatemanagement.TrustedBundleRO + ServerKeyPair certificatemanagement.KeyPairInterface + ClusterDomain string + + // TyphaClientKeyPair is the client keypair Serval presents to the in-cluster + // calico-typha. Its common name (typha-client) is one calico-typha accepts, + // so Serval relays felix connections to it as an ordinary Typha client. + TyphaClientKeyPair certificatemanagement.KeyPairInterface + + // K8sServiceEp is the Kubernetes API endpoint override (from the + // kubernetes-services-endpoint ConfigMap). Serval's own client needs it to + // reach the apiserver on clusters without kube-proxy (e.g. eBPF) or with a + // custom apiserver address. + K8sServiceEp k8sapi.ServiceEndpoint + + // Deleted renders the component for teardown: Objects returns serval's + // objects in the delete list so a NonClusterHost switched to the legacy + // (typhaEndpoint set) mode garbage-collects serval. + Deleted bool +} + +type Component struct { + cfg *Configuration + + calicoImage string +} + +func (c *Component) ResolveImages(is *operatorv1.ImageSet) error { + // Teardown renders object references only; no image is needed. + if c.cfg.Deleted { + return nil + } + reg := c.cfg.Installation.Registry + path := c.cfg.Installation.ImagePath + prefix := c.cfg.Installation.ImagePrefix + + var err error + c.calicoImage, err = components.GetReference(components.CombinedCalicoImage(c.cfg.Installation), reg, path, prefix, is) + return err +} + +func (c *Component) SupportedOSType() meta.OSType { + return meta.OSTypeLinux +} + +func (c *Component) Ready() bool { + return true +} + +func (c *Component) Objects() ([]client.Object, []client.Object) { + if c.cfg.Deleted { + return nil, c.objectsForDeletion() + } + objs := []client.Object{ + c.serviceAccount(), + c.clusterRole(), + c.clusterRoleBinding(), + c.service(), + c.deployment(), + c.networkPolicy(), + } + objs = append(objs, secret.ToRuntimeObjects(secret.CopyToNamespace(ServalNamespace, c.cfg.PullSecrets...)...)...) + return objs, nil +} + +// objectsForDeletion lists serval's objects by reference (no spec), for the +// teardown path when a NonClusterHost switches to the legacy Typha endpoint. +func (c *Component) objectsForDeletion() []client.Object { + return []client.Object{ + &corev1.ServiceAccount{TypeMeta: metav1.TypeMeta{Kind: "ServiceAccount", APIVersion: "v1"}, ObjectMeta: metav1.ObjectMeta{Name: ServalServiceAccountName, Namespace: ServalNamespace}}, + &rbacv1.ClusterRole{TypeMeta: metav1.TypeMeta{Kind: "ClusterRole", APIVersion: "rbac.authorization.k8s.io/v1"}, ObjectMeta: metav1.ObjectMeta{Name: ServalName}}, + &rbacv1.ClusterRoleBinding{TypeMeta: metav1.TypeMeta{Kind: "ClusterRoleBinding", APIVersion: "rbac.authorization.k8s.io/v1"}, ObjectMeta: metav1.ObjectMeta{Name: ServalName}}, + &corev1.Service{TypeMeta: metav1.TypeMeta{Kind: "Service", APIVersion: "v1"}, ObjectMeta: metav1.ObjectMeta{Name: ServalServiceName, Namespace: ServalNamespace}}, + &appsv1.Deployment{TypeMeta: metav1.TypeMeta{Kind: "Deployment", APIVersion: "apps/v1"}, ObjectMeta: metav1.ObjectMeta{Name: ServalDeploymentName, Namespace: ServalNamespace}}, + &v3.NetworkPolicy{TypeMeta: metav1.TypeMeta{Kind: "NetworkPolicy", APIVersion: "projectcalico.org/v3"}, ObjectMeta: metav1.ObjectMeta{Name: ServalPolicyName, Namespace: ServalNamespace}}, + // The operator-namespace copies are the source of truth for these keypairs, + // so they outlive the Deployment unless they go too. + &corev1.Secret{TypeMeta: metav1.TypeMeta{Kind: "Secret", APIVersion: "v1"}, ObjectMeta: metav1.ObjectMeta{Name: ServalKeyPairSecret, Namespace: common.OperatorNamespace()}}, + &corev1.Secret{TypeMeta: metav1.TypeMeta{Kind: "Secret", APIVersion: "v1"}, ObjectMeta: metav1.ObjectMeta{Name: ServalTyphaClientKeyPairSecret, Namespace: common.OperatorNamespace()}}, + &corev1.Secret{TypeMeta: metav1.TypeMeta{Kind: "Secret", APIVersion: "v1"}, ObjectMeta: metav1.ObjectMeta{Name: ServalKeyPairSecret, Namespace: ServalNamespace}}, + &corev1.Secret{TypeMeta: metav1.TypeMeta{Kind: "Secret", APIVersion: "v1"}, ObjectMeta: metav1.ObjectMeta{Name: ServalTyphaClientKeyPairSecret, Namespace: ServalNamespace}}, + } +} + +func (c *Component) serviceAccount() *corev1.ServiceAccount { + return &corev1.ServiceAccount{ + TypeMeta: metav1.TypeMeta{Kind: "ServiceAccount", APIVersion: "v1"}, + ObjectMeta: metav1.ObjectMeta{Name: ServalServiceAccountName, Namespace: ServalNamespace}, + } +} + +func (c *Component) clusterRole() *rbacv1.ClusterRole { + return &rbacv1.ClusterRole{ + TypeMeta: metav1.TypeMeta{Kind: "ClusterRole", APIVersion: "rbac.authorization.k8s.io/v1"}, + ObjectMeta: metav1.ObjectMeta{Name: ServalName}, + Rules: []rbacv1.PolicyRule{ + { + // The apiserver proxy forwards requests as the authenticated + // host user via impersonation headers. + APIGroups: []string{""}, + Resources: []string{"users", "groups", "serviceaccounts"}, + Verbs: []string{"impersonate"}, + }, + { + // Kubernetes service account bearer tokens are authenticated + // with TokenReviews. + APIGroups: []string{"authentication.k8s.io"}, + Resources: []string{"tokenreviews"}, + Verbs: []string{"create"}, + }, + { + // Log ingestion and Typha tunnel requests are authorized + // against the host's RBAC with SubjectAccessReviews. + APIGroups: []string{"authorization.k8s.io"}, + Resources: []string{"subjectaccessreviews"}, + Verbs: []string{"create"}, + }, + { + // The Tigera-JWT authenticator verifies that a token's + // subject service account exists. + APIGroups: []string{""}, + Resources: []string{"serviceaccounts"}, + Verbs: []string{"get"}, + }, + }, + } +} + +func (c *Component) clusterRoleBinding() *rbacv1.ClusterRoleBinding { + return &rbacv1.ClusterRoleBinding{ + TypeMeta: metav1.TypeMeta{Kind: "ClusterRoleBinding", APIVersion: "rbac.authorization.k8s.io/v1"}, + ObjectMeta: metav1.ObjectMeta{Name: ServalName}, + RoleRef: rbacv1.RoleRef{ + APIGroup: "rbac.authorization.k8s.io", + Kind: "ClusterRole", + Name: ServalName, + }, + Subjects: []rbacv1.Subject{ + { + Kind: "ServiceAccount", + Name: ServalServiceAccountName, + Namespace: ServalNamespace, + }, + }, + } +} + +// service is the in-cluster target for the customer's external load balancer +// or ingress; the operator does not provision the external exposure itself. +func (c *Component) service() *corev1.Service { + return &corev1.Service{ + TypeMeta: metav1.TypeMeta{Kind: "Service", APIVersion: "v1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: ServalServiceName, + Namespace: ServalNamespace, + }, + Spec: corev1.ServiceSpec{ + Selector: map[string]string{"k8s-app": ServalDeploymentName}, + Ports: []corev1.ServicePort{ + { + Name: ServalServicePortName, + Port: ServalServicePort, + TargetPort: intstr.FromInt32(ServalTargetPort), + Protocol: corev1.ProtocolTCP, + }, + }, + }, + } +} + +func (c *Component) ingestionEndpoint() string { + return fmt.Sprintf("https://%s.%s.svc.%s:%d", + render.FluentBitInputService, common.CalicoNamespace, c.cfg.ClusterDomain, logcollector.FluentBitInputPort) +} + +// typhaEndpoint is the in-cluster calico-typha address Serval relays felix +// tunnels to. +func (c *Component) typhaEndpoint() string { + return fmt.Sprintf("%s.%s.svc.%s:%d", + render.TyphaServiceName, common.CalicoNamespace, c.cfg.ClusterDomain, render.TyphaPort) +} + +// typhaRelayEnvVars configures the Typha client connection Serval relays felix +// tunnels onto: the in-cluster calico-typha address, the CA that verifies its +// server certificate, the required server CN, and Serval's own client keypair. +func (c *Component) typhaRelayEnvVars() []corev1.EnvVar { + return []corev1.EnvVar{ + {Name: "SERVAL_TYPHA_ENDPOINT", Value: c.typhaEndpoint()}, + {Name: "SERVAL_TYPHA_SERVER_NAME", Value: render.TyphaCommonName}, + {Name: "SERVAL_TYPHA_CA_BUNDLE_PATH", Value: c.cfg.TrustedCertBundle.MountPath()}, + {Name: "SERVAL_TYPHA_CLIENT_CERT_PATH", Value: c.cfg.TyphaClientKeyPair.VolumeMountCertificateFilePath()}, + {Name: "SERVAL_TYPHA_CLIENT_KEY_PATH", Value: c.cfg.TyphaClientKeyPair.VolumeMountKeyFilePath()}, + } +} + +func (c *Component) container() corev1.Container { + env := []corev1.EnvVar{ + {Name: "SERVAL_LOG_LEVEL", Value: "Info"}, + {Name: "SERVAL_PORT", Value: fmt.Sprintf("%d", ServalTargetPort)}, + {Name: "SERVAL_SERVER_CERT_PATH", Value: c.cfg.ServerKeyPair.VolumeMountCertificateFilePath()}, + {Name: "SERVAL_SERVER_KEY_PATH", Value: c.cfg.ServerKeyPair.VolumeMountKeyFilePath()}, + {Name: "SERVAL_INGESTION_ENDPOINT", Value: c.ingestionEndpoint()}, + {Name: "SERVAL_INGESTION_CA_BUNDLE_PATH", Value: c.cfg.TrustedCertBundle.MountPath()}, + // The log collector input requires an mTLS client certificate signed + // by the cluster CA; the server keypair doubles as it. + {Name: "SERVAL_INGESTION_CLIENT_CERT_PATH", Value: c.cfg.ServerKeyPair.VolumeMountCertificateFilePath()}, + {Name: "SERVAL_INGESTION_CLIENT_KEY_PATH", Value: c.cfg.ServerKeyPair.VolumeMountKeyFilePath()}, + {Name: "SERVAL_TIGERA_ISSUER_CA_BUNDLE_PATH", Value: c.cfg.TrustedCertBundle.MountPath()}, + {Name: "SERVAL_HEALTH_PORT", Value: fmt.Sprintf("%d", ServalHealthPort)}, + } + // The tunnel relays each felix connection to the in-cluster calico-typha. + env = append(env, c.typhaRelayEnvVars()...) + // KUBERNETES_SERVICE_HOST/PORT overrides so Serval's own client reaches the + // apiserver on clusters without kube-proxy or with a custom apiserver address. + env = append(env, c.cfg.K8sServiceEp.EnvVars()...) + + volumeMounts := []corev1.VolumeMount{ + c.cfg.ServerKeyPair.VolumeMount(c.SupportedOSType()), + c.cfg.TyphaClientKeyPair.VolumeMount(c.SupportedOSType()), + } + volumeMounts = append(volumeMounts, c.cfg.TrustedCertBundle.VolumeMounts(c.SupportedOSType())...) + + return corev1.Container{ + Name: ServalName, + Image: c.calicoImage, + Command: []string{components.CalicoBinaryPath, "component", "serval"}, + Env: env, + SecurityContext: securitycontext.NewNonRootContext(), + ReadinessProbe: &corev1.Probe{ + ProbeHandler: corev1.ProbeHandler{HTTPGet: &corev1.HTTPGetAction{ + Path: "/readiness", + Port: intstr.FromInt32(ServalHealthPort), + }}, + PeriodSeconds: 10, + }, + LivenessProbe: &corev1.Probe{ + ProbeHandler: corev1.ProbeHandler{HTTPGet: &corev1.HTTPGetAction{ + Path: "/liveness", + Port: intstr.FromInt32(ServalHealthPort), + }}, + PeriodSeconds: 10, + }, + VolumeMounts: volumeMounts, + } +} + +func (c *Component) deployment() *appsv1.Deployment { + tolerations := append(c.cfg.Installation.ControlPlaneTolerations, meta.TolerateCriticalAddonsAndControlPlane...) + if c.cfg.Installation.KubernetesProvider.IsGKE() { + tolerations = append(tolerations, meta.TolerateGKEARM64NoSchedule) + } + + d := &appsv1.Deployment{ + TypeMeta: metav1.TypeMeta{Kind: "Deployment", APIVersion: "apps/v1"}, + ObjectMeta: metav1.ObjectMeta{ + Name: ServalDeploymentName, + Namespace: ServalNamespace, + Annotations: map[string]string{ + c.cfg.ServerKeyPair.HashAnnotationKey(): c.cfg.ServerKeyPair.HashAnnotationValue(), + c.cfg.TyphaClientKeyPair.HashAnnotationKey(): c.cfg.TyphaClientKeyPair.HashAnnotationValue(), + }, + }, + Spec: appsv1.DeploymentSpec{ + // Replicas are owned by the HostEndpoint-count autoscaler: Serval's + // in-process Typha fans out to the registered hosts, so it scales with + // their number. The render deliberately leaves the count unset. + Template: corev1.PodTemplateSpec{ + ObjectMeta: metav1.ObjectMeta{ + Name: ServalDeploymentName, + Labels: map[string]string{ + "k8s-app": ServalDeploymentName, + }, + }, + Spec: corev1.PodSpec{ + // The "serval" Service would otherwise inject service-link + // variables such as SERVAL_PORT=tcp://..., which collide + // with the SERVAL_ envconfig prefix. + EnableServiceLinks: ptr.To(false), + NodeSelector: c.cfg.Installation.ControlPlaneNodeSelector, + ServiceAccountName: ServalServiceAccountName, + Tolerations: tolerations, + ImagePullSecrets: secret.GetReferenceList(c.cfg.PullSecrets), + Containers: []corev1.Container{c.container()}, + Volumes: []corev1.Volume{ + c.cfg.ServerKeyPair.Volume(), + c.cfg.TyphaClientKeyPair.Volume(), + c.cfg.TrustedCertBundle.Volume(), + }, + }, + }, + }, + } + + // The autoscaler may run more than one replica, so spread them across nodes. + d.Spec.Template.Spec.Affinity = podaffinity.NewPodAntiAffinity(ServalDeploymentName, []string{ServalNamespace}) + + return d +} + +func (c *Component) networkPolicy() *v3.NetworkPolicy { + // Hosts connect from outside the cluster, so ingress is unrestricted by + // source; every request is authenticated at the application layer. + ingressRules := []v3.Rule{ + { + Action: v3.Allow, + Protocol: &networkpolicy.TCPProtocol, + Destination: v3.EntityRule{ + Ports: networkpolicy.Ports(ServalTargetPort), + }, + }, + } + + egressRules := networkpolicy.AppendDNSEgressRules([]v3.Rule{}, c.cfg.OpenShift) + egressRules = append(egressRules, + // Serval's own client (token review, impersonation) and the tunnel relay + // reach the apiserver and the in-cluster Typha respectively. + v3.Rule{ + Action: v3.Allow, + Protocol: &networkpolicy.TCPProtocol, + Destination: networkpolicy.KubeAPIServerEntityRule, + }, + v3.Rule{ + // Serval relays to calico-typha through its ClusterIP, so match the + // Service (a plain endpoint selector does not match ClusterIP-DNAT'd + // traffic). A service selector implies the Service's ports, so Ports + // must not be set alongside it. + Action: v3.Allow, + Protocol: &networkpolicy.TCPProtocol, + Destination: networkpolicy.CreateServiceSelectorEntityRule(common.CalicoNamespace, render.TyphaServiceName), + }, + v3.Rule{ + Action: v3.Allow, + Protocol: &networkpolicy.TCPProtocol, + Destination: v3.EntityRule{ + Selector: networkpolicy.KubernetesAppSelector(logcollector.FluentBitNodeName), + Ports: networkpolicy.Ports(logcollector.FluentBitInputPort), + }, + }, + ) + + return &v3.NetworkPolicy{ + TypeMeta: metav1.TypeMeta{Kind: "NetworkPolicy", APIVersion: "projectcalico.org/v3"}, + ObjectMeta: metav1.ObjectMeta{Name: ServalPolicyName, Namespace: ServalNamespace}, + Spec: v3.NetworkPolicySpec{ + Order: &networkpolicy.HighPrecedenceOrder, + Tier: networkpolicy.CalicoTierName, + Selector: networkpolicy.KubernetesAppSelector(ServalDeploymentName), + Types: []v3.PolicyType{v3.PolicyTypeIngress, v3.PolicyTypeEgress}, + Ingress: ingressRules, + Egress: egressRules, + }, + } +} diff --git a/pkg/render/serval/component_test.go b/pkg/render/serval/component_test.go new file mode 100644 index 0000000000..94e3dd48e0 --- /dev/null +++ b/pkg/render/serval/component_test.go @@ -0,0 +1,177 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package serval_test + +import ( + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + v3 "github.com/tigera/api/pkg/apis/projectcalico/v3" + appsv1 "k8s.io/api/apps/v1" + corev1 "k8s.io/api/core/v1" + rbacv1 "k8s.io/api/rbac/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + + operatorv1 "github.com/tigera/operator/api/v1" + "github.com/tigera/operator/pkg/dns" + "github.com/tigera/operator/pkg/render/common/networkpolicy" + "github.com/tigera/operator/pkg/render/common/podaffinity" + rtest "github.com/tigera/operator/pkg/render/common/test" + "github.com/tigera/operator/pkg/render/serval" + "github.com/tigera/operator/pkg/tls/certificatemanagement" +) + +var _ = Describe("Serval rendering tests", func() { + var cfg *serval.Configuration + + BeforeEach(func() { + cfg = &serval.Configuration{ + Installation: &operatorv1.InstallationSpec{ + Variant: operatorv1.CalicoEnterprise, + }, + TrustedCertBundle: certificatemanagement.CreateTrustedBundle(nil), + ServerKeyPair: certificatemanagement.NewKeyPair(&corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: serval.ServalKeyPairSecret}}, nil, ""), + // The client keypair Serval presents to the in-cluster calico-typha. + TyphaClientKeyPair: certificatemanagement.NewKeyPair(&corev1.Secret{ObjectMeta: metav1.ObjectMeta{Name: serval.ServalTyphaClientKeyPairSecret}}, nil, ""), + ClusterDomain: dns.DefaultClusterDomain, + } + }) + + renderObjects := func() []client.Object { + component := serval.Serval(cfg) + Expect(component.ResolveImages(nil)).To(Succeed()) + toCreate, toDelete := component.Objects() + Expect(toDelete).To(BeEmpty()) + return toCreate + } + + envMap := func(env []corev1.EnvVar) map[string]string { + m := map[string]string{} + for _, e := range env { + m[e.Name] = e.Value + } + return m + } + + It("should render the expected objects", func() { + expectedResources := []struct { + name string + ns string + group string + version string + kind string + }{ + {name: "serval", ns: "calico-system", group: "", version: "v1", kind: "ServiceAccount"}, + {name: "serval", ns: "", group: "rbac.authorization.k8s.io", version: "v1", kind: "ClusterRole"}, + {name: "serval", ns: "", group: "rbac.authorization.k8s.io", version: "v1", kind: "ClusterRoleBinding"}, + {name: "serval", ns: "calico-system", group: "", version: "v1", kind: "Service"}, + {name: "serval", ns: "calico-system", group: "apps", version: "v1", kind: "Deployment"}, + {name: serval.ServalPolicyName, ns: "calico-system", group: "projectcalico.org", version: "v3", kind: "NetworkPolicy"}, + } + + toCreate := renderObjects() + Expect(toCreate).To(HaveLen(len(expectedResources))) + for i, expectedRes := range expectedResources { + rtest.ExpectResourceTypeAndObjectMetadata(toCreate[i], expectedRes.name, expectedRes.ns, expectedRes.group, expectedRes.version, expectedRes.kind) + } + }) + + It("should grant the RBAC Serval's auth stack needs", func() { + toCreate := renderObjects() + + clusterRole := rtest.GetResource(toCreate, "serval", "", "rbac.authorization.k8s.io", "v1", "ClusterRole").(*rbacv1.ClusterRole) + Expect(clusterRole.Rules).To(ContainElements( + rbacv1.PolicyRule{ + APIGroups: []string{""}, + Resources: []string{"users", "groups", "serviceaccounts"}, + Verbs: []string{"impersonate"}, + }, + rbacv1.PolicyRule{ + APIGroups: []string{"authentication.k8s.io"}, + Resources: []string{"tokenreviews"}, + Verbs: []string{"create"}, + }, + rbacv1.PolicyRule{ + APIGroups: []string{"authorization.k8s.io"}, + Resources: []string{"subjectaccessreviews"}, + Verbs: []string{"create"}, + }, + rbacv1.PolicyRule{ + APIGroups: []string{""}, + Resources: []string{"serviceaccounts"}, + Verbs: []string{"get"}, + }, + )) + }) + + It("should leave replicas to the HostEndpoint-count autoscaler", func() { + cfg.Installation.ControlPlaneReplicas = ptr.To(int32(3)) + toCreate := renderObjects() + deployment := rtest.GetResource(toCreate, "serval", "calico-system", "apps", "v1", "Deployment").(*appsv1.Deployment) + // Serval's in-process Typha fans out to the registered hosts, so the render + // leaves the replica count to the autoscaler rather than following + // controlPlaneReplicas. + Expect(deployment.Spec.Replicas).To(BeNil()) + // Pod anti-affinity is always set so autoscaled replicas spread across nodes. + Expect(deployment.Spec.Template.Spec.Affinity).To(Equal(podaffinity.NewPodAntiAffinity("serval", []string{"calico-system"}))) + }) + + It("should point the tunnel relay at the in-cluster Typha", func() { + toCreate := renderObjects() + + deployment := rtest.GetResource(toCreate, "serval", "calico-system", "apps", "v1", "Deployment").(*appsv1.Deployment) + container := deployment.Spec.Template.Spec.Containers[0] + Expect(container.Command).To(Equal([]string{"/usr/bin/calico", "component", "serval"})) + + env := envMap(container.Env) + Expect(env["SERVAL_INGESTION_ENDPOINT"]).To(Equal("https://calico-fluent-bit-http-input.calico-system.svc.cluster.local:9880")) + Expect(env["SERVAL_PORT"]).To(Equal("8449")) + // The tunnel relays to the in-cluster calico-typha as an ordinary Typha client. + Expect(env["SERVAL_TYPHA_ENDPOINT"]).To(Equal("calico-typha.calico-system.svc.cluster.local:5473")) + Expect(env["SERVAL_TYPHA_SERVER_NAME"]).To(Equal("typha-server")) + Expect(env["SERVAL_TYPHA_CLIENT_CERT_PATH"]).NotTo(BeEmpty()) + Expect(env["SERVAL_TYPHA_CLIENT_KEY_PATH"]).NotTo(BeEmpty()) + Expect(env["SERVAL_TYPHA_CA_BUNDLE_PATH"]).NotTo(BeEmpty()) + // Serval no longer embeds Typha, so it carries no TYPHA_* server config. + Expect(env).NotTo(HaveKey("TYPHA_DATASTORETYPE")) + Expect(env).NotTo(HaveKey("TYPHA_SERVERCERTFILE")) + }) + + It("should allow ingress to the serval port and egress to the apiserver, Typha, and the log input", func() { + toCreate := renderObjects() + + policy := rtest.GetResource(toCreate, serval.ServalPolicyName, "calico-system", "projectcalico.org", "v3", "NetworkPolicy").(*v3.NetworkPolicy) + Expect(policy.Spec.Selector).To(Equal(networkpolicy.KubernetesAppSelector("serval"))) + + Expect(policy.Spec.Ingress).To(HaveLen(1)) + Expect(policy.Spec.Ingress[0].Destination.Ports).To(Equal(networkpolicy.Ports(8449))) + + var egressSelectors []string + var typhaServiceMatched bool + for _, rule := range policy.Spec.Egress { + egressSelectors = append(egressSelectors, rule.Destination.Selector) + // The relay reaches calico-typha through its ClusterIP, so the rule + // must match the Service, not a plain endpoint selector. + if svc := rule.Destination.Services; svc != nil && svc.Name == "calico-typha" && svc.Namespace == "calico-system" { + typhaServiceMatched = true + } + } + Expect(typhaServiceMatched).To(BeTrue(), "expected a service-based egress rule to calico-typha") + Expect(egressSelectors).To(ContainElement(networkpolicy.KubernetesAppSelector("calico-fluent-bit"))) + // No egress to the legacy directly-exposed non-cluster-host Typha. + Expect(egressSelectors).NotTo(ContainElement(networkpolicy.KubernetesAppSelector("calico-typha-noncluster-host"))) + }) +}) diff --git a/pkg/render/serval/suite_test.go b/pkg/render/serval/suite_test.go new file mode 100644 index 0000000000..304a41c99b --- /dev/null +++ b/pkg/render/serval/suite_test.go @@ -0,0 +1,29 @@ +// Copyright (c) 2026 Tigera, Inc. All rights reserved. + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package serval_test + +import ( + "testing" + + "github.com/onsi/ginkgo/v2" + "github.com/onsi/gomega" +) + +func TestRender(t *testing.T) { + gomega.RegisterFailHandler(ginkgo.Fail) + suiteConfig, reporterConfig := ginkgo.GinkgoConfiguration() + reporterConfig.JUnitReport = "../../../report/ut/serval_suite.xml" + ginkgo.RunSpecs(t, "pkg/render/serval Suite", suiteConfig, reporterConfig) +} diff --git a/pkg/render/typha.go b/pkg/render/typha.go index 264851785f..9341407bb2 100644 --- a/pkg/render/typha.go +++ b/pkg/render/typha.go @@ -132,7 +132,45 @@ func (c *typhaComponent) Objects() ([]client.Object, []client.Object) { objs = append(objs, c.typhaPrometheusService()) } - return objs, nil + var objsToDelete []client.Object + if c.cfg.NonClusterHost == nil { + // The non-cluster-host Typha renders only while a NonClusterHost + // resource exists; with Serval, Typha runs in-process instead. When the + // resource is absent, delete any leftover non-cluster-host Typha objects + // rather than orphaning them. + objsToDelete = append(objsToDelete, c.nonClusterHostObjectsForDeletion()...) + } + + return objs, objsToDelete +} + +// nonClusterHostObjectsForDeletion returns the non-cluster-host Typha objects as +// deletion stubs (name and namespace only), used when no NonClusterHost +// resource exists so the operator garbage-collects them instead of leaving them +// orphaned. The component's v3 NetworkPolicy is deliberately not among them: v3 +// resources are only reconcilable while the API server is healthy, so it is +// deleted by TyphaNonClusterHostPolicyForDeletion alongside the other v3 +// policies instead. +func (c *typhaComponent) nonClusterHostObjectsForDeletion() []client.Object { + return []client.Object{ + &appsv1.Deployment{ + TypeMeta: metav1.TypeMeta{Kind: "Deployment", APIVersion: "apps/v1"}, + ObjectMeta: metav1.ObjectMeta{Name: common.TyphaDeploymentName + TyphaNonClusterHostSuffix, Namespace: common.CalicoNamespace}, + }, + &corev1.Service{ + TypeMeta: metav1.TypeMeta{Kind: "Service", APIVersion: "v1"}, + ObjectMeta: metav1.ObjectMeta{Name: TyphaServiceName + TyphaNonClusterHostSuffix, Namespace: common.CalicoNamespace}, + }, + } +} + +// TyphaNonClusterHostPolicyForDeletion returns the non-cluster-host Typha policy +// as a deletion stub, for the modes that do not render it. +func TyphaNonClusterHostPolicyForDeletion() client.Object { + return &v3.NetworkPolicy{ + TypeMeta: metav1.TypeMeta{Kind: "NetworkPolicy", APIVersion: "projectcalico.org/v3"}, + ObjectMeta: metav1.ObjectMeta{Name: TyphaNonClusterHostNetworkPolicyName, Namespace: common.CalicoNamespace}, + } } func NewTyphaNonClusterHostPolicy(cfg *TyphaConfiguration) Component { @@ -640,6 +678,16 @@ func (c *typhaComponent) typhaEnvVars(typhaSecret certificatemanagement.KeyPairI Value: c.cfg.Installation.CalicoNetwork.MultiInterfaceMode.Value(), }) } + + // Serval is a Typha client per replica rather than per node, so Typha counts + // its Service endpoints on top of the nodes when it sizes its connection + // limit. Set whether or not Serval is deployed: the lookup returns zero when + // it is absent, and making this conditional would restart Typha whenever a + // cluster moves between non-cluster host modes. + typhaEnv = append(typhaEnv, + corev1.EnvVar{Name: "TYPHA_K8SEXTRACLIENTSERVICENAME", Value: common.ServalName}, + corev1.EnvVar{Name: "TYPHA_K8SEXTRACLIENTPORTNAME", Value: common.ServalServicePortName}, + ) } // If host-local IPAM is in use, we need to configure typha to use the Kubernetes pod CIDR. diff --git a/pkg/render/typha_test.go b/pkg/render/typha_test.go index 6751058601..a9c61acf6b 100644 --- a/pkg/render/typha_test.go +++ b/pkg/render/typha_test.go @@ -218,6 +218,53 @@ var _ = Describe("Typha rendering tests", func() { Expect(d.Spec.Template.Spec.Containers[0].ReadinessProbe.ProbeHandler.HTTPGet.Host).To(BeEmpty()) }) + It("should mark the non-cluster-host Typha objects for deletion when no NonClusterHost resource exists", func() { + cfg.NonClusterHost = nil + component := render.Typha(&cfg) + create, del := component.Objects() + + // Not rendered for creation... + Expect(rtest.GetResource(create, "calico-typha-noncluster-host", "calico-system", "apps", "v1", "Deployment")).To(BeNil()) + // ...but returned for deletion so the operator garbage-collects any leftovers + // (Serval runs Typha in-process; there is no separate non-cluster-host Typha). + Expect(rtest.GetResource(del, "calico-typha-noncluster-host", "calico-system", "apps", "v1", "Deployment")).NotTo(BeNil()) + Expect(rtest.GetResource(del, "calico-typha-noncluster-host", "calico-system", "", "v1", "Service")).NotTo(BeNil()) + // The v3 policy is deliberately absent here: it is only reconcilable while + // the API server is healthy, so the core controller deletes it alongside + // the other v3 policies. Including it here made every reconcile fail on a + // cluster without the Calico API server. + Expect(rtest.GetResource(del, render.TyphaNonClusterHostNetworkPolicyName, "calico-system", "projectcalico.org", "v3", "NetworkPolicy")).To(BeNil()) + }) + + It("should tell Typha to count Serval's endpoints on enterprise", func() { + installation.Variant = operatorv1.CalicoEnterprise + + component := render.Typha(&cfg) + resources, _ := component.Objects() + + d := rtest.GetResource(resources, "calico-typha", "calico-system", "apps", "v1", "Deployment").(*appsv1.Deployment) + Expect(d.Spec.Template.Spec.Containers[0].Env).To(ContainElements( + corev1.EnvVar{Name: "TYPHA_K8SEXTRACLIENTSERVICENAME", Value: "serval"}, + corev1.EnvVar{Name: "TYPHA_K8SEXTRACLIENTPORTNAME", Value: "https"}, + )) + }) + + It("should not count Serval's endpoints on Calico", func() { + component := render.Typha(&cfg) + resources, _ := component.Objects() + + d := rtest.GetResource(resources, "calico-typha", "calico-system", "apps", "v1", "Deployment").(*appsv1.Deployment) + for _, e := range d.Spec.Template.Spec.Containers[0].Env { + Expect(e.Name).ToNot(HavePrefix("TYPHA_K8SEXTRACLIENT")) + } + }) + + It("should offer the non-cluster-host Typha policy as a separate deletion stub", func() { + policy := render.TyphaNonClusterHostPolicyForDeletion() + Expect(policy.GetName()).To(Equal(render.TyphaNonClusterHostNetworkPolicyName)) + Expect(policy.GetNamespace()).To(Equal("calico-system")) + }) + It("should strip the host-network apiserver endpoint from the non-cluster-host Typha and fall back to the default Service", func() { cfg.K8sServiceEp = k8sapi.ServiceEndpoint{Host: "proxy.local", Port: "6444"}