feedc0de/my-rook-config
1
0
Fork 0
forked from rook/rook
Code Pull Requests Activity

ceph: upgrade: always retrieve cluster version

Browse Source 
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 <seb@redhat.com>
This commit is contained in:
Sébastien Han
2019-07-17 00:39:18 +02:00
parent 7d86069f6e
commit 7eb7e8a192
9 changed files with 168 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/daemon/ceph/client/upgrade.go
+58
View File
@@ -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)
}
pkg/daemon/ceph/client/upgrade_test.go
+22
View File
@@ -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)
}
pkg/operator/ceph/cluster/mgr/mgr.go
+14 -1
View File
@@ -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)
}
}
pkg/operator/ceph/cluster/mon/mon.go
+14 -1
View File
@@ -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)
}
}
pkg/operator/ceph/cluster/osd/osd.go
+13 -2
View File
@@ -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)
}
}
pkg/operator/ceph/cluster/rbd/mirror.go
+17 -1
View File
@@ -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{})
pkg/operator/ceph/file/mds/mds.go
+14 -1
View File
@@ -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)
}
}
pkg/operator/ceph/nfs/nfs.go
+1
View File
@@ -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)
}
pkg/operator/ceph/object/rgw.go
+15 -1
View File
@@ -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)
}
}