diff --git a/pkg/daemon/ceph/osd/agent.go b/pkg/daemon/ceph/osd/agent.go
index 7753ff4e9..c08e9c6f6 100644
--- a/pkg/daemon/ceph/osd/agent.go
+++ b/pkg/daemon/ceph/osd/agent.go
@@ -375,15 +375,10 @@ func (a *OsdAgent) startOSD(context *clusterd.Context, cfg *osdConfig) error {
 		return err
 	}
 
-	newOSD := false
-	if isOSDDataNotExist(cfg.rootPath) {
-		// consider this a new osd if the "whoami" file is not found
-		newOSD = true
-
-		// ensure the config path exists
-		if err := os.MkdirAll(cfg.rootPath, 0744); err != nil {
-			return fmt.Errorf("failed to make osd %d config at %s: %+v", cfg.id, cfg.rootPath, err)
-		}
+	// prepare the osd root dir, which will tell us if it's a new osd
+	newOSD, err := prepareOSDRoot(cfg)
+	if err != nil {
+		return err
 	}
 
 	if newOSD {
@@ -414,7 +409,7 @@ func (a *OsdAgent) startOSD(context *clusterd.Context, cfg *osdConfig) error {
 			}
 		}
 
-		// osd_data_dir/whoami does not exist yet, create/initialize the OSD
+		// osd_data_dir/ready does not exist yet, create/initialize the OSD
 		err := initializeOSD(cfg, context, a.cluster, a.location)
 		if err != nil {
 			return fmt.Errorf("failed to initialize OSD at %s: %+v", cfg.rootPath, err)
@@ -426,7 +421,7 @@ func (a *OsdAgent) startOSD(context *clusterd.Context, cfg *osdConfig) error {
 			logger.Warningf("failed to update config file. %+v", err)
 		}
 
-		// osd_data_dir/whoami already exists, meaning the OSD is already set up.
+		// osd_data_dir/ready already exists, meaning the OSD is already set up.
 		// look up some basic information about it so we can run it.
 		err = loadOSDInfo(cfg)
 		if err != nil {
@@ -435,7 +430,7 @@ func (a *OsdAgent) startOSD(context *clusterd.Context, cfg *osdConfig) error {
 	}
 
 	// run the OSD in a child process now that it is fully initialized and ready to go
-	err := a.runOSD(context, a.cluster.Name, cfg)
+	err = a.runOSD(context, a.cluster.Name, cfg)
 	if err != nil {
 		return fmt.Errorf("failed to run osd %d: %+v", cfg.id, err)
 	}
@@ -443,6 +438,27 @@ func (a *OsdAgent) startOSD(context *clusterd.Context, cfg *osdConfig) error {
 	return nil
 }
 
+func prepareOSDRoot(cfg *osdConfig) (newOSD bool, err error) {
+	newOSD = isOSDDataNotExist(cfg.rootPath)
+	if !newOSD {
+		// osd is not new (it's ready), nothing to prepare
+		return newOSD, nil
+	}
+
+	// osd is new (it's not ready), make sure there is no stale state in the OSD dir by deleting the entire thing
+	logger.Infof("osd.%d appears to be new, cleaning the root dir at %s", cfg.id, cfg.rootPath)
+	if err := os.RemoveAll(cfg.rootPath); err != nil {
+		logger.Warningf("failed to clean osd.%d root dir at %s, will proceed with starting osd: %+v", cfg.id, cfg.rootPath, err)
+	}
+
+	// prepare the osd dir by creating it now
+	if err := os.MkdirAll(cfg.rootPath, 0744); err != nil {
+		return newOSD, fmt.Errorf("failed to make osd.%d config at %s: %+v", cfg.id, cfg.rootPath, err)
+	}
+
+	return newOSD, nil
+}
+
 // runs an OSD with the given config in a child process
 func (a *OsdAgent) runOSD(context *clusterd.Context, clusterName string, config *osdConfig) error {
 	// start the OSD daemon in the foreground with the given config
@@ -601,7 +617,7 @@ func waitForRebalance(context *clusterd.Context, clusterName string, osdID int,
 }
 
 func isOSDDataNotExist(osdDataPath string) bool {
-	_, err := os.Stat(filepath.Join(osdDataPath, "whoami"))
+	_, err := os.Stat(filepath.Join(osdDataPath, "ready"))
 	return os.IsNotExist(err)
 }
 
diff --git a/pkg/daemon/ceph/osd/agent_test.go b/pkg/daemon/ceph/osd/agent_test.go
index 6bb6aa237..863e5e11c 100644
--- a/pkg/daemon/ceph/osd/agent_test.go
+++ b/pkg/daemon/ceph/osd/agent_test.go
@@ -537,6 +537,40 @@ func TestGetPartitionSchemeDiskNameChanged(t *testing.T) {
 	assert.Equal(t, "nvme01", scheme.Entries[0].Partitions[config.DatabasePartitionType].Device)
 }
 
+func TestPrepareOSDRoot(t *testing.T) {
+	configDir, _ := ioutil.TempDir("", "")
+	defer os.RemoveAll(configDir)
+	os.MkdirAll(configDir, 0755)
+
+	cfg := &osdConfig{id: 516, configRoot: configDir}
+	cfg.rootPath = getOSDRootDir(cfg.configRoot, cfg.id)
+
+	// clean slate, definitely a new OSD
+	newOSD, err := prepareOSDRoot(cfg)
+	assert.Nil(t, err)
+	assert.True(t, newOSD)
+
+	// simulate the failure of a previous osd mkfs that left the osd dir in an intermediate state
+	// this should also be considered a new OSD and the previous (stale) state should have been cleaned
+	ioutil.WriteFile(filepath.Join(cfg.rootPath, "whoami"), []byte("516"), 0644)
+	newOSD, err = prepareOSDRoot(cfg)
+	assert.Nil(t, err)
+	assert.True(t, newOSD)
+	fis, err := ioutil.ReadDir(cfg.rootPath)
+	assert.Nil(t, err)
+	assert.Equal(t, 0, len(fis)) // osd dir should have been cleaned
+
+	// simulate a completed osd mkfs, where the osd is ready.  this should not be considered a new
+	// osd and the osd dir should be left intact
+	ioutil.WriteFile(filepath.Join(cfg.rootPath, "ready"), []byte("ready"), 0644)
+	newOSD, err = prepareOSDRoot(cfg)
+	assert.Nil(t, err)
+	assert.False(t, newOSD)
+	fis, err = ioutil.ReadDir(cfg.rootPath)
+	assert.Nil(t, err)
+	assert.Equal(t, 1, len(fis)) // osd dir should NOT have been cleaned
+}
+
 func verifyPartitionEntry(t *testing.T, actual *config.PerfSchemePartitionDetails, expectedDevice string,
 	expectedSize int, expectedOffset int) {