#6031

Committed 27 May 2026 05:09PM UTC coverage: 49.558% (-0.002%) from 49.56%

Build # #6031

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

cdi-deployment, cdi-operator: secure metrics endpoints with authentication (#4121)

* vendor: sigs.k8s.io/controller-runtime/pkg/metrics/filters/filters.go

Signed-off-by: Adi Aloni <aaloni@redhat.com>

* bazenl-generate: delete vendored BUILD.bazel to avoid gazelle conflicts

Delete vendored BUILD.bazel files that use external repo references
which are incompatible with vendored builds before gazelle runs so it
regenerates them with vendored-compatible paths.

Signed-off-by: Adi Aloni <aaloni@redhat.com>

* cdi: add metrics endpoint authentication

Use controller-runtime's WithAuthenticationAndAuthorization
FilterProvider to require authentication for the /metrics endpoints on
cdi-controller and cdi-operator. Requests are authenticated via
TokenReview and authorized via SubjectAccessReview against the k8s API
server.

Signed-off-by: Adi Aloni <aaloni@redhat.com>

---------

Signed-off-by: Adi Aloni <aaloni@redhat.com>

Coverage Stats

25 of 76 new or added lines in 6 files covered. (32.89%)

15018 of 30304 relevant lines covered (49.56%)

0.56 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/cmd/cdi-controller/controller.go

package main

import (
        "context"
        "crypto/rsa"
        "crypto/tls"
        "flag"
        "fmt"
        "os"
        "strconv"

        "github.com/kelseyhightower/envconfig"
        snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
        ocpconfigv1 "github.com/openshift/api/config/v1"
        imagev1 "github.com/openshift/api/image/v1"
        routev1 "github.com/openshift/api/route/v1"
        "github.com/pkg/errors"
        "go.uber.org/zap/zapcore"

        batchv1 "k8s.io/api/batch/v1"
        v1 "k8s.io/api/core/v1"
        networkingv1 "k8s.io/api/networking/v1"
        extv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
        "k8s.io/apimachinery/pkg/api/meta"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/fields"
        apiruntime "k8s.io/apimachinery/pkg/runtime"
        "k8s.io/client-go/kubernetes"
        clientgoscheme "k8s.io/client-go/kubernetes/scheme"
        "k8s.io/client-go/tools/clientcmd"
        "k8s.io/klog/v2"

        "sigs.k8s.io/controller-runtime/pkg/cache"
        "sigs.k8s.io/controller-runtime/pkg/client"
        "sigs.k8s.io/controller-runtime/pkg/client/config"
        logf "sigs.k8s.io/controller-runtime/pkg/log"
        "sigs.k8s.io/controller-runtime/pkg/log/zap"
        "sigs.k8s.io/controller-runtime/pkg/manager"
        "sigs.k8s.io/controller-runtime/pkg/manager/signals"
        metricsfilters "sigs.k8s.io/controller-runtime/pkg/metrics/filters"
        "sigs.k8s.io/controller-runtime/pkg/metrics/server"

        cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
        forklift "kubevirt.io/containerized-data-importer-api/pkg/apis/forklift/v1beta1"
        cdiclient "kubevirt.io/containerized-data-importer/pkg/client/clientset/versioned"
        "kubevirt.io/containerized-data-importer/pkg/common"
        "kubevirt.io/containerized-data-importer/pkg/controller"
        dvc "kubevirt.io/containerized-data-importer/pkg/controller/datavolume"
        "kubevirt.io/containerized-data-importer/pkg/controller/populators"
        "kubevirt.io/containerized-data-importer/pkg/controller/transfer"
        metrics "kubevirt.io/containerized-data-importer/pkg/monitoring/metrics/cdi-controller"
        "kubevirt.io/containerized-data-importer/pkg/util"
        "kubevirt.io/containerized-data-importer/pkg/util/cert"
        "kubevirt.io/containerized-data-importer/pkg/util/cert/fetcher"
        "kubevirt.io/containerized-data-importer/pkg/util/cert/generator"
        cryptowatch "kubevirt.io/containerized-data-importer/pkg/util/tls-crypto-watch"
)

const (
        readyFile = "/tmp/ready"
)

var (
        kubeconfig             string
        kubeURL                string
        importerImage          string
        clonerImage            string
        uploadServerImage      string
        uploadProxyServiceName string
        ovirtPopulatorImage    string
        configName             string
        pullPolicy             string
        verbose                string
        installerLabels        map[string]string
        log                    = logf.Log.WithName("controller")
        controllerEnvs         ControllerEnvs
        resourcesSchemeFuncs   = []func(*apiruntime.Scheme) error{
                clientgoscheme.AddToScheme,
                cdiv1.AddToScheme,
                extv1.AddToScheme,
                snapshotv1.AddToScheme,
                forklift.AddToScheme,
                imagev1.Install,
                ocpconfigv1.Install,
                routev1.Install,
        }
)

// ControllerEnvs contains environment variables read for setting custom cert paths
type ControllerEnvs struct {
        UploadServerKeyFile           string `default:"/var/run/certs/cdi-uploadserver-signer/tls.key" split_words:"true"`
        UploadServerCertFile          string `default:"/var/run/certs/cdi-uploadserver-signer/tls.crt" split_words:"true"`
        UploadClientKeyFile           string `default:"/var/run/certs/cdi-uploadserver-client-signer/tls.key" split_words:"true"`
        UploadClientCertFile          string `default:"/var/run/certs/cdi-uploadserver-client-signer/tls.crt" split_words:"true"`
        UploadServerCaBundleConfigMap string `default:"cdi-uploadserver-signer-bundle" split_words:"true"`
        UploadClientCaBundleConfigMap string `default:"cdi-uploadserver-client-signer-bundle" split_words:"true"`
}

// The importer and cloner images are obtained here along with the supported flags. IMPORTER_IMAGE, CLONER_IMAGE, and UPLOADSERVICE_IMAGE
// are required by the controller and will cause it to fail if not defined.
// Note: kubeconfig hierarchy is 1) -kubeconfig flag, 2) $KUBECONFIG exported var. If neither is
// specified we do an in-cluster config. For testing it's easiest to export KUBECONFIG.
func init() {
        // flags
        flag.StringVar(&kubeURL, "server", "", "(Optional) URL address of a remote api server.  Do not set for local clusters.")
        klog.InitFlags(nil)
        flag.Parse()

        if flag.Lookup("kubeconfig") != nil {
                kubeconfig = flag.Lookup("kubeconfig").Value.String()
        }
        importerImage = getRequiredEnvVar("IMPORTER_IMAGE")
        clonerImage = getRequiredEnvVar("CLONER_IMAGE")
        uploadServerImage = getRequiredEnvVar("UPLOADSERVER_IMAGE")
        ovirtPopulatorImage = getRequiredEnvVar("OVIRT_POPULATOR_IMAGE")
        uploadProxyServiceName = getRequiredEnvVar("UPLOADPROXY_SERVICE")
        installerLabels = map[string]string{}

        pullPolicy = common.DefaultPullPolicy
        if pp := os.Getenv(common.PullPolicy); len(pp) != 0 {
                pullPolicy = pp
        }

        // We will need to put those on every resource our controller creates
        if partOfVal := os.Getenv(common.InstallerPartOfLabel); len(partOfVal) != 0 {
                installerLabels[common.AppKubernetesPartOfLabel] = partOfVal
        }
        if versionVal := os.Getenv(common.InstallerVersionLabel); len(versionVal) != 0 {
                installerLabels[common.AppKubernetesVersionLabel] = versionVal
        }

        configName = common.ConfigName

        // NOTE we used to have a constant here and we're now just passing in the level directly
        // that should be fine since it was a constant and not a mutable variable
        defVerbose := fmt.Sprintf("%d", 1) // note flag values are strings
        verbose = defVerbose
        // visit actual flags passed in and if passed check -v and set verbose
        flag.Visit(func(f *flag.Flag) {
                if f.Name == "v" {
                        verbose = f.Value.String()
                }
        })
        if verbose == defVerbose {
                klog.V(1).Infof("Note: increase the -v level in the controller deployment for more detailed logging, eg. -v=%d or -v=%d\n", 2, 3)
        }

        // Setup metrics for our various controllers
        if err := metrics.SetupMetrics(); err != nil {
                klog.Errorf("failed to setup metrics: %v", err)
                os.Exit(1)
        }

        klog.V(3).Infof("init: complete: cdi controller will create importer using image %q\n", importerImage)
}

func getRequiredEnvVar(name string) string {
        val := os.Getenv(name)
        if val == "" {
                klog.Fatalf("Environment Variable %q undefined\n", name)
        }
        return val
}

func start() {
        klog.Info("Starting CDI controller components")

        namespace := util.GetNamespace()

        cfg, err := clientcmd.BuildConfigFromFlags(kubeURL, kubeconfig)
        if err != nil {
                klog.Fatalf("Unable to get kube config: %v\n", errors.WithStack(err))
        }

        k8sClient, err := kubernetes.NewForConfig(cfg)
        if err != nil {
                klog.Fatalf("Unable to get kube client: %v\n", errors.WithStack(err))
        }

        // Setup scheme for all resources
        scheme := apiruntime.NewScheme()
        for _, f := range resourcesSchemeFuncs {
                err := f(scheme)
                if err != nil {
                        klog.Errorf("Failed to add to scheme: %v", err)
                        os.Exit(1)
                }
        }

        // client.New() returns a client without cache
        // since we don't have a cached client before manager init
        apiClient, err := client.New(cfg, client.Options{
                Scheme: scheme,
        })
        if err != nil {
                klog.Fatalf("Unable to get uncached client: %v\n", errors.WithStack(err))
        }

        managedTLSWatcher := cryptowatch.NewManagedTLSWatcher(cdiclient.NewForConfigOrDie(cfg))

        opts := manager.Options{
                LeaderElection:             true,
                LeaderElectionNamespace:    namespace,
                LeaderElectionID:           "cdi-controller-leader-election-helper",
                LeaderElectionResourceLock: "leases",
                Cache:                      getCacheOptions(apiClient, namespace),
                Scheme:                     scheme,
                Metrics: server.Options{
                        BindAddress:    ":8443",
                        SecureServing:  true,
                        FilterProvider: metricsfilters.WithAuthenticationAndAuthorization,
                        // Disable HTTP/2 to prevent rapid reset vulnerability
                        // See CVE-2023-44487, CVE-2023-39325
                        TLSOpts: []func(*tls.Config){func(c *tls.Config) {
                                c.NextProtos = []string{"http/1.1"}
                                c.GetConfigForClient = func(t *tls.ClientHelloInfo) (*tls.Config, error) {
                                        config := c.Clone()
                                        if w := managedTLSWatcher.Watcher(); w != nil {
                                                cryptoConfig := w.GetCdiTLSConfig()
                                                config.CipherSuites = cryptoConfig.CipherSuites
                                                config.MinVersion = cryptoConfig.MinVersion
                                        }
                                        return config, nil
                                }
                        }},
                },
        }

        mgr, err := manager.New(config.GetConfigOrDie(), opts)
        if err != nil {
                klog.Errorf("Unable to setup controller manager: %v", err)
                os.Exit(1)
        }

        uploadClientCAFetcher := &fetcher.FileCertFetcher{KeyFileName: controllerEnvs.UploadClientKeyFile, CertFileName: controllerEnvs.UploadClientCertFile}
        uploadClientBundleFetcher := &fetcher.ConfigMapCertBundleFetcher{
                Name:   controllerEnvs.UploadClientCaBundleConfigMap,
                Client: k8sClient.CoreV1().ConfigMaps(namespace),
        }
        uploadClientCertGenerator := &generator.FetchCertGenerator{Fetcher: uploadClientCAFetcher}

        uploadServerCAFetcher := &fetcher.FileCertFetcher{KeyFileName: controllerEnvs.UploadServerKeyFile, CertFileName: controllerEnvs.UploadServerCertFile}
        uploadServerBundleFetcher := &fetcher.ConfigMapCertBundleFetcher{
                Name:   controllerEnvs.UploadServerCaBundleConfigMap,
                Client: k8sClient.CoreV1().ConfigMaps(namespace),
        }
        uploadServerCertGenerator := &generator.FetchCertGenerator{Fetcher: uploadServerCAFetcher}

        if _, err := controller.NewConfigController(mgr, log, uploadProxyServiceName, configName, installerLabels); err != nil {
                klog.Errorf("Unable to setup config controller: %v", err)
                os.Exit(1)
        }

        if _, err := controller.NewStorageProfileController(mgr, log, installerLabels); err != nil {
                klog.Errorf("Unable to setup storage profiles controller: %v", err)
                os.Exit(1)
        }

        if err := dvc.CreateCommonIndexes(mgr); err != nil {
                klog.Errorf("Unable to create shared indexes: %v", err)
                os.Exit(1)
        }

        ctx := signals.SetupSignalHandler()

        // TODO: Current DV controller had threadiness 3, should we do the same here, defaults to one thread.
        if _, err := dvc.NewImportController(ctx, mgr, log, installerLabels); err != nil {
                klog.Errorf("Unable to setup datavolume import controller: %v", err)
                os.Exit(1)
        }
        if _, err := dvc.NewUploadController(ctx, mgr, log, installerLabels); err != nil {
                klog.Errorf("Unable to setup datavolume upload controller: %v", err)
                os.Exit(1)
        }
        if _, err := dvc.NewPvcCloneController(ctx, mgr, log,
                clonerImage, importerImage, pullPolicy, getTokenPublicKey(), getTokenPrivateKey(), installerLabels); err != nil {
                klog.Errorf("Unable to setup datavolume pvc clone controller: %v", err)
                os.Exit(1)
        }
        if _, err := dvc.NewSnapshotCloneController(ctx, mgr, log,
                clonerImage, importerImage, pullPolicy, getTokenPublicKey(), getTokenPrivateKey(), installerLabels); err != nil {
                klog.Errorf("Unable to setup datavolume snapshot clone controller: %v", err)
                os.Exit(1)
        }
        if _, err := dvc.NewPopulatorController(ctx, mgr, log, installerLabels); err != nil {
                klog.Errorf("Unable to setup datavolume external-population controller: %v", err)
                os.Exit(1)
        }

        if _, err := controller.NewImportController(mgr, log, importerImage, pullPolicy, verbose, installerLabels); err != nil {
                klog.Errorf("Unable to setup import controller: %v", err)
                os.Exit(1)
        }

        if _, err := controller.NewCloneController(mgr, log, clonerImage, pullPolicy, verbose, uploadClientCertGenerator, uploadServerBundleFetcher, getTokenPublicKey(), installerLabels); err != nil {
                klog.Errorf("Unable to setup clone controller: %v", err)
                os.Exit(1)
        }

        if _, err := controller.NewUploadController(mgr, log, uploadServerImage, pullPolicy, verbose, uploadServerCertGenerator, uploadClientBundleFetcher, installerLabels); err != nil {
                klog.Errorf("Unable to setup upload controller: %v", err)
                os.Exit(1)
        }

        if _, err := transfer.NewObjectTransferController(mgr, log, installerLabels); err != nil {
                klog.Errorf("Unable to setup transfer controller: %v", err)
                os.Exit(1)
        }

        if _, err := controller.NewDataImportCronController(mgr, log, importerImage, pullPolicy, installerLabels); err != nil {
                klog.Errorf("Unable to setup dataimportcron controller: %v", err)
                os.Exit(1)
        }
        if _, err := controller.NewDataSourceController(mgr, log, installerLabels); err != nil {
                klog.Errorf("Unable to setup datasource controller: %v", err)
                os.Exit(1)
        }
        // Populator controllers and indexes
        if err := populators.CreateCommonPopulatorIndexes(mgr); err != nil {
                klog.Errorf("Unable to create common populator indexes: %v", err)
                os.Exit(1)
        }
        if _, err := populators.NewImportPopulator(ctx, mgr, log, installerLabels); err != nil {
                klog.Errorf("Unable to setup import populator: %v", err)
                os.Exit(1)
        }
        if _, err := populators.NewUploadPopulator(ctx, mgr, log, installerLabels); err != nil {
                klog.Errorf("Unable to setup upload populator: %v", err)
                os.Exit(1)
        }
        if _, err := populators.NewClonePopulator(ctx, mgr, log, clonerImage, pullPolicy, installerLabels, getTokenPublicKey()); err != nil {
                klog.Errorf("Unable to setup clone populator: %v", err)
                os.Exit(1)
        }
        if _, err := populators.NewForkliftPopulator(ctx, mgr, log, importerImage, ovirtPopulatorImage, installerLabels); err != nil {
                klog.Errorf("Unable to setup forklift populator: %v", err)
                os.Exit(1)
        }

        managedTLSWatcher.SetCache(mgr.GetCache())
        if err := mgr.Add(managedTLSWatcher); err != nil {
                log.Error(err, "unable to add watcher to manager")
                os.Exit(1)
        }

        klog.V(1).Infoln("created cdi controllers")

        if err := mgr.Start(ctx); err != nil {
                klog.Errorf("Error running manager: %v", err)
                os.Exit(1)
        }
}

func main() {
        defer klog.Flush()
        debug := false
        verbosityLevel, err := strconv.Atoi(verbose)
        if err == nil && verbosityLevel > 1 {
                debug = true
        }
        err = envconfig.Process("", &controllerEnvs)
        if err != nil {
                klog.Fatalf("Unable to get environment variables: %v\n", errors.WithStack(err))
        }

        logf.SetLogger(zap.New(zap.Level(zapcore.Level(-1*verbosityLevel)), zap.UseDevMode(debug)))
        logf.Log.WithName("main").Info("Verbosity level", "verbose", verbose, "debug", debug)

        if err = createReadyFile(); err != nil {
                klog.Fatalf("Error creating ready file: %+v", err)
        }

        start()

        deleteReadyFile()

        klog.V(2).Infoln("cdi controller exited")
}

func createReadyFile() error {
        f, err := os.Create(readyFile)
        if err != nil {
                return err
        }
        defer f.Close()
        return nil
}

func deleteReadyFile() {
        os.Remove(readyFile)
}

func getTokenPublicKey() *rsa.PublicKey {
        keyBytes, err := os.ReadFile(controller.TokenPublicKeyPath)
        if err != nil {
                klog.Fatalf("Error reading apiserver public key")
        }

        key, err := controller.DecodePublicKey(keyBytes)
        if err != nil {
                klog.Fatalf("Error decoding public key")
        }

        return key
}

func getTokenPrivateKey() *rsa.PrivateKey {
        bytes, err := os.ReadFile(controller.TokenPrivateKeyPath)
        if err != nil {
                klog.Fatalf("Error reading private key")
        }

        obj, err := cert.ParsePrivateKeyPEM(bytes)
        if err != nil {
                klog.Fatalf("Error decoding private key")
        }

        key, ok := obj.(*rsa.PrivateKey)
        if !ok {
                klog.Fatalf("Invalid private key format")
        }

        return key
}

// Restricts some types in the cache's ListWatch to specific fields/labels per GVK at the specified object,
// other types will continue working normally.
// Note: objects you read once with the controller runtime client are cached.
// TODO: Make our watches way more specific using labels, for example,
// at the point of writing this, we don't care about VolumeSnapshots without the CDI label
func getCacheOptions(apiClient client.Client, cdiNamespace string) cache.Options {
        namespaceSelector := fields.Set{"metadata.namespace": cdiNamespace}.AsSelector()

        cacheOptions := cache.Options{
                ByObject: map[client.Object]cache.ByObject{
                        &networkingv1.Ingress{}: {
                                Field: namespaceSelector,
                        },
                        &batchv1.CronJob{}: {
                                Field: namespaceSelector,
                        },
                        &batchv1.Job{}: {
                                Field: namespaceSelector,
                        },
                        &v1.ConfigMap{}: {
                                Field: namespaceSelector,
                        },
                        &v1.Secret{}: {
                                Field: namespaceSelector,
                        },
                },
        }

        cacheOptionsByObjectForOpenshift := map[client.Object]cache.ByObject{
                &routev1.Route{}: {
                        Field: namespaceSelector,
                },
        }

        // Currently controller-runtime will fail if types in here are not installed in the cluster
        // https://github.com/kubernetes-sigs/controller-runtime/issues/2456
        if isOpenShift(apiClient) {
                for k, v := range cacheOptionsByObjectForOpenshift {
                        cacheOptions.ByObject[k] = v
                }
        }

        return cacheOptions
}

func isOpenShift(apiClient client.Client) bool {
        clusterVersion := &ocpconfigv1.ClusterVersion{
                ObjectMeta: metav1.ObjectMeta{
                        Name: "version",
                },
        }
        if err := apiClient.Get(context.TODO(), client.ObjectKeyFromObject(clusterVersion), clusterVersion); err != nil {
                if !meta.IsNoMatchError(err) {
                        klog.Errorf("Error getting clusterVersion: %v", err)
                        os.Exit(1)
                }
                return false
        }

        return true
}

1	package main
2
3	import (
4	"context"
5	"crypto/rsa"
6	"crypto/tls"
7	"flag"
8	"fmt"
9	"os"
10	"strconv"
11
12	"github.com/kelseyhightower/envconfig"
13	snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
14	ocpconfigv1 "github.com/openshift/api/config/v1"
15	imagev1 "github.com/openshift/api/image/v1"
16	routev1 "github.com/openshift/api/route/v1"
17	"github.com/pkg/errors"
18	"go.uber.org/zap/zapcore"
19
20	batchv1 "k8s.io/api/batch/v1"
21	v1 "k8s.io/api/core/v1"
22	networkingv1 "k8s.io/api/networking/v1"
23	extv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
24	"k8s.io/apimachinery/pkg/api/meta"
25	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
26	"k8s.io/apimachinery/pkg/fields"
27	apiruntime "k8s.io/apimachinery/pkg/runtime"
28	"k8s.io/client-go/kubernetes"
29	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
30	"k8s.io/client-go/tools/clientcmd"
31	"k8s.io/klog/v2"
32
33	"sigs.k8s.io/controller-runtime/pkg/cache"
34	"sigs.k8s.io/controller-runtime/pkg/client"
35	"sigs.k8s.io/controller-runtime/pkg/client/config"
36	logf "sigs.k8s.io/controller-runtime/pkg/log"
37	"sigs.k8s.io/controller-runtime/pkg/log/zap"
38	"sigs.k8s.io/controller-runtime/pkg/manager"
39	"sigs.k8s.io/controller-runtime/pkg/manager/signals"
40	metricsfilters "sigs.k8s.io/controller-runtime/pkg/metrics/filters"
41	"sigs.k8s.io/controller-runtime/pkg/metrics/server"
42
43	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
44	forklift "kubevirt.io/containerized-data-importer-api/pkg/apis/forklift/v1beta1"
45	cdiclient "kubevirt.io/containerized-data-importer/pkg/client/clientset/versioned"
46	"kubevirt.io/containerized-data-importer/pkg/common"
47	"kubevirt.io/containerized-data-importer/pkg/controller"
48	dvc "kubevirt.io/containerized-data-importer/pkg/controller/datavolume"
49	"kubevirt.io/containerized-data-importer/pkg/controller/populators"
50	"kubevirt.io/containerized-data-importer/pkg/controller/transfer"
51	metrics "kubevirt.io/containerized-data-importer/pkg/monitoring/metrics/cdi-controller"
52	"kubevirt.io/containerized-data-importer/pkg/util"
53	"kubevirt.io/containerized-data-importer/pkg/util/cert"
54	"kubevirt.io/containerized-data-importer/pkg/util/cert/fetcher"
55	"kubevirt.io/containerized-data-importer/pkg/util/cert/generator"
56	cryptowatch "kubevirt.io/containerized-data-importer/pkg/util/tls-crypto-watch"
57	)
58
59	const (
60	readyFile = "/tmp/ready"
61	)
62
63	var (
64	kubeconfig string
65	kubeURL string
66	importerImage string
67	clonerImage string
68	uploadServerImage string
69	uploadProxyServiceName string
70	ovirtPopulatorImage string
71	configName string
72	pullPolicy string
73	verbose string
74	installerLabels map[string]string
75	log = logf.Log.WithName("controller")
76	controllerEnvs ControllerEnvs
77	resourcesSchemeFuncs = []func(*apiruntime.Scheme) error{
78	clientgoscheme.AddToScheme,
79	cdiv1.AddToScheme,
80	extv1.AddToScheme,
81	snapshotv1.AddToScheme,
82	forklift.AddToScheme,
83	imagev1.Install,
84	ocpconfigv1.Install,
85	routev1.Install,
86	}
87	)
88
89	// ControllerEnvs contains environment variables read for setting custom cert paths
90	type ControllerEnvs struct {
91	UploadServerKeyFile string `default:"/var/run/certs/cdi-uploadserver-signer/tls.key" split_words:"true"`
92	UploadServerCertFile string `default:"/var/run/certs/cdi-uploadserver-signer/tls.crt" split_words:"true"`
93	UploadClientKeyFile string `default:"/var/run/certs/cdi-uploadserver-client-signer/tls.key" split_words:"true"`
94	UploadClientCertFile string `default:"/var/run/certs/cdi-uploadserver-client-signer/tls.crt" split_words:"true"`
95	UploadServerCaBundleConfigMap string `default:"cdi-uploadserver-signer-bundle" split_words:"true"`
96	UploadClientCaBundleConfigMap string `default:"cdi-uploadserver-client-signer-bundle" split_words:"true"`
97	}
98
99	// The importer and cloner images are obtained here along with the supported flags. IMPORTER_IMAGE, CLONER_IMAGE, and UPLOADSERVICE_IMAGE
100	// are required by the controller and will cause it to fail if not defined.
101	// Note: kubeconfig hierarchy is 1) -kubeconfig flag, 2) $KUBECONFIG exported var. If neither is
102	// specified we do an in-cluster config. For testing it's easiest to export KUBECONFIG.
103	func init() {	×
104	// flags	×
105	flag.StringVar(&kubeURL, "server", "", "(Optional) URL address of a remote api server. Do not set for local clusters.")	×
106	klog.InitFlags(nil)	×
107	flag.Parse()	×
108		×
109	if flag.Lookup("kubeconfig") != nil {	×
110	kubeconfig = flag.Lookup("kubeconfig").Value.String()	×
111	}	×
112	importerImage = getRequiredEnvVar("IMPORTER_IMAGE")	×
113	clonerImage = getRequiredEnvVar("CLONER_IMAGE")	×
114	uploadServerImage = getRequiredEnvVar("UPLOADSERVER_IMAGE")	×
115	ovirtPopulatorImage = getRequiredEnvVar("OVIRT_POPULATOR_IMAGE")	×
116	uploadProxyServiceName = getRequiredEnvVar("UPLOADPROXY_SERVICE")	×
117	installerLabels = map[string]string{}	×
118		×
119	pullPolicy = common.DefaultPullPolicy	×
120	if pp := os.Getenv(common.PullPolicy); len(pp) != 0 {	×
121	pullPolicy = pp	×
122	}	×
123
124	// We will need to put those on every resource our controller creates
125	if partOfVal := os.Getenv(common.InstallerPartOfLabel); len(partOfVal) != 0 {	×
126	installerLabels[common.AppKubernetesPartOfLabel] = partOfVal	×
127	}	×
128	if versionVal := os.Getenv(common.InstallerVersionLabel); len(versionVal) != 0 {	×
129	installerLabels[common.AppKubernetesVersionLabel] = versionVal	×
130	}	×
131
132	configName = common.ConfigName	×
133		×
134	// NOTE we used to have a constant here and we're now just passing in the level directly	×
135	// that should be fine since it was a constant and not a mutable variable	×
136	defVerbose := fmt.Sprintf("%d", 1) // note flag values are strings	×
137	verbose = defVerbose	×
138	// visit actual flags passed in and if passed check -v and set verbose	×
139	flag.Visit(func(f *flag.Flag) {	×
140	if f.Name == "v" {	×
141	verbose = f.Value.String()	×
142	}	×
143	})
144	if verbose == defVerbose {	×
145	klog.V(1).Infof("Note: increase the -v level in the controller deployment for more detailed logging, eg. -v=%d or -v=%d\n", 2, 3)	×
146	}	×
147
148	// Setup metrics for our various controllers
149	if err := metrics.SetupMetrics(); err != nil {	×
150	klog.Errorf("failed to setup metrics: %v", err)	×
151	os.Exit(1)	×
152	}	×
153
154	klog.V(3).Infof("init: complete: cdi controller will create importer using image %q\n", importerImage)	×
155	}
156
157	func getRequiredEnvVar(name string) string {	×
158	val := os.Getenv(name)	×
159	if val == "" {	×
160	klog.Fatalf("Environment Variable %q undefined\n", name)	×
161	}	×
162	return val	×
163	}
164
165	func start() {	×
166	klog.Info("Starting CDI controller components")	×
167		×
168	namespace := util.GetNamespace()	×
169		×
170	cfg, err := clientcmd.BuildConfigFromFlags(kubeURL, kubeconfig)	×
171	if err != nil {	×
172	klog.Fatalf("Unable to get kube config: %v\n", errors.WithStack(err))	×
173	}	×
174
175	k8sClient, err := kubernetes.NewForConfig(cfg)	×
176	if err != nil {	×
177	klog.Fatalf("Unable to get kube client: %v\n", errors.WithStack(err))	×
178	}	×
179
180	// Setup scheme for all resources
181	scheme := apiruntime.NewScheme()	×
182	for _, f := range resourcesSchemeFuncs {	×
183	err := f(scheme)	×
184	if err != nil {	×
185	klog.Errorf("Failed to add to scheme: %v", err)	×
186	os.Exit(1)	×
187	}	×
188	}
189
190	// client.New() returns a client without cache
191	// since we don't have a cached client before manager init
192	apiClient, err := client.New(cfg, client.Options{	×
193	Scheme: scheme,	×
194	})	×
195	if err != nil {	×
196	klog.Fatalf("Unable to get uncached client: %v\n", errors.WithStack(err))	×
197	}	×
198
199	managedTLSWatcher := cryptowatch.NewManagedTLSWatcher(cdiclient.NewForConfigOrDie(cfg))	×
200		×
201	opts := manager.Options{	×
202	LeaderElection: true,	×
203	LeaderElectionNamespace: namespace,	×
204	LeaderElectionID: "cdi-controller-leader-election-helper",	×
205	LeaderElectionResourceLock: "leases",	×
206	Cache: getCacheOptions(apiClient, namespace),	×
207	Scheme: scheme,	×
208	Metrics: server.Options{	×
NEW 209	BindAddress: ":8443",	×
NEW 210	SecureServing: true,	×
NEW 211	FilterProvider: metricsfilters.WithAuthenticationAndAuthorization,	×
212	// Disable HTTP/2 to prevent rapid reset vulnerability	×
213	// See CVE-2023-44487, CVE-2023-39325	×
214	TLSOpts: []func(tls.Config){func(c tls.Config) {	×
215	c.NextProtos = []string{"http/1.1"}	×
216	c.GetConfigForClient = func(t tls.ClientHelloInfo) (tls.Config, error) {	×
217	config := c.Clone()	×
218	if w := managedTLSWatcher.Watcher(); w != nil {	×
219	cryptoConfig := w.GetCdiTLSConfig()	×
220	config.CipherSuites = cryptoConfig.CipherSuites	×
221	config.MinVersion = cryptoConfig.MinVersion	×
222	}	×
223	return config, nil	×
224	}
225	}},
226	},
227	}
228
229	mgr, err := manager.New(config.GetConfigOrDie(), opts)	×
230	if err != nil {	×
231	klog.Errorf("Unable to setup controller manager: %v", err)	×
232	os.Exit(1)	×
233	}	×
234
235	uploadClientCAFetcher := &fetcher.FileCertFetcher{KeyFileName: controllerEnvs.UploadClientKeyFile, CertFileName: controllerEnvs.UploadClientCertFile}	×
236	uploadClientBundleFetcher := &fetcher.ConfigMapCertBundleFetcher{	×
237	Name: controllerEnvs.UploadClientCaBundleConfigMap,	×
238	Client: k8sClient.CoreV1().ConfigMaps(namespace),	×
239	}	×
240	uploadClientCertGenerator := &generator.FetchCertGenerator{Fetcher: uploadClientCAFetcher}	×
241		×
242	uploadServerCAFetcher := &fetcher.FileCertFetcher{KeyFileName: controllerEnvs.UploadServerKeyFile, CertFileName: controllerEnvs.UploadServerCertFile}	×
243	uploadServerBundleFetcher := &fetcher.ConfigMapCertBundleFetcher{	×
244	Name: controllerEnvs.UploadServerCaBundleConfigMap,	×
245	Client: k8sClient.CoreV1().ConfigMaps(namespace),	×
246	}	×
247	uploadServerCertGenerator := &generator.FetchCertGenerator{Fetcher: uploadServerCAFetcher}	×
248		×
249	if _, err := controller.NewConfigController(mgr, log, uploadProxyServiceName, configName, installerLabels); err != nil {	×
250	klog.Errorf("Unable to setup config controller: %v", err)	×
251	os.Exit(1)	×
252	}	×
253
254	if _, err := controller.NewStorageProfileController(mgr, log, installerLabels); err != nil {	×
255	klog.Errorf("Unable to setup storage profiles controller: %v", err)	×
256	os.Exit(1)	×
257	}	×
258
259	if err := dvc.CreateCommonIndexes(mgr); err != nil {	×
260	klog.Errorf("Unable to create shared indexes: %v", err)	×
261	os.Exit(1)	×
262	}	×
263
264	ctx := signals.SetupSignalHandler()	×
265		×
266	// TODO: Current DV controller had threadiness 3, should we do the same here, defaults to one thread.	×
267	if _, err := dvc.NewImportController(ctx, mgr, log, installerLabels); err != nil {	×
268	klog.Errorf("Unable to setup datavolume import controller: %v", err)	×
269	os.Exit(1)	×
270	}	×
271	if _, err := dvc.NewUploadController(ctx, mgr, log, installerLabels); err != nil {	×
272	klog.Errorf("Unable to setup datavolume upload controller: %v", err)	×
273	os.Exit(1)	×
274	}	×
275	if _, err := dvc.NewPvcCloneController(ctx, mgr, log,	×
276	clonerImage, importerImage, pullPolicy, getTokenPublicKey(), getTokenPrivateKey(), installerLabels); err != nil {	×
277	klog.Errorf("Unable to setup datavolume pvc clone controller: %v", err)	×
278	os.Exit(1)	×
279	}	×
280	if _, err := dvc.NewSnapshotCloneController(ctx, mgr, log,	×
281	clonerImage, importerImage, pullPolicy, getTokenPublicKey(), getTokenPrivateKey(), installerLabels); err != nil {	×
282	klog.Errorf("Unable to setup datavolume snapshot clone controller: %v", err)	×
283	os.Exit(1)	×
284	}	×
285	if _, err := dvc.NewPopulatorController(ctx, mgr, log, installerLabels); err != nil {	×
286	klog.Errorf("Unable to setup datavolume external-population controller: %v", err)	×
287	os.Exit(1)	×
288	}	×
289
290	if _, err := controller.NewImportController(mgr, log, importerImage, pullPolicy, verbose, installerLabels); err != nil {	×
291	klog.Errorf("Unable to setup import controller: %v", err)	×
292	os.Exit(1)	×
293	}	×
294
295	if _, err := controller.NewCloneController(mgr, log, clonerImage, pullPolicy, verbose, uploadClientCertGenerator, uploadServerBundleFetcher, getTokenPublicKey(), installerLabels); err != nil {	×
296	klog.Errorf("Unable to setup clone controller: %v", err)	×
297	os.Exit(1)	×
298	}	×
299
300	if _, err := controller.NewUploadController(mgr, log, uploadServerImage, pullPolicy, verbose, uploadServerCertGenerator, uploadClientBundleFetcher, installerLabels); err != nil {	×
301	klog.Errorf("Unable to setup upload controller: %v", err)	×
302	os.Exit(1)	×
303	}	×
304
305	if _, err := transfer.NewObjectTransferController(mgr, log, installerLabels); err != nil {	×
306	klog.Errorf("Unable to setup transfer controller: %v", err)	×
307	os.Exit(1)	×
308	}	×
309
310	if _, err := controller.NewDataImportCronController(mgr, log, importerImage, pullPolicy, installerLabels); err != nil {	×
311	klog.Errorf("Unable to setup dataimportcron controller: %v", err)	×
312	os.Exit(1)	×
313	}	×
314	if _, err := controller.NewDataSourceController(mgr, log, installerLabels); err != nil {	×
315	klog.Errorf("Unable to setup datasource controller: %v", err)	×
316	os.Exit(1)	×
317	}	×
318	// Populator controllers and indexes
319	if err := populators.CreateCommonPopulatorIndexes(mgr); err != nil {	×
320	klog.Errorf("Unable to create common populator indexes: %v", err)	×
321	os.Exit(1)	×
322	}	×
323	if _, err := populators.NewImportPopulator(ctx, mgr, log, installerLabels); err != nil {	×
324	klog.Errorf("Unable to setup import populator: %v", err)	×
325	os.Exit(1)	×
326	}	×
327	if _, err := populators.NewUploadPopulator(ctx, mgr, log, installerLabels); err != nil {	×
328	klog.Errorf("Unable to setup upload populator: %v", err)	×
329	os.Exit(1)	×
330	}	×
331	if _, err := populators.NewClonePopulator(ctx, mgr, log, clonerImage, pullPolicy, installerLabels, getTokenPublicKey()); err != nil {	×
332	klog.Errorf("Unable to setup clone populator: %v", err)	×
333	os.Exit(1)	×
334	}	×
335	if _, err := populators.NewForkliftPopulator(ctx, mgr, log, importerImage, ovirtPopulatorImage, installerLabels); err != nil {	×
336	klog.Errorf("Unable to setup forklift populator: %v", err)	×
337	os.Exit(1)	×
338	}	×
339
340	managedTLSWatcher.SetCache(mgr.GetCache())	×
341	if err := mgr.Add(managedTLSWatcher); err != nil {	×
342	log.Error(err, "unable to add watcher to manager")	×
343	os.Exit(1)	×
344	}	×
345
346	klog.V(1).Infoln("created cdi controllers")	×
347		×
348	if err := mgr.Start(ctx); err != nil {	×
349	klog.Errorf("Error running manager: %v", err)	×
350	os.Exit(1)	×
351	}	×
352	}
353
354	func main() {	×
355	defer klog.Flush()	×
356	debug := false	×
357	verbosityLevel, err := strconv.Atoi(verbose)	×
358	if err == nil && verbosityLevel > 1 {	×
359	debug = true	×
360	}	×
361	err = envconfig.Process("", &controllerEnvs)	×
362	if err != nil {	×
363	klog.Fatalf("Unable to get environment variables: %v\n", errors.WithStack(err))	×
364	}	×
365
366	logf.SetLogger(zap.New(zap.Level(zapcore.Level(-1*verbosityLevel)), zap.UseDevMode(debug)))	×
367	logf.Log.WithName("main").Info("Verbosity level", "verbose", verbose, "debug", debug)	×
368		×
369	if err = createReadyFile(); err != nil {	×
370	klog.Fatalf("Error creating ready file: %+v", err)	×
371	}	×
372
373	start()	×
374		×
375	deleteReadyFile()	×
376		×
377	klog.V(2).Infoln("cdi controller exited")	×
378	}
379
380	func createReadyFile() error {	×
381	f, err := os.Create(readyFile)	×
382	if err != nil {	×
383	return err	×
384	}	×
385	defer f.Close()	×
386	return nil	×
387	}
388
389	func deleteReadyFile() {	×
390	os.Remove(readyFile)	×
391	}	×
392
393	func getTokenPublicKey() *rsa.PublicKey {	×
394	keyBytes, err := os.ReadFile(controller.TokenPublicKeyPath)	×
395	if err != nil {	×
396	klog.Fatalf("Error reading apiserver public key")	×
397	}	×
398
399	key, err := controller.DecodePublicKey(keyBytes)	×
400	if err != nil {	×
401	klog.Fatalf("Error decoding public key")	×
402	}	×
403
404	return key	×
405	}
406
407	func getTokenPrivateKey() *rsa.PrivateKey {	×
408	bytes, err := os.ReadFile(controller.TokenPrivateKeyPath)	×
409	if err != nil {	×
410	klog.Fatalf("Error reading private key")	×
411	}	×
412
413	obj, err := cert.ParsePrivateKeyPEM(bytes)	×
414	if err != nil {	×
415	klog.Fatalf("Error decoding private key")	×
416	}	×
417
418	key, ok := obj.(*rsa.PrivateKey)	×
419	if !ok {	×
420	klog.Fatalf("Invalid private key format")	×
421	}	×
422
423	return key	×
424	}
425
426	// Restricts some types in the cache's ListWatch to specific fields/labels per GVK at the specified object,
427	// other types will continue working normally.
428	// Note: objects you read once with the controller runtime client are cached.
429	// TODO: Make our watches way more specific using labels, for example,
430	// at the point of writing this, we don't care about VolumeSnapshots without the CDI label
431	func getCacheOptions(apiClient client.Client, cdiNamespace string) cache.Options {	×
432	namespaceSelector := fields.Set{"metadata.namespace": cdiNamespace}.AsSelector()	×
433		×
434	cacheOptions := cache.Options{	×
435	ByObject: map[client.Object]cache.ByObject{	×
436	&networkingv1.Ingress{}: {	×
437	Field: namespaceSelector,	×
438	},	×
439	&batchv1.CronJob{}: {	×
440	Field: namespaceSelector,	×
441	},	×
442	&batchv1.Job{}: {	×
443	Field: namespaceSelector,	×
444	},	×
445	&v1.ConfigMap{}: {	×
446	Field: namespaceSelector,	×
447	},	×
448	&v1.Secret{}: {	×
449	Field: namespaceSelector,	×
450	},	×
451	},	×
452	}	×
453		×
454	cacheOptionsByObjectForOpenshift := map[client.Object]cache.ByObject{	×
455	&routev1.Route{}: {	×
456	Field: namespaceSelector,	×
457	},	×
458	}	×
459		×
460	// Currently controller-runtime will fail if types in here are not installed in the cluster	×
461	// https://github.com/kubernetes-sigs/controller-runtime/issues/2456	×
462	if isOpenShift(apiClient) {	×
463	for k, v := range cacheOptionsByObjectForOpenshift {	×
464	cacheOptions.ByObject[k] = v	×
465	}	×
466	}
467
468	return cacheOptions	×
469	}
470
471	func isOpenShift(apiClient client.Client) bool {	×
472	clusterVersion := &ocpconfigv1.ClusterVersion{	×
473	ObjectMeta: metav1.ObjectMeta{	×
474	Name: "version",	×
475	},	×
476	}	×
477	if err := apiClient.Get(context.TODO(), client.ObjectKeyFromObject(clusterVersion), clusterVersion); err != nil {	×
478	if !meta.IsNoMatchError(err) {	×
479	klog.Errorf("Error getting clusterVersion: %v", err)	×
480	os.Exit(1)	×
481	}	×
482	return false	×
483	}
484
485	return true	×
486	}

kubevirt / containerized-data-importer / #6031

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous