ceph: skip osd prepare job creation if osd daemon exists for the pvc

pkg/operator/ceph/cluster/osd/osd.go

@@ -263,13 +263,41 @@ func (c *Cluster) startProvisioningOverPVCs(config *provisionConfig) {
 			continue
 		}
 
+		//Skip OSD prepare if deployment already exists for the PVC
+		listOpts := metav1.ListOptions{LabelSelector: fmt.Sprintf("%s=%s,%s=%s",
+			k8sutil.AppAttr, AppName,
+			OSDOverPVCLabelKey, volume.PersistentVolumeClaimSource.ClaimName,
+		)}
+
+		osdDeployments, err := c.context.Clientset.AppsV1().Deployments(c.Namespace).List(listOpts)
+		if err != nil {
+			config.addError("failed to check if OSD daemon exists for pvc %q. %+v", osdProps.crushHostname, err)
+			continue
+		}
+
+		if len(osdDeployments.Items) != 0 {
+			logger.Infof("skip OSD prepare pod creation as OSD daemon already exists for %q", osdProps.crushHostname)
+			osds, err := getOSDInfo(&osdDeployments.Items[0])
+			if err != nil {
+				config.addError("failed to get osdInfo for pvc %q. %+v", osdProps.crushHostname, err)
+				continue
+			}
+			// update the orchestration status of this pvc to the completed state
+			status = OrchestrationStatus{OSDs: osds, Status: OrchestrationStatusCompleted, PvcBackedOSD: true}
+			if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil {
+				config.addError("failed to update pvc %q status. %+v", osdProps.crushHostname, err)
+				continue
+			}
+			continue
+		}
+
 		job, err := c.makeJob(osdProps)
 		if err != nil {
 			message := fmt.Sprintf("failed to create prepare job for pvc %s: %v", osdProps.crushHostname, err)
 			config.addError(message)
 			status := OrchestrationStatus{Status: OrchestrationStatusCompleted, Message: message, PvcBackedOSD: true}
 			if err := c.updateOSDStatus(osdProps.crushHostname, status); err != nil {
-				config.addError("failed to update pvc %s status. %+v", osdProps.crushHostname, err)
+				config.addError("failed to update pvc %q status. %+v", osdProps.crushHostname, err)
 				continue
 			}
 		}
@@ -828,3 +856,40 @@ func (c *Cluster) getPVCHostName(pvcName string) (string, error) {
 	}
 	return "", err
 }
+
+func getOSDInfo(d *apps.Deployment) ([]OSDInfo, error) {
+	container := d.Spec.Template.Spec.Containers[0]
+	var osd OSDInfo
+
+	osdID, err := strconv.Atoi(d.Labels[OsdIdLabelKey])
+	if err != nil {
+		return []OSDInfo{}, fmt.Errorf("error parsing ceph-osd-id. %+v", err)
+	}
+	osd.ID = osdID
+
+	for _, envVar := range d.Spec.Template.Spec.Containers[0].Env {
+		if envVar.Name == "ROOK_OSD_UUID" {
+			osd.UUID = envVar.Value
+		}
+		if envVar.Name == "ROOK_LV_PATH" {
+			osd.LVPath = envVar.Value
+		}
+	}
+
+	for i, a := range container.Args {
+		if strings.HasPrefix(a, "--setuser-match-path") {
+			if len(container.Args) >= i+1 {
+				osd.DataPath = container.Args[i+1]
+				break
+			}
+		}
+	}
+
+	osd.CephVolumeInitiated = true
+
+	if osd.DataPath == "" || osd.UUID == "" || osd.LVPath == "" {
+		return []OSDInfo{}, fmt.Errorf("failed to get required osdInfo. %+v", osd)
+	}
+
+	return []OSDInfo{osd}, nil
+}

pkg/operator/ceph/cluster/osd/osd_test.go

@@ -369,3 +369,30 @@ func TestAddNodeFailure(t *testing.T) {
 	assert.True(t, startCompleted)
 	assert.NotNil(t, startErr)
 }
+
+func TestGetOSDInfo(t *testing.T) {
+	c := New(&cephconfig.ClusterInfo{}, &clusterd.Context{}, "ns", "myversion", cephv1.CephVersionSpec{},
+		rookalpha.StorageScopeSpec{}, "", rookalpha.Placement{}, rookalpha.Annotations{}, cephv1.NetworkSpec{}, v1.ResourceRequirements{}, v1.ResourceRequirements{}, metav1.OwnerReference{}, false, false)
+
+	node := "n1"
+	osd1 := OSDInfo{ID: 3, UUID: "osd-uuid", LVPath: "dev/logical-volume-path", DataPath: "/rook/path", CephVolumeInitiated: true}
+	osd2 := OSDInfo{ID: 3, UUID: "osd-uuid", LVPath: "", DataPath: "/rook/path", CephVolumeInitiated: true}
+	osdProp := osdProperties{
+		crushHostname: node,
+		pvc:           v1.PersistentVolumeClaimVolumeSource{ClaimName: "pvc"},
+		selection:     rookalpha.Selection{},
+		resources:     v1.ResourceRequirements{},
+		storeConfig:   config.StoreConfig{},
+	}
+	d1, _ := c.makeDeployment(osdProp, osd1)
+	osds1, _ := getOSDInfo(d1)
+	assert.Equal(t, 1, len(osds1))
+	assert.Equal(t, osd1.ID, osds1[0].ID)
+	assert.Equal(t, osd1.LVPath, osds1[0].LVPath)
+
+	d2, _ := c.makeDeployment(osdProp, osd2)
+	osds2, err := getOSDInfo(d2)
+	assert.Equal(t, 0, len(osds2))
+	assert.NotNil(t, err)
+
+}