From d9ff9e2d6a5269094ca5e52848111da1a3c76a18 Mon Sep 17 00:00:00 2001 From: Bala FA Date: Mon, 19 Sep 2022 21:34:10 +0530 Subject: [PATCH] fix controller/node-server command line arguments (#642) Signed-off-by: Bala.FA Signed-off-by: Bala.FA --- cmd/directpv/cmd.go | 131 ------------------------ cmd/directpv/controller.go | 75 ++++++++++++++ cmd/directpv/main.go | 99 +++++++++++++++++- cmd/directpv/{run.go => node-server.go} | 107 +++++++++---------- pkg/consts/consts.go | 3 + pkg/consts/consts.go.in | 3 + pkg/installer/config.go | 4 +- pkg/installer/const.go | 26 ++--- pkg/installer/daemonset.go | 101 ++++++------------ pkg/installer/deployment.go | 66 +++--------- pkg/installer/psp.go | 4 +- pkg/installer/vars.go | 52 ++++++++++ pkg/k8s/k8s.go | 1 - pkg/volume/list_test.go | 12 +-- 14 files changed, 356 insertions(+), 328 deletions(-) delete mode 100644 cmd/directpv/cmd.go create mode 100644 cmd/directpv/controller.go rename cmd/directpv/{run.go => node-server.go} (67%) create mode 100644 pkg/installer/vars.go diff --git a/cmd/directpv/cmd.go b/cmd/directpv/cmd.go deleted file mode 100644 index 7eb29508f..000000000 --- a/cmd/directpv/cmd.go +++ /dev/null @@ -1,131 +0,0 @@ -// This file is part of MinIO DirectPV -// Copyright (c) 2021, 2022 MinIO, Inc. -// -// This program is free software: you can redistribute it and/or modify -// it under the terms of the GNU Affero General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful, -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Affero General Public License for more details. -// -// You should have received a copy of the GNU Affero General Public License -// along with this program. If not, see . - -package main - -import ( - "context" - "flag" - "fmt" - "os" - "path/filepath" - - "github.com/minio/directpv/pkg/client" - "github.com/minio/directpv/pkg/consts" - "github.com/spf13/cobra" - "github.com/spf13/viper" - "k8s.io/klog/v2" -) - -// Version of this application populated by `go build` -// e.g. $ go build -ldflags="-X main.Version=v4.0.1" -var Version string - -// flags -var ( - identity = consts.Identity - nodeID = "" - rack = "default" - zone = "default" - region = "default" - endpoint = "unix://csi/csi.sock" - kubeconfig = "" - controller = false - driver = false - procfs = consts.ProcFSDir - showVersion = false - conversionHealthzURL = "" - metricsPort = consts.MetricsPort - readinessPort = consts.ReadinessPort -) - -var driverCmd = &cobra.Command{ - Use: filepath.Base(os.Args[0]), - Short: "CSI driver for provisioning from JBOD(s) directly", - Long: fmt.Sprintf(`This Container Storage Interface (CSI) driver provides just a bunch of drives (JBODs) as volumes consumable within containers. This driver does not manage the lifecycle of the data or the backing of the disk itself. It only acts as the middle-man between a drive and a container runtime. - -This driver is rack, region and zone aware i.e., a workload requesting volumes with constraints on rack, region or zone will be scheduled to run only within the constraints. This is useful for requesting volumes that need to be within a specified failure domain (rack, region or zone) - -For more information, use '%s --help' -`, os.Args[0]), - SilenceUsage: true, - RunE: func(c *cobra.Command, args []string) error { - if showVersion { - fmt.Println(Version) - return nil - } - - switch { - case controller: - case driver: - default: - return fmt.Errorf("one among [--controller, --driver] should be set") - } - - client.Init() - return run(c.Context(), args) - }, -} - -func init() { - if Version == "" { - Version = "dev" - } - - viper.AutomaticEnv() - - flag.Set("alsologtostderr", "true") - kflags := flag.NewFlagSet("klog", flag.ExitOnError) - klog.InitFlags(kflags) - - // parse the go default flagset to get flags for glog and other packages in future - driverCmd.PersistentFlags().AddGoFlagSet(flag.CommandLine) - driverCmd.PersistentFlags().AddGoFlagSet(kflags) - // defaulting this to true so that logs are printed to console - flag.Set("logtostderr", "true") - - driverCmd.PersistentFlags().StringVarP(&kubeconfig, "kubeconfig", "k", kubeconfig, "path to kubeconfig") - driverCmd.Flags().StringVarP(&identity, "identity", "i", identity, "identity of this "+consts.AppPrettyName) - driverCmd.Flags().BoolVarP(&showVersion, "version", "", showVersion, "version of "+consts.AppPrettyName) - driverCmd.Flags().StringVarP(&endpoint, "endpoint", "e", endpoint, "endpoint at which "+consts.AppPrettyName+" is listening") - driverCmd.Flags().StringVarP(&nodeID, "node-id", "n", nodeID, "identity of the node in which "+consts.AppPrettyName+" is running") - driverCmd.Flags().StringVarP(&rack, "rack", "", rack, "identity of the rack in which this "+consts.AppPrettyName+" is running") - driverCmd.Flags().StringVarP(&zone, "zone", "", zone, "identity of the zone in which this "+consts.AppPrettyName+" is running") - driverCmd.Flags().StringVarP(®ion, "region", "", region, "identity of the region in which this "+consts.AppPrettyName+" is running") - driverCmd.Flags().StringVarP(&procfs, "procfs", "", procfs, "path to host "+consts.ProcFSDir+" for accessing mount information") - driverCmd.Flags().BoolVarP(&controller, "controller", "", controller, "running in controller mode") - driverCmd.Flags().BoolVarP(&driver, "driver", "", driver, "run in driver mode") - driverCmd.Flags().StringVarP(&conversionHealthzURL, "conversion-healthz-url", "", conversionHealthzURL, "The URL of the conversion webhook healthz endpoint") - driverCmd.Flags().IntVarP(&metricsPort, "metrics-port", "", metricsPort, fmt.Sprintf("Metrics port for scraping. default is %v", consts.MetricsPort)) - driverCmd.Flags().IntVarP(&readinessPort, "readiness-port", "", readinessPort, fmt.Sprintf("Readiness port. default is %v", consts.ReadinessPort)) - - driverCmd.PersistentFlags().MarkHidden("alsologtostderr") - driverCmd.PersistentFlags().MarkHidden("log_backtrace_at") - driverCmd.PersistentFlags().MarkHidden("log_dir") - driverCmd.PersistentFlags().MarkHidden("logtostderr") - driverCmd.PersistentFlags().MarkHidden("master") - driverCmd.PersistentFlags().MarkHidden("stderrthreshold") - driverCmd.PersistentFlags().MarkHidden("vmodule") - - // suppress the incorrect prefix in glog output - flag.CommandLine.Parse([]string{}) - viper.BindPFlags(driverCmd.PersistentFlags()) -} - -// Execute executes driver command. -func Execute(ctx context.Context) error { - return driverCmd.ExecuteContext(ctx) -} diff --git a/cmd/directpv/controller.go b/cmd/directpv/controller.go new file mode 100644 index 000000000..a54b44e87 --- /dev/null +++ b/cmd/directpv/controller.go @@ -0,0 +1,75 @@ +// This file is part of MinIO DirectPV +// Copyright (c) 2021, 2022 MinIO, Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package main + +import ( + "context" + + "github.com/container-storage-interface/spec/lib/go/csi" + "github.com/minio/directpv/pkg/consts" + "github.com/minio/directpv/pkg/controller" + pkgidentity "github.com/minio/directpv/pkg/identity" + "github.com/spf13/cobra" + "k8s.io/klog/v2" +) + +var controllerCmd = &cobra.Command{ + Use: "controller", + Short: "Start controller server of " + consts.AppPrettyName + ".", + SilenceUsage: true, + SilenceErrors: true, + RunE: func(c *cobra.Command, args []string) error { + return startController(c.Context(), args) + }, +} + +func startController(ctx context.Context, args []string) error { + var cancel context.CancelFunc + ctx, cancel = context.WithCancel(ctx) + defer cancel() + + idServer, err := pkgidentity.NewServer(identity, Version, map[string]string{}) + if err != nil { + return err + } + klog.V(3).Infof("Identity server started") + + var ctrlServer csi.ControllerServer + ctrlServer, err = controller.NewServer(ctx, identity, kubeNodeName, rack, zone, region) + if err != nil { + return err + } + klog.V(3).Infof("Controller server started") + + errCh := make(chan error) + + go func() { + if err := runServers(ctx, csiEndpoint, idServer, ctrlServer, nil); err != nil { + klog.ErrorS(err, "unable to start GRPC servers") + errCh <- err + } + }() + + go func() { + if err := serveReadinessEndpoint(ctx); err != nil { + klog.ErrorS(err, "unable to start readiness endpoint") + errCh <- err + } + }() + + return <-errCh +} diff --git a/cmd/directpv/main.go b/cmd/directpv/main.go index f879fe473..460cb542d 100644 --- a/cmd/directpv/main.go +++ b/cmd/directpv/main.go @@ -18,14 +18,110 @@ package main import ( "context" + "flag" + "fmt" "os" "os/signal" "syscall" "time" + "github.com/minio/directpv/pkg/client" + "github.com/minio/directpv/pkg/consts" + "github.com/spf13/cobra" + "github.com/spf13/viper" "k8s.io/klog/v2" ) +// Version of this application populated by `go build` +// e.g. $ go build -ldflags="-X main.Version=v4.0.1" +var Version string + +// flags +var ( + identity = consts.Identity + kubeNodeName = "" + rack = "default" + zone = "default" + region = "default" + csiEndpoint = consts.UnixCSIEndpoint + kubeconfig = "" + conversionHealthzURL = "" + metricsPort = consts.MetricsPort + readinessPort = consts.ReadinessPort +) + +var mainCmd = &cobra.Command{ + Use: consts.AppName, + Short: "Start " + consts.AppPrettyName + " controller and driver. This binary is usually executed by Kubernetes.", + SilenceUsage: true, + SilenceErrors: false, + Version: Version, + CompletionOptions: cobra.CompletionOptions{ + DisableDefaultCmd: true, + DisableNoDescFlag: true, + DisableDescriptions: true, + HiddenDefaultCmd: true, + }, + PersistentPreRunE: func(c *cobra.Command, args []string) error { + if kubeNodeName == "" { + return fmt.Errorf("value to --kube-node-name must be provided") + } + + client.Init() + return nil + }, +} + +func init() { + if mainCmd.Version == "" { + mainCmd.Version = "dev" + } + + viper.AutomaticEnv() + + kflags := flag.NewFlagSet("klog", flag.ExitOnError) + klog.InitFlags(kflags) + + // parse the go default flagset to get flags for glog and other packages in future + mainCmd.PersistentFlags().AddGoFlagSet(flag.CommandLine) + mainCmd.PersistentFlags().AddGoFlagSet(kflags) + + flag.Set("logtostderr", "true") + flag.Set("alsologtostderr", "true") + + mainCmd.PersistentFlags().StringVarP(&kubeconfig, "kubeconfig", "k", kubeconfig, "Path to the kubeconfig file to use for Kubernetes requests.") + mainCmd.PersistentFlags().StringVarP(&identity, "identity", "", identity, "Identity of "+consts.AppPrettyName+" instances") + mainCmd.PersistentFlags().StringVarP(&csiEndpoint, "csi-endpoint", "", csiEndpoint, "CSI endpoint") + mainCmd.PersistentFlags().StringVarP(&kubeNodeName, "kube-node-name", "", kubeNodeName, "Kubernetes node name (MUST BE SET)") + mainCmd.PersistentFlags().StringVarP(&rack, "rack", "", rack, "Rack ID of "+consts.AppPrettyName+" instances") + mainCmd.PersistentFlags().StringVarP(&zone, "zone", "", zone, "Zone ID of "+consts.AppPrettyName+" instances") + mainCmd.PersistentFlags().StringVarP(®ion, "region", "", region, "Region ID of "+consts.AppPrettyName+" instances") + mainCmd.PersistentFlags().StringVarP(&conversionHealthzURL, "conversion-healthz-url", "", conversionHealthzURL, "URL to conversion webhook health endpoint") + mainCmd.PersistentFlags().IntVarP(&readinessPort, "readiness-port", "", readinessPort, "Readiness port at "+consts.AppPrettyName+" exports readiness of services") + + mainCmd.PersistentFlags().MarkHidden("alsologtostderr") + mainCmd.PersistentFlags().MarkHidden("add_dir_header") + mainCmd.PersistentFlags().MarkHidden("log_file") + mainCmd.PersistentFlags().MarkHidden("log_file_max_size") + mainCmd.PersistentFlags().MarkHidden("one_output") + mainCmd.PersistentFlags().MarkHidden("skip_headers") + mainCmd.PersistentFlags().MarkHidden("skip_log_headers") + mainCmd.PersistentFlags().MarkHidden("v") + mainCmd.PersistentFlags().MarkHidden("log_backtrace_at") + mainCmd.PersistentFlags().MarkHidden("log_dir") + mainCmd.PersistentFlags().MarkHidden("logtostderr") + mainCmd.PersistentFlags().MarkHidden("master") + mainCmd.PersistentFlags().MarkHidden("stderrthreshold") + mainCmd.PersistentFlags().MarkHidden("vmodule") + + // suppress the incorrect prefix in glog output + flag.CommandLine.Parse([]string{}) + viper.BindPFlags(mainCmd.PersistentFlags()) + + mainCmd.AddCommand(controllerCmd) + mainCmd.AddCommand(nodeServerCmd) +} + func main() { sigs := make(chan os.Signal, 1) signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM, syscall.SIGSEGV) @@ -39,7 +135,8 @@ func main() { os.Exit(1) }() - if err := Execute(ctx); err != nil { + if err := mainCmd.ExecuteContext(ctx); err != nil { + klog.ErrorS(err, "unable to execute command") os.Exit(1) } } diff --git a/cmd/directpv/run.go b/cmd/directpv/node-server.go similarity index 67% rename from cmd/directpv/run.go rename to cmd/directpv/node-server.go index 0dcceaa9c..3faaf14d5 100644 --- a/cmd/directpv/run.go +++ b/cmd/directpv/node-server.go @@ -24,18 +24,32 @@ import ( "github.com/container-storage-interface/spec/lib/go/csi" "github.com/google/uuid" "github.com/minio/directpv/pkg/consts" - pkgcontroller "github.com/minio/directpv/pkg/controller" pkgidentity "github.com/minio/directpv/pkg/identity" "github.com/minio/directpv/pkg/node" "github.com/minio/directpv/pkg/sys" "github.com/minio/directpv/pkg/volume" "github.com/minio/directpv/pkg/xfs" + "github.com/spf13/cobra" losetup "gopkg.in/freddierice/go-losetup.v1" "k8s.io/klog/v2" ) var errMountFailure = errors.New("could not mount the drive") +var nodeServerCmd = &cobra.Command{ + Use: "node-server", + Short: "Start node server of " + consts.AppPrettyName + ".", + SilenceUsage: true, + SilenceErrors: true, + RunE: func(c *cobra.Command, args []string) error { + return startNodeServer(c.Context(), args) + }, +} + +func init() { + nodeServerCmd.PersistentFlags().IntVarP(&metricsPort, "metrics-port", "", metricsPort, "Metrics port at "+consts.AppPrettyName+" exports metrics data") +} + func checkXFS(ctx context.Context, reflinkSupport bool) error { mountPoint, err := os.MkdirTemp("", "xfs.check.mnt.") if err != nil { @@ -89,10 +103,10 @@ func getReflinkSupport(ctx context.Context) (reflinkSupport bool, err error) { return } -func run(ctxMain context.Context, args []string) error { - ctx, cancel := context.WithCancel(ctxMain) +func startNodeServer(ctx context.Context, args []string) error { + var cancel context.CancelFunc + ctx, cancel = context.WithCancel(ctx) defer cancel() - errChan := make(chan error) idServer, err := pkgidentity.NewServer(identity, Version, map[string]string{}) if err != nil { @@ -100,72 +114,59 @@ func run(ctxMain context.Context, args []string) error { } klog.V(3).Infof("Identity server started") - var nodeServer csi.NodeServer - if driver { - reflinkSupport, err := getReflinkSupport(ctx) - if err != nil { - return err - } + reflinkSupport, err := getReflinkSupport(ctx) + if err != nil { + return err + } - if reflinkSupport { - klog.V(3).Infof("reflink support is ENABLED for XFS formatting and mounting") - } else { - klog.V(3).Infof("reflink support is DISABLED for XFS formatting and mounting") - } + if reflinkSupport { + klog.V(3).Infof("reflink support is ENABLED for XFS formatting and mounting") + } else { + klog.V(3).Infof("reflink support is DISABLED for XFS formatting and mounting") + } - go func() { - if err := volume.StartController(ctx, nodeID); err != nil { - klog.ErrorS(err, "unable to start volume controller") - errChan <- err - } - }() - - nodeServer, err = node.NewServer(ctx, - identity, - nodeID, - rack, - zone, - region, - reflinkSupport, - metricsPort, - ) - if err != nil { - return err - } - klog.V(3).Infof("Node server started") + errCh := make(chan error) - if err = os.Mkdir(consts.MountRootDir, 0o777); err != nil && !errors.Is(err, os.ErrExist) { - return err + go func() { + if err := volume.StartController(ctx, kubeNodeName); err != nil { + klog.ErrorS(err, "unable to start volume controller") + errCh <- err } + }() + + var nodeServer csi.NodeServer + nodeServer, err = node.NewServer( + ctx, + identity, + kubeNodeName, + rack, + zone, + region, + reflinkSupport, + metricsPort, + ) + if err != nil { + return err } + klog.V(3).Infof("Node server started") - var ctrlServer csi.ControllerServer - if controller { - ctrlServer, err = pkgcontroller.NewServer(ctx, identity, nodeID, rack, zone, region) - if err != nil { - return err - } - klog.V(3).Infof("Controller server started") + if err = os.Mkdir(consts.MountRootDir, 0o777); err != nil && !errors.Is(err, os.ErrExist) { + return err } go func() { - if err := runServers(ctx, endpoint, idServer, ctrlServer, nodeServer); err != nil { + if err := runServers(ctx, csiEndpoint, idServer, nil, nodeServer); err != nil { klog.ErrorS(err, "unable to start GRPC servers") - errChan <- err + errCh <- err } }() go func() { if err := serveReadinessEndpoint(ctx); err != nil { klog.ErrorS(err, "unable to start readiness endpoint") - errChan <- err + errCh <- err } }() - err = <-errChan - if err != nil { - cancel() - return err - } - return nil + return <-errCh } diff --git a/pkg/consts/consts.go b/pkg/consts/consts.go index 87e2fc16a..702ca99cd 100644 --- a/pkg/consts/consts.go +++ b/pkg/consts/consts.go @@ -99,4 +99,7 @@ const ( // MountRootDir is mount root directory. MountRootDir = AppRootDir + "/mnt" + + // UnixCSIEndpoint is Unix CSI endpoint + UnixCSIEndpoint = "unix:///csi/csi.sock" ) diff --git a/pkg/consts/consts.go.in b/pkg/consts/consts.go.in index 411440c6d..5c6d769d3 100644 --- a/pkg/consts/consts.go.in +++ b/pkg/consts/consts.go.in @@ -97,4 +97,7 @@ const ( // MountRootDir is mount root directory. MountRootDir = AppRootDir + "/mnt" + + // UnixCSIEndpoint is Unix CSI endpoint + UnixCSIEndpoint = "unix:///csi/csi.sock" ) diff --git a/pkg/installer/config.go b/pkg/installer/config.go index 964930373..2b0ddefc4 100644 --- a/pkg/installer/config.go +++ b/pkg/installer/config.go @@ -130,11 +130,11 @@ func (c *Config) csiDriverName() string { } func (c *Config) daemonsetName() string { - return c.Identity + return "node-server" } func (c *Config) deploymentName() string { - return c.Identity + return "controller" } func (c *Config) getPSPName() string { diff --git a/pkg/installer/const.go b/pkg/installer/const.go index 35ef83ef4..4039b237b 100644 --- a/pkg/installer/const.go +++ b/pkg/installer/const.go @@ -37,6 +37,7 @@ const ( appRootDir = consts.AppRootDir + "/" nodeDriverRegistrarContainerName = "node-driver-registrar" healthZContainerPortName = "healthz" + healthZContainerPort = 9898 livenessProbeContainerName = "liveness-probe" volumeNameSysDir = "sysfs" volumePathSysDir = consts.SysFSDir @@ -59,18 +60,19 @@ const ( validationWebhookConfigName = "drive.validation.controller" // Common - volumeNameSocketDir = "socket-dir" - selectorKey = "selector." + consts.GroupName - containerName = consts.AppName - driveDiscoveryContainerName = consts.AppName + "-drive-discovery" - kubeNodeNameEnvVar = "KUBE_NODE_NAME" - endpointEnvVarCSI = "CSI_ENDPOINT" - kubeletDirPath = "/var/lib/kubelet" - pluginName = "kubectl-" + consts.AppName - selectorValueEnabled = "enabled" - serviceSelector = "selector." + consts.GroupName + ".service" - healthZContainerPortPath = "/healthz" - deleteProtectionFinalizer = "/delete-protection" + volumeNameSocketDir = "socket-dir" + socketDir = "/csi" + socketFile = "/csi.sock" + selectorKey = "selector." + consts.GroupName + containerName = consts.AppName + kubeNodeNameEnvVarName = "KUBE_NODE_NAME" + csiEndpointEnvVarName = "CSI_ENDPOINT" + kubeletDirPath = "/var/lib/kubelet" + pluginName = "kubectl-" + consts.AppName + selectorValueEnabled = "enabled" + serviceSelector = "selector." + consts.GroupName + ".service" + healthZContainerPortPath = "/healthz" + deleteProtectionFinalizer = "/delete-protection" // debug log level default logLevel = 3 diff --git a/pkg/installer/daemonset.go b/pkg/installer/daemonset.go index f15ccfdee..5af166b89 100644 --- a/pkg/installer/daemonset.go +++ b/pkg/installer/daemonset.go @@ -57,28 +57,25 @@ func createDaemonSet(ctx context.Context, c *Config) error { } volumes := []corev1.Volume{ - newHostPathVolume(volumeNameSocketDir, newPluginsSocketDir(kubeletDirPath, c.daemonsetName())), + newHostPathVolume(volumeNameSocketDir, newPluginsSocketDir(kubeletDirPath, c.identity())), newHostPathVolume(volumeNameMountpointDir, kubeletDirPath+"/pods"), newHostPathVolume(volumeNameRegistrationDir, kubeletDirPath+"/plugins_registry"), newHostPathVolume(volumeNamePluginDir, kubeletDirPath+"/plugins"), newHostPathVolume(volumeNameAppRootDir, appRootDir), + newHostPathVolume(volumeNameSysDir, volumePathSysDir), + newHostPathVolume(volumeNameDevDir, volumePathDevDir), + newHostPathVolume(volumeNameRunUdevData, volumePathRunUdevData), } volumeMounts := []corev1.VolumeMount{ - newVolumeMount(volumeNameSocketDir, "/csi", corev1.MountPropagationNone, false), + newVolumeMount(volumeNameSocketDir, socketDir, corev1.MountPropagationNone, false), newVolumeMount(volumeNameMountpointDir, kubeletDirPath+"/pods", corev1.MountPropagationBidirectional, false), newVolumeMount(volumeNamePluginDir, kubeletDirPath+"/plugins", corev1.MountPropagationBidirectional, false), newVolumeMount(volumeNameAppRootDir, appRootDir, corev1.MountPropagationBidirectional, false), + newVolumeMount(volumeNameSysDir, volumePathSysDir, corev1.MountPropagationBidirectional, true), + newVolumeMount(volumeNameDevDir, volumePathDevDir, corev1.MountPropagationHostToContainer, true), + newVolumeMount(volumeNameRunUdevData, volumePathRunUdevData, corev1.MountPropagationBidirectional, true), } - volumes = append(volumes, newHostPathVolume(volumeNameSysDir, volumePathSysDir)) - volumeMounts = append(volumeMounts, newVolumeMount(volumeNameSysDir, volumePathSysDir, corev1.MountPropagationBidirectional, true)) - - volumes = append(volumes, newHostPathVolume(volumeNameDevDir, volumePathDevDir)) - volumeMounts = append(volumeMounts, newVolumeMount(volumeNameDevDir, volumePathDevDir, corev1.MountPropagationHostToContainer, true)) - - volumes = append(volumes, newHostPathVolume(volumeNameRunUdevData, volumePathRunUdevData)) - volumeMounts = append(volumeMounts, newVolumeMount(volumeNameRunUdevData, volumePathRunUdevData, corev1.MountPropagationBidirectional, true)) - podSpec := corev1.PodSpec{ ServiceAccountName: c.serviceAccountName(), HostIPC: false, @@ -91,82 +88,44 @@ func createDaemonSet(ctx context.Context, c *Config) error { Image: path.Join(c.ContainerRegistry, c.ContainerOrg, c.getNodeDriverRegistrarImage()), Args: []string{ fmt.Sprintf("--v=%d", logLevel), - "--csi-address=unix:///csi/csi.sock", + "--csi-address=" + consts.UnixCSIEndpoint, fmt.Sprintf("--kubelet-registration-path=%s", - newPluginsSocketDir(kubeletDirPath, c.daemonsetName())+"/csi.sock"), - }, - Env: []corev1.EnvVar{ - { - Name: kubeNodeNameEnvVar, - ValueFrom: &corev1.EnvVarSource{ - FieldRef: &corev1.ObjectFieldSelector{ - APIVersion: "v1", - FieldPath: "spec.nodeName", - }, - }, - }, + newPluginsSocketDir(kubeletDirPath, c.identity())+socketFile), }, + Env: []corev1.EnvVar{kubeNodeNameEnvVar}, VolumeMounts: []corev1.VolumeMount{ - newVolumeMount(volumeNameSocketDir, "/csi", corev1.MountPropagationNone, false), + newVolumeMount(volumeNameSocketDir, socketDir, corev1.MountPropagationNone, false), newVolumeMount(volumeNameRegistrationDir, "/registration", corev1.MountPropagationNone, false), }, TerminationMessagePolicy: corev1.TerminationMessageFallbackToLogsOnError, TerminationMessagePath: "/var/log/driver-registrar-termination-log", }, { - Name: containerName, + Name: "node-server", Image: path.Join(c.ContainerRegistry, c.ContainerOrg, c.ContainerImage), Args: func() []string { args := []string{ - fmt.Sprintf("--identity=%s", c.daemonsetName()), + "node-server", fmt.Sprintf("-v=%d", logLevel), - fmt.Sprintf("--endpoint=$(%s)", endpointEnvVarCSI), - fmt.Sprintf("--node-id=$(%s)", kubeNodeNameEnvVar), - fmt.Sprintf("--metrics-port=%d", consts.MetricsPort), + fmt.Sprintf("--identity=%s", c.identity()), + fmt.Sprintf("--csi-endpoint=$(%s)", csiEndpointEnvVarName), + fmt.Sprintf("--kube-node-name=$(%s)", kubeNodeNameEnvVarName), fmt.Sprintf("--readiness-port=%d", consts.ReadinessPort), - "--driver", + fmt.Sprintf("--metrics-port=%d", consts.MetricsPort), } return args }(), - SecurityContext: securityContext, - Env: []corev1.EnvVar{ - { - Name: kubeNodeNameEnvVar, - ValueFrom: &corev1.EnvVarSource{ - FieldRef: &corev1.ObjectFieldSelector{ - APIVersion: "v1", - FieldPath: "spec.nodeName", - }, - }, - }, - { - Name: endpointEnvVarCSI, - Value: "unix:///csi/csi.sock", - }, - }, + SecurityContext: securityContext, + Env: []corev1.EnvVar{kubeNodeNameEnvVar, csiEndpointEnvVar}, TerminationMessagePolicy: corev1.TerminationMessageFallbackToLogsOnError, TerminationMessagePath: "/var/log/driver-termination-log", VolumeMounts: volumeMounts, - Ports: []corev1.ContainerPort{ - { - ContainerPort: 9898, - Name: "healthz", - Protocol: corev1.ProtocolTCP, - }, - { - ContainerPort: consts.ReadinessPort, - Name: readinessPortName, - Protocol: corev1.ProtocolTCP, - }, - { - ContainerPort: consts.MetricsPort, - Name: metricsPortName, - Protocol: corev1.ProtocolTCP, - }, - }, - ReadinessProbe: &corev1.Probe{ - ProbeHandler: getReadinessHandler(), - }, + Ports: append(commonContainerPorts, corev1.ContainerPort{ + ContainerPort: consts.MetricsPort, + Name: metricsPortName, + Protocol: corev1.ProtocolTCP, + }), + ReadinessProbe: &corev1.Probe{ProbeHandler: getReadinessHandler()}, LivenessProbe: &corev1.Probe{ FailureThreshold: 5, InitialDelaySeconds: 300, @@ -184,13 +143,13 @@ func createDaemonSet(ctx context.Context, c *Config) error { Name: livenessProbeContainerName, Image: path.Join(c.ContainerRegistry, c.ContainerOrg, c.getLivenessProbeImage()), Args: []string{ - "--csi-address=/csi/csi.sock", - "--health-port=9898", + "--csi-address=" + socketDir + socketFile, + "--health-port=" + fmt.Sprintf("%v", healthZContainerPort), }, TerminationMessagePolicy: corev1.TerminationMessageFallbackToLogsOnError, TerminationMessagePath: "/var/log/driver-liveness-termination-log", VolumeMounts: []corev1.VolumeMount{ - newVolumeMount(volumeNameSocketDir, "/csi", corev1.MountPropagationNone, false), + newVolumeMount(volumeNameSocketDir, socketDir, corev1.MountPropagationNone, false), }, }, }, @@ -205,7 +164,7 @@ func createDaemonSet(ctx context.Context, c *Config) error { annotations["container.apparmor.security.beta.kubernetes.io/"+consts.AppName] = c.ApparmorProfile } - generatedSelectorValue := generateSanitizedUniqueNameFrom(c.daemonsetName()) + generatedSelectorValue := generateSanitizedUniqueNameFrom(c.identity()) daemonset := &appsv1.DaemonSet{ TypeMeta: metav1.TypeMeta{ APIVersion: "apps/v1", diff --git a/pkg/installer/deployment.go b/pkg/installer/deployment.go index b797dc8e1..392c87770 100644 --- a/pkg/installer/deployment.go +++ b/pkg/installer/deployment.go @@ -110,7 +110,7 @@ func createDeployment(ctx context.Context, c *Config) error { newHostPathVolume(volumeNameSocketDir, newPluginsSocketDir(kubeletDirPath, fmt.Sprintf("%s-controller", c.deploymentName()))), } volumeMounts := []corev1.VolumeMount{ - newVolumeMount(volumeNameSocketDir, "/csi", corev1.MountPropagationNone, false), + newVolumeMount(volumeNameSocketDir, socketDir, corev1.MountPropagationNone, false), } if c.AdmissionControl { @@ -129,19 +129,14 @@ func createDeployment(ctx context.Context, c *Config) error { Args: []string{ fmt.Sprintf("--v=%d", logLevel), "--timeout=300s", - fmt.Sprintf("--csi-address=$(%s)", endpointEnvVarCSI), + fmt.Sprintf("--csi-address=$(%s)", csiEndpointEnvVarName), "--leader-election", "--feature-gates=Topology=true", "--strict-topology", }, - Env: []corev1.EnvVar{ - { - Name: endpointEnvVarCSI, - Value: "unix:///csi/csi.sock", - }, - }, + Env: []corev1.EnvVar{csiEndpointEnvVar}, VolumeMounts: []corev1.VolumeMount{ - newVolumeMount(volumeNameSocketDir, "/csi", corev1.MountPropagationNone, false), + newVolumeMount(volumeNameSocketDir, socketDir, corev1.MountPropagationNone, false), }, TerminationMessagePolicy: corev1.TerminationMessageFallbackToLogsOnError, TerminationMessagePath: "/var/log/controller-provisioner-termination-log", @@ -163,54 +158,27 @@ func createDeployment(ctx context.Context, c *Config) error { }, }, { - Name: containerName, + Name: "controller", Image: path.Join(c.ContainerRegistry, c.ContainerOrg, c.ContainerImage), Args: []string{ + "controller", fmt.Sprintf("-v=%d", logLevel), fmt.Sprintf("--identity=%s", c.deploymentName()), - fmt.Sprintf("--endpoint=$(%s)", endpointEnvVarCSI), + fmt.Sprintf("--csi-endpoint=$(%s)", csiEndpointEnvVarName), + fmt.Sprintf("--kube-node-name=$(%s)", kubeNodeNameEnvVarName), fmt.Sprintf("--readiness-port=%d", consts.ReadinessPort), - "--controller", }, SecurityContext: &corev1.SecurityContext{ Privileged: &privileged, }, - Ports: []corev1.ContainerPort{ - { - ContainerPort: admissionControllerWebhookPort, - Name: admissionControllerWebhookName, - Protocol: corev1.ProtocolTCP, - }, - { - ContainerPort: consts.ReadinessPort, - Name: readinessPortName, - Protocol: corev1.ProtocolTCP, - }, - { - ContainerPort: 9898, - Name: "healthz", - Protocol: corev1.ProtocolTCP, - }, - }, - ReadinessProbe: &corev1.Probe{ - ProbeHandler: getReadinessHandler(), - }, - Env: []corev1.EnvVar{ - { - Name: kubeNodeNameEnvVar, - ValueFrom: &corev1.EnvVarSource{ - FieldRef: &corev1.ObjectFieldSelector{ - APIVersion: "v1", - FieldPath: "spec.nodeName", - }, - }, - }, - { - Name: endpointEnvVarCSI, - Value: "unix:///csi/csi.sock", - }, - }, - VolumeMounts: volumeMounts, + Ports: append(commonContainerPorts, corev1.ContainerPort{ + ContainerPort: admissionControllerWebhookPort, + Name: admissionControllerWebhookName, + Protocol: corev1.ProtocolTCP, + }), + ReadinessProbe: &corev1.Probe{ProbeHandler: getReadinessHandler()}, + Env: []corev1.EnvVar{kubeNodeNameEnvVar, csiEndpointEnvVar}, + VolumeMounts: volumeMounts, }, }, } @@ -259,7 +227,7 @@ func createDeployment(ctx context.Context, c *Config) error { }, Status: appsv1.DeploymentStatus{}, } - deployment.ObjectMeta.Finalizers = []string{ + deployment.Finalizers = []string{ c.namespace() + deleteProtectionFinalizer, } diff --git a/pkg/installer/psp.go b/pkg/installer/psp.go index 11978089a..9cf3fda26 100644 --- a/pkg/installer/psp.go +++ b/pkg/installer/psp.go @@ -53,8 +53,8 @@ func createPodSecurityPolicy(ctx context.Context, i *Config) error { {PathPrefix: consts.ProcFSDir, ReadOnly: true}, {PathPrefix: consts.SysFSDir, ReadOnly: true}, {PathPrefix: consts.AppRootDir}, - {PathPrefix: "/csi"}, - {PathPrefix: "/var/lib/kubelet"}, + {PathPrefix: socketDir}, + {PathPrefix: kubeletDirPath}, }, SELinux: policy.SELinuxStrategyOptions{ Rule: policy.SELinuxStrategyRunAsAny, diff --git a/pkg/installer/vars.go b/pkg/installer/vars.go new file mode 100644 index 000000000..cbdf47bf6 --- /dev/null +++ b/pkg/installer/vars.go @@ -0,0 +1,52 @@ +// This file is part of MinIO DirectPV +// Copyright (c) 2022 MinIO, Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +package installer + +import ( + "github.com/minio/directpv/pkg/consts" + corev1 "k8s.io/api/core/v1" +) + +var ( + kubeNodeNameEnvVar = corev1.EnvVar{ + Name: kubeNodeNameEnvVarName, + ValueFrom: &corev1.EnvVarSource{ + FieldRef: &corev1.ObjectFieldSelector{ + APIVersion: "v1", + FieldPath: "spec.nodeName", + }, + }, + } + + csiEndpointEnvVar = corev1.EnvVar{ + Name: csiEndpointEnvVarName, + Value: consts.UnixCSIEndpoint, + } + + commonContainerPorts = []corev1.ContainerPort{ + { + ContainerPort: consts.ReadinessPort, + Name: readinessPortName, + Protocol: corev1.ProtocolTCP, + }, + { + ContainerPort: healthZContainerPort, + Name: healthZContainerPortName, + Protocol: corev1.ProtocolTCP, + }, + } +) diff --git a/pkg/k8s/k8s.go b/pkg/k8s/k8s.go index eb4e81c1c..a48bf5ef7 100644 --- a/pkg/k8s/k8s.go +++ b/pkg/k8s/k8s.go @@ -79,7 +79,6 @@ func GetGroupVersionKind(group, kind string, versions ...string) (*schema.GroupV versions..., ) if err != nil { - klog.ErrorS(err, "unable to get REST mapping") return nil, err } diff --git a/pkg/volume/list_test.go b/pkg/volume/list_test.go index e25f9a862..455d0ba5b 100644 --- a/pkg/volume/list_test.go +++ b/pkg/volume/list_test.go @@ -99,13 +99,13 @@ func TestGetSortedVolumeList(t *testing.T) { if err != nil { t.Fatalf("unexpected error: %v", err) } - if volumes[0].ObjectMeta.Namespace != "AAA" { - t.Fatalf("expected volume to be in Namespace : AAA, got: %v", volumes[0].ObjectMeta.Namespace) + if volumes[0].Namespace != "AAA" { + t.Fatalf("expected volume to be in Namespace : AAA, got: %v", volumes[0].Namespace) } - if volumes[4].ObjectMeta.Namespace != "BBB" { - t.Fatalf("expected volume to be in Namespace : BBB, got: %v", volumes[3].ObjectMeta.Namespace) + if volumes[4].Namespace != "BBB" { + t.Fatalf("expected volume to be in Namespace : BBB, got: %v", volumes[3].Namespace) } - if volumes[8].ObjectMeta.Namespace != "CCC" { - t.Fatalf("expected volume to be in Namespace : CCC, got: %v", volumes[7].ObjectMeta.Namespace) + if volumes[8].Namespace != "CCC" { + t.Fatalf("expected volume to be in Namespace : CCC, got: %v", volumes[7].Namespace) } }