diff --git a/pkg/daemon/ceph/client/upgrade.go b/pkg/daemon/ceph/client/upgrade.go index 6c6abb2d1..20017b7fd 100644 --- a/pkg/daemon/ceph/client/upgrade.go +++ b/pkg/daemon/ceph/client/upgrade.go @@ -351,3 +351,61 @@ func stringInSlice(a string, list []string) bool { } return false } + +// LeastUptodateDaemonVersion returns the ceph version of the least updated daemon type +// So if we invoke this method function with "mon", it will look for the least recent version +// Assume the following: +// +// "mon": { +// "ceph version 13.2.5 (cbff874f9007f1869bfd3821b7e33b2a6ffd4988) mimic (stable)": 1, +// "ceph version 14.2.0 (3a54b2b6d167d4a2a19e003a705696d4fe619afc) nautilus (stable)": 2 +// } +// +// In the case we will pick: "ceph version 13.2.5 (cbff874f9007f1869bfd3821b7e33b2a6ffd4988) mimic (stable)": 1, +// And eventually return 13.2.5 +func LeastUptodateDaemonVersion(context *clusterd.Context, clusterName, daemonType string) (cephver.CephVersion, error) { + var r map[string]int + var vv cephver.CephVersion + + // 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) + } else { + r, err = daemonMapEntry(versions, daemonType) + if err != nil { + return vv, fmt.Errorf("failed to find daemon map entry %+v", err) + } + for v := range r { + version, err := cephver.ExtractCephVersion(v) + if err != nil { + return vv, fmt.Errorf("failed to extract ceph version. %+v", err) + } + vv = *version + // break right after the first iteration + // the first one is always the least up-to-date + break + } + } + + return vv, nil +} + +func daemonMapEntry(versions *CephDaemonsVersions, daemonType string) (map[string]int, error) { + switch daemonType { + case "mon": + return versions.Mon, nil + case "mgr": + return versions.Mgr, nil + case "mds": + return versions.Mds, nil + case "osd": + return versions.Osd, nil + case "rgw": + return versions.Rgw, nil + case "mirror": + return versions.RbdMirror, nil + } + + return nil, fmt.Errorf("invalid daemonType %s", daemonType) +} diff --git a/pkg/daemon/ceph/client/upgrade_test.go b/pkg/daemon/ceph/client/upgrade_test.go index eb6e86bf6..cd0f8551e 100644 --- a/pkg/daemon/ceph/client/upgrade_test.go +++ b/pkg/daemon/ceph/client/upgrade_test.go @@ -17,6 +17,7 @@ limitations under the License. package client import ( + "encoding/json" "fmt" "testing" @@ -180,3 +181,24 @@ func TestFindFSName(t *testing.T) { fsName = findFSName("rook-ceph-mds-my-super-fs-a", "rook-ceph") assert.Equal(t, "my-super-fs", fsName) } + +func TestDaemonMapEntry(t *testing.T) { + dummyVersionsRaw := []byte(` + { + "mon": { + "ceph version 13.2.5 (cbff874f9007f1869bfd3821b7e33b2a6ffd4988) mimic (stable)": 1, + "ceph version 14.2.0 (3a54b2b6d167d4a2a19e003a705696d4fe619afc) nautilus (stable)": 2 + } + }`) + + var dummyVersions CephDaemonsVersions + err := json.Unmarshal([]byte(dummyVersionsRaw), &dummyVersions) + assert.Nil(t, err) + + m, err := daemonMapEntry(&dummyVersions, "mon") + assert.Nil(t, err) + assert.Equal(t, dummyVersions.Mon, m) + + m, err = daemonMapEntry(&dummyVersions, "dummy") + assert.NotNil(t, err) +} diff --git a/pkg/operator/ceph/cluster/mgr/mgr.go b/pkg/operator/ceph/cluster/mgr/mgr.go index 07b1f0cd9..9b86d98c8 100644 --- a/pkg/operator/ceph/cluster/mgr/mgr.go +++ b/pkg/operator/ceph/cluster/mgr/mgr.go @@ -32,6 +32,7 @@ import ( "github.com/rook/rook/pkg/operator/ceph/cluster/mon" "github.com/rook/rook/pkg/operator/ceph/config" opspec "github.com/rook/rook/pkg/operator/ceph/spec" + cephver "github.com/rook/rook/pkg/operator/ceph/version" "github.com/rook/rook/pkg/operator/k8sutil" v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" @@ -152,7 +153,19 @@ func (c *Cluster) Start() error { return fmt.Errorf("failed to create mgr deployment %s. %+v", resourceName, err) } logger.Infof("deployment for mgr %s already exists. updating if needed", resourceName) - if err := updateDeploymentAndWait(c.context, d, c.Namespace, c.clusterInfo.Name, c.clusterInfo.CephVersion); err != nil { + // Always invoke ceph version before an upgrade so we are sure to be up-to-date + daemon := "mgr" + 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.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion") + cephVersionToUse = c.clusterInfo.CephVersion + } else { + logger.Debugf("current cluster version for mgrs before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion + } + if err := updateDeploymentAndWait(c.context, d, c.Namespace, c.clusterInfo.Name, cephVersionToUse); err != nil { return fmt.Errorf("failed to update mgr deployment %s. %+v", resourceName, err) } } diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index 267f37f0e..9f5649b21 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/pkg/operator/ceph/cluster/mon/mon.go @@ -627,7 +627,20 @@ func (c *Cluster) startMon(m *monConfig, hostname string) error { return fmt.Errorf("failed to create mon deployment %s. %+v", m.ResourceName, err) } logger.Infof("deployment for mon %s already exists. updating if needed", m.ResourceName) - if err := updateDeploymentAndWait(c.context, d, c.Namespace, c.clusterInfo.Name, c.clusterInfo.CephVersion); err != nil { + + // Always invoke ceph version before an upgrade so we are sure to be up-to-date + daemon := "mon" + 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.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion") + cephVersionToUse = c.clusterInfo.CephVersion + } else { + logger.Debugf("current cluster version for monitors before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion + } + if err := updateDeploymentAndWait(c.context, d, c.Namespace, c.clusterInfo.Name, cephVersionToUse); err != nil { return fmt.Errorf("failed to update mon deployment %s. %+v", m.ResourceName, err) } } diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go index 441946993..700f5f229 100644 --- a/pkg/operator/ceph/cluster/osd/osd.go +++ b/pkg/operator/ceph/cluster/osd/osd.go @@ -322,11 +322,22 @@ func (c *Cluster) startOSDDaemonsOnNode(nodeName string, config *provisionConfig continue } logger.Infof("deployment for osd %d already exists. updating if needed", osd.ID) - if err = updateDeploymentAndWait(c.context, dp, c.Namespace, c.clusterInfo.Name, c.clusterInfo.CephVersion); err != nil { + // Always invoke ceph version before an upgrade so we are sure to be up-to-date + daemon := "osd" + 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.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion") + cephVersionToUse = c.clusterInfo.CephVersion + } else { + logger.Debugf("current cluster version for osds before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion + } + if err = updateDeploymentAndWait(c.context, dp, c.Namespace, c.clusterInfo.Name, cephVersionToUse); err != nil { config.addError(fmt.Sprintf("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) } } diff --git a/pkg/operator/ceph/cluster/rbd/mirror.go b/pkg/operator/ceph/cluster/rbd/mirror.go index b5884f401..22e6c0b39 100644 --- a/pkg/operator/ceph/cluster/rbd/mirror.go +++ b/pkg/operator/ceph/cluster/rbd/mirror.go @@ -24,10 +24,12 @@ import ( cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2" "github.com/rook/rook/pkg/clusterd" + "github.com/rook/rook/pkg/daemon/ceph/client" cephconfig "github.com/rook/rook/pkg/daemon/ceph/config" "github.com/rook/rook/pkg/operator/ceph/cluster/mon" "github.com/rook/rook/pkg/operator/ceph/config" opspec "github.com/rook/rook/pkg/operator/ceph/spec" + cephver "github.com/rook/rook/pkg/operator/ceph/version" "github.com/rook/rook/pkg/operator/k8sutil" v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/api/errors" @@ -120,7 +122,21 @@ func (m *Mirroring) Start() error { return fmt.Errorf("failed to create %s deployment. %+v", resourceName, err) } logger.Infof("deployment for rbd-mirror %s already exists. updating if needed", resourceName) - if err := updateDeploymentAndWait(m.context, d, m.Namespace, m.ClusterInfo.Name, m.ClusterInfo.CephVersion); err != nil { + + // Always invoke ceph version before an upgrade so we are sure to be up-to-date + daemon := "mirror" + var cephVersionToUse cephver.CephVersion + 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.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with m.ClusterInfo.CephVersion") + cephVersionToUse = m.ClusterInfo.CephVersion + + } else { + logger.Debugf("current cluster version for rbd mirrors before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion + } + if err := updateDeploymentAndWait(m.context, d, m.Namespace, m.ClusterInfo.Name, cephVersionToUse); 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) err = m.context.Clientset.AppsV1().Deployments(m.Namespace).Delete(d.Name, &metav1.DeleteOptions{}) diff --git a/pkg/operator/ceph/file/mds/mds.go b/pkg/operator/ceph/file/mds/mds.go index ba6cc2ea6..60c86c008 100644 --- a/pkg/operator/ceph/file/mds/mds.go +++ b/pkg/operator/ceph/file/mds/mds.go @@ -156,7 +156,20 @@ func (c *Cluster) Start() error { // keyring must be generated before update-and-wait since no keyring will prevent the // deployment from reaching ready state if createErr != nil && errors.IsAlreadyExists(createErr) { - if err = UpdateDeploymentAndWait(c.context, d, c.fs.Namespace, c.clusterInfo.Name, c.clusterInfo.CephVersion); err != nil { + // Always invoke ceph version before an upgrade so we are sure to be up-to-date + daemon := "mds" + 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.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion") + cephVersionToUse = c.clusterInfo.CephVersion + + } else { + logger.Debugf("current cluster version for mdss before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion + } + if err = UpdateDeploymentAndWait(c.context, d, c.fs.Namespace, c.clusterInfo.Name, cephVersionToUse); err != nil { return fmt.Errorf("failed to update mds deployment %s. %+v", d.Name, err) } } diff --git a/pkg/operator/ceph/nfs/nfs.go b/pkg/operator/ceph/nfs/nfs.go index a84e66ae4..e7016e1bb 100644 --- a/pkg/operator/ceph/nfs/nfs.go +++ b/pkg/operator/ceph/nfs/nfs.go @@ -68,6 +68,7 @@ func (c *CephNFSController) upCephNFS(n cephv1.CephNFS, oldActive int) error { return fmt.Errorf("failed to create ganesha deployment. %+v", err) } logger.Infof("ganesha deployment %s already exists. updating if needed", deployment.Name) + // We don't invoke ceph versions here since nfs do not show up in the service map (yet?) if err := updateDeploymentAndWait(c.context, deployment, n.Namespace, c.clusterInfo.Name, c.clusterInfo.CephVersion); err != nil { return fmt.Errorf("failed to update ganesha deployment %s. %+v", deployment.Name, err) } diff --git a/pkg/operator/ceph/object/rgw.go b/pkg/operator/ceph/object/rgw.go index 987176bcb..eb06c5af2 100644 --- a/pkg/operator/ceph/object/rgw.go +++ b/pkg/operator/ceph/object/rgw.go @@ -28,6 +28,7 @@ import ( "github.com/rook/rook/pkg/operator/ceph/cluster/mon" "github.com/rook/rook/pkg/operator/ceph/config" "github.com/rook/rook/pkg/operator/ceph/pool" + cephver "github.com/rook/rook/pkg/operator/ceph/version" "github.com/rook/rook/pkg/operator/k8sutil" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -144,7 +145,20 @@ func (c *clusterConfig) startRGWPods() error { // Generate the mime.types file after the rep. controller as well for the same reason as keyring if createErr != nil && errors.IsAlreadyExists(createErr) { - if err := updateDeploymentAndWait(c.context, deployment, c.store.Namespace, c.clusterInfo.Name, c.clusterInfo.CephVersion); err != nil { + // Always invoke ceph version before an upgrade so we are sure to be up-to-date + daemon := "rgw" + 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.Debug("could not detect ceph version during update, this is likely an initial bootstrap, proceeding with c.clusterInfo.CephVersion") + cephVersionToUse = c.clusterInfo.CephVersion + + } else { + logger.Debugf("current cluster version for rgws before upgrading is: %+v", currentCephVersion) + cephVersionToUse = currentCephVersion + } + if err := updateDeploymentAndWait(c.context, deployment, c.store.Namespace, c.clusterInfo.Name, cephVersionToUse); err != nil { return fmt.Errorf("failed to update object store %s deployment %s. %+v", c.store.Name, deployment.Name, err) } }