diff --git a/cluster/charts/rook-ceph/templates/resources.yaml b/cluster/charts/rook-ceph/templates/resources.yaml
index f47311e8b..5ff481ad8 100644
--- a/cluster/charts/rook-ceph/templates/resources.yaml
+++ b/cluster/charts/rook-ceph/templates/resources.yaml
@@ -63,6 +63,8 @@ spec:
                   type: boolean
                 continueUpgradeAfterChecksEvenIfNotHealthy:
                   type: boolean
+                waitTimeoutForHealthyOSDInMinutes:
+                  type: integer
                 mon:
                   type: object
                   properties:
@@ -1411,6 +1413,8 @@ spec:
               type: boolean
             continueUpgradeAfterChecksEvenIfNotHealthy:
               type: boolean
+            waitTimeoutForHealthyOSDInMinutes:
+              type: integer
             mon:
               properties:
                 allowMultiplePerNode:
diff --git a/cluster/examples/kubernetes/ceph/cluster.yaml b/cluster/examples/kubernetes/ceph/cluster.yaml
index 0fda1e94b..77b9e23ff 100644
--- a/cluster/examples/kubernetes/ceph/cluster.yaml
+++ b/cluster/examples/kubernetes/ceph/cluster.yaml
@@ -37,6 +37,12 @@ spec:
   skipUpgradeChecks: false
   # Whether or not continue if PGs are not clean during an upgrade
   continueUpgradeAfterChecksEvenIfNotHealthy: false
+  # WaitTimeoutForHealthyOSDInMinutes defines the time (in minutes) the operator would wait before an OSD can be stopped for upgrade or restart.
+  # If the timeout exceeds and OSD is not ok to stop, then the operator would skip upgrade for the current OSD and proceed with the next one
+  # if `continueUpgradeAfterChecksEvenIfNotHealthy` is `false`. If `continueUpgradeAfterChecksEvenIfNotHealthy` is `true`, then opertor would
+  # continue with the upgrade of an OSD even if its not ok to stop after the timeout. This timeout won't be applied if `skipUpgradeChecks` is `true`.
+  # The default wait timeout is 10 minutes.
+  waitTimeoutForHealthyOSDInMinutes: 10
   mon:
     # Set the number of mons to be started. Must be an odd number, and is generally recommended to be 3.
     count: 3
diff --git a/cluster/examples/kubernetes/ceph/crds.yaml b/cluster/examples/kubernetes/ceph/crds.yaml
index 02e511461..3fa604a27 100644
--- a/cluster/examples/kubernetes/ceph/crds.yaml
+++ b/cluster/examples/kubernetes/ceph/crds.yaml
@@ -65,6 +65,8 @@ spec:
                   type: boolean
                 continueUpgradeAfterChecksEvenIfNotHealthy:
                   type: boolean
+                waitTimeoutForHealthyOSDInMinutes:
+                  type: integer
                 mon:
                   type: object
                   properties:
diff --git a/cluster/examples/kubernetes/ceph/pre-k8s-1.16/crds.yaml b/cluster/examples/kubernetes/ceph/pre-k8s-1.16/crds.yaml
index 1c2aec7f8..dd4945515 100644
--- a/cluster/examples/kubernetes/ceph/pre-k8s-1.16/crds.yaml
+++ b/cluster/examples/kubernetes/ceph/pre-k8s-1.16/crds.yaml
@@ -73,6 +73,8 @@ spec:
               type: boolean
             continueUpgradeAfterChecksEvenIfNotHealthy:
               type: boolean
+            waitTimeoutForHealthyOSDInMinutes:
+              type: integer
             mon:
               properties:
                 allowMultiplePerNode:
diff --git a/pkg/apis/ceph.rook.io/v1/types.go b/pkg/apis/ceph.rook.io/v1/types.go
index 091bf9294..20eb59f58 100755
--- a/pkg/apis/ceph.rook.io/v1/types.go
+++ b/pkg/apis/ceph.rook.io/v1/types.go
@@ -98,6 +98,13 @@ type ClusterSpec struct {
 	// ContinueUpgradeAfterChecksEvenIfNotHealthy defines if an upgrade should continue even if PGs are not clean
 	ContinueUpgradeAfterChecksEvenIfNotHealthy bool `json:"continueUpgradeAfterChecksEvenIfNotHealthy,omitempty"`
 
+	// WaitTimeoutForHealthyOSDInMinutes defines the time the operator would wait before an OSD can be stopped for upgrade or restart.
+	// If the timeout exceeds and OSD is not ok to stop, then the operator would skip upgrade for the current OSD and proceed with the next one
+	// if `continueUpgradeAfterChecksEvenIfNotHealthy` is `false`. If `continueUpgradeAfterChecksEvenIfNotHealthy` is `true`, then opertor would
+	// continue with the upgrade of an OSD even if its not ok to stop after the timeout. This timeout won't be applied if `skipUpgradeChecks` is `true`.
+	// The default wait timeout is 10 minutes.
+	WaitTimeoutForHealthyOSDInMinutes time.Duration `json:"waitTimeoutForHealthyOSDInMinutes,omitempty"`
+
 	// A spec for configuring disruption management.
 	DisruptionManagement DisruptionManagementSpec `json:"disruptionManagement,omitempty"`
 
diff --git a/pkg/daemon/ceph/client/info.go b/pkg/daemon/ceph/client/info.go
index 818d77074..8eaf54b08 100644
--- a/pkg/daemon/ceph/client/info.go
+++ b/pkg/daemon/ceph/client/info.go
@@ -19,6 +19,7 @@ package client
 import (
 	"fmt"
 	"net"
+	"time"
 
 	cephver "github.com/rook/rook/pkg/operator/ceph/version"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -37,7 +38,8 @@ type ClusterInfo struct {
 	OwnerRef      metav1.OwnerReference
 	// Hide the name of the cluster since in 99% of uses we want to use the cluster namespace.
 	// If the CR name is needed, access it through the NamespacedName() method.
-	name string
+	name              string
+	OsdUpgradeTimeout time.Duration
 }
 
 // MonInfo is a collection of information about a Ceph mon.
diff --git a/pkg/daemon/ceph/client/upgrade.go b/pkg/daemon/ceph/client/upgrade.go
index 679507f56..4bccf2592 100644
--- a/pkg/daemon/ceph/client/upgrade.go
+++ b/pkg/daemon/ceph/client/upgrade.go
@@ -27,6 +27,12 @@ import (
 	"github.com/rook/rook/pkg/util"
 )
 
+const (
+	defaultMaxRetries    = 10
+	defaultRetryDelay    = 60 * time.Second
+	defaultOSDRetryDelay = 10 * time.Second
+)
+
 // CephDaemonsVersions is a structure that can be used to parsed the output of the 'ceph versions' command
 type CephDaemonsVersions struct {
 	Mon       map[string]int `json:"mon,omitempty"`
@@ -131,6 +137,7 @@ func EnableReleaseOSDFunctionality(context *clusterd.Context, clusterInfo *Clust
 
 // OkToStop determines if it's ok to stop an upgrade
 func OkToStop(context *clusterd.Context, clusterInfo *ClusterInfo, deployment, daemonType, daemonName string) error {
+	okToStopRetries, okToStopDelay := getRetryConfig(clusterInfo, daemonType)
 	versions, err := GetAllCephDaemonVersions(context, clusterInfo)
 	if err != nil {
 		return errors.Wrap(err, "failed to get ceph daemons versions")
@@ -161,11 +168,14 @@ func OkToStop(context *clusterd.Context, clusterInfo *ClusterInfo, deployment, d
 			return nil
 		}
 	}
+
 	// we don't implement any checks for mon, rgw and rbdmirror since:
 	//  - mon: the is done in the monitor code since it ensures all the mons are always in quorum before continuing
 	//  - rgw: the pod spec has a liveness probe so if the pod successfully start
 	//  - rbdmirror: you can chain as many as you want like mdss but there is no ok-to-stop logic yet
-	err = okToStopDaemon(context, clusterInfo, deployment, daemonType, daemonName)
+	err = util.Retry(okToStopRetries, okToStopDelay, func() error {
+		return okToStopDaemon(context, clusterInfo, deployment, daemonType, daemonName)
+	})
 	if err != nil {
 		return errors.Wrapf(err, "failed to check if %s was ok to stop", deployment)
 	}
@@ -177,14 +187,6 @@ func OkToStop(context *clusterd.Context, clusterInfo *ClusterInfo, deployment, d
 func OkToContinue(context *clusterd.Context, clusterInfo *ClusterInfo, deployment, daemonType, daemonName string) error {
 	// the mon case is handled directly in the deployment where the mon checks for quorum
 	switch daemonType {
-	case "osd":
-		if osdDoNothing(context, clusterInfo) {
-			return nil
-		}
-		err := okToContinueOSDDaemon(context, clusterInfo)
-		if err != nil {
-			return errors.Wrapf(err, "failed to check if %s was ok to continue", deployment)
-		}
 	case "mds":
 		err := okToContinueMDSDaemon(context, clusterInfo, deployment, daemonType, daemonName)
 		if err != nil {
@@ -214,25 +216,12 @@ func okToStopDaemon(context *clusterd.Context, clusterInfo *ClusterInfo, deploym
 	return nil
 }
 
-// okToContinueOSDDaemon determines whether it's fine to go to the next osd during an upgrade
-// This basically makes sure all the PGs have settled
-func okToContinueOSDDaemon(context *clusterd.Context, clusterInfo *ClusterInfo) error {
-	// Reconciliating PGs should not take too long so let's wait up to 10 minutes
-	err := util.Retry(10, 60*time.Second, func() error {
-		return IsClusterCleanError(context, clusterInfo)
-	})
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
 // okToContinueMDSDaemon determines whether it's fine to go to the next mds during an upgrade
 // mostly a placeholder function for the future but since we have standby mds this shouldn't be needed
 func okToContinueMDSDaemon(context *clusterd.Context, clusterInfo *ClusterInfo, deployment, daemonType, daemonName string) error {
 	// wait for the MDS to be active again or in standby-replay
-	err := util.Retry(10, 15*time.Second, func() error {
+	retries, delay := getRetryConfig(clusterInfo, "mds")
+	err := util.Retry(retries, delay, func() error {
 		return MdsActiveOrStandbyReplay(context, clusterInfo, findFSName(deployment))
 	})
 	if err != nil {
@@ -400,3 +389,14 @@ func osdDoNothing(context *clusterd.Context, clusterInfo *ClusterInfo) bool {
 
 	return false
 }
+
+func getRetryConfig(clusterInfo *ClusterInfo, daemonType string) (int, time.Duration) {
+	switch daemonType {
+	case "osd":
+		return int(clusterInfo.OsdUpgradeTimeout / defaultOSDRetryDelay), defaultOSDRetryDelay
+	case "mds":
+		return defaultMaxRetries, 15 * time.Second
+	}
+
+	return defaultMaxRetries, defaultRetryDelay
+}
diff --git a/pkg/daemon/ceph/client/upgrade_test.go b/pkg/daemon/ceph/client/upgrade_test.go
index 748ec37d7..e6aaf1518 100644
--- a/pkg/daemon/ceph/client/upgrade_test.go
+++ b/pkg/daemon/ceph/client/upgrade_test.go
@@ -19,6 +19,7 @@ package client
 import (
 	"encoding/json"
 	"testing"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/rook/rook/pkg/clusterd"
@@ -250,3 +251,49 @@ func TestBuildHostListFromTree(t *testing.T) {
 	_, err = buildHostListFromTree(dummyEmptyNodeTree)
 	assert.NoError(t, err)
 }
+
+func TestGetRetryConfig(t *testing.T) {
+	testcases := []struct {
+		label           string
+		clusterInfo     *ClusterInfo
+		daemonType      string
+		expectedRetries int
+		expectedDelay   time.Duration
+	}{
+		{
+			label:           "case 1: mon daemon",
+			clusterInfo:     &ClusterInfo{},
+			daemonType:      "mon",
+			expectedRetries: 10,
+			expectedDelay:   60 * time.Second,
+		},
+		{
+			label:           "case 2: osd daemon with 5 minutes delay",
+			clusterInfo:     &ClusterInfo{OsdUpgradeTimeout: 5 * time.Minute},
+			daemonType:      "osd",
+			expectedRetries: 30,
+			expectedDelay:   10 * time.Second,
+		},
+		{
+			label:           "case 3: osd daemon with 10 minutes delay",
+			clusterInfo:     &ClusterInfo{OsdUpgradeTimeout: 10 * time.Minute},
+			daemonType:      "osd",
+			expectedRetries: 60,
+			expectedDelay:   10 * time.Second,
+		},
+		{
+			label:           "case 4: mds daemon",
+			clusterInfo:     &ClusterInfo{},
+			daemonType:      "mds",
+			expectedRetries: 10,
+			expectedDelay:   15 * time.Second,
+		},
+	}
+
+	for _, tc := range testcases {
+		actualRetries, actualDelay := getRetryConfig(tc.clusterInfo, tc.daemonType)
+
+		assert.Equal(t, tc.expectedRetries, actualRetries, "[%s] failed to get correct retry count", tc.label)
+		assert.Equalf(t, tc.expectedDelay, actualDelay, "[%s] failed to get correct delays between retries", tc.label)
+	}
+}
diff --git a/pkg/operator/ceph/cluster/osd/osd.go b/pkg/operator/ceph/cluster/osd/osd.go
index 6f43cefce..e69d2658c 100644
--- a/pkg/operator/ceph/cluster/osd/osd.go
+++ b/pkg/operator/ceph/cluster/osd/osd.go
@@ -23,6 +23,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"time"
 
 	"k8s.io/client-go/kubernetes"
 
@@ -61,10 +62,11 @@ const (
 	// AppName is the "app" label on osd pods
 	AppName = "rook-ceph-osd"
 	// FailureDomainKey is the label key whose value is the failure domain of the OSD
-	FailureDomainKey  = "failure-domain"
-	prepareAppName    = "rook-ceph-osd-prepare"
-	prepareAppNameFmt = "rook-ceph-osd-prepare-%s"
-	osdAppNameFmt     = "rook-ceph-osd-%d"
+	FailureDomainKey                = "failure-domain"
+	prepareAppName                  = "rook-ceph-osd-prepare"
+	prepareAppNameFmt               = "rook-ceph-osd-prepare-%s"
+	osdAppNameFmt                   = "rook-ceph-osd-%d"
+	defaultWaitTimeoutForHealthyOSD = 10 * time.Minute
 	// OsdIdLabelKey is the OSD label key
 	OsdIdLabelKey                  = "ceph-osd-id"
 	serviceAccountName             = "rook-ceph-osd"
@@ -181,6 +183,13 @@ func (c *Cluster) Start() error {
 		logger.Warningf("useAllNodes is set to false and no nodes, driveGroups, storageClassDevicesets or volumeSources are specified, no OSD pods are going to be created")
 	}
 
+	if c.spec.WaitTimeoutForHealthyOSDInMinutes != 0 {
+		c.clusterInfo.OsdUpgradeTimeout = c.spec.WaitTimeoutForHealthyOSDInMinutes * time.Minute
+	} else {
+		c.clusterInfo.OsdUpgradeTimeout = defaultWaitTimeoutForHealthyOSD
+	}
+	logger.Infof("wait timeout for healthy OSDs during upgrade or restart is %q", c.clusterInfo.OsdUpgradeTimeout)
+
 	// start the jobs to provision the OSD devices
 	logger.Infof("start provisioning the osds on pvcs, if needed")
 	c.startProvisioningOverPVCs(config)
diff --git a/pkg/operator/k8sutil/deployment.go b/pkg/operator/k8sutil/deployment.go
index 5de304326..3338d71c2 100644
--- a/pkg/operator/k8sutil/deployment.go
+++ b/pkg/operator/k8sutil/deployment.go
@@ -24,7 +24,6 @@ import (
 	"github.com/banzaicloud/k8s-objectmatcher/patch"
 
 	"github.com/rook/rook/pkg/clusterd"
-	"github.com/rook/rook/pkg/util"
 	apps "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
@@ -84,10 +83,7 @@ func UpdateDeploymentAndWait(clusterdContext *clusterd.Context, modifiedDeployme
 		logger.Infof("updating deployment %q after verifying it is safe to stop", modifiedDeployment.Name)
 
 		// Let's verify the deployment can be stopped
-		// retry for 5 times, every minute
-		err = util.Retry(5, 60*time.Second, func() error {
-			return verifyCallback("stop")
-		})
+		err = verifyCallback("stop")
 		if err != nil {
 			return nil, fmt.Errorf("failed to check if deployment %q can be updated. %v", modifiedDeployment.Name, err)
 		}