Merge pull request #3495 from SUSE/legacy-osd-integration-test

ceph: in upgrade test, ensure legacy osds run

tests/framework/installer/ceph_installer.go

@@ -29,11 +29,13 @@ import (
 
 	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/tests/framework/utils"
 	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 )
 
@@ -176,14 +178,7 @@ func (h *CephInstaller) CreateK8sRookToolbox(namespace string) (err error) {
 	return nil
 }
 
-func (h *CephInstaller) CreateK8sRookCluster(namespace, systemNamespace string, storeType string) (err error) {
-	return h.CreateK8sRookClusterWithHostPathAndDevices(namespace, systemNamespace, storeType, false,
-		cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, true, /* startWithAllNodes */
-		1, /* rbd workers */
-		NautilusVersion)
-}
-
-// CreateK8sRookCluster creates rook cluster via kubectl
+// CreateK8sRookClusterWithHostPathAndDevices creates rook cluster via kubectl
 func (h *CephInstaller) CreateK8sRookClusterWithHostPathAndDevices(namespace, systemNamespace, storeType string,
 	useAllDevices bool, mon cephv1.MonSpec, startWithAllNodes bool, rbdMirrorWorkers int, cephVersion cephv1.CephVersionSpec) error {
 
@@ -210,6 +205,10 @@ func (h *CephInstaller) CreateK8sRookClusterWithHostPathAndDevices(namespace, sy
 		}
 	}
 
+	if err := h.WipeClusterDisks(namespace); err != nil {
+		return fmt.Errorf("failed to wipe cluster disks. %+v", err)
+	}
+
 	logger.Infof("Starting Rook Cluster with yaml")
 	settings := &ClusterSettings{namespace, storeType, dataDirHostPath, useAllDevices, mon.Count, rbdMirrorWorkers, cephVersion}
 	rookCluster := h.Manifests.GetRookCluster(settings)
@@ -258,6 +257,7 @@ func (h *CephInstaller) initTestDir(namespace string) (string, error) {
 	return testDir, nil
 }
 
+// GetNodeHostnames returns the list of nodes in the k8s cluster
 func (h *CephInstaller) GetNodeHostnames() ([]string, error) {
 	nodes, err := h.k8shelper.Clientset.CoreV1().Nodes().List(metav1.ListOptions{})
 	if err != nil {
@@ -276,7 +276,7 @@ func (h *CephInstaller) InstallRookOnK8sWithHostPathAndDevices(namespace, storeT
 	useDevices bool, mon cephv1.MonSpec, startWithAllNodes bool, rbdMirrorWorkers int) (bool, error) {
 
 	var err error
-	// flag used for local debuggin purpose, when rook is pre-installed
+	// flag used for local debugging purpose, when rook is pre-installed
 	if Env.SkipInstallRook {
 		return true, nil
 	}
@@ -314,15 +314,6 @@ func (h *CephInstaller) InstallRookOnK8sWithHostPathAndDevices(namespace, storeT
 		return false, err
 	}
 
-	if forceUseDevices {
-		logger.Infof("Forcing the use of devices")
-		useDevices = true
-	} else if useDevices {
-		// This check only looks at the local machine for devices. If you want to force using devices,
-		// set the forceUseDevices flag
-		useDevices = IsAdditionalDeviceAvailableOnCluster()
-	}
-
 	// Create rook cluster
 	err = h.CreateK8sRookClusterWithHostPathAndDevices(namespace, onamespace, storeType,
 		useDevices, cephv1.MonSpec{Count: mon.Count, AllowMultiplePerNode: mon.AllowMultiplePerNode}, startWithAllNodes,
@@ -351,12 +342,12 @@ func (h *CephInstaller) InstallRookOnK8sWithHostPathAndDevices(namespace, storeT
 	return true, nil
 }
 
-// UninstallRookFromK8s uninstalls rook from k8s
+// UninstallRook uninstalls rook from k8s
 func (h *CephInstaller) UninstallRook(namespace string, gatherLogs bool) {
 	h.UninstallRookFromMultipleNS(gatherLogs, SystemNamespace(namespace), namespace)
 }
 
-// UninstallRookFromK8s uninstalls rook from multiple namespaces in k8s
+// UninstallRookFromMultipleNS uninstalls rook from multiple namespaces in k8s
 func (h *CephInstaller) UninstallRookFromMultipleNS(gatherLogs bool, systemNamespace string, namespaces ...string) {
 	// flag used for local debugging purpose, when rook is pre-installed
 	if Env.SkipInstallRook {
@@ -602,3 +593,137 @@ func NewCephInstaller(t func() *testing.T, clientset *kubernetes.Clientset, useH
 	flag.Parse()
 	return h
 }
+
+// WipeClusterDisks runs a disk wipe job on all nodes in the k8s cluster.
+func (h *CephInstaller) WipeClusterDisks(namespace string) error {
+	wipeJobName := func(node string) string {
+		return k8sutil.TruncateNodeName("rook-ceph-disk-wipe-%s", node)
+	}
+
+	// Wipe clean disks on all nodes
+	nodes, err := h.GetNodeHostnames()
+	if err != nil {
+		return fmt.Errorf("failed to get node hostnames. %+v", err)
+	}
+	for _, node := range nodes {
+		job := h.GetDiskWipeJob(node, wipeJobName(node), namespace)
+		_, err := h.k8shelper.KubectlWithStdin(job, createFromStdinArgs...)
+		if err != nil {
+			return fmt.Errorf("failed to create disk wipe job for host %s. %+v", node, err)
+		}
+	}
+
+	allJobsAreComplete := func() (done bool, err error) {
+		for _, node := range nodes {
+			j, err := h.k8shelper.Clientset.BatchV1().Jobs(namespace).Get(wipeJobName(node), metav1.GetOptions{})
+			if err != nil {
+				return false, nil
+			}
+			if j.Status.Failed > 0 {
+				return false, fmt.Errorf("job %s failed", wipeJobName(node))
+			}
+			if j.Status.Succeeded == 0 {
+				return false, nil
+			}
+		}
+		return true, nil
+	}
+
+	if err = wait.Poll(5*time.Second, 90*time.Second, allJobsAreComplete); err != nil {
+		return fmt.Errorf("failed to wait for wipe jobs to complete. %+v", err)
+	}
+
+	for _, node := range nodes {
+		// if delete fails, don't worry about the error; delete only on best-effort basis
+		h.k8shelper.Clientset.BatchV1().Jobs(namespace).Delete(wipeJobName(node), &metav1.DeleteOptions{})
+	}
+
+	return nil
+}
+
+// GetDiskWipeJob returns a YAML manifest string for a job which will wipe clean the extra
+// (non-boot) disks on a node, allowing Ceph to use the disk(s) during its testing.
+func (h *CephInstaller) GetDiskWipeJob(nodeName, jobName, namespace string) string {
+	// put the wipe job in the cluster namespace so that logs get picked up in failure conditions
+	return `apiVersion: batch/v1
+kind: Job
+metadata:
+  name: ` + jobName + `
+  namespace: ` + namespace + `
+spec:
+    template:
+      spec:
+          restartPolicy: OnFailure
+          containers:
+              - name: disk-wipe
+                image: ` + h.CephVersion.Image + `
+                securityContext:
+                    privileged: true
+                volumeMounts:
+                    - name: dev
+                      mountPath: /dev
+                    - name: run-udev
+                      mountPath: /run/udev
+                command:
+                    - "sh"
+                    - "-c"
+                    - |
+                      set -xEeuo pipefail
+                      # Wipe LVM data from disks
+                      #
+                      udevadm trigger || true
+                      vgimport -a || true
+                      for vg in $(vgs --noheadings --readonly --separator=' ' -o vg_name); do
+                        lvremove --yes --force "$vg"
+                        vgremove --yes --force "$vg"
+                      done
+                      for pv in $(pvs --noheadings --readonly --separator=' ' -o pv_name); do
+                        pvremove --yes --force "$pv"
+                      done
+                      #
+                      # we CANNOT wipe the boot disk if it exists
+                      fdisk -l # show the output here for helping debug log output
+                      lsblk
+                      parted --script --list
+                      df /
+                      boot_disk=""
+                      all_disks="$(lsblk --paths --nodeps --output=NAME --noheadings)"
+                      for disk in ${all_disks}; do
+                        # parted returns an error if the disk has an unknown label, which we don't
+                        # care about. ceph containers have a very old version of lsblk which makes
+                        # it difficult to ascertain the boot disk programmatically, so part is used
+                        if (parted --script ${disk} print || true) | grep boot; then
+                            boot_disk="${disk}"
+                            break
+                        fi
+                      done
+                      # in cloud environments, the disk could possibly be xvd[a-z]
+                      rook_disks="$(find /dev -regex '/dev/x?[vs]d[a-z]+$' -and -not -wholename "${boot_disk}")"
+                      #
+                      # zap the disks to a fresh, usable state after LVM info is deleted
+                      # (zap-all is important, b/c MBR has to be clean)
+                      for disk in ${rook_disks}; do
+                        wipefs --all "${disk}"
+                        # lvm metadata can be a lot of sectors
+                        dd if=/dev/zero of="${disk}" bs=512 count=2500
+                        sgdisk --zap-all "${disk}"
+                      done
+                      #
+                      # some devices might still be mapped that lock the disks
+                      # this can fail with long-gone vestigial devices, so just assume this is successful
+                      ls /dev/mapper/ceph-* | xargs -I% -- dmsetup remove --force % || true
+                      rm -rf /dev/mapper/ceph-*  # clutter
+                      #
+                      # ceph-volume setup also leaves ceph-UUID directories in /dev (just clutter)
+                      rm -rf /dev/ceph-*
+          nodeSelector:
+            kubernetes.io/hostname: ` + nodeName + `
+          volumes:
+              - name: dev
+                hostPath:
+                  path: /dev
+              - name: run-udev
+                hostPath:
+                  path: /run/udev
+`
+}

tests/framework/installer/ceph_manifests.go

@@ -1670,6 +1670,11 @@ subjects:
 
 // GetRookCluster returns rook-cluster manifest
 func (m *CephManifestsMaster) GetRookCluster(settings *ClusterSettings) string {
+	store := "# storeType not specified; Rook will use default store types"
+	if settings.StoreType != "" {
+		store = `storeType: "` + settings.StoreType + `"`
+	}
+
 	return `apiVersion: ceph.rook.io/v1
 kind: CephCluster
 metadata:
@@ -1697,7 +1702,7 @@ spec:
     - path: ` + settings.DataDirHostPath + /* simulate legacy fallback osd behavior so existing tests still work */ `
     deviceFilter:
     config:
-      storeType: "` + settings.StoreType + `"
+      ` + store + `
       databaseSizeMB: "1024"
       journalSizeMB: "1024"
   mgr:

tests/framework/installer/ceph_manifests_v1.0.go

@@ -1168,6 +1168,11 @@ subjects:
 
 // GetRookCluster returns rook-cluster manifest
 func (m *CephManifestsV1_0) GetRookCluster(settings *ClusterSettings) string {
+	store := "# storeType not specified; Rook will use default store types"
+	if settings.StoreType != "" {
+		store = `storeType: "` + settings.StoreType + `"`
+	}
+
 	return `apiVersion: ceph.rook.io/v1
 kind: CephCluster
 metadata:
@@ -1196,7 +1201,7 @@ spec:
     deviceFilter:
     location:
     config:
-      storeType: "` + settings.StoreType + `"
+      ` + store + `
       databaseSizeMB: "1024"
       journalSizeMB: "1024"`
 }

tests/framework/installer/installer.go

@@ -30,8 +30,8 @@ import (
 const (
 	// Version tag for the latest manifests
 	VersionMaster = "master"
-	// Version tag for Rook v0.9
-	Version1_0 = "v1.0.1"
+	// Version tag for Rook v1.0
+	Version1_0 = "v1.0.6"
 	// test suite names
 	CassandraTestSuite   = "cassandra"
 	CephTestSuite        = "ceph"
@@ -44,7 +44,6 @@ const (
 var (
 	// ** Variables that might need to be changed depending on the dev environment. The init function below will modify some of them automatically. **
 	baseTestDir       string
-	forceUseDevices   = false
 	createBaseTestDir = true
 	// ** end of Variables to modify
 	logger              = capnslog.NewPackageLogger("github.com/rook/rook", "installer")

tests/framework/utils/k8s_helper.go

@@ -33,6 +33,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/util/exec"
 	"github.com/stretchr/testify/require"
+	apps "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -54,7 +55,7 @@ type K8sHelper struct {
 
 const (
 	// RetryLoop params for tests.
-	RetryLoop = 40
+	RetryLoop = 80
 	// RetryInterval param for test - wait time while in RetryLoop
 	RetryInterval = 5
 	// TestMountPath is the path inside a test pod where storage is mounted
@@ -1619,3 +1620,45 @@ func (k8sh *K8sHelper) WaitForDeploymentCount(label, namespace string, count int
 	}
 	return fmt.Errorf("giving up waiting for %d deployments with label %s in namespace %s", count, label, namespace)
 }
+
+// WaitForLabeledDeploymentsToBeReady waits for all deployments matching the given label selector to
+// be fully ready with a default timeout.
+func (k8sh *K8sHelper) WaitForLabeledDeploymentsToBeReady(label, namespace string) error {
+	return k8sh.WaitForLabeledDeploymentsToBeReadyWithRetries(label, namespace, RetryLoop)
+}
+
+// WaitForLabeledDeploymentsToBeReadyWithRetries waits for all deployments matching the given label
+// selector to be fully ready. Retry the number of times given.
+func (k8sh *K8sHelper) WaitForLabeledDeploymentsToBeReadyWithRetries(label, namespace string, retries int) error {
+	listOpts := metav1.ListOptions{LabelSelector: label}
+	var lastDep apps.Deployment
+	for i := 0; i < retries; i++ {
+		deps, err := k8sh.Clientset.AppsV1().Deployments(namespace).List(listOpts)
+		ready := 0
+		if err == nil && len(deps.Items) > 0 {
+			for _, dep := range deps.Items {
+				if dep.Status.Replicas == dep.Status.ReadyReplicas {
+					ready++
+				} else {
+					lastDep = dep // make it the last non-ready dep
+				}
+				if ready == len(deps.Items) {
+					logger.Infof("all %d deployments with label %s are running", len(deps.Items), label)
+					return nil
+				}
+			}
+		}
+		logger.Infof("waiting for deployment(s) with label %s in namespace %s to be running. ready=%d/%d, err=%+v",
+			label, namespace, ready, len(deps.Items), err)
+		time.Sleep(RetryInterval * time.Second)
+	}
+	if len(lastDep.Name) == 0 {
+		logger.Infof("no deployment was found with label %s", label)
+	} else {
+		r, err := k8sh.Kubectl("-n", namespace, "get", "-o", "yaml", "deployments", "--selector", label)
+		if err != nil {
+			logger.Infof("deployments with label %s:\n%s", label, r)
+		}
+	}
+	return fmt.Errorf("giving up waiting for deployment(s) with label %s in namespace %s to be ready", label, namespace)
+}

tests/integration/ceph_base_file_test.go

@@ -18,7 +18,10 @@ package integration
 
 import (
 	"fmt"
+	"strings"
+	"time"
 
+	cephclient "github.com/rook/rook/pkg/daemon/ceph/client"
 	"github.com/rook/rook/tests/framework/clients"
 	"github.com/rook/rook/tests/framework/installer"
 	"github.com/rook/rook/tests/framework/utils"
@@ -163,6 +166,7 @@ func getFilesystemTestPod(podName, namespace, filesystemName, driverName string,
         mountUser: ` + fileMountUser + `
         mountSecret: ` + fileMountSecret
 	}
+	// Bash's sleep signal handling: http://mywiki.wooledge.org/SignalTrap#When_is_the_signal_handled.3F
 	return `apiVersion: v1
 kind: Pod
 metadata:
@@ -171,11 +175,19 @@ metadata:
 spec:
   containers:
   - name: ` + podName + `
-    image: busybox
+    image: krallin/ubuntu-tini
     command:
-        - sh
+        - "/usr/local/bin/tini"
+        - "--"
+        - "bash"
         - "-c"
-        - "sleep 300"
+        - |
+          pid=
+          trap 'exit 0' SIGTERM
+          trap '[[ $pid ]] && kill "$pid"' EXIT
+          sleep 1800 & pid=$!
+          wait $pid
+          pid=
     imagePullPolicy: IfNotPresent
     env:
     volumeMounts:
@@ -188,9 +200,8 @@ spec:
       fsType: ceph
       options:
         fsName: ` + filesystemName + `
-        clusterNamespace: ` + namespace +
-		mountUserInsert + `
-  restartPolicy: Never
+        clusterNamespace: ` + namespace + mountUserInsert + `
+  restartPolicy: Always
 `
 }
 
@@ -264,3 +275,24 @@ subjects:
   namespace: ` + namespace + `
 `
 }
+
+func waitForFilesystemActive(k8sh *utils.K8sHelper, namespace, filesystemName string) error {
+	cmd := cephclient.NewCephCommand(k8sh.MakeContext(), namespace, []string{"fs", "status", filesystemName})
+	var stat []byte
+	var err error
+	logger.Infof("waiting for filesystem %q to be active", filesystemName)
+	for i := 0; i < utils.RetryLoop; i++ {
+		stat, err := cmd.Run()
+		if err != nil {
+			logger.Warningf("failed to get filesystem %q status. %+v", filesystemName, err)
+		}
+		// as long as at least one mds is active, it's okay
+		if strings.Contains(string(stat), "active") {
+			logger.Infof("done waiting for filesystem %q to be active", filesystemName)
+			return nil
+		}
+		logger.Infof("waiting for filesystem %q to be active", filesystemName)
+		time.Sleep(utils.RetryInterval * time.Second)
+	}
+	return fmt.Errorf("gave up waiting to get filesystem %q status [err: %+v] Status returned:\n%s", filesystemName, err, string(stat))
+}

tests/integration/ceph_helm_test.go

@@ -69,7 +69,7 @@ func (hs *HelmSuite) SetupSuite() {
 	hs.namespace = "helm-ns"
 	mons := 1
 	rbdMirrorWorkers := 1
-	hs.op, hs.kh = StartTestCluster(hs.T, helmMinimalTestVersion, hs.namespace, "bluestore", true, false, mons, rbdMirrorWorkers, installer.VersionMaster, installer.NautilusVersion)
+	hs.op, hs.kh = StartTestCluster(hs.T, helmMinimalTestVersion, hs.namespace, "bluestore", true, true, mons, rbdMirrorWorkers, installer.VersionMaster, installer.NautilusVersion)
 	hs.helper = clients.CreateTestClient(hs.kh, hs.op.installer.Manifests)
 }
 

tests/integration/ceph_multi_cluster_test.go

@@ -21,6 +21,7 @@ import (
 	"testing"
 	"time"
 
+	cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/tests/framework/clients"
 	"github.com/rook/rook/tests/framework/installer"
 	"github.com/rook/rook/tests/framework/utils"
@@ -191,7 +192,12 @@ func (o MCTestOperations) Teardown() {
 
 func (o MCTestOperations) startCluster(namespace, store string) error {
 	logger.Infof("starting cluster %s", namespace)
-	if err := o.installer.CreateK8sRookCluster(namespace, o.systemNamespace, store); err != nil {
+	useDevices := false
+	// do not use disks for this cluster, otherwise the 2 test clusters will each try to use the
+	// same disks.
+	err := o.installer.CreateK8sRookClusterWithHostPathAndDevices(namespace, o.systemNamespace, store,
+		useDevices, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, true, 1, installer.NautilusVersion)
+	if err != nil {
 		o.T().Fail()
 		o.installer.GatherAllRookLogs(o.T().Name(), namespace, o.systemNamespace)
 		return fmt.Errorf("failed to create cluster %s. %+v", namespace, err)

tests/integration/ceph_upgrade_test.go

@@ -18,9 +18,11 @@ package integration
 
 import (
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
+	rookcephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	"github.com/rook/rook/tests/framework/clients"
 	"github.com/rook/rook/tests/framework/installer"
@@ -64,7 +66,22 @@ func (s *UpgradeSuite) SetupSuite() {
 	useDevices := true
 	mons := 3
 	rbdMirrorWorkers := 0
-	s.op, s.k8sh = StartTestCluster(s.T, upgradeMinimalTestVersion, s.namespace, "bluestore", false, useDevices, mons, rbdMirrorWorkers, installer.Version1_0, installer.MimicVersion)
+	cephVersion := rookcephv1.CephVersionSpec{
+		// Ceph v13.2.3 got ceph-volume features needed for provisioning
+		// test that when Rook upgrades, it can still run non-ceph-volume osds
+		Image: "ceph/ceph:v13.2.0-20190410",
+	}
+	s.op, s.k8sh = StartTestCluster(s.T,
+		upgradeMinimalTestVersion,
+		s.namespace,
+		"",
+		false,
+		useDevices,
+		mons,
+		rbdMirrorWorkers,
+		installer.Version1_0,
+		cephVersion,
+	)
 	s.helper = clients.CreateTestClient(s.k8sh, s.op.installer.Manifests)
 }
 
@@ -98,18 +115,30 @@ func (s *UpgradeSuite) TestUpgradeToMaster() {
 	objectStoreName := "upgraded-object"
 	runObjectE2ETestLite(s.helper, s.k8sh, s.Suite, s.namespace, objectStoreName, 1)
 
-	// verify that we're actually running 0.8 before the upgrade
+	// verify that we're actually running the right pre-upgrade image
 	operatorContainer := "rook-ceph-operator"
 	version, err := k8sutil.GetDeploymentImage(s.k8sh.Clientset, systemNamespace, operatorContainer, operatorContainer)
-	assert.Nil(s.T(), err)
+	assert.NoError(s.T(), err)
 	assert.Equal(s.T(), "rook/ceph:"+installer.Version1_0, version)
 
 	message := "my simple message"
 	preFilename := "pre-upgrade-file"
-	assert.Nil(s.T(), s.k8sh.WriteToPod("", podName, preFilename, message))
-	assert.Nil(s.T(), s.k8sh.ReadFromPod("", podName, preFilename, message))
-	assert.Nil(s.T(), s.k8sh.WriteToPod(s.namespace, filePodName, preFilename, message))
-	assert.Nil(s.T(), s.k8sh.ReadFromPod(s.namespace, filePodName, preFilename, message))
+	assert.NoError(s.T(), s.k8sh.WriteToPod("", podName, preFilename, message))
+	assert.NoError(s.T(), s.k8sh.ReadFromPod("", podName, preFilename, message))
+	assert.NoError(s.T(), s.k8sh.WriteToPod(s.namespace, filePodName, preFilename, message))
+	assert.NoError(s.T(), s.k8sh.ReadFromPod(s.namespace, filePodName, preFilename, message))
+
+	// Gather logs before Ceph upgrade to help with debugging
+	if installer.Env.Logs == "all" {
+		s.k8sh.PrintPodDescribe(s.namespace)
+	}
+	n := strings.Replace(s.T().Name(), "/", "_", -1) + "_before_ceph_upgrade"
+	s.op.installer.GatherAllRookLogs(n, systemNamespace, s.namespace)
+
+	// Get some info about the currently deployed mons to determine later if they are all updated
+	monDepList, err := k8sutil.GetDeployments(s.k8sh.Clientset, s.namespace, "app=rook-ceph-mon")
+	require.NoError(s.T(), err)
+	numMons := len(monDepList.Items)
 
 	// Get some info about the currently deployed OSDs to determine later if they are all updated
 	osdDepList, err := k8sutil.GetDeployments(s.k8sh.Clientset, s.namespace, "app=rook-ceph-osd")
@@ -119,46 +148,99 @@ func (s *UpgradeSuite) TestUpgradeToMaster() {
 	require.True(s.T(), numOSDs > 0)
 	d := osdDeps[0]
 	oldRookVersion := d.Labels["rook-version"] // upgraded OSDs should not have this version label
+	oldCephVersion := d.Labels["ceph-version"] // upgraded OSDs should not have this version label
 
 	// Update to the next version of cluster roles before the operator is restarted
 	err = s.updateClusterRoles()
-	require.Nil(s.T(), err)
+	require.NoError(s.T(), err)
 
-	// Upgrade to master
-	require.Nil(s.T(), s.k8sh.SetDeploymentVersion(systemNamespace, operatorContainer, operatorContainer, installer.VersionMaster))
+	// Upgrade Ceph version
+	s.k8sh.Kubectl("-n", s.namespace, "patch", "CephCluster", s.namespace, "--type=merge",
+		"-p", fmt.Sprintf(`{"spec": {"cephVersion": {"image": "%s"}}}`, installer.NautilusVersion.Image))
+
+	// we need to make sure Ceph is fully updated (including RGWs and MDSes) before proceeding to
+	// upgrade rook; we do not support upgrading Ceph simultaneously with Rook upgrade
+	monsNotOldVersion := fmt.Sprintf("app=rook-ceph-mon,ceph-version!=%s", oldCephVersion)
+	err = s.k8sh.WaitForDeploymentCount(monsNotOldVersion, s.namespace, numMons)
+	require.NoError(s.T(), err)
+	err = s.k8sh.WaitForLabeledDeploymentsToBeReady(monsNotOldVersion, s.namespace)
+	require.NoError(s.T(), err)
+
+	osdsNotOldVersion := fmt.Sprintf("app=rook-ceph-osd,ceph-version!=%s", oldCephVersion)
+	err = s.k8sh.WaitForDeploymentCount(osdsNotOldVersion, s.namespace, numOSDs)
+	require.NoError(s.T(), err)
+	err = s.k8sh.WaitForLabeledDeploymentsToBeReady(osdsNotOldVersion, s.namespace)
+	require.NoError(s.T(), err)
+
+	mdsesNotOldVersion := fmt.Sprintf("app=rook-ceph-mds,ceph-version!=%s", oldCephVersion)
+	err = s.k8sh.WaitForDeploymentCount(mdsesNotOldVersion, s.namespace, 4 /* always expect 4 mdses */)
+	require.NoError(s.T(), err)
+	err = s.k8sh.WaitForLabeledDeploymentsToBeReady(mdsesNotOldVersion, s.namespace)
+	require.NoError(s.T(), err)
+
+	rgwsNotOldVersion := fmt.Sprintf("app=rook-ceph-rgw,ceph-version!=%s", oldCephVersion)
+	err = s.k8sh.WaitForDeploymentCount(rgwsNotOldVersion, s.namespace, 1 /* always expect 1 rgw */)
+	require.NoError(s.T(), err)
+	err = s.k8sh.WaitForLabeledDeploymentsToBeReady(rgwsNotOldVersion, s.namespace)
+	require.NoError(s.T(), err)
+
+	// Gather logs after Ceph upgrade to help with debugging
+	if installer.Env.Logs == "all" {
+		s.k8sh.PrintPodDescribe(s.namespace)
+	}
+	n = strings.Replace(s.T().Name(), "/", "_", -1) + "_after_ceph_upgrade"
+	s.op.installer.GatherAllRookLogs(n, systemNamespace, s.namespace)
+
+	// Upgrade Rook to master
+	require.NoError(s.T(), s.k8sh.SetDeploymentVersion(systemNamespace, operatorContainer, operatorContainer, installer.VersionMaster))
 
 	// verify that the operator spec is updated
 	version, err = k8sutil.GetDeploymentImage(s.k8sh.Clientset, systemNamespace, operatorContainer, operatorContainer)
-	assert.Nil(s.T(), err)
+	assert.NoError(s.T(), err)
 	assert.Equal(s.T(), "rook/ceph:"+installer.VersionMaster, version)
 
 	// wait for the mon pods to be running. we can no longer check a version in the pod spec, so we just wait a bit.
 	time.Sleep(15 * time.Second)
 
-	err = s.k8sh.WaitForLabeledPodsToRun("app=rook-ceph-mon", s.namespace)
-	require.Nil(s.T(), err)
+	monsNotOldVersion = fmt.Sprintf("app=rook-ceph-mon,rook-version!=%s", oldRookVersion)
+	err = s.k8sh.WaitForDeploymentCount(monsNotOldVersion, s.namespace, numMons)
+	require.NoError(s.T(), err)
+	err = s.k8sh.WaitForLabeledDeploymentsToBeReady(monsNotOldVersion, s.namespace)
+	require.NoError(s.T(), err)
 
 	// wait for the osd pods to be updated
-	osdsNotOldVersion := fmt.Sprintf("app=rook-ceph-osd,rook-version!=%s", oldRookVersion)
+	osdsNotOldVersion = fmt.Sprintf("app=rook-ceph-osd,rook-version!=%s", oldRookVersion)
 	err = s.k8sh.WaitForDeploymentCount(osdsNotOldVersion, s.namespace, numOSDs)
 	require.NoError(s.T(), err)
+	err = s.k8sh.WaitForLabeledDeploymentsToBeReady(osdsNotOldVersion, s.namespace)
+	require.NoError(s.T(), err)
+
+	mdsesNotOldVersion = fmt.Sprintf("app=rook-ceph-mds,rook-version!=%s", oldRookVersion)
+	err = s.k8sh.WaitForDeploymentCount(mdsesNotOldVersion, s.namespace, 4 /* always expect 4 mdses */)
+	require.NoError(s.T(), err)
+	err = s.k8sh.WaitForLabeledDeploymentsToBeReady(mdsesNotOldVersion, s.namespace)
+	require.NoError(s.T(), err)
 
 	logger.Infof("Done with automatic upgrade to master")
 
 	// Give a few seconds for the daemons to settle down after the upgrade
 	time.Sleep(5 * time.Second)
 
+	// wait for filesystem to be active
+	err = waitForFilesystemActive(s.k8sh, s.namespace, filesystemName)
+	require.NoError(s.T(), err)
+
 	// Test writing and reading from the pod with cephfs mounted that was created before the upgrade.
 	postFilename := "post-upgrade-file"
-	assert.Nil(s.T(), s.k8sh.ReadFromPodRetry(s.namespace, filePodName, preFilename, message, 5))
-	assert.Nil(s.T(), s.k8sh.WriteToPodRetry(s.namespace, filePodName, postFilename, message, 5))
-	assert.Nil(s.T(), s.k8sh.ReadFromPodRetry(s.namespace, filePodName, postFilename, message, 5))
+	assert.NoError(s.T(), s.k8sh.ReadFromPodRetry(s.namespace, filePodName, preFilename, message, 5))
+	assert.NoError(s.T(), s.k8sh.WriteToPodRetry(s.namespace, filePodName, postFilename, message, 5))
+	assert.NoError(s.T(), s.k8sh.ReadFromPodRetry(s.namespace, filePodName, postFilename, message, 5))
 
 	// Test writing and reading from the pod with rbd mounted that was created before the upgrade.
 	// There is some unreliability right after the upgrade when there is only one osd, so we will retry if needed
-	assert.Nil(s.T(), s.k8sh.ReadFromPodRetry("", podName, preFilename, message, 5))
-	assert.Nil(s.T(), s.k8sh.WriteToPodRetry("", podName, postFilename, message, 5))
-	assert.Nil(s.T(), s.k8sh.ReadFromPodRetry("", podName, postFilename, message, 5))
+	assert.NoError(s.T(), s.k8sh.ReadFromPodRetry("", podName, preFilename, message, 5))
+	assert.NoError(s.T(), s.k8sh.WriteToPodRetry("", podName, postFilename, message, 5))
+	assert.NoError(s.T(), s.k8sh.ReadFromPodRetry("", podName, postFilename, message, 5))
 }
 
 // Update the clusterroles that have been modified in master from the previous release