#6098

Committed 02 Jul 2026 03:15PM UTC coverage: 49.826% (+0.04%) from 49.79%

Build # #6098

Build Type

Pull #4193

travis-ci

Committed by

HoustonBoston

Commit Message

Changed types of some member variables in structs wherever possible and added nolint:gosec declarative if it required more work

Signed-off-by: HoustonBoston <rorajesh@redhat.com>

Pull Request Pull Request #4193: Solve G115 Linter Errors

Coverage Stats

22 of 28 new or added lines in 7 files covered. (78.57%)

4 existing lines in 1 file now uncovered.

15157 of 30420 relevant lines covered (49.83%)

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"
        "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))
        }

        // using nolint since Atoi, used for verbosityLevel, strictly returns an int
        logf.SetLogger(zap.New(zap.Level(zapcore.Level(-1*int8(verbosityLevel))), zap.UseDevMode(debug))) //nolint:gosec
        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,	×
214	ExtraHandlers: map[string]http.Handler{	×
215	"/debug/pprof/": http.HandlerFunc(pprof.Index),	×
216	"/debug/pprof/cmdline": http.HandlerFunc(pprof.Cmdline),	×
217	"/debug/pprof/profile": http.HandlerFunc(pprof.Profile),	×
218	"/debug/pprof/symbol": http.HandlerFunc(pprof.Symbol),	×
219	"/debug/pprof/trace": http.HandlerFunc(pprof.Trace),	×
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	// using nolint since Atoi, used for verbosityLevel, strictly returns an int
NEW 376	logf.SetLogger(zap.New(zap.Level(zapcore.Level(-1*int8(verbosityLevel))), zap.UseDevMode(debug))) //nolint:gosec	×
377	logf.Log.WithName("main").Info("Verbosity level", "verbose", verbose, "debug", debug)	×
378		×
379	if err = createReadyFile(); err != nil {	×
380	klog.Fatalf("Error creating ready file: %+v", err)	×
381	}	×
382
383	start()	×
384		×
385	deleteReadyFile()	×
386		×
387	klog.V(2).Infoln("cdi controller exited")	×
388	}
389
390	func createReadyFile() error {	×
391	f, err := os.Create(readyFile)	×
392	if err != nil {	×
393	return err	×
394	}	×
395	defer f.Close()	×
396	return nil	×
397	}
398
399	func deleteReadyFile() {	×
400	os.Remove(readyFile)	×
401	}	×
402
403	func getTokenPublicKey() *rsa.PublicKey {	×
404	keyBytes, err := os.ReadFile(controller.TokenPublicKeyPath)	×
405	if err != nil {	×
406	klog.Fatalf("Error reading apiserver public key")	×
407	}	×
408
409	key, err := controller.DecodePublicKey(keyBytes)	×
410	if err != nil {	×
411	klog.Fatalf("Error decoding public key")	×
412	}	×
413
414	return key	×
415	}
416
417	func getTokenPrivateKey() *rsa.PrivateKey {	×
418	bytes, err := os.ReadFile(controller.TokenPrivateKeyPath)	×
419	if err != nil {	×
420	klog.Fatalf("Error reading private key")	×
421	}	×
422
423	obj, err := cert.ParsePrivateKeyPEM(bytes)	×
424	if err != nil {	×
425	klog.Fatalf("Error decoding private key")	×
426	}	×
427
428	key, ok := obj.(*rsa.PrivateKey)	×
429	if !ok {	×
430	klog.Fatalf("Invalid private key format")	×
431	}	×
432
433	return key	×
434	}
435
436	// Restricts some types in the cache's ListWatch to specific fields/labels per GVK at the specified object,
437	// other types will continue working normally.
438	// Note: objects you read once with the controller runtime client are cached.
439	// TODO: Make our watches way more specific using labels, for example,
440	// at the point of writing this, we don't care about VolumeSnapshots without the CDI label
441	func getCacheOptions(apiClient client.Client, cdiNamespace string) cache.Options {	×
442	namespaceSelector := fields.Set{"metadata.namespace": cdiNamespace}.AsSelector()	×
443		×
444	cacheOptions := cache.Options{	×
445	ByObject: map[client.Object]cache.ByObject{	×
446	&networkingv1.Ingress{}: {	×
447	Field: namespaceSelector,	×
448	},	×
449	&batchv1.CronJob{}: {	×
450	Field: namespaceSelector,	×
451	},	×
452	&batchv1.Job{}: {	×
453	Field: namespaceSelector,	×
454	},	×
455	&v1.ConfigMap{}: {	×
456	Field: namespaceSelector,	×
457	},	×
458	&v1.Secret{}: {	×
459	Field: namespaceSelector,	×
460	},	×
461	},	×
462	}	×
463		×
464	cacheOptionsByObjectForOpenshift := map[client.Object]cache.ByObject{	×
465	&routev1.Route{}: {	×
466	Field: namespaceSelector,	×
467	},	×
468	}	×
469		×
470	// Currently controller-runtime will fail if types in here are not installed in the cluster	×
471	// https://github.com/kubernetes-sigs/controller-runtime/issues/2456	×
472	if isOpenShift(apiClient) {	×
473	for k, v := range cacheOptionsByObjectForOpenshift {	×
474	cacheOptions.ByObject[k] = v	×
475	}	×
476	}
477
478	return cacheOptions	×
479	}
480
481	func isOpenShift(apiClient client.Client) bool {	×
482	clusterVersion := &ocpconfigv1.ClusterVersion{	×
483	ObjectMeta: metav1.ObjectMeta{	×
484	Name: "version",	×
485	},	×
486	}	×
487	if err := apiClient.Get(context.TODO(), client.ObjectKeyFromObject(clusterVersion), clusterVersion); err != nil {	×
488	if !meta.IsNoMatchError(err) {	×
489	klog.Errorf("Error getting clusterVersion: %v", err)	×
490	os.Exit(1)	×
491	}	×
492	return false	×
493	}
494
495	return true	×
496	}

kubevirt / containerized-data-importer / #6098

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