From 7eb7e8a192eb3613c5dab06bf4a85fa2c26c7875 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Wed, 26 Jun 2019 14:38:19 +0200 Subject: [PATCH] ceph: upgrade: always retrieve cluster version MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Do not consider clusterInfo Ceph version since it's the one picked by the spec image. The CephVersion provided here needs to be the current running version of Ceph for when it is later passed to OkToStop and checked with IsAtLeast. If you are upgrading from a pre 14.2.1 version to a version 14.2.1 or newer, IsAtLeast will incorrectly return true and then fail trying to run mon ok-to-stop on a mon pod running a pre 14.2.1 version. Signed-off-by: Sébastien Han --- pkg/daemon/ceph/client/upgrade.go | 58 +++++++++++++++++++++++++ pkg/daemon/ceph/client/upgrade_test.go | 22 ++++++++++ pkg/operator/ceph/cluster/mgr/mgr.go | 15 ++++++- pkg/operator/ceph/cluster/mon/mon.go | 15 ++++++- pkg/operator/ceph/cluster/osd/osd.go | 15 ++++++- pkg/operator/ceph/cluster/rbd/mirror.go | 18 +++++++- pkg/operator/ceph/file/mds/mds.go | 15 ++++++- pkg/operator/ceph/nfs/nfs.go | 1 + pkg/operator/ceph/object/rgw.go | 16 ++++++- 9 files changed, 168 insertions(+), 7 deletions(-) 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) } }