Merge pull request #4511 from leseb/error-withmessage

ceph: do not print extended format for loggers
This commit is contained in:
Travis Nielsen authored and GitHub committed 2019-12-16 16:51:46 -07:00
commit cfa34d2973
57 files changed
+356 -356

No files matched your search

+5 -5
View File
@@ -82,12 +82,12 @@ func (c *ClusterController) onDelete(obj interface{}) {
func (c *ClusterController) handleClusterDelete(cluster *cephv1.CephCluster, retryInterval time.Duration) {
node := os.Getenv(k8sutil.NodeNameEnvVar)
agentNamespace := os.Getenv(k8sutil.PodNamespaceEnvVar)
logger.Infof("cluster in namespace %s is being deleted, agent on node %s will attempt to clean up.", cluster.Namespace, node)
logger.Infof("cluster in namespace %q is being deleted, agent on node %q will attempt to clean up.", cluster.Namespace, node)
// TODO: filter this List operation by node name and cluster namespace on the server side
vols, err := c.volumeAttachment.List(agentNamespace)
if err != nil {
logger.Errorf("failed to get volume attachments for agent namespace %s: %+v", agentNamespace, err)
logger.Errorf("failed to get volume attachments for agent namespace %s. %v", agentNamespace, err)
}
var waitGroup sync.WaitGroup
@@ -97,7 +97,7 @@ func (c *ClusterController) handleClusterDelete(cluster *cephv1.CephCluster, ret
for _, vol := range vols.Items {
for _, a := range vol.Attachments {
if a.Node == node && a.ClusterName == cluster.Namespace {
logger.Infof("volume %s has an attachment belonging to deleted cluster %s, will clean it up now. mountDir: %s",
logger.Infof("volume %q has an attachment belonging to deleted cluster %q, will clean it up now. mountDir: %q",
vol.Name, cluster.Namespace, a.MountDir)
// we will perform all the cleanup asynchronously later on. Right now, just add this one
@@ -115,9 +115,9 @@ func (c *ClusterController) handleClusterDelete(cluster *cephv1.CephCluster, ret
go func(mountDir string) {
defer waitGroup.Done()
if err := c.cleanupVolumeAttachment(mountDir, retryInterval); err != nil {
logger.Errorf("failed to clean up attachment for mountDir %s: %+v", mountDir, err)
logger.Errorf("failed to clean up attachment for mountDir %q. %v", mountDir, err)
} else {
logger.Infof("cleaned up attachment for mountDir %s", mountDir)
logger.Infof("cleaned up attachment for mountDir %q", mountDir)
}
}(cleanupList[i])
}
@@ -64,10 +64,10 @@ func (c *crd) Create(volumeAttachment *rookalpha.Volume) error {
func (c *crd) Update(volumeAttachment *rookalpha.Volume) error {
_, err := c.context.RookClientset.RookV1alpha2().Volumes(volumeAttachment.Namespace).Update(volumeAttachment)
if err != nil {
logger.Errorf("failed to update Volume CRD. %+v", err)
logger.Errorf("failed to update Volume CRD. %v", err)
return err
}
logger.Infof("updated Volumeattach CRD %s", volumeAttachment.ObjectMeta.Name)
logger.Infof("updated Volumeattach CRD %q", volumeAttachment.ObjectMeta.Name)
return nil
}
@@ -82,7 +82,7 @@ func (vm *VolumeManager) Init() error {
// check to see if the rbd kernel module has single_major support
hasSingleMajor, err := sys.CheckKernelModuleParam(rbdKernelModuleName, "single_major", vm.context.Executor)
if err != nil {
logger.Noticef("failed %s single_major check, assuming it's unsupported: %+v", rbdKernelModuleName, err)
logger.Noticef("failed %q single_major check, assuming it's unsupported. %v", rbdKernelModuleName, err)
hasSingleMajor = false
}
@@ -93,7 +93,7 @@ func (vm *VolumeManager) Init() error {
// load the rbd kernel module with options
if err := sys.LoadKernelModule(rbdKernelModuleName, opts, vm.context.Executor); err != nil {
logger.Noticef("failed to load kernel module %s: %+v", rbdKernelModuleName, err)
logger.Noticef("failed to load kernel module %q. %v", rbdKernelModuleName, err)
return err
}
@@ -157,7 +157,7 @@ func (vm *VolumeManager) Attach(image, pool, id, key, clusterNamespace string) (
return "", errors.Wrapf(err, "exceeded retry count while finding device path")
}
logger.Infof("failed to find device path, sleeping 1 second: %+v", err)
logger.Infof("failed to find device path, sleeping 1 second. %v", err)
<-time.After(time.Second)
}
}
+15 -15
View File
@@ -94,17 +94,17 @@ func (s *FlexvolumeServer) Start(driverVendor, driverName string) error {
listener, err := net.Listen("unix", unixSocketFile)
if err != nil {
return errors.Wrapf(err, "unable to listen at %s", unixSocketFile)
return errors.Wrapf(err, "unable to listen at %q", unixSocketFile)
}
s.listeners[unixSocketFile] = listener
if err := os.Chmod(unixSocketFile, 0770); err != nil {
return errors.Wrapf(err, "unable to set file permission to unix socket %s", unixSocketFile)
return errors.Wrapf(err, "unable to set file permission to unix socket %q", unixSocketFile)
}
go rpc.Accept(listener)
logger.Infof("Listening on unix socket for Kubernetes volume attach commands: %s", unixSocketFile)
logger.Infof("listening on unix socket for Kubernetes volume attach commands %q", unixSocketFile)
return nil
}
@@ -113,12 +113,12 @@ func (s *FlexvolumeServer) StopAll() {
logger.Infof("Stopping %d unix socket rpc server(s).", len(s.listeners))
for unixSocketFile, listener := range s.listeners {
if err := listener.Close(); err != nil {
logger.Errorf("Failed to stop unix socket rpc server: %+v", err)
logger.Errorf("failed to stop unix socket rpc server. %v", err)
}
// closing the listener should remove the unix socket file. But lets try it remove it just in case.
if _, err := os.Stat(unixSocketFile); !os.IsNotExist(err) {
logger.Infof("Deleting unix domain socket file %s.", unixSocketFile)
logger.Infof("deleting unix domain socket file %q.", unixSocketFile)
os.Remove(unixSocketFile)
}
}
@@ -135,7 +135,7 @@ func RookDriverName(context *clusterd.Context) (string, error) {
// TouchFlexDrivers causes k8s to reload the flex volumes. Needed periodically due to a k8s race condition with flex driver loading.
func TouchFlexDrivers(vendor, driverName string) {
filename := path.Join(fmt.Sprintf(flexMountPath, vendor, driverName), driverName)
logger.Debugf("reloading flex drivers. touching %s", filename)
logger.Debugf("reloading flex drivers. touching %q", filename)
currenttime := time.Now().Local()
err := os.Chtimes(filename, currenttime, currenttime)
@@ -195,7 +195,7 @@ func configureFlexVolume(driverFile, driverDir, driverName string) error {
if _, err := os.Stat(driverDir); os.IsNotExist(err) {
err := os.Mkdir(driverDir, 0755)
if err != nil {
logger.Errorf("failed to create dir %s. %+v", driverDir, err)
logger.Errorf("failed to create dir %q. %v", driverDir, err)
}
}
@@ -203,7 +203,7 @@ func configureFlexVolume(driverFile, driverDir, driverName string) error {
finalDestFile := path.Join(driverDir, driverName) // /flextmnt/rook.io~rook-system/rook-system
err := copyFile(driverFile, destFile)
if err != nil {
return errors.Wrapf(err, "unable to copy flexvolume from %s to %s", driverFile, destFile)
return errors.Wrapf(err, "unable to copy flexvolume from %q to %q", driverFile, destFile)
}
// renaming flex volume. Rename is an atomic execution while copying is not.
@@ -211,34 +211,34 @@ func configureFlexVolume(driverFile, driverDir, driverName string) error {
// Delete old plugin if it exists
err = os.Remove(finalDestFile)
if err != nil {
logger.Warningf("Could not delete old Rook Flexvolume driver at %s: %v", finalDestFile, err)
logger.Warningf("Could not delete old Rook Flexvolume driver at %q. %v", finalDestFile, err)
}
}
if err := os.Rename(destFile, finalDestFile); err != nil {
return errors.Wrapf(err, "failed to rename %s to %s", destFile, finalDestFile)
return errors.Wrapf(err, "failed to rename %q to %q", destFile, finalDestFile)
}
// Write the flex configuration
enableSELinuxRelabeling, err := strconv.ParseBool(os.Getenv(agent.RookEnableSelinuxRelabelingEnv))
if err != nil {
logger.Errorf("invalid value for disabling SELinux relabeling. %+v", err)
logger.Errorf("invalid value for disabling SELinux relabeling. %v", err)
enableSELinuxRelabeling = true
}
enableFSGroup, err := strconv.ParseBool(os.Getenv(agent.RookEnableFSGroupEnv))
if err != nil {
logger.Errorf("invalid value for disabling fs group. %+v", err)
logger.Errorf("invalid value for disabling fs group. %v", err)
enableFSGroup = true
}
settings, err := generateFlexSettings(enableSELinuxRelabeling, enableFSGroup)
if err != nil {
logger.Errorf("invalid flex settings. %+v", err)
logger.Errorf("invalid flex settings. %v", err)
} else {
if err := ioutil.WriteFile(path.Join(driverDir, settingsFilename), settings, 0644); err != nil {
logger.Errorf("failed to write settings file %s. %+v", settingsFilename, err)
logger.Errorf("failed to write settings file %q. %v", settingsFilename, err)
} else {
logger.Debugf("flex settings: %s", string(settings))
logger.Debugf("flex settings: %q", string(settings))
}
}
+8 -8
View File
@@ -136,7 +136,7 @@ func CreateFilesystem(context *clusterd.Context, clusterName, name, metadataPool
_, err = NewCephCommand(context, clusterName, args).Run()
if err != nil {
// continue if this fails
logger.Warning("failed enabling multiple file systems. %+v", err)
logger.Warning("failed enabling multiple file systems. %v", err)
}
}
@@ -145,11 +145,11 @@ func CreateFilesystem(context *clusterd.Context, clusterName, name, metadataPool
// Force to use pre-existing pools
if force {
args = append(args, "--force")
logger.Infof("Filesystem %s will reuse pre-existing pools", name)
logger.Infof("Filesystem %q will reuse pre-existing pools", name)
}
_, err = NewCephCommand(context, clusterName, args).Run()
if err != nil {
return errors.Wrapf(err, "failed enabling ceph fs %s", name)
return errors.Wrapf(err, "failed enabling ceph fs %q", name)
}
// add each additional pool
@@ -158,7 +158,7 @@ func CreateFilesystem(context *clusterd.Context, clusterName, name, metadataPool
args = []string{"fs", "add_data_pool", name, poolName}
_, err = NewCephCommand(context, clusterName, args).Run()
if err != nil {
logger.Errorf("failed to add pool %s to file system %s. %+v", poolName, name, err)
logger.Errorf("failed to add pool %q to file system %q. %v", poolName, name, err)
}
}
@@ -243,7 +243,7 @@ func deactivateMdsWithRetry(context *clusterd.Context, mdsGid int, namespace, fs
time.Sleep(retrySleep)
}
// report most recent error with additional err info
return errors.Wrapf(err, "failed to deactivate mds w/ gid %d for filesystem %s", mdsGid, fsName)
return errors.Wrapf(err, "failed to deactivate mds w/ gid %d for filesystem %q", mdsGid, fsName)
}
// WaitForActiveRanks waits for the filesystem's number of active ranks to equal the desired count.
@@ -266,14 +266,14 @@ func WaitForActiveRanks(
fs, err := GetFilesystem(context, clusterName, fsName)
if err != nil {
logger.Errorf(
"Error getting filesystem %s details while waiting for num mds ranks to become %d: %+v",
"Error getting filesystem %q details while waiting for num mds ranks to become %d. %v",
fsName, desiredActiveRanks, err)
} else if fs.MDSMap.MaxMDS == int(desiredActiveRanks) &&
activeRanksSuccess(len(fs.MDSMap.Up), int(desiredActiveRanks), moreIsOkay) {
// Both max_mds and number of up MDS daemons must equal desired number of ranks to
// prevent a false positive when Ceph has got the correct number of mdses up but is
// trying to change the number of mdses up to an undesired number.
logger.Debugf("mds ranks for filesystem %s successfully became %d", fsName, desiredActiveRanks)
logger.Debugf("mds ranks for filesystem %q successfully became %d", fsName, desiredActiveRanks)
return true, nil
// continue to inf loop after send ready; only return when get quit signal to
// prevent deadlock
@@ -281,7 +281,7 @@ func WaitForActiveRanks(
return false, nil
})
if err != nil {
return errors.Errorf("timeout waiting for number active mds daemons for filesystem %s to become %s",
return errors.Errorf("timeout waiting for number active mds daemons for filesystem %q to become %q",
fsName, countText)
}
return nil
+5 -5
View File
@@ -189,29 +189,29 @@ func DeletePool(context *clusterd.Context, clusterName string, name string) erro
// check if the pool exists
pool, err := GetPoolDetails(context, clusterName, name)
if err != nil {
logger.Infof("pool %s not found for deletion. %+v", name, err)
logger.Infof("pool %q not found for deletion. %v", name, err)
return nil
}
err = checkForImagesInPool(context, name, clusterName)
if err != nil {
return errors.Wrapf(err, "failed to delete pool %s", name)
return errors.Wrapf(err, "failed to delete pool %q", name)
}
logger.Infof("purging pool %s (id=%d)", name, pool.Number)
args := []string{"osd", "pool", "delete", name, name, reallyConfirmFlag}
_, err = NewCephCommand(context, clusterName, args).Run()
if err != nil {
return errors.Wrapf(err, "failed to delete pool %s", name)
return errors.Wrapf(err, "failed to delete pool %q", name)
}
// remove the crush rule for this pool and ignore the error in case the rule is still in use or not found
args = []string{"osd", "crush", "rule", "rm", name}
_, err = NewCephCommand(context, clusterName, args).Run()
if err != nil {
logger.Infof("did not delete crush rule %s. %+v", name, err)
logger.Infof("did not delete crush rule %q. %v", name, err)
}
logger.Infof("purge completed for pool %s", name)
logger.Infof("purge completed for pool %q", name)
return nil
}
+4 -4
View File
@@ -221,10 +221,10 @@ func getMDSRank(status CephStatus, clusterName, fsName string) (int, error) {
}
}
// if the mds is not shown in the map one reason might be because it's in standby
// if not in standby there is something else going wron
// if not in standby there is something else going wrong
if mdsRank < 0 && status.Fsmap.UpStandby < 1 {
// it might seem strange to log an error since this could be a warning too
// it is a warning until we reach the timeout, this should give enough time to the mds to transtion its state
// it is a warning until we reach the timeout, this should give enough time to the mds to transition its state
// after the timeout we consider that the mds might be gone or the timeout was not long enough...
return mdsRank, errors.Errorf("mds %s not found in fsmap, this likely means mdss are transitioning between active and standby states", fsName)
}
@@ -244,7 +244,7 @@ func MdsActiveOrStandbyReplay(context *clusterd.Context, clusterName, fsName str
return errors.Cause(err)
}
// this MDS is in standby so let's return immediatly
// this MDS is in standby so let's return immediately
if mdsRank < 0 {
logger.Infof("mds %s is in standby, nothing to check", fsName)
return nil
@@ -263,7 +263,7 @@ func MdsActiveOrStandbyReplay(context *clusterd.Context, clusterName, fsName str
func IsCephHealthy(context *clusterd.Context, clusterName string) bool {
cephStatus, err := Status(context, clusterName, false)
if err != nil {
logger.Errorf("failed to detect if Ceph is healthy. failed to get ceph status. %+v", err)
logger.Errorf("failed to detect if Ceph is healthy. failed to get ceph status. %v", err)
return false
}
+2 -2
View File
@@ -277,7 +277,7 @@ func LeastUptodateDaemonVersion(context *clusterd.Context, clusterName, daemonTy
// Always invoke ceph version before an upgrade so we are sure to be up-to-date
versions, err := GetAllCephDaemonVersions(context, clusterName)
if err != nil {
logger.Warningf("failed to get ceph daemons versions. %+v, this likely means there is no cluster yet.", err)
logger.Warningf("failed to get ceph daemons versions, this likely means there is no cluster yet. %v", err)
} else {
r, err = daemonMapEntry(versions, daemonType)
if err != nil {
@@ -380,7 +380,7 @@ func buildHostListFromTree(tree OsdTree) (OsdTree, error) {
func osdDoNothing(context *clusterd.Context, clusterName string) bool {
versions, err := GetAllCephDaemonVersions(context, clusterName)
if err != nil {
logger.Warningf("failed to get ceph daemons versions. %+v, this likely means there is no cluster yet.", err)
logger.Warningf("failed to get ceph daemons versions, this likely means there is no cluster yet. %v", err)
return true
}
+1 -1
View File
@@ -142,7 +142,7 @@ func GenerateConfigFile(context *clusterd.Context, cluster *ClusterInfo, pathRoo
// create the config directory
if err := os.MkdirAll(pathRoot, 0744); err != nil {
logger.Warningf("failed to create config directory at %s: %+v", pathRoot, err)
logger.Warningf("failed to create config directory at %q. %v", pathRoot, err)
}
configFile, err := createGlobalConfigFileSection(context, cluster, globalConfig)
+6 -6
View File
@@ -114,7 +114,7 @@ func (a *OsdAgent) configureDirs(context *clusterd.Context, dirs map[string]int)
osd, err := a.prepareOSD(context, cfg)
if err != nil {
logger.Errorf("failed to config osd in path %s. %+v", dirPath, err)
logger.Errorf("failed to config osd in path %q. %v", dirPath, err)
lastErr = err
} else {
succeeded++
@@ -135,7 +135,7 @@ func (a *OsdAgent) configureDevices(context *clusterd.Context, devices *DeviceOs
} else {
cvSupported, err = getCephVolumeSupported(context)
if err != nil {
logger.Errorf("failed to detect if ceph-volume is available. %+v", err)
logger.Errorf("failed to detect if ceph-volume is available. %v", err)
}
}
@@ -167,7 +167,7 @@ func (a *OsdAgent) configureDevices(context *clusterd.Context, devices *DeviceOs
// compute an OSD layout scheme that will optimize performance
scheme, cvDevices, err := a.getPartitionPerfScheme(context, devices, cvSupported)
logger.Debugf("partition scheme: %+v, err: %+v", scheme, err)
logger.Debugf("partition scheme: %+v. %v", scheme, err)
if err != nil {
return osds, errors.Wrapf(err, "failed to get OSD partition scheme")
}
@@ -447,7 +447,7 @@ func (a *OsdAgent) prepareOSD(context *clusterd.Context, cfg *osdConfig) (*oposd
// update the osd config file
err := writeConfigFile(cfg, context, a.cluster)
if err != nil {
logger.Warningf("failed to update config file. %+v", err)
logger.Warningf("failed to update config file. %v", err)
}
// osd_data_dir/ready already exists, meaning the OSD is already set up.
@@ -478,7 +478,7 @@ func prepareOSDRoot(cfg *osdConfig) (newOSD bool, err error) {
// osd is new (it's not ready), make sure there is no stale state in the OSD dir by deleting the entire thing
logger.Infof("osd.%d appears to be new, cleaning the root dir at %s", cfg.id, cfg.rootPath)
if err := os.RemoveAll(cfg.rootPath); err != nil {
logger.Warningf("failed to clean osd.%d root dir at %s, will proceed with starting osd: %+v", cfg.id, cfg.rootPath, err)
logger.Warningf("failed to clean osd.%d root dir at %q, will proceed with starting osd. %v", cfg.id, cfg.rootPath, err)
}
// prepare the osd dir by creating it now
@@ -517,7 +517,7 @@ func (a *OsdAgent) removeOSDConfigDir(configRoot string, id int) error {
osdRootDir := getOSDRootDir(configRoot, id)
logger.Infof("deleting osd dir: %s", osdRootDir)
if err := os.RemoveAll(osdRootDir); err != nil {
logger.Warningf("failed to delete osd.%d root dir from %s, it may need to be cleaned up manually: %+v",
logger.Warningf("failed to delete osd.%d root dir from %q, it may need to be cleaned up manually. %v",
id, osdRootDir, err)
}
+9 -9
View File
@@ -48,7 +48,7 @@ func StartOSD(context *clusterd.Context, osdType, osdID, osdUUID, lvPath string,
configDir := fmt.Sprintf("/var/lib/ceph/osd/ceph-%s", osdID)
err := os.Mkdir(configDir, 0755)
if err != nil {
logger.Errorf("failed to create config dir %s. %+v", configDir, err)
logger.Errorf("failed to create config dir %q. %v", configDir, err)
}
// Update LVM config at runtime
@@ -60,7 +60,7 @@ func StartOSD(context *clusterd.Context, osdType, osdID, osdUUID, lvPath string,
if pvcBackedOSD && !lvBackedPV {
volumeGroupName, err = getVolumeGroupName(lvPath)
if err != nil {
return errors.Wrapf(err, "error fetching volume group name for OSD %s", osdID)
return errors.Wrapf(err, "error fetching volume group name for OSD %q", osdID)
}
go handleTerminate(context, lvPath, volumeGroupName)
@@ -82,7 +82,7 @@ func StartOSD(context *clusterd.Context, osdType, osdID, osdUUID, lvPath string,
// run the ceph-osd daemon
if err := context.Executor.ExecuteCommand(false, "", "ceph-osd", cephArgs...); err != nil {
// Instead of returning, we want to allow the lvm release to happen below, so we just log the err
logger.Errorf("failed to start osd or shutting down. %+v", err)
logger.Errorf("failed to start osd or shutting down. %v", err)
}
if pvcBackedOSD && !lvBackedPV {
@@ -114,7 +114,7 @@ func killCephOSDProcess(context *clusterd.Context, lvPath string) error {
pid, err := context.Executor.ExecuteCommandWithOutput(false, "", "fuser", "-a", lvPath)
if err != nil {
return errors.Wrapf(err, "failed to retrieve process ID for %s", lvPath)
return errors.Wrapf(err, "failed to retrieve process ID for %q", lvPath)
}
logger.Infof("process ID for ceph-osd: %s", pid)
@@ -181,13 +181,13 @@ func Provision(context *clusterd.Context, agent *OsdAgent, crushLocation string)
}
err = ioutil.WriteFile(cephconfig.DefaultConfigFilePath(), src, 0444)
if err != nil {
return errors.Wrapf(err, "failed to copy connection config to /etc/ceph. failed to write %s", cephconfig.DefaultConfigFilePath())
return errors.Wrapf(err, "failed to copy connection config to /etc/ceph. failed to write %q", cephconfig.DefaultConfigFilePath())
}
dst, err := ioutil.ReadFile(cephconfig.DefaultConfigFilePath())
if err == nil {
logger.Debugf("config file @ %s: %s", cephconfig.DefaultConfigFilePath(), dst)
} else {
logger.Warningf("wrote and copied config file but failed to read it back from %s for logging. %+v", cephconfig.DefaultConfigFilePath(), err)
logger.Warningf("wrote and copied config file but failed to read it back from %s for logging. %v", cephconfig.DefaultConfigFilePath(), err)
}
logger.Infof("discovering hardware")
@@ -313,7 +313,7 @@ func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevi
// the desired devices is a regular expression
matched, err = regexp.Match(desiredDevice.Name, []byte(device.Name))
if err != nil {
logger.Errorf("regex failed on device %q and filter %q. %+v", device.Name, desiredDevice.Name, err)
logger.Errorf("regex failed on device %q and filter %q. %v", device.Name, desiredDevice.Name, err)
continue
}
logger.Infof("device %q matches device filter %q: %t", device.Name, desiredDevice.Name, matched)
@@ -322,7 +322,7 @@ func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevi
for _, pathname := range pathnames {
matched, err = regexp.Match(desiredDevice.Name, []byte(pathname))
if err != nil {
logger.Errorf("regex failed on device %q and filter %q. %+v", device.Name, desiredDevice.Name, err)
logger.Errorf("regex failed on device %q and filter %q. %v", device.Name, desiredDevice.Name, err)
continue
}
if matched {
@@ -345,7 +345,7 @@ func getAvailableDevices(context *clusterd.Context, desiredDevices []DesiredDevi
logger.Infof("device %q is selected by the device filter/name %q", device.Name, matchedDevice.Name)
deviceInfo = &DeviceOsdIDEntry{Data: unassignedOSDID, Config: matchedDevice, PersistentDevicePaths: strings.Fields(device.DevLinks)}
} else {
logger.Infof("skipping device %q that does not match the device filter/list (%v). %+v", device.Name, desiredDevices, err)
logger.Infof("skipping device %q that does not match the device filter/list (%v). %v", device.Name, desiredDevices, err)
}
} else {
logger.Infof("skipping device %q until the admin specifies it can be used by an osd", device.Name)
+1 -1
View File
@@ -274,7 +274,7 @@ func prepareFilestoreDevice(context *clusterd.Context, cfg *osdConfig, doFormat
if doFormat {
// perform the format and retry if needed
if err = sys.FormatDevice(dataPartPath, context.Executor); err != nil {
logger.Warningf("first attempt to format partition %s on device %s failed. Waiting 2 seconds then retrying: %+v",
logger.Warningf("first attempt to format partition %q on device %q failed. Waiting 2 seconds then retrying. %v",
dataPartDetails.PartitionUUID, dataPartDetails.Device, err)
<-time.After(2 * time.Second)
if err = sys.FormatDevice(dataPartPath, context.Executor); err != nil {
+3 -3
View File
@@ -174,12 +174,12 @@ func backupOSDFileSystem(config *osdConfig, clusterName string) error {
content, err := ioutil.ReadFile(filepath.Join(config.rootPath, fi.Name()))
if err != nil {
logger.Warningf("failed to read file %s: %+v", fi.Name(), err)
logger.Warningf("failed to read file %q. %v", fi.Name(), err)
continue
}
if err := config.kv.SetValue(storeName, fi.Name(), string(content)); err != nil {
logger.Warningf("failed to backup file %s: %+v", fi.Name(), err)
logger.Warningf("failed to backup file %q. %v", fi.Name(), err)
continue
}
}
@@ -207,7 +207,7 @@ func repairOSDFileSystem(config *osdConfig) error {
for fileName, content := range store {
filePath := filepath.Join(config.rootPath, fileName)
if err := ioutil.WriteFile(filePath, []byte(content), 0644); err != nil {
logger.Warningf("failed to restore file %s: %+v", filePath, err)
logger.Warningf("failed to restore file %q. %v", filePath, err)
continue
}
}
+4 -4
View File
@@ -62,7 +62,7 @@ func (a *OsdAgent) configureCVDevices(context *clusterd.Context, devices *Device
logger.Infof("no new devices to configure. returning devices already configured with ceph-volume.")
osds, err = getCephVolumeOSDs(context, a.cluster.Name, a.cluster.FSID, lv, false, lvBackedPV)
if err != nil {
logger.Infof("failed to get devices already provisioned by ceph-volume. %+v", err)
logger.Infof("failed to get devices already provisioned by ceph-volume. %v", err)
}
return osds, nil
}
@@ -457,14 +457,14 @@ func getCephVolumeOSDs(context *clusterd.Context, clusterName string, cephfsid s
for name, osdInfo := range cephVolumeResult {
id, err := strconv.Atoi(name)
if err != nil {
logger.Errorf("bad osd returned from ceph-volume: %s", name)
logger.Errorf("bad osd returned from ceph-volume: %q", name)
continue
}
var osdFSID string
isFilestore := false
for _, osd := range osdInfo {
if osd.Tags.ClusterFSID != cephfsid {
logger.Infof("skipping osd%d: %s running on a different ceph cluster: %s", id, osd.Tags.OSDFSID, osd.Tags.ClusterFSID)
logger.Infof("skipping osd%d: %q running on a different ceph cluster %q", id, osd.Tags.OSDFSID, osd.Tags.ClusterFSID)
continue
}
osdFSID = osd.Tags.OSDFSID
@@ -473,7 +473,7 @@ func getCephVolumeOSDs(context *clusterd.Context, clusterName string, cephfsid s
}
}
if len(osdFSID) == 0 {
logger.Infof("Skipping osd%d as no instances are running on ceph cluster: %s", id, cephfsid)
logger.Infof("Skipping osd%d as no instances are running on ceph cluster %q", id, cephfsid)
continue
}
logger.Infof("osdInfo has %d elements. %+v", len(osdInfo), osdInfo)
+3 -3
View File
@@ -68,7 +68,7 @@ func FindRBDMappedFile(imageName, poolName, sysBusDir string) (string, error) {
func GetIPFromEndpoint(endpoint string) string {
host, _, err := net.SplitHostPort(endpoint)
if err != nil {
logger.Errorf("failed to split ip and port for endpoint %s", endpoint)
logger.Errorf("failed to split ip and port for endpoint %q. %v", endpoint, err)
}
return host
}
@@ -78,11 +78,11 @@ func GetPortFromEndpoint(endpoint string) int32 {
var port int
_, portString, err := net.SplitHostPort(endpoint)
if err != nil {
logger.Errorf("failed to split host and port for endpoint %s, assuming default Ceph port %s", endpoint, portString)
logger.Errorf("failed to split host and port for endpoint %q, assuming default Ceph port %q. %v", endpoint, portString, err)
} else {
port, err = strconv.Atoi(portString)
if err != nil {
logger.Errorf("failed to convert %s to integer", portString)
logger.Errorf("failed to convert %q to integer. %v", portString, err)
}
}
return int32(port)
+3 -3
View File
@@ -247,7 +247,7 @@ func (a *Agent) createAgentDaemonSet(namespace, agentImage, serviceAccount strin
tolerationsRaw := os.Getenv(agentDaemonsetTolerationsEnv)
tolerations, err := k8sutil.YamlToTolerations(tolerationsRaw)
if err != nil {
logger.Warningf("failed to parse %s. %+v", tolerationsRaw, err)
logger.Warningf("failed to parse %q. %v", tolerationsRaw, err)
}
ds.Spec.Template.Spec.Tolerations = append(ds.Spec.Template.Spec.Tolerations, tolerations...)
@@ -256,7 +256,7 @@ func (a *Agent) createAgentDaemonSet(namespace, agentImage, serviceAccount strin
if nodeAffinity != "" {
v1NodeAffinity, err := k8sutil.GenerateNodeAffinity(nodeAffinity)
if err != nil {
logger.Errorf("failed to create NodeAffinity. %+v", err)
logger.Errorf("failed to create NodeAffinity. %v", err)
} else {
ds.Spec.Template.Spec.Affinity = &v1.Affinity{
NodeAffinity: v1NodeAffinity,
@@ -304,7 +304,7 @@ func (a *Agent) discoverFlexvolumeDir() (flexvolumeDirPath, source string) {
// unmarshal to a KubeletConfiguration
kubeletConfiguration := KubeletConfiguration{}
if err := json.Unmarshal(nodeConfig, &kubeletConfiguration); err != nil {
logger.Warningf("unable to parse node config as kubelet configuration: %+v", err)
logger.Warningf("unable to parse node config as kubelet configuration. %v", err)
} else {
flexvolumeDirPath = kubeletConfiguration.KubeletConfig.VolumePluginDir
}
+10 -10
View File
@@ -75,7 +75,7 @@ func (c *ClientController) StartWatch(stopCh chan struct{}) error {
DeleteFunc: c.onDelete,
}
logger.Infof("start watching client resources in namespace %s", c.namespace)
logger.Infof("start watching client resources in namespace %q", c.namespace)
watcher := opkit.NewWatcher(ClientResource, c.namespace, resourceHandlerFuncs, c.context.RookClientset.CephV1().RESTClient())
go watcher.Watch(&cephv1.CephClient{}, stopCh)
@@ -87,13 +87,13 @@ func (c *ClientController) onAdd(obj interface{}) {
client, err := getClientObject(obj)
if err != nil {
logger.Errorf("failed to get client object: %+v", err)
logger.Errorf("failed to get client object. %v", err)
return
}
err = createClient(c.context, client)
if err != nil {
logger.Errorf("failed to create client %s. %+v", client.ObjectMeta.Name, err)
logger.Errorf("failed to create client %q. %v", client.ObjectMeta.Name, err)
}
}
@@ -189,27 +189,27 @@ func updateClient(context *clusterd.Context, p *cephv1.CephClient) error {
func (c *ClientController) onUpdate(oldObj, newObj interface{}) {
oldClient, err := getClientObject(oldObj)
if err != nil {
logger.Errorf("failed to get old client object: %+v", err)
logger.Errorf("failed to get old client object. %v", err)
return
}
client, err := getClientObject(newObj)
if err != nil {
logger.Errorf("failed to get new client object: %+v", err)
logger.Errorf("failed to get new client object. %v", err)
return
}
if oldClient.Name != client.Name {
logger.Errorf("failed to update client %s. name update not allowed", client.Name)
logger.Errorf("failed to update client %q. name update not allowed", client.Name)
return
}
if !clientChanged(oldClient.Spec, client.Spec) {
logger.Debugf("client %s not changed", client.Name)
logger.Debugf("client %q not changed", client.Name)
return
}
logger.Infof("updating client %s", client.Name)
if err := updateClient(c.context, client); err != nil {
logger.Errorf("failed to update client %s. %+v", client.ObjectMeta.Name, err)
logger.Errorf("failed to update client %q. %v", client.ObjectMeta.Name, err)
}
}
@@ -229,13 +229,13 @@ func clientChanged(old, new cephv1.ClientSpec) bool {
func (c *ClientController) onDelete(obj interface{}) {
client, err := getClientObject(obj)
if err != nil {
logger.Errorf("failed to get client object: %+v", err)
logger.Errorf("failed to get client object. %v", err)
return
}
logger.Infof("Going to remove client object %s", client.Name)
if err := deleteClient(c.context, client); err != nil {
logger.Errorf("failed to delete client %s. %+v", client.ObjectMeta.Name, err)
logger.Errorf("failed to delete client %q. %v", client.ObjectMeta.Name, err)
}
logger.Infof("Removed client %s", client.Name)
}
+2 -2
View File
@@ -83,13 +83,13 @@ func (c *cephStatusChecker) checkStatus() {
logger.Debugf("checking health of cluster")
status, err := client.Status(c.context, c.namespace, true)
if err != nil {
logger.Errorf("failed to get ceph status. %+v", err)
logger.Errorf("failed to get ceph status. %v", err)
return
}
logger.Debugf("Cluster status: %+v", status)
if err := c.updateCephStatus(&status); err != nil {
logger.Errorf("failed to query cluster status in namespace %s", c.namespace)
logger.Errorf("failed to query cluster status in namespace %q. %v", c.namespace, err)
}
}
+5 -5
View File
@@ -153,7 +153,7 @@ func (c *cluster) validateCephVersion(version *cephver.CephVersion) error {
// Write connection info (ceph config file and keyring) for ceph commands
err = mon.WriteConnectionConfig(c.context, clusterInfo)
if err != nil {
logger.Errorf("failed to write config. Attempting to continue. %+v", err)
logger.Errorf("failed to write config. attempting to continue. %v", err)
}
}
@@ -180,14 +180,14 @@ func (c *cluster) validateCephVersion(version *cephver.CephVersion) error {
// Get cluster running versions
versions, err := client.GetAllCephDaemonVersions(c.context, c.Namespace)
if err != nil {
logger.Errorf("failed to get ceph daemons versions. %+v", err)
logger.Errorf("failed to get ceph daemons versions. %v", err)
return nil
}
runningVersions := *versions
differentImages, err := diffImageSpecAndClusterRunningVersion(*version, runningVersions)
if err != nil {
logger.Errorf("failed to determine if we should upgrade or not. %+v", err)
logger.Errorf("failed to determine if we should upgrade or not. %v", err)
// we shouldn't block the orchestration if we can't determine the version of the image spec, we proceed anyway in best effort
// we won't be able to check if there is an update or not and what to do, so we don't check the cluster status either
// This will happen if someone uses ceph/daemon:latest-master for instance
@@ -225,7 +225,7 @@ func (c *cluster) createInstance(rookImage string, cephVersion cephver.CephVersi
// while no other goroutine is already running a cluster update
for c.checkSetOrchestrationStatus() == true {
if err != nil {
logger.Errorf("There was an orchestration error, but there is another orchestration pending; proceeding with next orchestration run (which may succeed). %+v", err)
logger.Errorf("There was an orchestration error, but there is another orchestration pending; proceeding with next orchestration run (which may succeed). %v", err)
}
// Use a DeepCopy of the spec to avoid using an inconsistent data-set
spec := c.Spec.DeepCopy()
@@ -397,7 +397,7 @@ func diffImageSpecAndClusterRunningVersion(imageSpecVersion cephver.CephVersion,
for v := range runningVersions.Overall {
version, err := cephver.ExtractCephVersion(v)
if err != nil {
logger.Errorf("failed to extract ceph version. %+v", err)
logger.Errorf("failed to extract ceph version. %v", err)
return false, err
}
clusterRunningVersion := *version
+52 -52
View File
@@ -226,7 +226,7 @@ func (c *ClusterController) onK8sNodeAdd(obj interface{}) {
logger.Debugf("Adding %s to cluster %s", newNode.Labels[v1.LabelHostname], cluster.Namespace)
err := cluster.createInstance(c.rookImage, cluster.Info.CephVersion)
if err != nil {
logger.Errorf("Failed to update cluster in namespace %s. Was not able to add %s. %+v", cluster.Namespace, newNode.Labels[v1.LabelHostname], err)
logger.Errorf("failed to update cluster in namespace %q. was not able to add %q. %v", cluster.Namespace, newNode.Labels[v1.LabelHostname], err)
}
} else {
logger.Infof("Could not add host %s . It is not valid", newNode.Labels[v1.LabelHostname])
@@ -239,12 +239,12 @@ func (c *ClusterController) onK8sNodeAdd(obj interface{}) {
func (c *ClusterController) onAdd(obj interface{}) {
clusterObj, err := getClusterObject(obj)
if err != nil {
logger.Errorf("failed to get cluster object: %+v", err)
logger.Errorf("failed to get cluster object. %v", err)
return
}
if existing, ok := c.clusterMap[clusterObj.Namespace]; ok {
logger.Errorf("Failed to add cluster cr %s in namespace %s. Cluster cr %s already exists in this namespace. Only one cluster cr per namespace is supported.",
logger.Errorf("failed to add cluster cr %q in namespace %q. Cluster cr %q already exists in this namespace. Only one cluster cr per namespace is supported.",
clusterObj.Name, clusterObj.Namespace, existing.crdName)
return
}
@@ -260,7 +260,7 @@ func (c *ClusterController) onAdd(obj interface{}) {
for _, callback := range c.addClusterCallbacks {
if err := callback(cluster.Spec); err != nil {
logger.Errorf("%+v", err)
logger.Errorf("%v", err)
}
}
c.csiConfigMutex.Unlock()
@@ -285,7 +285,7 @@ func (c *ClusterController) configureExternalCephCluster(namespace, name string,
if cluster.Spec.CephVersion.Image == "" {
err = mon.WriteConnectionConfig(c.context, cluster.Info)
if err != nil {
logger.Errorf("failed to write config. Attempting to continue. %+v", err)
logger.Errorf("failed to write config. attempting to continue. %v", err)
}
}
@@ -354,7 +354,7 @@ func (c *ClusterController) configureLocalCephCluster(namespace, name string, cl
func() (bool, error) {
cephVersion, canRetry, err := c.detectAndValidateCephVersion(cluster, cluster.Spec.CephVersion.Image)
if err != nil {
failedMessage = fmt.Sprintf("failed the ceph version check. %+v", err)
failedMessage = fmt.Sprintf("failed the ceph version check. %v", err)
logger.Errorf(failedMessage)
if !canRetry {
// it may seem strange to exit true but we don't want to retry if the version is not supported
@@ -367,7 +367,7 @@ func (c *ClusterController) configureLocalCephCluster(namespace, name string, cl
err = cluster.createInstance(c.rookImage, *cephVersion)
if err != nil {
failedMessage = fmt.Sprintf("failed to create cluster in namespace %s. %+v", cluster.Namespace, err)
failedMessage = fmt.Sprintf("failed to create cluster in namespace %q. %v", cluster.Namespace, err)
logger.Errorf(failedMessage)
return false, nil
}
@@ -398,19 +398,19 @@ func (c *ClusterController) initializeCluster(cluster *cluster, clusterObj *ceph
cleanDataDirHostPath := path.Clean(cluster.Spec.DataDirHostPath)
for _, b := range disallowedHostDirectories {
if cleanDataDirHostPath == b {
logger.Errorf("dataDirHostPath (given: %s) must not be used, conflicts with %s internal path", cluster.Spec.DataDirHostPath, b)
logger.Errorf("dataDirHostPath (given: %q) must not be used, conflicts with %q internal path", cluster.Spec.DataDirHostPath, b)
return
}
}
if !cluster.Spec.External.Enable {
if err := c.configureLocalCephCluster(clusterObj.Namespace, clusterObj.Name, cluster, clusterObj); err != nil {
logger.Errorf("failed to configure local ceph cluster. %+v", err)
logger.Errorf("failed to configure local ceph cluster. %v", err)
return
}
} else {
if err := c.configureExternalCephCluster(clusterObj.Namespace, clusterObj.Name, cluster); err != nil {
logger.Errorf("failed to configure external ceph cluster. %+v", err)
logger.Errorf("failed to configure external ceph cluster. %v", err)
return
}
}
@@ -474,7 +474,7 @@ func (c *ClusterController) initializeCluster(cluster *cluster, clusterObj *ceph
// add the finalizer to the crd
err := c.addFinalizer(clusterObj.Namespace, clusterObj.Name)
if err != nil {
logger.Errorf("failed to add finalizer to cluster crd. %+v", err)
logger.Errorf("failed to add finalizer to cluster crd. %v", err)
}
}
@@ -522,33 +522,33 @@ func (c *ClusterController) onK8sNodeUpdate(oldObj, newObj interface{}) {
logger.Debugf("Adding %s to cluster %s", newNode.Labels[v1.LabelHostname], cluster.Namespace)
err := cluster.createInstance(c.rookImage, cluster.Info.CephVersion)
if err != nil {
logger.Errorf("Failed adding the updated node %s to cluster in namespace %s. %+v", newNode.Labels[v1.LabelHostname], cluster.Namespace, err)
logger.Errorf("Failed adding the updated node %q to cluster in namespace %q. %v", newNode.Labels[v1.LabelHostname], cluster.Namespace, err)
continue
}
} else {
logger.Infof("Updated node %s is not valid and could not get added to cluster in namespace %s.", newNode.Labels[v1.LabelHostname], cluster.Namespace)
logger.Infof("Updated node %q is not valid and could not get added to cluster in namespace %q.", newNode.Labels[v1.LabelHostname], cluster.Namespace)
continue
}
logger.Infof("Added updated node %s to cluster %s", newNode.Labels[v1.LabelHostname], cluster.Namespace)
logger.Infof("Added updated node %q to cluster %q", newNode.Labels[v1.LabelHostname], cluster.Namespace)
}
}
func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
oldClust, err := getClusterObject(oldObj)
if err != nil {
logger.Errorf("failed to get old cluster object: %+v", err)
logger.Errorf("failed to get old cluster object. %v", err)
return
}
newClust, err := getClusterObject(newObj)
if err != nil {
logger.Errorf("failed to get new cluster object: %+v", err)
logger.Errorf("failed to get new cluster object. %v", err)
return
}
logger.Debugf("update event for cluster %s", newClust.Namespace)
if existing, ok := c.clusterMap[newClust.Namespace]; ok && existing.crdName != newClust.Name {
logger.Errorf("Skipping update of cluster cr %s in namespace %s. Cluster cr %s already exists in this namespace. Only one cluster cr per namespace is supported.",
logger.Errorf("skipping update of cluster cr %q in namespace %q. Cluster cr %q already exists in this namespace. Only one cluster cr per namespace is supported.",
newClust.Name, newClust.Namespace, existing.crdName)
return
}
@@ -557,10 +557,10 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
// When a cluster is requested for deletion, K8s will only set the deletion timestamp if there are any finalizers in the list.
// K8s will only delete the crd and child resources when the finalizers have been removed from the crd.
if newClust.DeletionTimestamp != nil {
logger.Infof("cluster %s has a deletion timestamp", newClust.Namespace)
logger.Infof("cluster %q has a deletion timestamp", newClust.Namespace)
err := c.handleDelete(newClust, time.Duration(clusterDeleteRetryInterval)*time.Second)
if err != nil {
logger.Errorf("failed finalizer for cluster. %+v", err)
logger.Errorf("failed finalizer for cluster. %v", err)
return
}
// remove the finalizer from the crd, which indicates to k8s that the resource can safely be deleted
@@ -569,25 +569,25 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
}
cluster, ok := c.clusterMap[newClust.Namespace]
if !ok {
logger.Errorf("Cannot update cluster %s that does not exist", newClust.Namespace)
logger.Errorf("cannot update cluster %q that does not exist", newClust.Namespace)
return
}
// If the cluster was never initialized during the OnAdd() method due to a failure, we must
// treat the cluster as if it was just created.
if !cluster.initialized() {
logger.Infof("Update event for uninitialized cluster %s. Initializing...", newClust.Namespace)
logger.Infof("update event for uninitialized cluster %q. Initializing...", newClust.Namespace)
c.initializeCluster(cluster, newClust)
return
}
changed, _ := clusterChanged(oldClust.Spec, newClust.Spec, cluster)
if !changed {
logger.Debugf("update event for cluster %s is not supported", newClust.Namespace)
logger.Debugf("update event for cluster %q is not supported", newClust.Namespace)
return
}
logger.Infof("update event for cluster %s is supported, orchestrating update now", newClust.Namespace)
logger.Infof("update event for cluster %q is supported, orchestrating update now", newClust.Namespace)
if oldClust.Spec.RemoveOSDsIfOutAndSafeToRemove != newClust.Spec.RemoveOSDsIfOutAndSafeToRemove {
logger.Infof("removeOSDsIfOutAndSafeToRemove is set to %t", newClust.Spec.RemoveOSDsIfOutAndSafeToRemove)
@@ -602,10 +602,10 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
// if the image changed, we need to detect the new image version
versionChanged := false
if oldClust.Spec.CephVersion.Image != newClust.Spec.CephVersion.Image {
logger.Infof("the ceph version changed from %s to %s", oldClust.Spec.CephVersion.Image, newClust.Spec.CephVersion.Image)
logger.Infof("the ceph version changed from %q to %q", oldClust.Spec.CephVersion.Image, newClust.Spec.CephVersion.Image)
version, _, err := c.detectAndValidateCephVersion(cluster, newClust.Spec.CephVersion.Image)
if err != nil {
logger.Errorf("unknown ceph major version. %+v", err)
logger.Errorf("unknown ceph major version. %q", err)
return
}
versionChanged = true
@@ -615,7 +615,7 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
// Get cluster running versions
versions, err := client.GetAllCephDaemonVersions(c.context, cluster.Namespace)
if err != nil {
logger.Errorf("failed to get ceph daemons versions. %+v", err)
logger.Errorf("failed to get ceph daemons versions. %q", err)
return
}
runningVersions := *versions
@@ -629,7 +629,7 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
// so don't get confused by the name of the function and its arguments
updateOrNot, err := diffImageSpecAndClusterRunningVersion(cluster.Info.CephVersion, runningVersions)
if err != nil {
logger.Errorf("failed to determine if we should upgrade or not. %+v", err)
logger.Errorf("failed to determine if we should upgrade or not. %v", err)
return
}
@@ -641,7 +641,7 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
if cluster.Spec.SkipUpgradeChecks {
logger.Warning("ceph is not healthy but SkipUpgradeChecks is set, forcing upgrade.")
} else {
logger.Errorf("ceph status in namespace %s is not healthy, refusing to upgrade. fix the cluster and re-edit the cluster CR to trigger a new orchestation update", cluster.Namespace)
logger.Errorf("ceph status in namespace %q is not healthy, refusing to upgrade. fix the cluster and re-edit the cluster CR to trigger a new orchestation update", cluster.Namespace)
return
}
}
@@ -664,7 +664,7 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
})
if err != nil {
c.updateClusterStatus(newClust.Namespace, newClust.Name, cephv1.ClusterStateError,
fmt.Sprintf("giving up trying to update cluster in namespace %s after %s. %+v", cluster.Namespace, updateClusterTimeout, err))
fmt.Sprintf("giving up trying to update cluster in namespace %q after %q. %v", cluster.Namespace, updateClusterTimeout, err))
return
}
@@ -690,7 +690,7 @@ func (c *ClusterController) handleUpdate(crdName string, cluster *cluster) (bool
c.updateClusterStatus(cluster.Namespace, crdName, cephv1.ClusterStateUpdating, "")
if err := cluster.createInstance(c.rookImage, cluster.Info.CephVersion); err != nil {
logger.Errorf("failed to update cluster in namespace %s. %+v", cluster.Namespace, err)
logger.Errorf("failed to update cluster in namespace %q. %v", cluster.Namespace, err)
return false, nil
}
@@ -750,7 +750,7 @@ func (c *ClusterController) onDeviceCMUpdate(oldObj, newObj interface{}) {
logger.Infof("Running orchestration for namespace %s after device change", cluster.Namespace)
err := cluster.createInstance(c.rookImage, cluster.Info.CephVersion)
if err != nil {
logger.Errorf("Failed orchestration after device change in namespace %s. %+v", cluster.Namespace, err)
logger.Errorf("Failed orchestration after device change in namespace %q. %v", cluster.Namespace, err)
continue
}
}
@@ -763,21 +763,21 @@ func (c *ClusterController) onDeviceCMUpdate(oldObj, newObj interface{}) {
func (c *ClusterController) onDelete(obj interface{}) {
clust, err := getClusterObject(obj)
if err != nil {
logger.Errorf("failed to get cluster object: %+v", err)
logger.Errorf("failed to get cluster object. %v", err)
return
}
if existing, ok := c.clusterMap[clust.Namespace]; ok && existing.crdName != clust.Name {
logger.Errorf("Skipping deletion of cluster cr %s in namespace %s. Cluster cr %s already exists in this namespace. Only one cluster cr per namespace is supported.",
logger.Errorf("Skipping deletion of cluster cr %q in namespace %q. Cluster cr %q already exists in this namespace. Only one cluster cr per namespace is supported.",
clust.Name, clust.Namespace, existing.crdName)
return
}
logger.Infof("delete event for cluster %s in namespace %s", clust.Name, clust.Namespace)
logger.Infof("delete event for cluster %q in namespace %q", clust.Name, clust.Namespace)
err = c.handleDelete(clust, time.Duration(clusterDeleteRetryInterval)*time.Second)
if err != nil {
logger.Errorf("failed to delete cluster. %+v", err)
logger.Errorf("failed to delete cluster. %v", err)
}
// Note that this lock is held through the callback process, as this deletes CSI resources, but we must lock in
@@ -790,7 +790,7 @@ func (c *ClusterController) onDelete(obj interface{}) {
}
for _, callback := range c.removeClusterCallbacks {
if err := callback(); err != nil {
logger.Errorf("%+v", err)
logger.Errorf("%v", err)
}
}
c.csiConfigMutex.Unlock()
@@ -803,7 +803,7 @@ func (c *ClusterController) onDelete(obj interface{}) {
} else {
err := purgeExternalCluster(c.context.Clientset, clust.Namespace)
if err != nil {
logger.Errorf("failed to purge external cluster ressources. %+v", err)
logger.Errorf("failed to purge external cluster ressources. %v", err)
}
return
}
@@ -815,7 +815,7 @@ func (c *ClusterController) waitForFlexVolumeCleanup(cluster *cephv1.CephCluster
// TODO: filter this List operation by cluster namespace on the server side
vols, err := c.volumeAttachment.List(operatorNamespace)
if err != nil {
return errors.Wrapf(err, "failed to get volume attachments for operator namespace %s", operatorNamespace)
return errors.Wrapf(err, "failed to get volume attachments for operator namespace %q", operatorNamespace)
}
// find volume attachments in the deleted cluster
@@ -839,13 +839,13 @@ func (c *ClusterController) waitForFlexVolumeCleanup(cluster *cephv1.CephCluster
retryCount++
if retryCount == clusterDeleteMaxRetries {
logger.Warningf(
"exceeded retry count while waiting for volume attachments for cluster %s to be cleaned up. vols: %+v",
"exceeded retry count while waiting for volume attachments for cluster %q to be cleaned up. vols: %+v",
cluster.Namespace,
vols.Items)
break
}
logger.Infof("waiting for volume attachments in cluster %s to be cleaned up. Retrying in %s.",
logger.Infof("waiting for volume attachments in cluster %q to be cleaned up. Retrying in %q.",
cluster.Namespace, retryInterval.String())
<-time.After(retryInterval)
}
@@ -860,7 +860,7 @@ func (c *ClusterController) checkPVPresentInCluster(drivers []string, clusterID
for _, p := range pv.Items {
if p.Spec.CSI == nil {
logger.Errorf("Spec.CSI is nil for PV %s", p.Name)
logger.Errorf("Spec.CSI is nil for PV %q", p.Name)
continue
}
if p.Spec.CSI.VolumeAttributes["clusterID"] == clusterID {
@@ -880,7 +880,7 @@ func (c *ClusterController) waitForCSIVolumeCleanup(cluster *cephv1.CephCluster,
retryCount := 0
drivers := []string{csi.CephFSDriverName, csi.RBDDriverName}
for {
logger.Infof("checking any PVC created by drivers %s and %s with clusterID %s", csi.CephFSDriverName, csi.RBDDriverName, cluster.Namespace)
logger.Infof("checking any PVC created by drivers %q and %q with clusterID %q", csi.CephFSDriverName, csi.RBDDriverName, cluster.Namespace)
// check any PV is created in this cluster
attachmentsExist, err := c.checkPVPresentInCluster(drivers, cluster.Namespace)
if err != nil {
@@ -1000,7 +1000,7 @@ func (c *ClusterController) removeFinalizer(obj interface{}) {
// Get the latest cluster instead of using the same instance in case it has been changed
cluster, err := c.context.RookClientset.CephV1().CephClusters(cl.Namespace).Get(cl.Name, metav1.GetOptions{})
if err != nil {
logger.Errorf("failed to remove finalizer. failed to get cluster. %+v", err)
logger.Errorf("failed to remove finalizer. failed to get cluster. %v", err)
return
}
objectMeta := &cluster.ObjectMeta
@@ -1015,13 +1015,13 @@ func (c *ClusterController) removeFinalizer(obj interface{}) {
}
}
if !found {
logger.Infof("finalizer %s not found in the cluster crd '%s'", finalizerName, objectMeta.Name)
logger.Infof("finalizer %q not found in the cluster crd %q", finalizerName, objectMeta.Name)
return
}
_, err = c.context.RookClientset.CephV1().CephClusters(cluster.Namespace).Update(cluster)
if err != nil {
logger.Errorf("failed to remove finalizer %s from cluster %s. %+v", finalizerName, objectMeta.Name, err)
logger.Errorf("failed to remove finalizer %q from cluster %q. %v", finalizerName, objectMeta.Name, err)
time.Sleep(retrySeconds)
continue
}
@@ -1034,12 +1034,12 @@ func (c *ClusterController) removeFinalizer(obj interface{}) {
// updateClusterStatus updates the status of the cluster custom resource, whether it is being updated or is completed
func (c *ClusterController) updateClusterStatus(namespace, name string, state cephv1.ClusterState, message string) {
logger.Infof("CephCluster %s status: %s. %s", namespace, state, message)
logger.Infof("CephCluster %q status: %q. %s", namespace, state, message)
// get the most recent cluster CRD object
cluster, err := c.context.RookClientset.CephV1().CephClusters(namespace).Get(name, metav1.GetOptions{})
if err != nil {
logger.Errorf("failed to get cluster from namespace %s prior to updating its status to %s. %+v", namespace, state, err)
logger.Errorf("failed to get cluster from namespace %q prior to updating its status to %q. %v", namespace, state, err)
}
// update the status on the retrieved cluster object
@@ -1047,7 +1047,7 @@ func (c *ClusterController) updateClusterStatus(namespace, name string, state ce
cluster.Status.State = state
cluster.Status.Message = message
if _, err := c.context.RookClientset.CephV1().CephClusters(namespace).Update(cluster); err != nil {
logger.Errorf("failed to update cluster %s status: %+v", namespace, err)
logger.Errorf("failed to update cluster %q status: %v", namespace, err)
}
}
@@ -1065,7 +1065,7 @@ func ClusterOwnerRef(clusterName, clusterID string) metav1.OwnerReference {
func printOverallCephVersion(context *clusterd.Context, namespace string) {
versions, err := client.GetAllCephDaemonVersions(context, namespace)
if err != nil {
logger.Errorf("failed to get ceph daemons versions. %+v", err)
logger.Errorf("failed to get ceph daemons versions. %v", err)
return
}
@@ -1073,15 +1073,15 @@ func printOverallCephVersion(context *clusterd.Context, namespace string) {
for v := range versions.Overall {
version, err := cephver.ExtractCephVersion(v)
if err != nil {
logger.Errorf("failed to extract ceph version. %+v", err)
logger.Errorf("failed to extract ceph version. %v", err)
return
}
vv := *version
logger.Infof("successfully upgraded cluster to version: %s", vv.String())
logger.Infof("successfully upgraded cluster to version: %q", vv.String())
}
} else {
// This shouldn't happen, but let's log just in case
logger.Warningf("upgrade orchestration completed but somehow we still have more than one Ceph version running. %+v:", versions.Overall)
logger.Warningf("upgrade orchestration completed but somehow we still have more than one Ceph version running. %v:", versions.Overall)
}
}
+2 -2
View File
@@ -71,9 +71,9 @@ func (c *Cluster) generateKeyring(m *mgrConfig) (string, error) {
err = c.context.Clientset.CoreV1().Secrets(c.Namespace).Delete(m.ResourceName, &metav1.DeleteOptions{})
if err != nil {
if errors.IsNotFound(err) {
logger.Debugf("legacy mgr key %s is already removed", m.ResourceName)
logger.Debugf("legacy mgr key %q is already removed", m.ResourceName)
} else {
logger.Warningf("legacy mgr key %s could not be removed: %+v", m.ResourceName, err)
logger.Warningf("legacy mgr key %q could not be removed. %v", m.ResourceName, err)
}
}
+1 -1
View File
@@ -96,7 +96,7 @@ func (c *Cluster) configureDashboardModules() error {
}
} else {
if err := client.MgrDisableModule(c.context, c.Namespace, dashboardModuleName); err != nil {
logger.Errorf("failed to disable mgr dashboard module. %+v", err)
logger.Errorf("failed to disable mgr dashboard module. %v", err)
}
return nil
}
+8 -8
View File
@@ -183,7 +183,7 @@ func (c *Cluster) Start() error {
if c.isUpgrade {
currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon)
if err != nil {
logger.Warningf("failed to retrieve current ceph %s version. %+v", daemon, err)
logger.Warningf("failed to retrieve current ceph %q version. %v", daemon, err)
logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion")
cephVersionToUse = c.clusterInfo.CephVersion
} else {
@@ -193,20 +193,20 @@ func (c *Cluster) Start() error {
}
if err := updateDeploymentAndWait(c.context, d, c.Namespace, daemon, mgrConfig.DaemonID, cephVersionToUse, c.isUpgrade, c.skipUpgradeChecks); err != nil {
logger.Errorf("failed to update mgr deployment %q. %+v", resourceName, err)
logger.Errorf("failed to update mgr deployment %q. %v", resourceName, err)
}
}
if existingDeployment, err := c.context.Clientset.AppsV1().Deployments(c.Namespace).Get(d.GetName(), metav1.GetOptions{}); err != nil {
logger.Warningf("failed to find mgr deployment %s for keyring association: %+v", resourceName, err)
logger.Warningf("failed to find mgr deployment %q for keyring association. %v", resourceName, err)
} else {
if err = c.associateKeyring(keyring, existingDeployment); err != nil {
logger.Warningf("failed to associate keyring with mgr deployment %s: %+v", resourceName, err)
logger.Warningf("failed to associate keyring with mgr deployment %q. %v", resourceName, err)
}
}
}
if err := c.configureDashboardService(); err != nil {
logger.Errorf("failed to enable dashboard. %+v", err)
logger.Errorf("failed to enable dashboard. %v", err)
}
// configure the mgr modules
@@ -229,7 +229,7 @@ func (c *Cluster) Start() error {
logger.Infof("starting monitoring deployment")
// servicemonitor takes some metadata from the service for easy mapping
if err := c.enableServiceMonitor(service); err != nil {
logger.Errorf("failed to enable service monitor. %+v", err)
logger.Errorf("failed to enable service monitor. %v", err)
} else {
logger.Infof("servicemonitor enabled")
}
@@ -240,7 +240,7 @@ func (c *Cluster) Start() error {
namespace = c.Namespace
}
if err := c.deployPrometheusRule(prometheusRuleName, namespace); err != nil {
logger.Errorf("failed to deploy prometheus rule. %+v", err)
logger.Errorf("failed to deploy prometheus rule. %v", err)
} else {
logger.Infof("prometheusRule deployed")
}
@@ -274,7 +274,7 @@ func startModuleConfiguration(wg *sync.WaitGroup, description string, configureM
go func() {
err := configureModules()
if err != nil {
logger.Errorf("failed modules: %s. %+v", description, err)
logger.Errorf("failed modules: %q. %v", description, err)
} else {
logger.Infof("successful modules: %s", description)
}
+2 -2
View File
@@ -157,7 +157,7 @@ func loadMonConfig(clientset kubernetes.Interface, namespace string) (map[string
if id, ok := cm.Data[MaxMonIDKey]; ok {
maxMonID, err = strconv.Atoi(id)
if err != nil {
logger.Errorf("invalid max mon id %s. %+v", id, err)
logger.Errorf("invalid max mon id %q. %v", id, err)
}
}
@@ -171,7 +171,7 @@ func loadMonConfig(clientset kubernetes.Interface, namespace string) (map[string
err = json.Unmarshal([]byte(cm.Data[MappingKey]), &monMapping)
if err != nil {
logger.Errorf("invalid JSON in mon mapping. %+v", err)
logger.Errorf("invalid JSON in mon mapping. %v", err)
}
logger.Infof("loaded: maxMonID=%d, mons=%+v, mapping=%+v", maxMonID, monEndpointMap, monMapping)
+4 -4
View File
@@ -69,7 +69,7 @@ func (hc *HealthChecker) Check(stopCh chan struct{}) {
logger.Debugf("checking health of mons")
err := hc.monCluster.checkHealth()
if err != nil {
logger.Warningf("failed to check mon health. %+v", err)
logger.Warningf("failed to check mon health. %v", err)
}
}
}
@@ -192,18 +192,18 @@ func (c *Cluster) failMon(monCount, desiredMonCount int, name string) {
if monCount > desiredMonCount {
// no need to create a new mon since we have an extra
if err := c.removeMon(name); err != nil {
logger.Errorf("failed to remove mon %s. %+v", name, err)
logger.Errorf("failed to remove mon %q. %v", name, err)
}
} else {
// bring up a new mon to replace the unhealthy mon
if err := c.failoverMon(name); err != nil {
logger.Errorf("failed to failover mon %s. %+v", name, err)
logger.Errorf("failed to failover mon %q. %v", name, err)
}
}
}
func (c *Cluster) failoverMon(name string) error {
logger.Infof("Failing over monitor %s", name)
logger.Infof("Failing over monitor %q", name)
// Start a new monitor
m := c.newMonConfig(c.maxMonID + 1)
+9 -9
View File
@@ -233,7 +233,7 @@ func (c *Cluster) startMons(targetCount int) error {
// fixed by updating the mon deployments. Instead of returning error here, log a
// warning, and retry setting this later.
setConfigsNeedsRetry = true
logger.Warningf("failed to set Rook and/or user-defined Ceph config options before starting mons; will retry after starting mons. %+v", err)
logger.Warningf("failed to set Rook and/or user-defined Ceph config options before starting mons; will retry after starting mons. %v", err)
}
}
@@ -564,7 +564,7 @@ func (c *Cluster) assignMons(mons []*monConfig) error {
logger.Infof("assignmon: cleaning up canary deployment %s and canary pvc %s", result.CanaryDeployment, result.CanaryPVC)
if result.CanaryDeployment != "" {
if err := k8sutil.DeleteDeployment(c.context.Clientset, c.Namespace, result.CanaryDeployment); err != nil {
logger.Infof("assignmon: error deleting canary deployment %s. %+v", result.CanaryDeployment, err)
logger.Infof("assignmon: error deleting canary deployment %s. %v", result.CanaryDeployment, err)
}
}
if result.CanaryPVC != "" {
@@ -573,7 +573,7 @@ func (c *Cluster) assignMons(mons []*monConfig) error {
options := &metav1.DeleteOptions{GracePeriodSeconds: &gracePeriod, PropagationPolicy: &propagation}
err := c.context.Clientset.CoreV1().PersistentVolumeClaims(c.Namespace).Delete(result.CanaryPVC, options)
if err != nil {
logger.Infof("assignmon: error removing canary monitor %s pvc %s. %+v", result.CanaryDeployment, result.CanaryPVC, err)
logger.Infof("assignmon: error removing canary monitor %s pvc %s. %v", result.CanaryDeployment, result.CanaryPVC, err)
}
}
}
@@ -645,7 +645,7 @@ func (c *Cluster) startDeployments(mons []*monConfig, requireAllInQuorum bool) e
logger.Infof("0 of %d expected mon deployments exist. creating new deployment(s).", len(mons))
onlyCheckQuorumOnce = true
} else {
logger.Warningf("failed to list mon deployments. attempting to continue. %+v", err)
logger.Warningf("failed to list mon deployments. attempting to continue. %v", err)
}
} else if len(deployments.Items) < len(mons) {
logger.Infof("%d of %d expected mon deployments exist. creating new deployment(s).", len(deployments.Items), len(mons))
@@ -666,7 +666,7 @@ func (c *Cluster) startDeployments(mons []*monConfig, requireAllInQuorum bool) e
// we need to do everything possible to verify the basic health of a cluster, complete the first orchestration,
// and start watching for all the CRs. If mons still have quorum we can continue with the orchestration even
// if they aren't all up.
logger.Errorf("attempting to continue after failing to start mon %q. %+v", mons[i].DaemonName, err)
logger.Errorf("attempting to continue after failing to start mon %q. %v", mons[i].DaemonName, err)
}
// For the initial deployment (first creation) it's expected to not have all the monitors in quorum
@@ -687,7 +687,7 @@ func (c *Cluster) startDeployments(mons []*monConfig, requireAllInQuorum bool) e
// Only do this when monitors versions are different so we don't block the orchestration if a mon is down.
versions, err := client.GetAllCephDaemonVersions(c.context, c.ClusterInfo.Name)
if err != nil {
logger.Warningf("failed to get ceph daemons versions; this likely means there is no cluster yet. %+v", err)
logger.Warningf("failed to get ceph daemons versions; this likely means there is no cluster yet. %v", err)
} else {
if len(versions.Mon) != 1 {
requireAllInQuorum = true
@@ -785,7 +785,7 @@ func (c *Cluster) updateMon(m *monConfig, d *apps.Deployment) error {
if c.isUpgrade {
currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.ClusterInfo.Name, daemonType)
if err != nil {
logger.Warningf("failed to retrieve current ceph %s version. %+v", daemonType, err)
logger.Warningf("failed to retrieve current ceph %q version. %v", daemonType, err)
logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with %+v", c.ClusterInfo.CephVersion)
cephVersionToUse = c.ClusterInfo.CephVersion
} else {
@@ -937,7 +937,7 @@ func waitForQuorumWithMons(context *clusterd.Context, clusterName string, mons [
for _, m := range mons {
running, err := k8sutil.PodsRunningWithLabel(context.Clientset, clusterName, fmt.Sprintf("app=%s,mon=%s", AppName, m))
if err != nil {
logger.Infof("failed to query mon pod status, trying again. %+v", err)
logger.Infof("failed to query mon pod status, trying again. %v", err)
continue
}
if running > 0 {
@@ -957,7 +957,7 @@ func waitForQuorumWithMons(context *clusterd.Context, clusterName string, mons [
// their quorum status
monQuorumStatusResp, err := client.GetMonQuorumStatus(context, clusterName, false)
if err != nil {
logger.Debugf("failed to get quorum_status. %+v", err)
logger.Debugf("failed to get quorum_status. %v", err)
continue
}
+3 -3
View File
@@ -60,7 +60,7 @@ func (c *Cluster) createService(mon *monConfig) (string, error) {
}
if s == nil {
logger.Errorf("service ip not found for mon %s. if this is not a unit test, this is an error", mon.ResourceName)
logger.Errorf("service ip not found for mon %q. if this is not a unit test, this is an error", mon.ResourceName)
return "", nil
}
@@ -68,9 +68,9 @@ func (c *Cluster) createService(mon *monConfig) (string, error) {
// however it's interesting to show that monitors can be addressed via 2 different ports
// in the end the service has msgr1 and msgr2 ports configured so it's not entirely wrong
if c.ClusterInfo.CephVersion.IsAtLeastNautilus() {
logger.Infof("mon %s endpoint are [v2:%s:%s,v1:%s:%d]", mon.DaemonName, s.Spec.ClusterIP, strconv.Itoa(int(DefaultMsgr2Port)), s.Spec.ClusterIP, mon.Port)
logger.Infof("mon %q endpoint are [v2:%s:%s,v1:%s:%d]", mon.DaemonName, s.Spec.ClusterIP, strconv.Itoa(int(DefaultMsgr2Port)), s.Spec.ClusterIP, mon.Port)
} else {
logger.Infof("mon %s endpoint is %s:%d", mon.DaemonName, s.Spec.ClusterIP, mon.Port)
logger.Infof("mon %q endpoint is %s:%d", mon.DaemonName, s.Spec.ClusterIP, mon.Port)
}
return s.Spec.ClusterIP, nil
}
+1 -1
View File
@@ -34,7 +34,7 @@ func (c *Cluster) prepareStorageClassDeviceSets(config *provisionConfig) []rooka
for i := 0; i < storageClassDeviceSet.Count; i++ {
pvc, err := c.createStorageClassDeviceSetPVC(storageClassDeviceSet, i)
if err != nil {
config.addError("%+v", err)
config.addError("%v", err)
config.addError("OSD creation for storageClassDeviceSet %v failed for count %v", storageClassDeviceSet.Name, i)
continue
}
+2 -2
View File
@@ -60,7 +60,7 @@ func (m *Monitor) Start(stopCh chan struct{}) {
logger.Debug("Checking osd processes status.")
err := m.osdStatus()
if err != nil {
logger.Warningf("Failed OSD status check: %+v", err)
logger.Warningf("failed OSD status check. %v", err)
}
case <-stopCh:
@@ -108,7 +108,7 @@ func (m *Monitor) osdStatus() error {
logger.Debugf("osd.%d is marked 'OUT'", id)
if m.removeOSDsIfOUTAndSafeToRemove {
if err := m.handleOSDMarkedOut(id); err != nil {
logger.Errorf("Error handling marked out osd osd.%d: %v", id, err)
logger.Errorf("error handling marked out osd osd.%d. %v", id, err)
}
}
}
+29 -29
View File
@@ -207,7 +207,7 @@ func (c *Cluster) Start() error {
if c.clusterInfo.CephVersion.IsAtLeastNautilus() {
versions, err := client.GetAllCephDaemonVersions(c.context, c.clusterInfo.Name)
if err != nil {
logger.Warningf("failed to get ceph daemons versions; this likely means there are no osds yet. %+v", err)
logger.Warningf("failed to get ceph daemons versions; this likely means there are no osds yet. %v", err)
} else {
// If length is one, this clearly indicates that all the osds are running the same version
// If this is the first time we are creating a cluster length will be 0
@@ -248,7 +248,7 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig) {
//check k8s version
k8sVersion, err := k8sutil.GetK8SVersion(c.context.Clientset)
if err != nil {
config.addError("error finding Kubernetes version. %+v", err)
config.addError("error finding Kubernetes version. %v", err)
return
}
if !k8sVersion.AtLeast(version.MustParseSemantic("v1.13.0")) {
@@ -268,7 +268,7 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig) {
// update the orchestration status of this pvc to the starting state
status := OrchestrationStatus{Status: OrchestrationStatusStarting, PvcBackedOSD: true}
if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil {
config.addError("failed to set orchestration starting status for pvc %s: %+v", osdProps.crushHostname, err)
config.addError("failed to set orchestration starting status for pvc %q. %v", osdProps.crushHostname, err)
continue
}
@@ -280,7 +280,7 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig) {
osdDeployments, err := c.context.Clientset.AppsV1().Deployments(c.Namespace).List(listOpts)
if err != nil {
config.addError("failed to check if OSD daemon exists for pvc %q. %+v", osdProps.crushHostname, err)
config.addError("failed to check if OSD daemon exists for pvc %q. %v", osdProps.crushHostname, err)
continue
}
@@ -288,13 +288,13 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig) {
logger.Infof("skip OSD prepare pod creation as OSD daemon already exists for %q", osdProps.crushHostname)
osds, err := getOSDInfo(&osdDeployments.Items[0])
if err != nil {
config.addError("failed to get osdInfo for pvc %q. %+v", osdProps.crushHostname, err)
config.addError("failed to get osdInfo for pvc %q. %v", osdProps.crushHostname, err)
continue
}
// update the orchestration status of this pvc to the completed state
status = OrchestrationStatus{OSDs: osds, Status: OrchestrationStatusCompleted, PvcBackedOSD: true}
if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil {
config.addError("failed to update pvc %q status. %+v", osdProps.crushHostname, err)
config.addError("failed to update pvc %q status. %v", osdProps.crushHostname, err)
continue
}
continue
@@ -306,7 +306,7 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig) {
config.addError(message)
status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message, PvcBackedOSD: true}
if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil {
config.addError("failed to update pvc %q status. %+v", osdProps.crushHostname, err)
config.addError("failed to update pvc %q status. %v", osdProps.crushHostname, err)
continue
}
}
@@ -318,7 +318,7 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig) {
PvcBackedOSD: true,
}
if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil {
config.addError("failed to update osd %s status. %+v", osdProps.crushHostname, err)
config.addError("failed to update osd %q status. %v", osdProps.crushHostname, err)
}
}
}
@@ -379,7 +379,7 @@ func (c *Cluster) startProvisioningOverNodes(config *provisionConfig) {
// update the orchestration status of this node to the starting state
status := OrchestrationStatus{Status: OrchestrationStatusStarting}
if err := c.updateOSDStatus(n.Name, status); err != nil {
config.addError("failed to set orchestration starting status for node %s: %+v", n.Name, err)
config.addError("failed to set orchestration starting status for node %q. %v", n.Name, err)
continue
}
@@ -396,11 +396,11 @@ func (c *Cluster) startProvisioningOverNodes(config *provisionConfig) {
}
job, err := c.makeJob(osdProps, config)
if err != nil {
message := fmt.Sprintf("failed to create prepare job node %s: %v", n.Name, err)
message := fmt.Sprintf("failed to create prepare job node %q. %v", n.Name, err)
config.addError(message)
status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message}
if err := c.updateOSDStatus(n.Name, status); err != nil {
config.addError("failed to update node %s status. %+v", n.Name, err)
config.addError("failed to update node %q status. %v", n.Name, err)
continue
}
}
@@ -408,7 +408,7 @@ func (c *Cluster) startProvisioningOverNodes(config *provisionConfig) {
if !c.runJob(job, n.Name, config, "provision") {
status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: fmt.Sprintf("failed to start osd provisioning on node %s", n.Name)}
if err := c.updateOSDStatus(n.Name, status); err != nil {
config.addError("failed to update node %s status. %+v", n.Name, err)
config.addError("failed to update node %q status. %v", n.Name, err)
}
}
}
@@ -420,7 +420,7 @@ func (c *Cluster) runJob(job *batch.Job, nodeName string, config *provisionConfi
if err := k8sutil.RunReplaceableJob(c.context.Clientset, job, false); err != nil {
if !kerrors.IsAlreadyExists(err) {
// we failed to create job, update the orchestration status for this node
message := fmt.Sprintf("failed to create %s job for node %s. %+v", action, nodeName, err)
message := fmt.Sprintf("failed to create %q job for node %q. %v", action, nodeName, err)
c.handleOrchestrationFailure(config, nodeName, message)
return false
}
@@ -439,7 +439,7 @@ func (c *Cluster) startOSDDaemonsOnPVC(pvcName string, config *provisionConfig,
storeConfig := osdconfig.ToStoreConfig(conf)
osdProps, err := c.getOSDPropsForPVC(pvcName)
if err != nil {
config.addError(fmt.Sprintf("%+v", err))
config.addError(fmt.Sprintf("%v", err))
return
}
@@ -451,7 +451,7 @@ func (c *Cluster) startOSDDaemonsOnPVC(pvcName string, config *provisionConfig,
// in intermittent failure of first-attempt OSD pods.
keyring, err := c.generateKeyring(osd.ID)
if err != nil {
errMsg := fmt.Sprintf("failed to create keyring for pvc %s, osd %v: %+v", osdProps.crushHostname, osd, err)
errMsg := fmt.Sprintf("failed to create keyring for pvc %q, osd %v. %v", osdProps.crushHostname, osd, err)
config.addError(errMsg)
continue
}
@@ -467,20 +467,20 @@ func (c *Cluster) startOSDDaemonsOnPVC(pvcName string, config *provisionConfig,
if createErr != nil {
if !kerrors.IsAlreadyExists(createErr) {
// we failed to create job, update the orchestration status for this pvc
logger.Warningf("failed to create osd deployment for pvc %s, osd %v: %+v", osdProps.pvc.ClaimName, osd, createErr)
logger.Warningf("failed to create osd deployment for pvc %q, osd %v. %v", osdProps.pvc.ClaimName, osd, createErr)
continue
}
logger.Infof("deployment for osd %d already exists. updating if needed", osd.ID)
createdDeployment, err = c.context.Clientset.AppsV1().Deployments(c.Namespace).Get(dp.Name, metav1.GetOptions{})
if err != nil {
logger.Warningf("failed to get existing OSD deployment %q for update: %+v", dp.Name, err)
logger.Warningf("failed to get existing OSD deployment %q for update. %v", dp.Name, err)
continue
}
}
err = c.associateKeyring(keyring, createdDeployment)
if err != nil {
logger.Errorf("failed to associate keyring for pvc %s, osd %v: %+v", osdProps.pvc.ClaimName, osd, err)
logger.Errorf("failed to associate keyring for pvc %q, osd %v. %v", osdProps.pvc.ClaimName, osd, err)
}
if createErr != nil && kerrors.IsAlreadyExists(createErr) {
@@ -492,7 +492,7 @@ func (c *Cluster) startOSDDaemonsOnPVC(pvcName string, config *provisionConfig,
if c.isUpgrade {
currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon)
if err != nil {
logger.Warningf("failed to retrieve current ceph %s version. %+v", daemon, err)
logger.Warningf("failed to retrieve current ceph %q version. %v", daemon, err)
logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion")
cephVersionToUse = c.clusterInfo.CephVersion
} else {
@@ -502,7 +502,7 @@ func (c *Cluster) startOSDDaemonsOnPVC(pvcName string, config *provisionConfig,
}
if err = updateDeploymentAndWait(c.context, dp, c.Namespace, daemon, strconv.Itoa(osd.ID), cephVersionToUse, c.isUpgrade, c.skipUpgradeChecks); err != nil {
logger.Errorf("failed to update osd deployment %d. %+v", osd.ID, err)
logger.Errorf("failed to update osd deployment %d. %v", osd.ID, err)
}
}
logger.Infof("started deployment for osd %d (dir=%t, type=%s)", osd.ID, osd.IsDirectory, storeConfig.StoreType)
@@ -517,7 +517,7 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig
// fully resolve the storage config and resources for this node
n := c.resolveNode(nodeName)
if n == nil {
logger.Errorf("node %s did not resolve to start osds", nodeName)
logger.Errorf("node %q did not resolve to start osds", nodeName)
return
}
storeConfig := osdconfig.ToStoreConfig(n.Config)
@@ -540,7 +540,7 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig
// in intermittent failure of first-attempt OSD pods.
keyring, err := c.generateKeyring(osd.ID)
if err != nil {
errMsg := fmt.Sprintf("failed to create keyring for node %s, osd %v: %+v", n.Name, osd, err)
errMsg := fmt.Sprintf("failed to create keyring for node %q, osd %v. %v", n.Name, osd, err)
config.addError(errMsg)
continue
}
@@ -556,20 +556,20 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig
if createErr != nil {
if !kerrors.IsAlreadyExists(createErr) {
// we failed to create job, update the orchestration status for this node
logger.Warningf("failed to create osd deployment for node %s, osd %v: %+v", n.Name, osd, createErr)
logger.Warningf("failed to create osd deployment for node %q, osd %+v. %v", n.Name, osd, createErr)
continue
}
logger.Infof("deployment for osd %d already exists. updating if needed", osd.ID)
createdDeployment, err = c.context.Clientset.AppsV1().Deployments(c.Namespace).Get(dp.Name, metav1.GetOptions{})
if err != nil {
logger.Warningf("failed to get existing OSD deployment %s for update: %+v", dp.Name, err)
logger.Warningf("failed to get existing OSD deployment %q for update. %v", dp.Name, err)
continue
}
}
err = c.associateKeyring(keyring, createdDeployment)
if err != nil {
logger.Errorf("failed to associate keyring for node %s, osd %v: %+v", n.Name, osd, err)
logger.Errorf("failed to associate keyring for node %q, osd %v. %v", n.Name, osd, err)
}
if createErr != nil && kerrors.IsAlreadyExists(createErr) {
@@ -581,7 +581,7 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig
if c.isUpgrade {
currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon)
if err != nil {
logger.Warningf("failed to retrieve current ceph %s version. %+v", daemon, err)
logger.Warningf("failed to retrieve current ceph %q version. %v", daemon, err)
logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion")
cephVersionToUse = c.clusterInfo.CephVersion
} else {
@@ -591,7 +591,7 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig
}
if err = updateDeploymentAndWait(c.context, dp, c.Namespace, daemon, strconv.Itoa(osd.ID), cephVersionToUse, c.isUpgrade, c.skipUpgradeChecks); err != nil {
logger.Errorf("failed to update osd deployment %d. %+v", osd.ID, err)
logger.Errorf("failed to update osd deployment %d. %v", osd.ID, err)
}
}
logger.Infof("started deployment for osd %d (dir=%t, type=%s)", osd.ID, osd.IsDirectory, storeConfig.StoreType)
@@ -634,12 +634,12 @@ func getIDFromDeployment(deployment *apps.Deployment) int {
if idstr, ok := deployment.Labels[OsdIdLabelKey]; ok {
id, err := strconv.Atoi(idstr)
if err != nil {
logger.Errorf("unknown osd id from label %s", idstr)
logger.Errorf("unknown osd id from label %q", idstr)
return unknownID
}
return id
}
logger.Errorf("unknown osd id for deployment %s", deployment.Name)
logger.Errorf("unknown osd id for deployment %q", deployment.Name)
return unknownID
}
+2 -2
View File
@@ -80,7 +80,7 @@ func (c *Cluster) removeOSD(deploymentName string, id int) error {
// delete any backups of the OSD filesystem
if err := deleteOSDFileSystem(c.context.Clientset, c.Namespace, id); err != nil {
logger.Warningf("failed to delete osd.%d filesystem, it may need to be cleaned up manually: %+v", id, err)
logger.Warningf("failed to delete osd.%d filesystem, it may need to be cleaned up manually. %v", id, err)
}
return nil
@@ -240,7 +240,7 @@ func (c *Cluster) cleanUpNodeResources(nodeName, nodeCrushName string) error {
// clean up node config store
configStoreName := config.GetConfigStoreName(nodeName)
if err := c.kv.ClearStore(configStoreName); err != nil {
logger.Warningf("failed to delete node config store %s, may need to be cleaned up manually: %+v", configStoreName, err)
logger.Warningf("failed to delete node config store %q, may need to be cleaned up manually. %v", configStoreName, err)
}
return nil
+5 -5
View File
@@ -90,7 +90,7 @@ func (c *Cluster) handleOrchestrationFailure(config *provisionConfig, nodeName,
config.addError(message)
status := OrchestrationStatus{Status: OrchestrationStatusFailed, Message: message}
if err := c.updateOSDStatus(nodeName, status); err != nil {
config.addError("failed to update status for node %s. %+v", nodeName, err)
config.addError("failed to update status for node %q. %v", nodeName, err)
}
}
@@ -111,7 +111,7 @@ func parseOrchestrationStatus(data map[string]string) *OrchestrationStatus {
// we have status for this node, unmarshal it
var status OrchestrationStatus
if err := json.Unmarshal([]byte(statusRaw), &status); err != nil {
logger.Warningf("failed to unmarshal orchestration status. status: %s. %+v", statusRaw, err)
logger.Warningf("failed to unmarshal orchestration status. status: %s. %v", statusRaw, err)
return nil
}
@@ -136,7 +136,7 @@ func (c *Cluster) checkNodesCompleted(selector string, config *provisionConfig,
statuses, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).List(opts)
if err != nil {
if !kerrors.IsNotFound(err) {
config.addError("failed to get config status. %+v", err)
config.addError("failed to get config status. %v", err)
return 0, remainingNodes, false, statuses, err
}
// the status map doesn't exist yet, watching below is still an OK thing to do
@@ -185,7 +185,7 @@ func (c *Cluster) completeOSDsForAllNodes(config *provisionConfig, configOSDs bo
w, err := c.context.Clientset.CoreV1().ConfigMaps(c.Namespace).Watch(opts)
if err != nil {
logger.Warningf("failed to start watch on osd status, trying again. %+v", err)
logger.Warningf("failed to start watch on osd status, trying again. %v", err)
time.Sleep(5 * time.Second)
continue
}
@@ -243,7 +243,7 @@ func (c *Cluster) completeOSDsForAllNodes(config *provisionConfig, configOSDs bo
for remainingNode := range remainingNodes.Iter() {
clearNodeName := k8sutil.TruncateNodeName(orchestrationStatusMapName, remainingNode)
if err := c.kv.ClearStore(clearNodeName); err != nil {
config.addError("failed to clear node %s status with name %s. %+v", remainingNode, clearNodeName, err)
config.addError("failed to clear node %q status with name %q. %v", remainingNode, clearNodeName, err)
}
}
return false
+2 -2
View File
@@ -56,9 +56,9 @@ func (m *Mirroring) generateKeyring(daemonConfig *daemonConfig) (string, error)
err = m.context.Clientset.CoreV1().Secrets(m.Namespace).Delete(daemonConfig.ResourceName, &metav1.DeleteOptions{})
if err != nil {
if errors.IsNotFound(err) {
logger.Debugf("legacy rbd-mirror key %s is already removed", daemonConfig.ResourceName)
logger.Debugf("legacy rbd-mirror key %q is already removed", daemonConfig.ResourceName)
} else {
logger.Warningf("legacy rbd-mirror key %s could not be removed: %+v", daemonConfig.ResourceName, err)
logger.Warningf("legacy rbd-mirror key %q could not be removed. %v", daemonConfig.ResourceName, err)
}
}
+6 -6
View File
@@ -143,7 +143,7 @@ func (m *Mirroring) Start() error {
if m.isUpgrade {
currentCephVersion, err := client.LeastUptodateDaemonVersion(m.context, m.ClusterInfo.Name, daemon)
if err != nil {
logger.Warningf("failed to retrieve current ceph %s version. %+v", daemon, err)
logger.Warningf("failed to retrieve current ceph %q version. %v", daemon, err)
logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with m.ClusterInfo.CephVersion")
cephVersionToUse = m.ClusterInfo.CephVersion
@@ -155,7 +155,7 @@ func (m *Mirroring) Start() error {
if err := updateDeploymentAndWait(m.context, d, m.Namespace, daemon, daemonConf.DaemonID, cephVersionToUse, m.isUpgrade, m.skipUpgradeChecks); err != nil {
// fail could be an issue updating label selector (immutable), so try del and recreate
logger.Debugf("updateDeploymentAndWait failed for rbd-mirror %s. Attempting del-and-recreate. %+v", resourceName, err)
logger.Debugf("updateDeploymentAndWait failed for rbd-mirror %q. Attempting del-and-recreate. %v", resourceName, err)
err = m.context.Clientset.AppsV1().Deployments(m.Namespace).Delete(d.Name, &metav1.DeleteOptions{})
if err != nil {
return errors.Wrapf(err, "failed to delete rbd-mirror %s during del-and-recreate update attempt", resourceName)
@@ -167,10 +167,10 @@ func (m *Mirroring) Start() error {
}
if existingDeployment, err := m.context.Clientset.AppsV1().Deployments(m.Namespace).Get(d.GetName(), metav1.GetOptions{}); err != nil {
logger.Warningf("failed to find rbd-mirror deployment %s for keyring association: %+v", resourceName, err)
logger.Warningf("failed to find rbd-mirror deployment %q for keyring association. %v", resourceName, err)
} else {
if err = m.associateKeyring(keyring, existingDeployment); err != nil {
logger.Warningf("failed to associate keyring with rbd-mirror deployment %s: %+v", resourceName, err)
logger.Warningf("failed to associate keyring with rbd-mirror deployment %q. %v", resourceName, err)
}
}
logger.Infof("%s deployment started", resourceName)
@@ -179,7 +179,7 @@ func (m *Mirroring) Start() error {
// Remove extra rbd-mirror deployments if necessary
err = m.removeExtraMirrors()
if err != nil {
logger.Errorf("failed to remove extra mirrors. %+v", err)
logger.Errorf("failed to remove extra mirrors. %v", err)
}
return nil
@@ -214,7 +214,7 @@ func (m *Mirroring) removeExtraMirrors() error {
propagation := metav1.DeletePropagationForeground
deleteOpts := metav1.DeleteOptions{GracePeriodSeconds: &gracePeriod, PropagationPolicy: &propagation}
if err = m.context.Clientset.AppsV1().Deployments(m.Namespace).Delete(deploy.Name, &deleteOpts); err != nil {
logger.Warningf("failed to delete rbd-mirror %s. %+v", daemonName, err)
logger.Warningf("failed to delete rbd-mirror %q. %v", daemonName, err)
}
logger.Infof("removed rbd-mirror %s", daemonName)
+3 -3
View File
@@ -84,8 +84,8 @@ func (k *SecretStore) GenerateKey(user string, access []string) (string, error)
// get-or-create-key for the user account
key, err := client.AuthGetOrCreateKey(k.context, k.namespace, user, access)
if err != nil {
logger.Infof("Error getting or creating key for %s. "+
"Attempting to update capabilities in case the user already exists. %+v", user, err)
logger.Infof("Error getting or creating key for %q. "+
"Attempting to update capabilities in case the user already exists. %v", user, err)
uErr := client.AuthUpdateCaps(k.context, k.namespace, user, access)
if uErr != nil {
return "", errors.Wrapf(err, "failed to get, create, or update auth key for %s", user)
@@ -121,7 +121,7 @@ func (k *SecretStore) Delete(resourceName string) error {
secretName := keyringSecretName(resourceName)
err := k.context.Clientset.CoreV1().Secrets(k.namespace).Delete(secretName, &metav1.DeleteOptions{})
if err != nil && !kerrors.IsNotFound(err) {
logger.Warningf("failed to delete keyring secret for %s. user may need to delete the resource manually. %+v", secretName, err)
logger.Warningf("failed to delete keyring secret for %q. user may need to delete the resource manually. %v", secretName, err)
}
return nil
+1 -1
View File
@@ -75,7 +75,7 @@ func (m *MonStore) SetAll(options ...Option) error {
if len(errs) > 0 {
retErr := errors.New("failed to set one or more Ceph configs")
for _, err := range errs {
retErr = errors.Wrapf(err, "%+v", retErr)
retErr = errors.Wrapf(err, "%v", retErr)
}
return retErr
}
+5 -5
View File
@@ -35,19 +35,19 @@ func (o *Operator) startManager(stopCh <-chan struct{}) {
logger.Info("setting up the controller-runtime manager")
kubeConfig, err := config.GetConfig()
if err != nil {
logger.Errorf("unable to get client config for controller-runtime manager: %+v", err)
logger.Errorf("unable to get client config for controller-runtime manager. %v", err)
return
}
mgr, err := manager.New(kubeConfig, mgrOpts)
if err != nil {
logger.Errorf("unable to set up overall controller-runtime manager: %+v", err)
logger.Errorf("unable to set up overall controller-runtime manager. %v", err)
return
}
// Add the registered controllers to the manager (entrypoint for controllers)
err = cluster.AddToManager(mgr)
if err != nil {
logger.Errorf("Can't add controllers to controller-runtime manager: %+v", err)
logger.Errorf("cannot add controllers to controller-runtime manager. %v", err)
}
// options to pass to the controllers
controllerOpts := &controllerconfig.Context{
@@ -59,12 +59,12 @@ func (o *Operator) startManager(stopCh <-chan struct{}) {
// Add the registered controllers to the manager (entrypoint for controllers)
err = controllers.AddToManager(mgr, controllerOpts)
if err != nil {
logger.Errorf("Can't add controllers to controller-runtime manager: %+v", err)
logger.Errorf("cannot add controllers to controller-runtime manager. %v", err)
}
logger.Info("starting the controller-runtime manager")
if err := mgr.Start(stopCh); err != nil {
logger.Errorf("unable to run the controller-runtime manager: %+v", err)
logger.Errorf("unable to run the controller-runtime manager. %v", err)
return
}
}
+3 -3
View File
@@ -117,7 +117,7 @@ func getToleration(provisioner bool) []corev1.Toleration {
if tolerationsRaw != "" {
tolerations, err = k8sutil.YamlToTolerations(tolerationsRaw)
if err != nil {
logger.Warningf("failed to parse %s. %+v", tolerationsRaw, err)
logger.Warningf("failed to parse %q. %v", tolerationsRaw, err)
}
}
for i := range tolerations {
@@ -143,7 +143,7 @@ func getNodeAffinity(provisioner bool) *corev1.NodeAffinity {
if nodeAffinity != "" {
v1NodeAffinity, err := k8sutil.GenerateNodeAffinity(nodeAffinity)
if err != nil {
logger.Warningf("failed to parse %s. %+v", nodeAffinity, err)
logger.Warningf("failed to parse %q. %v", nodeAffinity, err)
}
return v1NodeAffinity
}
@@ -164,7 +164,7 @@ func getPortFromENV(env string, defaultPort uint16) uint16 {
}
p, err := strconv.ParseUint(port, 10, 64)
if err != nil {
logger.Debugf("failed to parse port value for env %s. using default port %d. %+v", env, defaultPort, err)
logger.Debugf("failed to parse port value for env %q. using default port %d. %v", env, defaultPort, err)
return defaultPort
}
if p > 65535 {
@@ -67,7 +67,7 @@ func Add(mgr manager.Manager, context *controllerconfig.Context) error {
// The name will be populated in the reconcile
namespace := obj.Meta.GetNamespace()
if len(namespace) == 0 {
logger.Errorf("enqueByNamespace recieved an obj without a namespace: %+v", obj)
logger.Errorf("enqueByNamespace received an obj without a namespace. %+v", obj)
return []reconcile.Request{}
}
req := reconcile.Request{NamespacedName: types.NamespacedName{Namespace: namespace}}
@@ -89,7 +89,7 @@ func Add(mgr manager.Manager, context *controllerconfig.Context) error {
_, ok := obj.Object.(*policyv1beta1.PodDisruptionBudget)
if !ok {
// not a pdb, returning empty
logger.Errorf("PDB handler recieved non-PDB")
logger.Errorf("PDB handler received non-PDB")
return []reconcile.Request{}
}
labels := obj.Meta.GetLabels()
@@ -119,7 +119,7 @@ func Add(mgr manager.Manager, context *controllerconfig.Context) error {
_, ok := obj.Object.(*appsv1.Deployment)
if !ok {
// not a Deployment, returning empty
logger.Errorf("Deployment handler recieved non-Deployment")
logger.Errorf("deployment handler received non-Deployment")
return []reconcile.Request{}
}
@@ -135,7 +135,7 @@ func (r *ReconcileClusterDisruption) initializePDBState(request reconcile.Reques
for _, osdData := range osdDataList {
err := r.createPDBForOSD(osdData.Deployment)
if err != nil {
return pdbStateMap, errors.Wrapf(err, "failed to create pdb for osd deployment %s. %+v", osdData.Deployment.ObjectMeta.GetName(), err)
return pdbStateMap, errors.Wrapf(err, "failed to create pdb for osd deployment %q. %v", osdData.Deployment.ObjectMeta.GetName(), err)
}
}
pdbStateMap.Data = map[string]string{disabledPDBKey: ""}
@@ -181,7 +181,7 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
if disabledPDBTimeSet {
disabledPDBTime, err = time.Parse(time.RFC3339, timeString)
if err != nil {
logger.Errorf("Could not parse timestamp %v: %v", disabledPDBTime, err)
logger.Errorf("could not parse timestamp %v. %v", disabledPDBTime, err)
disabledPDBTime = time.Now()
pdbStateMap.Data[disabledPDBTimeKey] = disabledPDBTime.Format(time.RFC3339)
}
@@ -190,7 +190,7 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
shouldChange := clean && !recentlyChanged
activeDrains := len(drainingFailureDomains) != 0
if activeDrains {
logger.Infof("pg health: %s. detected drains on %ss: %v", pgHealthMsg, poolFailureDomain, drainingFailureDomains)
logger.Infof("pg health: %q. detected drains on %q: %v", pgHealthMsg, poolFailureDomain, drainingFailureDomains)
}
if shouldChange {
if activeDrains {
@@ -204,12 +204,12 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
err = r.updateNoout(pdbStateMap, allFailureDomainsMap)
if err != nil {
logger.Errorf("could not update maintenance noout in cluster %s with ceph image : %+v", request, err)
logger.Errorf("could not update maintenance noout in cluster %q with ceph image. %v", request, err)
}
err = r.client.Update(context.TODO(), pdbStateMap)
if err != nil {
return errors.Wrapf(err, "could not update %s in cluster %s", pdbStateMapName, request)
return errors.Wrapf(err, "could not update %q in cluster %q", pdbStateMapName, request)
}
drainingFailureDomain, ok := pdbStateMap.Data[disabledPDBKey]
if ok && clean && len(drainingFailureDomain) > 0 {
@@ -227,7 +227,7 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
if time.Since(drainingCanary.GetCreationTimestamp().Time) > time.Minute && drainingCanary.Status.ReadyReplicas < 1 {
err := r.client.Delete(context.TODO(), &drainingCanary)
if err != nil {
logger.Warningf("could not delete canary deployment %q in namespace %q: %+v", drainingCanary.GetName(), drainingCanary.GetNamespace(), err)
logger.Warningf("could not delete canary deployment %q in namespace %q. %v", drainingCanary.GetName(), drainingCanary.GetNamespace(), err)
}
}
}
@@ -241,7 +241,7 @@ func (r *ReconcileClusterDisruption) reconcilePDBsForOSDs(
err = r.createPDBForOSD(osdData.Deployment)
}
if err != nil {
return errors.Wrapf(err, "failed to reconcile pdb for osd deployment %s. %+v", osdData.Deployment.ObjectMeta.GetName(), err)
return errors.Wrapf(err, "failed to reconcile pdb for osd deployment %q. %v", osdData.Deployment.ObjectMeta.GetName(), err)
}
}
}
@@ -123,7 +123,7 @@ func (r *ReconcileClusterDisruption) getOngoingDrains(request reconcile.Request)
if deployment.Status.ReadyReplicas < 1 {
nodeHostname, ok := deployment.Spec.Template.Spec.NodeSelector[corev1.LabelHostname]
if !ok {
logger.Errorf("could not find a the nodeSelector key %s for canary deployment %s", corev1.LabelHostname, deployment.ObjectMeta.Name)
logger.Errorf("could not find a the nodeSelector key %q for canary deployment %q", corev1.LabelHostname, deployment.ObjectMeta.Name)
continue
}
@@ -76,7 +76,7 @@ func (r *ReconcileClusterDisruption) Reconcile(request reconcile.Request) (recon
func (r *ReconcileClusterDisruption) reconcile(request reconcile.Request) (reconcile.Result, error) {
if len(request.Namespace) == 0 {
return reconcile.Result{}, errors.Errorf("request did not have namespace: %s", request.NamespacedName)
return reconcile.Result{}, errors.Errorf("request did not have namespace: %q", request.NamespacedName)
}
// ensure that the cluster name is populated
@@ -85,23 +85,23 @@ func (r *ReconcileClusterDisruption) reconcile(request reconcile.Request) (recon
if !found && r.namelessRetries < maxNamelessRetries {
// ensure that a deleted cluster doesn't result in infinite retries
r.namelessRetries++
logger.Infof("clusterName is not known yet for namespace %s", request.Namespace)
logger.Infof("clusterName is not known yet for namespace %q", request.Namespace)
return reconcile.Result{Requeue: true, RequeueAfter: 5 * time.Second}, errors.New("clusterName for this namespace not yet known")
}
request.Name = clusterName
logger.Debugf("discovered NamespacedName: %s", request.NamespacedName)
logger.Debugf("discovered NamespacedName: %q", request.NamespacedName)
}
r.namelessRetries = 0
logger.Debugf("reconciling %s", request.NamespacedName)
logger.Debugf("reconciling %q", request.NamespacedName)
// get the ceph cluster
cephCluster := &cephv1.CephCluster{}
err := r.client.Get(context.TODO(), request.NamespacedName, cephCluster)
if kerrors.IsNotFound(err) {
logger.Errorf("cephcluster %s seems to be deleted, not requeuing until triggered again", request)
logger.Errorf("cephcluster %q seems to be deleted, not requeuing until triggered again", request)
return reconcile.Result{}, nil
} else if err != nil {
return reconcile.Result{}, errors.Wrapf(err, "could not get the ceph cluster %s", request.NamespacedName)
return reconcile.Result{}, errors.Wrapf(err, "could not get the ceph cluster %q", request.NamespacedName)
}
// update the clustermap with the cluster's name so that
@@ -121,12 +121,12 @@ func (r *ReconcileNode) reconcile(request reconcile.Request) (reconcile.Result,
var deployment appsv1.Deployment
err := r.client.List(context.TODO(), deploymentList, client.MatchingLabels(labelSelector), client.InNamespace(osdPod.GetNamespace()))
if err != nil || len(deploymentList.Items) < 1 {
logger.Errorf("Cannot find deployment for osd id %s in namespace %s", labels[osd.OsdIdLabelKey], osdPod.GetNamespace())
logger.Errorf("cannot find deployment for osd id %q in namespace %q", labels[osd.OsdIdLabelKey], osdPod.GetNamespace())
} else {
deployment = deploymentList.Items[0]
}
if len(deploymentList.Items) > 1 {
logger.Errorf("Found multiple deployments for osd id %s in namespace %s: %+v", labels[osd.OsdIdLabelKey], osdPod.GetNamespace(), deploymentList)
logger.Errorf("found multiple deployments for osd id %q in namespace %q: %+v", labels[osd.OsdIdLabelKey], osdPod.GetNamespace(), deploymentList)
}
occupiedByOSD = true
@@ -134,7 +134,7 @@ func (r *ReconcileNode) reconcile(request reconcile.Request) (reconcile.Result,
for _, osdToleration := range deployment.Spec.Template.Spec.Tolerations {
if osdToleration.Key == "node.kubernetes.io/unschedulable" {
logger.Errorf(
"osd %s in namespace %s tolerates the drain taint, but the drain canary will not.",
"osd %q in namespace %q tolerates the drain taint, but the drain canary will not.",
labels[osd.OsdIdLabelKey],
labels[k8sutil.ClusterAttr],
)
@@ -166,11 +166,11 @@ func (r *ReconcileNode) reconcile(request reconcile.Request) (reconcile.Result,
}
ownerReferences := []metav1.OwnerReference{}
// get the operator Deployment to use as an owner refreence
// get the operator Deployment to use as an owner reference
operatorPodKey := types.NamespacedName{Name: os.Getenv(k8sutil.PodNameEnvVar), Namespace: r.context.OperatorNamespace}
operatorDeployment, err := getDeploymentForPod(r.client, operatorPodKey)
if err != nil {
logger.Errorf("could not find rook operator deployment for pod %+v: %+v", operatorPodKey, err)
logger.Errorf("could not find rook operator deployment for pod %+v. %v", operatorPodKey, err)
} else {
operatorDeployment := operatorDeployment
controllerBool := true
+11 -11
View File
@@ -106,7 +106,7 @@ func (c *FilesystemController) onAdd(obj interface{}) {
filesystem, err := getFilesystemObject(obj)
if err != nil {
logger.Errorf("failed to get filesystem object: %+v", err)
logger.Errorf("failed to get filesystem object. %v", err)
return
}
@@ -118,14 +118,14 @@ func (c *FilesystemController) onAdd(obj interface{}) {
if err != nil {
// This handles the case where the operator is running, the external cluster has been upgraded and a CR creation is called
// If that's a major version upgrade we fail, if it's a minor version, we continue, it's not ideal but not critical
logger.Errorf("refusing to run new crd. %+v", err)
logger.Errorf("refusing to run new crd. %v", err)
return
}
}
err = createFilesystem(c.clusterInfo, c.context, *filesystem, c.rookVersion, c.clusterSpec, c.filesystemOwner(filesystem), c.clusterSpec.DataDirHostPath, c.isUpgrade)
if err != nil {
logger.Errorf("failed to create filesystem %s: %+v", filesystem.Name, err)
logger.Errorf("failed to create filesystem %q. %v", filesystem.Name, err)
}
}
@@ -137,12 +137,12 @@ func (c *FilesystemController) onUpdate(oldObj, newObj interface{}) {
oldFS, err := getFilesystemObject(oldObj)
if err != nil {
logger.Errorf("failed to get old filesystem object: %+v", err)
logger.Errorf("failed to get old filesystem object. %v", err)
return
}
newFS, err := getFilesystemObject(newObj)
if err != nil {
logger.Errorf("failed to get new filesystem object: %+v", err)
logger.Errorf("failed to get new filesystem object. %v", err)
return
}
@@ -158,7 +158,7 @@ func (c *FilesystemController) onUpdate(oldObj, newObj interface{}) {
logger.Infof("updating filesystem %s", newFS.Name)
err = createFilesystem(c.clusterInfo, c.context, *newFS, c.rookVersion, c.clusterSpec, c.filesystemOwner(newFS), c.clusterSpec.DataDirHostPath, c.isUpgrade)
if err != nil {
logger.Errorf("failed to create (modify) filesystem %s: %+v", newFS.Name, err)
logger.Errorf("failed to create (modify) filesystem %q. %v", newFS.Name, err)
}
}
@@ -179,16 +179,16 @@ func (c *FilesystemController) ParentClusterChanged(cluster cephv1.ClusterSpec,
c.clusterSpec.CephVersion = cluster.CephVersion
filesystems, err := c.context.RookClientset.CephV1().CephFilesystems(c.namespace).List(metav1.ListOptions{})
if err != nil {
logger.Errorf("failed to retrieve filesystems to update the ceph version. %+v", err)
logger.Errorf("failed to retrieve filesystems to update the ceph version. %v", err)
return
}
for _, fs := range filesystems.Items {
logger.Infof("updating the ceph version for filesystem %s to %s", fs.Name, c.clusterSpec.CephVersion.Image)
err = createFilesystem(c.clusterInfo, c.context, fs, c.rookVersion, c.clusterSpec, c.filesystemOwner(&fs), c.clusterSpec.DataDirHostPath, c.isUpgrade)
if err != nil {
logger.Errorf("failed to update filesystem %s. %+v", fs.Name, err)
logger.Errorf("failed to update filesystem %q. %v", fs.Name, err)
} else {
logger.Infof("updated filesystem %s to ceph version %s", fs.Name, c.clusterSpec.CephVersion.Image)
logger.Infof("updated filesystem %q to ceph version %q", fs.Name, c.clusterSpec.CephVersion.Image)
}
}
}
@@ -201,7 +201,7 @@ func (c *FilesystemController) onDelete(obj interface{}) {
filesystem, err := getFilesystemObject(obj)
if err != nil {
logger.Errorf("failed to get filesystem object: %+v", err)
logger.Errorf("failed to get filesystem object. %v", err)
return
}
@@ -210,7 +210,7 @@ func (c *FilesystemController) onDelete(obj interface{}) {
err = deleteFilesystem(c.context, c.clusterInfo.CephVersion, *filesystem)
if err != nil {
logger.Errorf("failed to delete filesystem %s: %+v", filesystem.Name, err)
logger.Errorf("failed to delete filesystem %q. %v", filesystem.Name, err)
}
}
+3 -3
View File
@@ -88,7 +88,7 @@ func createFilesystem(
// set the number of active mds instances
if fs.Spec.MetadataServer.ActiveCount > 1 {
if err = client.SetNumMDSRanks(context, clusterInfo.CephVersion, fs.Namespace, fs.Name, fs.Spec.MetadataServer.ActiveCount); err != nil {
logger.Warningf("failed setting active mds count to %d. %+v", fs.Spec.MetadataServer.ActiveCount, err)
logger.Warningf("failed setting active mds count to %d. %v", fs.Spec.MetadataServer.ActiveCount, err)
}
}
@@ -177,7 +177,7 @@ func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, cephVersion c
fmt.Sprintf("failed to set num mds ranks (max_mds) to %d for filesystem %s, still continuing. ", f.activeMDSCount, f.Name) +
"this error is not critical, but mdses may not be as failure tolerant as desired. " +
fmt.Sprintf("USER should verify that the number of active mdses is %d with 'ceph fs get %s'", f.activeMDSCount, f.Name) +
fmt.Sprintf(": %+v", err),
fmt.Sprintf(". %v", err),
)
}
return nil
@@ -228,7 +228,7 @@ func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, cephVersion c
if pool.Type == model.ErasureCoded {
// An erasure coded data pool used for a filesystem must allow overwrites
if err := client.SetPoolProperty(context, clusterName, pool.Name, "allow_ec_overwrites", "true"); err != nil {
logger.Warningf("failed to set ec pool property: %+v", err)
logger.Warningf("failed to set ec pool property. %v", err)
}
}
}
+1 -1
View File
@@ -57,7 +57,7 @@ func (c *Cluster) generateKeyring(m *mdsConfig) (string, error) {
if kerrors.IsNotFound(err) {
logger.Debugf("legacy mds key %s is already removed", m.ResourceName)
} else {
logger.Warningf("legacy mds key %s could not be removed: %+v", m.ResourceName, err)
logger.Warningf("legacy mds key %q could not be removed. %v", m.ResourceName, err)
}
}
+5 -5
View File
@@ -112,7 +112,7 @@ func (c *Cluster) Start() error {
defer func() {
if fsPreparedForUpgrade {
if err := finishedWithDaemonUpgrade(c.context, c.clusterInfo.CephVersion, c.fs.Namespace, c.fs.Name, c.fs.Spec.MetadataServer.ActiveCount); err != nil {
logger.Errorf("for filesystem %s, USER should make sure the Ceph fs max_mds property is set to %d: %+v",
logger.Errorf("for filesystem %q, USER should make sure the Ceph fs max_mds property is set to %d. %v",
c.fs.Name, c.fs.Spec.MetadataServer.ActiveCount, err)
}
}
@@ -172,7 +172,7 @@ func (c *Cluster) Start() error {
}
if err := c.associateKeyring(keyring, createdDeployment); err != nil {
logger.Warningf("failed to associate keyring with deployment for %q. %+v", resourceName, err)
logger.Warningf("failed to associate keyring with deployment for %q. %v", resourceName, err)
}
// keyring must be generated before update-and-wait since no keyring will prevent the
@@ -183,7 +183,7 @@ func (c *Cluster) Start() error {
var cephVersionToUse cephver.CephVersion
currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon)
if err != nil {
logger.Warningf("failed to retrieve current ceph %s version. %+v", daemon, err)
logger.Warningf("failed to retrieve current ceph %q version. %v", daemon, err)
logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion")
cephVersionToUse = c.clusterInfo.CephVersion
@@ -236,13 +236,13 @@ func (c *Cluster) scaleDownDeployments(replicas int32, desiredDeployments map[st
"number of active mds ranks is not as desired. it is potentially unsafe to continue with extraneous mds deletion, so stopping. " +
fmt.Sprintf("USER should delete undesired mds daemons once filesystem %s is healthy. ", c.fs.Name) +
fmt.Sprintf("desired mds deployments for this filesystem are %+v", desiredDeployments) +
fmt.Sprintf(": %+v", err),
fmt.Sprintf(". %v", err),
)
break // stop trying to delete daemons, but continue to reporting any errors below
}
if err := deleteMdsDeployment(c.context, c.fs.Namespace, &d); err != nil {
errCount++
logger.Errorf("error during deletion of extraneous mds deployments: %+v", err)
logger.Errorf("error during deletion of extraneous mds deployments. %v", err)
}
}
}
+12 -12
View File
@@ -91,7 +91,7 @@ func (c *CephNFSController) onAdd(obj interface{}) {
nfs := obj.(*cephv1.CephNFS).DeepCopy()
if !c.clusterInfo.CephVersion.IsAtLeastNautilus() {
logger.Errorf("Ceph NFS is only supported with Nautilus or newer. CRD %s will be ignored.", nfs.Name)
logger.Errorf("Ceph NFS is only supported with Nautilus or newer. CRD %q will be ignored.", nfs.Name)
return
}
@@ -100,7 +100,7 @@ func (c *CephNFSController) onAdd(obj interface{}) {
err := c.upCephNFS(*nfs, 0)
if err != nil {
logger.Errorf("failed to create NFS Ganesha %s. %+v", nfs.Name, err)
logger.Errorf("failed to create NFS Ganesha %q. %v", nfs.Name, err)
}
}
@@ -113,12 +113,12 @@ func (c *CephNFSController) onUpdate(oldObj, newObj interface{}) {
oldNFS := oldObj.(*cephv1.CephNFS).DeepCopy()
newNFS := newObj.(*cephv1.CephNFS).DeepCopy()
if !c.clusterInfo.CephVersion.IsAtLeastNautilus() {
logger.Errorf("Ceph NFS is only supported with Nautilus or newer. CRD %s will be ignored.", newNFS.Name)
logger.Errorf("Ceph NFS is only supported with Nautilus or newer. CRD %q will be ignored.", newNFS.Name)
return
}
if !nfsChanged(oldNFS.Spec, newNFS.Spec) {
logger.Debugf("nfs ganesha %s not updated", newNFS.Name)
logger.Debugf("nfs ganesha %q not updated", newNFS.Name)
return
}
@@ -129,12 +129,12 @@ func (c *CephNFSController) onUpdate(oldObj, newObj interface{}) {
if oldNFS.Spec.Server.Active < newNFS.Spec.Server.Active {
err := c.upCephNFS(*newNFS, oldNFS.Spec.Server.Active)
if err != nil {
logger.Errorf("Failed to start daemons for CephNFS %s. %+v", newNFS.Name, err)
logger.Errorf("Failed to start daemons for CephNFS %q. %v", newNFS.Name, err)
}
} else {
err := c.downCephNFS(*oldNFS, newNFS.Spec.Server.Active)
if err != nil {
logger.Errorf("Failed to stop daemons for CephNFS %s. %+v", newNFS.Name, err)
logger.Errorf("Failed to stop daemons for CephNFS %q. %v", newNFS.Name, err)
}
}
@@ -148,7 +148,7 @@ func (c *CephNFSController) onDelete(obj interface{}) {
nfs := obj.(*cephv1.CephNFS).DeepCopy()
if !c.clusterInfo.CephVersion.IsAtLeastNautilus() {
logger.Errorf("Ceph NFS is only supported with Nautilus or newer. CRD %s cleanup will be ignored.", nfs.Name)
logger.Errorf("Ceph NFS is only supported with Nautilus or newer. CRD %q cleanup will be ignored.", nfs.Name)
return
}
@@ -157,7 +157,7 @@ func (c *CephNFSController) onDelete(obj interface{}) {
err := c.downCephNFS(*nfs, 0)
if err != nil {
logger.Errorf("failed to delete file system %s. %+v", nfs.Name, err)
logger.Errorf("failed to delete file system %s. %v", nfs.Name, err)
}
}
@@ -180,16 +180,16 @@ func (c *CephNFSController) ParentClusterChanged(cluster cephv1.ClusterSpec, clu
c.clusterSpec.CephVersion = cluster.CephVersion
nfses, err := c.context.RookClientset.CephV1().CephNFSes(c.namespace).List(metav1.ListOptions{})
if err != nil {
logger.Errorf("failed to retrieve NFSes to update the ceph version. %+v", err)
logger.Errorf("failed to retrieve NFSes to update the ceph version. %v", err)
return
}
for _, nfs := range nfses.Items {
logger.Infof("updating the ceph version for nfs %s to %s", nfs.Name, c.clusterSpec.CephVersion.Image)
logger.Infof("updating the ceph version for nfs %q to %q", nfs.Name, c.clusterSpec.CephVersion.Image)
err := c.upCephNFS(nfs, 0)
if err != nil {
logger.Errorf("failed to update nfs %s. %+v", nfs.Name, err)
logger.Errorf("failed to update nfs %q. %v", nfs.Name, err)
} else {
logger.Infof("updated nfs %s to ceph version %s", nfs.Name, c.clusterSpec.CephVersion.Image)
logger.Infof("updated nfs %q to ceph version %q", nfs.Name, c.clusterSpec.CephVersion.Image)
}
}
}
+3 -3
View File
@@ -134,15 +134,15 @@ func (c *CephNFSController) addServerToDatabase(nfs cephv1.CephNFS, name string)
logger.Infof("Adding ganesha %s to grace db", name)
if err := c.runGaneshaRadosGrace(nfs, name, "add"); err != nil {
logger.Errorf("failed to add %s to grace db. %+v", name, err)
logger.Errorf("failed to add %q to grace db. %v", name, err)
}
}
func (c *CephNFSController) removeServerFromDatabase(nfs cephv1.CephNFS, name string) {
logger.Infof("Removing ganesha %s from grace db", name)
logger.Infof("Removing ganesha %q from grace db", name)
if err := c.runGaneshaRadosGrace(nfs, name, "remove"); err != nil {
logger.Errorf("failed to remove %s from grace db. %+v", name, err)
logger.Errorf("failed to remove %q from grace db. %v", name, err)
}
}
@@ -39,7 +39,7 @@ func (p *Provisioner) getStorageClassWithBackoff(name string) (class *storagev1.
if kerrors.IsNotFound(err) {
return true, err
}
logger.Errorf("error getting class %s, retrying: %v", name, err)
logger.Errorf("error getting class %q, retrying. %v", name, err)
return false, nil
})
if err != nil {
@@ -65,7 +65,7 @@ func (p *Provisioner) getSecretWithBackoff(namespace, name string) (secret *v1.S
if kerrors.IsNotFound(err) {
return true, err
}
logger.Errorf("error getting class %s, retrying: %v", name, err)
logger.Errorf("error getting class %q, retrying. %v", name, err)
return false, nil
})
return
@@ -243,7 +243,7 @@ func (p Provisioner) Revoke(ob *bktv1alpha1.ObjectBucket) error {
logger.Errorf("no bucket policy for bucket %q, so no need to drop policy", p.bucketName)
} else {
logger.Errorf("error getting policy for bucket %q: %v", p.bucketName, err)
logger.Errorf("error getting policy for bucket %q. %v", p.bucketName, err)
return err
}
}
@@ -277,7 +277,7 @@ func (p *Provisioner) initializeCreateOrGrant(options *apibkt.BucketOptions) err
scName := options.ObjectBucketClaim.Spec.StorageClassName
sc, err := p.getStorageClassWithBackoff(scName)
if err != nil {
logger.Errorf("failed to get storage class for OBC \"%s/%s\": %v", obc.Namespace, obc.Name, err)
logger.Errorf("failed to get storage class for OBC %q in namespace %q. %v", obc.Name, obc.Namespace, err)
return err
}
+13 -13
View File
@@ -104,7 +104,7 @@ func (c *ObjectStoreController) onAdd(obj interface{}) {
objectstore, err := getObjectStoreObject(obj)
if err != nil {
logger.Errorf("failed to get objectstore object: %+v", err)
logger.Errorf("failed to get objectstore object. %v", err)
return
}
@@ -116,7 +116,7 @@ func (c *ObjectStoreController) onAdd(obj interface{}) {
if err != nil {
// This handles the case where the operator is running, the external cluster has been upgraded and a CR creation is called
// If that's a major version upgrade we fail, if it's a minor version, we continue, it's not ideal but not critical
logger.Errorf("refusing to run new crd. %+v", err)
logger.Errorf("refusing to run new crd. %v", err)
return
}
}
@@ -132,17 +132,17 @@ func (c *ObjectStoreController) onUpdate(oldObj, newObj interface{}) {
// if the object store spec is modified, update the object store
oldStore, err := getObjectStoreObject(oldObj)
if err != nil {
logger.Errorf("failed to get old objectstore object: %+v", err)
logger.Errorf("failed to get old objectstore object. %v", err)
return
}
newStore, err := getObjectStoreObject(newObj)
if err != nil {
logger.Errorf("failed to get new objectstore object: %+v", err)
logger.Errorf("failed to get new objectstore object. %v", err)
return
}
if !storeChanged(oldStore.Spec, newStore.Spec) {
logger.Debugf("object store %s did not change", newStore.Name)
logger.Debugf("object store %q did not change", newStore.Name)
return
}
@@ -153,7 +153,7 @@ func (c *ObjectStoreController) onUpdate(oldObj, newObj interface{}) {
}
func (c *ObjectStoreController) createOrUpdateStore(objectstore *cephv1.CephObjectStore) {
logger.Infof("creating object store %s", objectstore.Name)
logger.Infof("creating object store %q", objectstore.Name)
cfg := clusterConfig{
clusterInfo: c.clusterInfo,
context: c.context,
@@ -165,7 +165,7 @@ func (c *ObjectStoreController) createOrUpdateStore(objectstore *cephv1.CephObje
isUpgrade: c.isUpgrade,
}
if err := cfg.createOrUpdate(); err != nil {
logger.Errorf("failed to create or update object store %s. %+v", objectstore.Name, err)
logger.Errorf("failed to create or update object store %s. %v", objectstore.Name, err)
}
}
@@ -177,7 +177,7 @@ func (c *ObjectStoreController) onDelete(obj interface{}) {
objectstore, err := getObjectStoreObject(obj)
if err != nil {
logger.Errorf("failed to get objectstore object: %+v", err)
logger.Errorf("failed to get objectstore object. %v", err)
return
}
@@ -186,7 +186,7 @@ func (c *ObjectStoreController) onDelete(obj interface{}) {
cfg := clusterConfig{context: c.context, store: *objectstore}
if err = cfg.deleteStore(); err != nil {
logger.Errorf("failed to delete object store %s. %+v", objectstore.Name, err)
logger.Errorf("failed to delete object store %q. %v", objectstore.Name, err)
}
}
@@ -207,16 +207,16 @@ func (c *ObjectStoreController) ParentClusterChanged(cluster cephv1.ClusterSpec,
c.clusterSpec.CephVersion = cluster.CephVersion
objectStores, err := c.context.RookClientset.CephV1().CephObjectStores(c.namespace).List(metav1.ListOptions{})
if err != nil {
logger.Errorf("failed to retrieve object stores to update the ceph version. %+v", err)
logger.Errorf("failed to retrieve object stores to update the ceph version. %v", err)
return
}
for _, store := range objectStores.Items {
logger.Infof("updating the ceph version for object store %s to %s", store.Name, c.clusterSpec.CephVersion.Image)
logger.Infof("updating the ceph version for object store %q to %q", store.Name, c.clusterSpec.CephVersion.Image)
c.createOrUpdateStore(&store)
if err != nil {
logger.Errorf("failed to update object store %s. %+v", store.Name, err)
logger.Errorf("failed to update object store %q. %v", store.Name, err)
} else {
logger.Infof("updated object store %s to ceph version %s", store.Name, c.clusterSpec.CephVersion.Image)
logger.Infof("updated object store %q to ceph version %q", store.Name, c.clusterSpec.CephVersion.Image)
}
}
}
+4 -4
View File
@@ -171,17 +171,17 @@ func deleteRealm(context *Context) error {
// <name>
_, err := runAdminCommand(context, "realm", "delete", "--rgw-realm", context.Name)
if err != nil {
logger.Warningf("failed to delete rgw realm %s. %+v", context.Name, err)
logger.Warningf("failed to delete rgw realm %q. %v", context.Name, err)
}
_, err = runAdminCommand(context, "zonegroup", "delete", "--rgw-zonegroup", context.Name)
if err != nil {
logger.Warningf("failed to delete rgw zonegroup %s. %+v", context.Name, err)
logger.Warningf("failed to delete rgw zonegroup %q. %v", context.Name, err)
}
_, err = runAdminCommand(context, "zone", "delete", "--rgw-zone", context.Name)
if err != nil {
logger.Warningf("failed to delete rgw zone %s. %+v", context.Name, err)
logger.Warningf("failed to delete rgw zone %q. %v", context.Name, err)
}
return nil
@@ -224,7 +224,7 @@ func deletePools(context *Context, lastStore bool) error {
for _, pool := range pools {
name := poolName(context.Name, pool)
if err := ceph.DeletePool(context.Context, context.ClusterName, name); err != nil {
logger.Warningf("failed to delete pool %s. %+v", name, err)
logger.Warningf("failed to delete pool %q. %v", name, err)
}
}
+20 -20
View File
@@ -162,7 +162,7 @@ func (c *clusterConfig) startRGWPods() error {
err = c.associateKeyring(keyring, resourceControllerOwnerRef)
if err != nil {
logger.Warningf("failed to associate keyring with rgw deployment %q: %+v", createdDeployment.Name, err)
logger.Warningf("failed to associate keyring with rgw deployment %q. %v", createdDeployment.Name, err)
}
// Generate the mime.types file after the rep. controller as well for the same reason as keyring
@@ -172,7 +172,7 @@ func (c *clusterConfig) startRGWPods() error {
var cephVersionToUse cephver.CephVersion
currentCephVersion, err := client.LeastUptodateDaemonVersion(c.context, c.clusterInfo.Name, daemon)
if err != nil {
logger.Warningf("failed to retrieve current ceph %s version. %+v", daemon, err)
logger.Warningf("failed to retrieve current ceph %s version. %v", daemon, err)
logger.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion")
cephVersionToUse = c.clusterInfo.CephVersion
@@ -193,18 +193,18 @@ func (c *clusterConfig) startRGWPods() error {
// scale down scenario
deps, err := k8sutil.GetDeployments(c.context.Clientset, c.store.Namespace, c.storeLabelSelector())
if err != nil {
logger.Warning("could not get deployments for object store %s (matching label selector '%s'): %+v", c.store.Name, c.storeLabelSelector(), err)
logger.Warning("could not get deployments for object store %q (matching label selector %q). %v", c.store.Name, c.storeLabelSelector(), err)
}
currentRgwInstances := int(len(deps.Items))
if currentRgwInstances > desiredRgwInstances {
logger.Infof("found more rgw deployments %d than desired %d in object store %s, scaling down", currentRgwInstances, c.store.Spec.Gateway.Instances, c.store.Name)
logger.Infof("found more rgw deployments %d than desired %d in object store %q, scaling down", currentRgwInstances, c.store.Spec.Gateway.Instances, c.store.Name)
diffCount := currentRgwInstances - desiredRgwInstances
for i := 0; i < diffCount; {
depIDToRemove := currentRgwInstances - 1
depNameToRemove := fmt.Sprintf("%s-%s-%s", AppName, c.store.Name, k8sutil.IndexToName(depIDToRemove))
if err := k8sutil.DeleteDeployment(c.context.Clientset, c.store.Namespace, depNameToRemove); err != nil {
logger.Warning("error during deletion of deployment %s resource: %+v", depNameToRemove, err)
logger.Warning("error during deletion of deployment %q resource. %v", depNameToRemove, err)
}
currentRgwInstances = currentRgwInstances - 1
i++
@@ -212,17 +212,17 @@ func (c *clusterConfig) startRGWPods() error {
// Delete the auth key
err = client.AuthDelete(c.context, c.store.Namespace, generateCephXUser(depNameToRemove))
if err != nil {
logger.Infof("failed to delete rgw key %s. %+v", depNameToRemove, err)
logger.Infof("failed to delete rgw key %q. %v", depNameToRemove, err)
}
}
// verify scale down was successful
deps, err = k8sutil.GetDeployments(c.context.Clientset, c.store.Namespace, c.storeLabelSelector())
if err != nil {
logger.Warning("could not get deployments for object store %s (matching label selector '%s'): %+v", c.store.Name, c.storeLabelSelector(), err)
logger.Warning("could not get deployments for object store %q (matching label selector %q). %v", c.store.Name, c.storeLabelSelector(), err)
}
currentRgwInstances = len(deps.Items)
if currentRgwInstances == desiredRgwInstances {
logger.Infof("successfully scaled down rgw deployments to %d in object store %s", desiredRgwInstances, c.store.Name)
logger.Infof("successfully scaled down rgw deployments to %d in object store %q", desiredRgwInstances, c.store.Name)
}
}
@@ -235,39 +235,39 @@ func (c *clusterConfig) deleteLegacyDaemons() {
// Make a best effort to delete the rgw pods daemonsets
daemons, err := k8sutil.GetDaemonsets(c.context.Clientset, c.store.Namespace, c.storeLabelSelector())
if err != nil {
logger.Warningf("could not get deployments for object store %s (matching label selector '%s'): %+v", c.store.Name, c.storeLabelSelector(), err)
logger.Warningf("could not get deployments for object store %q (matching label selector %q). %v", c.store.Name, c.storeLabelSelector(), err)
}
daemonsetNum := len(daemons.Items)
if daemonsetNum > 0 {
for _, d := range daemons.Items {
// Delete any existing daemonset
if err := k8sutil.DeleteDaemonset(c.context.Clientset, c.store.Namespace, d.Name); err != nil {
logger.Errorf("error during deletion of daemonset %s resource: %+v", d.Name, err)
logger.Errorf("error during deletion of daemonset %q resource. %v", d.Name, err)
}
}
// Delete legacy rgw key
err = client.AuthDelete(c.context, c.store.Namespace, oldRgwKeyName)
if err != nil {
logger.Infof("failed to delete legacy rgw key %s. %+v", oldRgwKeyName, err)
logger.Infof("failed to delete legacy rgw key %q. %v", oldRgwKeyName, err)
}
}
// legacy deployment detection
logger.Debugf("looking for legacy deployment in object store %s", c.store.Name)
logger.Debugf("looking for legacy deployment in object store %q", c.store.Name)
deps, err := k8sutil.GetDeployments(c.context.Clientset, c.store.Namespace, c.storeLabelSelector())
if err != nil {
logger.Warning("could not get deployments for object store %s (matching label selector '%s'): %+v", c.store.Name, c.storeLabelSelector(), err)
logger.Warning("could not get deployments for object store %q (matching label selector %q). %v", c.store.Name, c.storeLabelSelector(), err)
}
for _, d := range deps.Items {
if d.Name == c.instanceName() {
logger.Infof("legacy deployment in object store %s found %s", c.store.Name, d.Name)
logger.Infof("legacy deployment in object store %q found %q", c.store.Name, d.Name)
if err := k8sutil.DeleteDeployment(c.context.Clientset, c.store.Namespace, d.Name); err != nil {
logger.Warning("error during deletion of deployment %s resource: %+v", d.Name, err)
logger.Warning("error during deletion of deployment %q resource. %v", d.Name, err)
}
// Delete legacy rgw key
err = client.AuthDelete(c.context, c.store.Namespace, oldRgwKeyName)
if err != nil {
logger.Infof("failed to delete legacy rgw key %s. %+v", oldRgwKeyName, err)
logger.Infof("failed to delete legacy rgw key %q. %v", oldRgwKeyName, err)
}
}
}
@@ -285,24 +285,24 @@ func (c *clusterConfig) deleteStore() error {
// Delete the rgw service
err := c.context.Clientset.CoreV1().Services(c.store.Namespace).Delete(c.instanceName(), options)
if err != nil && !kerrors.IsNotFound(err) {
logger.Warningf("failed to delete rgw service. %+v", err)
logger.Warningf("failed to delete rgw service. %v", err)
}
// Make a best effort to delete the rgw pods deployments
deps, err := k8sutil.GetDeployments(c.context.Clientset, c.store.Namespace, c.storeLabelSelector())
if err != nil {
logger.Warning("could not get deployments for object store %s (matching label selector '%s'): %+v", c.store.Namespace, c.storeLabelSelector(), err)
logger.Warning("could not get deployments for object store %s (matching label selector %q). %v", c.store.Namespace, c.storeLabelSelector(), err)
}
for _, d := range deps.Items {
if err := k8sutil.DeleteDeployment(c.context.Clientset, c.store.Namespace, d.Name); err != nil {
logger.Warning("error during deletion of deployment %s resource: %+v", d.Name, err)
logger.Warning("error during deletion of deployment %s resource. %v", d.Name, err)
}
}
// Delete the rgw config map keyrings
err = c.context.Clientset.CoreV1().Secrets(c.store.Namespace).Delete(c.instanceName(), options)
if err != nil && !kerrors.IsNotFound(err) {
logger.Warningf("failed to delete rgw secret. %+v", err)
logger.Warningf("failed to delete rgw secret. %v", err)
}
// Delete rgw CephX keys
+8 -8
View File
@@ -97,12 +97,12 @@ func (c *ObjectStoreUserController) onAdd(obj interface{}) {
user, err := getObjectStoreUserObject(obj)
if err != nil {
logger.Errorf("failed to get objectstoreuser object: %+v", err)
logger.Errorf("failed to get objectstoreuser object. %v", err)
return
}
if err = c.createUser(c.context, user); err != nil {
logger.Errorf("failed to create object store user %s. %+v", user.Name, err)
logger.Errorf("failed to create object store user %q. %v", user.Name, err)
}
}
@@ -122,12 +122,12 @@ func (c *ObjectStoreUserController) onDelete(obj interface{}) {
user, err := getObjectStoreUserObject(obj)
if err != nil {
logger.Errorf("failed to get objectstoreuser object: %+v", err)
logger.Errorf("failed to get objectstoreuser object. %v", err)
return
}
if err = deleteUser(c.context, user); err != nil {
logger.Errorf("failed to delete object store user %s. %+v", user.Name, err)
logger.Errorf("failed to delete object store user %q. %v", user.Name, err)
}
}
@@ -176,7 +176,7 @@ func (c *ObjectStoreUserController) createUser(context *clusterd.Context, u *cep
initialized, err := objectStoreInitialized(objContext)
if err != nil {
logger.Errorf("failed to detect if object store is initialized. %+v", err)
logger.Errorf("failed to detect if object store is initialized. %v", err)
} else if !initialized {
err := wait.Poll(time.Second*15, time.Minute*5, func() (ok bool, err error) {
initialized, err := objectStoreInitialized(objContext)
@@ -186,7 +186,7 @@ func (c *ObjectStoreUserController) createUser(context *clusterd.Context, u *cep
return initialized, nil
})
if err != nil {
logger.Errorf("err or timed out while waiting for objectstore %s to be ready. %+v", u.Spec.Store, err)
logger.Errorf("err or timed out while waiting for objectstore %q to be ready. %v", u.Spec.Store, err)
}
}
@@ -241,7 +241,7 @@ func objectStoreInitialized(context *object.Context) (bool, error) {
_, err := context.Context.RookClientset.CephV1().CephObjectStores(context.ClusterName).Get(context.Name, metav1.GetOptions{})
if err != nil {
if kerrors.IsNotFound(err) {
logger.Warningf("CephObjectStore %s could not be found. %+v", context.Name, err)
logger.Warningf("CephObjectStore %s could not be found. %v", context.Name, err)
return false, nil
}
return false, err
@@ -283,7 +283,7 @@ func deleteUser(context *clusterd.Context, u *cephv1.CephObjectStoreUser) error
err = context.Clientset.CoreV1().Secrets(u.Namespace).Delete(fmt.Sprintf("rook-ceph-object-user-%s-%s", u.Spec.Store, u.Name), &metav1.DeleteOptions{})
if err != nil {
logger.Warningf("failed to delete user %s secret. %+v", fmt.Sprintf("rook-ceph-object-user-%s-%s", u.Spec.Store, u.Name), err)
logger.Warningf("failed to delete user %s secret. %v", fmt.Sprintf("rook-ceph-object-user-%s-%s", u.Spec.Store, u.Name), err)
}
logger.Infof("user %s deleted successfully", u.Name)
+15 -15
View File
@@ -87,13 +87,13 @@ func (c *PoolController) onAdd(obj interface{}) {
pool, err := getPoolObject(obj)
if err != nil {
logger.Errorf("failed to get pool object: %+v", err)
logger.Errorf("failed to get pool object. %v", err)
return
}
err = createPool(c.context, pool)
if err != nil {
logger.Errorf("failed to create pool %s. %+v", pool.ObjectMeta.Name, err)
logger.Errorf("failed to create pool %q. %v", pool.ObjectMeta.Name, err)
}
}
@@ -105,32 +105,32 @@ func (c *PoolController) onUpdate(oldObj, newObj interface{}) {
oldPool, err := getPoolObject(oldObj)
if err != nil {
logger.Errorf("failed to get old pool object: %+v", err)
logger.Errorf("failed to get old pool object. %v", err)
return
}
pool, err := getPoolObject(newObj)
if err != nil {
logger.Errorf("failed to get new pool object: %+v", err)
logger.Errorf("failed to get new pool object. %v", err)
return
}
if oldPool.Name != pool.Name {
logger.Errorf("failed to update pool %s. name update not allowed", pool.Name)
logger.Errorf("failed to update pool %q. name update not allowed", pool.Name)
return
}
if pool.Spec.ErasureCoded.CodingChunks != 0 && pool.Spec.ErasureCoded.DataChunks != 0 {
logger.Errorf("failed to update pool %s. erasurecoded update not allowed", pool.Name)
logger.Errorf("failed to update pool %q. erasurecoded update not allowed", pool.Name)
return
}
if !poolChanged(oldPool.Spec, pool.Spec) {
logger.Debugf("pool %s not changed", pool.Name)
logger.Debugf("pool %q not changed", pool.Name)
return
}
// if the pool is modified, allow the pool to be created if it wasn't already
logger.Infof("updating pool %s", pool.Name)
logger.Infof("updating pool %q", pool.Name)
if err := createPool(c.context, pool); err != nil {
logger.Errorf("failed to create (modify) pool %s. %+v", pool.ObjectMeta.Name, err)
logger.Errorf("failed to create (modify) pool %q. %v", pool.ObjectMeta.Name, err)
}
}
@@ -155,11 +155,11 @@ func (c *PoolController) onDelete(obj interface{}) {
pool, err := getPoolObject(obj)
if err != nil {
logger.Errorf("failed to get pool object: %+v", err)
logger.Errorf("failed to get pool object. %v", err)
return
}
if err := deletePool(c.context, pool); err != nil {
logger.Errorf("failed to delete pool %s. %+v", pool.ObjectMeta.Name, err)
logger.Errorf("failed to delete pool %q. %v", pool.ObjectMeta.Name, err)
}
}
@@ -167,16 +167,16 @@ func (c *PoolController) onDelete(obj interface{}) {
func createPool(context *clusterd.Context, p *cephv1.CephBlockPool) error {
// validate the pool settings
if err := ValidatePool(context, p); err != nil {
return errors.Wrapf(err, "invalid pool %s arguments", p.Name)
return errors.Wrapf(err, "invalid pool %q arguments", p.Name)
}
// create the pool
logger.Infof("creating pool %s in namespace %s", p.Name, p.Namespace)
logger.Infof("creating pool %q in namespace %q", p.Name, p.Namespace)
if err := ceph.CreatePoolWithProfile(context, p.Namespace, *p.Spec.ToModel(p.Name), poolApplicationNameRBD); err != nil {
return errors.Wrapf(err, "failed to create pool %s", p.Name)
return errors.Wrapf(err, "failed to create pool %q", p.Name)
}
logger.Infof("created pool %s", p.Name)
logger.Infof("created pool %q", p.Name)
return nil
}
+4 -4
View File
@@ -265,7 +265,7 @@ func IsInferior(a, b CephVersion) bool {
// ValidateCephVersionsBetweenLocalAndExternalClusters makes sure an external cluster can be connected
// by checking the external ceph versions available and comparing it with the local image provided
func ValidateCephVersionsBetweenLocalAndExternalClusters(localVersion, externalVersion CephVersion) error {
logger.Debugf("local version is %s, external version is %s", localVersion.String(), externalVersion.String())
logger.Debugf("local version is %q, external version is %q", localVersion.String(), externalVersion.String())
// We only support Nautilus or newer
if !externalVersion.IsAtLeastNautilus() {
@@ -279,20 +279,20 @@ func ValidateCephVersionsBetweenLocalAndExternalClusters(localVersion, externalV
// Local version must never be higher than the external one
if IsSuperior(localVersion, externalVersion) {
return errors.Errorf("local cluster ceph version is higher %s than the external cluster %s, this must never happen", externalVersion.String(), localVersion.String())
return errors.Errorf("local cluster ceph version is higher %q than the external cluster %q, this must never happen", externalVersion.String(), localVersion.String())
}
// External cluster was updated to a minor version higher, consider updating too!
if localVersion.Major == externalVersion.Major {
if IsSuperior(externalVersion, localVersion) {
logger.Warningf("external cluster ceph version is a minor version higher %s than the local cluster %s, consider upgrading", externalVersion.String(), localVersion.String())
logger.Warningf("external cluster ceph version is a minor version higher %q than the local cluster %q, consider upgrading", externalVersion.String(), localVersion.String())
return nil
}
}
// The external cluster was upgraded, consider upgrading too!
if localVersion.Major < externalVersion.Major {
logger.Errorf("external cluster ceph version is a major version higher %s than the local cluster %s, consider upgrading", externalVersion.String(), localVersion.String())
logger.Errorf("external cluster ceph version is a major version higher %q than the local cluster %q, consider upgrading", externalVersion.String(), localVersion.String())
return nil
}