pool: return error if device class update fails

Updating the device class swallowed any error if updated
for the pool. The error was not even logged, so we couldn't
troubleshoot why the new crush rule was not applied.
Log the error for troubleshooting and also fail the pool
reconcile since the desired configuration was not applied.

Signed-off-by: Travis Nielsen <tnielsen@redhat.com>

pkg/daemon/ceph/client/pool.go

@@ -456,7 +456,7 @@ func createReplicatedPoolForApp(context *clusterd.Context, clusterInfo *ClusterI
 
 	if checkFailureDomain || pool.PoolSpec.DeviceClass != "" {
 		if err = updatePoolCrushRule(context, clusterInfo, clusterSpec, pool); err != nil {
-			return nil
+			return errors.Wrapf(err, "failed to update crush rule for pool %q", pool.Name)
 		}
 	}
 	return nil
@@ -696,9 +696,9 @@ func createReplicationCrushRule(context *clusterd.Context, clusterInfo *ClusterI
 		args = append(args, deviceClass)
 	}
 
-	_, err := NewCephCommand(context, clusterInfo, args).Run()
+	output, err := NewCephCommand(context, clusterInfo, args).Run()
 	if err != nil {
-		return errors.Wrapf(err, "failed to create crush rule %s", ruleName)
+		return errors.Wrapf(err, "failed to create crush rule %s. %s", ruleName, string(output))
 	}
 
 	return nil

pkg/daemon/ceph/client/pool_test.go

@@ -179,6 +179,11 @@ func testCreateReplicaPool(t *testing.T, failureDomain, crushRoot, deviceClass,
 				assert.Equal(t, "12345", args[8])
 				return "", nil
 			}
+			if args[2] == "get" {
+				assert.Equal(t, "mypool", args[3])
+				assert.Equal(t, "all", args[4])
+				return `{"pool":"replicapool","pool_id":2,"size":1,"min_size":1,"crush_rule":"replicapool_osd"}`, nil
+			}
 			if args[2] == "set" {
 				assert.Equal(t, "mypool", args[3])
 				if args[4] == "size" {
@@ -203,8 +208,12 @@ func testCreateReplicaPool(t *testing.T, failureDomain, crushRoot, deviceClass,
 		if args[1] == "crush" {
 			crushRuleCreated = true
 			assert.Equal(t, "rule", args[2])
+			if args[3] == "dump" {
+				assert.Equal(t, "replicapool_osd", args[4])
+				return `{"rule_id": 3,"rule_name": "replicapool_osd","type": 1}`, nil
+			}
 			assert.Equal(t, "create-replicated", args[3])
-			assert.Equal(t, "mypool", args[4])
+			assert.Contains(t, args[4], "mypool")
 			if crushRoot == "" {
 				assert.Equal(t, "cluster-crush-root", args[5])
 			} else {

pkg/operator/ceph/pool/controller.go

@@ -364,7 +364,7 @@ func (r *ReconcileCephBlockPool) reconcileCreatePool(clusterInfo *cephclient.Clu
 	poolSpec := cephBlockPool.ToNamedPoolSpec()
 	err := createPool(r.context, clusterInfo, cephCluster, &poolSpec)
 	if err != nil {
-		return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "failed to create pool %q.", cephBlockPool.GetName())
+		return opcontroller.ImmediateRetryResult, errors.Wrapf(err, "failed to configure pool %q.", cephBlockPool.GetName())
 	}
 
 	// Let's return here so that on the initial creation we don't check for update right away
@@ -380,7 +380,7 @@ func createPool(context *clusterd.Context, clusterInfo *cephclient.ClusterInfo,
 	// create the pool
 	logger.Infof("creating pool %q in namespace %q", p.Name, clusterInfo.Namespace)
 	if err := cephclient.CreatePool(context, clusterInfo, clusterSpec, *p); err != nil {
-		return errors.Wrapf(err, "failed to create pool %q", p.Name)
+		return errors.Wrapf(err, "failed to configure pool %q", p.Name)
 	}
 
 	if p.Application != poolApplicationNameRBD {