#5856

Committed 08 Mar 2026 08:28PM UTC coverage: 49.3% (-0.06%) from 49.364%

Build # #5856

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

metrics-server: inherit TLS options from CDI TLSSecurityProfile (#4033)

Previously the metrics-server would be initiated with default TLS
configuration. This change makes it so the TLS configuration is updated
during runtime for every request according to the CDI
TLSSecurityProfile.

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

Coverage Stats

0 of 59 new or added lines in 3 files covered. (0.0%)

14722 of 29862 relevant lines covered (49.3%)

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

kubevirt / containerized-data-importer / #5856

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