#6034

Committed 30 May 2026 10:12AM UTC coverage: 49.464% (-0.09%) from 49.558%

Build # #6034

Build Type

Pull #4161

travis-ci

Committed by

Acedus

Commit Message

cdi: expose secure pprof endpoints for controller and operator

pprof is Go's built-in profiling toolkit that exposes runtime
diagnostics (e.g., heap allocations, goroutine stacks, CPU profiles,
etc.) over HTTP. Exposing these endpoints enables live debugging of
memory leaks, goroutine leaks, and CPU hotspots in running controller
and operator pods without redeployment or recompilation.

This commit exposes the handlers via the existing metrics server by
relying on controller-runtime's ExtraHandlers along with the previously
introduced WithAuthenticationAndAuthorization which serves to secure
the handlers with TLS and RBAC auth. A new ClusterRole (cdi-pprof-reader)
is added to grant pprof access independently of metrics scraping.

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

Pull Request Pull Request #4161: cdi: expose secure pprof endpoints for controller and operator

Coverage Stats

0 of 37 new or added lines in 4 files covered. (0.0%)

10 existing lines in 2 files now uncovered.

15008 of 30341 relevant lines covered (49.46%)

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

kubevirt / containerized-data-importer / #6034

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