Merge pull request #1651 from tonya11en/ec_overwrite

Enable EC overwrite for RBD and filesystem.

pkg/daemon/ceph/client/pool.go

@@ -134,7 +134,13 @@ func CreatePoolWithProfile(context *clusterd.Context, clusterName string, newPoo
 		}
 	}
 
-	return CreatePoolForApp(context, clusterName, newPool, appName)
+	isReplicatedPool := newPool.ErasureCodeProfile == "" && newPool.Size > 0
+	if isReplicatedPool {
+		return CreateReplicatedPoolForApp(context, clusterName, newPool, appName)
+	} else {
+		// If the pool is not a replicated pool, then the only other option is an erasure coded pool.
+		return CreateECPoolForApp(context, clusterName, newPool, appName, true /* enableECOverwrite */)
+	}
 }
 
 func DeletePool(context *clusterd.Context, clusterName string, name string) error {
@@ -163,48 +169,65 @@ func DeletePool(context *clusterd.Context, clusterName string, name string) erro
 	return nil
 }
 
-func CreatePool(context *clusterd.Context, clusterName string, newPool CephStoragePoolDetails) error {
-	// for a generic/custom pool, just reuse the pool name for its app name
-	return CreatePoolForApp(context, clusterName, newPool, newPool.Name)
+func givePoolAppTag(context *clusterd.Context, clusterName string, poolName string, appName string) error {
+	args := []string{"osd", "pool", "application", "enable", poolName, appName, confirmFlag}
+	_, err := ExecuteCephCommand(context, clusterName, args)
+	if err != nil {
+		return fmt.Errorf("failed to enable application %s on pool %s. %+v", appName, poolName, err)
+	}
+
+	return nil
 }
 
-func CreatePoolForApp(context *clusterd.Context, clusterName string, newPool CephStoragePoolDetails, appName string) error {
-	// create a crush rule for a replicated pool, if a failure domain is specified
-	replicated := newPool.ErasureCodeProfile == "" && newPool.Size > 0
-	ruleName := newPool.Name
-	if replicated {
-		if err := createReplicationCrushRule(context, clusterName, newPool, ruleName); err != nil {
-			return err
-		}
-	}
-
-	args := []string{"osd", "pool", "create", newPool.Name, strconv.Itoa(newPool.Number)}
-	if newPool.ErasureCodeProfile != "" {
-		args = append(args, "erasure", newPool.ErasureCodeProfile)
-	} else {
-		args = append(args, "replicated", ruleName)
-	}
+func CreateECPoolForApp(context *clusterd.Context, clusterName string, newPool CephStoragePoolDetails, appName string, enableECOverwrite bool) error {
+	args := []string{"osd", "pool", "create", newPool.Name, strconv.Itoa(newPool.Number), "erasure", newPool.ErasureCodeProfile}
 
 	buf, err := ExecuteCephCommand(context, clusterName, args)
 	if err != nil {
-		return fmt.Errorf("failed to create pool %s. %+v", newPool.Name, err)
+		return fmt.Errorf("failed to create EC pool %s. %+v", newPool.Name, err)
 	}
 
-	if replicated {
-		// the pool is type replicated, set the size for the pool now that it's been created
-		if err = SetPoolProperty(context, clusterName, newPool.Name, "size", strconv.FormatUint(uint64(newPool.Size), 10)); err != nil {
-			return err
+	if enableECOverwrite {
+		err = SetPoolProperty(context, clusterName, newPool.Name, "allow_ec_overwrites", "true")
+		if err != nil {
+			return fmt.Errorf("failed to allow EC overwrite for pool %s. %+v", newPool.Name, err)
 		}
 	}
 
-	// ensure that the newly created pool gets an application tag
-	args = []string{"osd", "pool", "application", "enable", newPool.Name, appName, confirmFlag}
-	_, err = ExecuteCephCommand(context, clusterName, args)
+	err = givePoolAppTag(context, clusterName, newPool.Name, appName)
 	if err != nil {
-		return fmt.Errorf("failed to enable application %s on pool %s. %+v", appName, newPool.Name, err)
+		return err
 	}
 
-	logger.Infof("creating pool %s succeeded, buf: %s", newPool.Name, string(buf))
+	logger.Infof("creating EC pool %s succeeded, buf: %s", newPool.Name, string(buf))
+	return nil
+}
+
+func CreateReplicatedPoolForApp(context *clusterd.Context, clusterName string, newPool CephStoragePoolDetails, appName string) error {
+	// create a crush rule for a replicated pool, if a failure domain is specified
+	if err := createReplicationCrushRule(context, clusterName, newPool, newPool.Name); err != nil {
+		return err
+	}
+
+	args := []string{"osd", "pool", "create", newPool.Name, strconv.Itoa(newPool.Number), "replicated", newPool.Name}
+
+	buf, err := ExecuteCephCommand(context, clusterName, args)
+	if err != nil {
+		return fmt.Errorf("failed to create replicated pool %s. %+v", newPool.Name, err)
+	}
+
+	// the pool is type replicated, set the size for the pool now that it's been created
+	if err = SetPoolProperty(context, clusterName, newPool.Name, "size", strconv.FormatUint(uint64(newPool.Size), 10)); err != nil {
+		return err
+	}
+
+	// ensure that the newly created pool gets an application tag
+	err = givePoolAppTag(context, clusterName, newPool.Name, appName)
+	if err != nil {
+		return err
+	}
+
+	logger.Infof("creating replicated pool %s succeeded, buf: %s", newPool.Name, string(buf))
 	return nil
 }
 

pkg/daemon/ceph/client/pool_test.go

@@ -25,7 +25,7 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 )
 
-func TestCreateECPool(t *testing.T) {
+func TestCreateECPoolWithOverwrites(t *testing.T) {
 	p := CephStoragePoolDetails{Name: "mypool", Size: 12345, ErasureCodeProfile: "myecprofile", FailureDomain: "host"}
 	executor := &exectest.MockExecutor{}
 	context := &clusterd.Context{Executor: executor}
@@ -40,8 +40,8 @@ func TestCreateECPool(t *testing.T) {
 			}
 			if args[2] == "set" {
 				assert.Equal(t, "mypool", args[3])
-				assert.Equal(t, "size", args[4])
-				assert.Equal(t, "12345", args[5])
+				assert.Equal(t, "allow_ec_overwrites", args[4])
+				assert.Equal(t, "true", args[5])
 				return "", nil
 			}
 			if args[2] == "application" {
@@ -54,7 +54,34 @@ func TestCreateECPool(t *testing.T) {
 		return "", fmt.Errorf("unexpected ceph command '%v'", args)
 	}
 
-	err := CreatePoolForApp(context, "myns", p, "myapp")
+	err := CreateECPoolForApp(context, "myns", p, "myapp", true)
+	assert.Nil(t, err)
+}
+
+func TestCreateECPoolWithoutOverwrites(t *testing.T) {
+	p := CephStoragePoolDetails{Name: "mypool", Size: 12345, ErasureCodeProfile: "myecprofile", FailureDomain: "host"}
+	executor := &exectest.MockExecutor{}
+	context := &clusterd.Context{Executor: executor}
+	executor.MockExecuteCommandWithOutputFile = func(debug bool, actionName, command, outputFile string, args ...string) (string, error) {
+		logger.Infof("Command: %s %v", command, args)
+		if args[1] == "pool" {
+			if args[2] == "create" {
+				assert.Equal(t, "mypool", args[3])
+				assert.Equal(t, "erasure", args[5])
+				assert.Equal(t, p.ErasureCodeProfile, args[6])
+				return "", nil
+			}
+			if args[2] == "application" {
+				assert.Equal(t, "enable", args[3])
+				assert.Equal(t, "mypool", args[4])
+				assert.Equal(t, "myapp", args[5])
+				return "", nil
+			}
+		}
+		return "", fmt.Errorf("unexpected ceph command '%v'", args)
+	}
+
+	err := CreateECPoolForApp(context, "myns", p, "myapp", false)
 	assert.Nil(t, err)
 }
 
@@ -111,7 +138,7 @@ func testCreateReplicaPool(t *testing.T, failureDomain, crushRoot string) {
 	}
 
 	p := CephStoragePoolDetails{Name: "mypool", Size: 12345, FailureDomain: failureDomain, CrushRoot: crushRoot}
-	err := CreatePoolForApp(context, "myns", p, "myapp")
+	err := CreateReplicatedPoolForApp(context, "myns", p, "myapp")
 	assert.Nil(t, err)
 	assert.True(t, crushRuleCreated)
 }

pkg/daemon/ceph/rgw/objectstore.go

@@ -253,7 +253,8 @@ func createPools(context *Context, metadataSpec, dataSpec model.Pool) error {
 func createSimilarPools(context *Context, pools []string, poolSpec model.Pool) error {
 	poolSpec.Name = context.Name
 	cephConfig := ceph.ModelPoolToCephPool(poolSpec)
-	if cephConfig.ErasureCodeProfile != "" {
+	isECPool := cephConfig.ErasureCodeProfile != ""
+	if isECPool {
 		// create a new erasure code profile for the new pool
 		if err := ceph.CreateErasureCodeProfile(context.context, context.ClusterName, poolSpec.ErasureCodedConfig, cephConfig.ErasureCodeProfile,
 			poolSpec.FailureDomain, poolSpec.CrushRoot); err != nil {
@@ -266,7 +267,16 @@ func createSimilarPools(context *Context, pools []string, poolSpec model.Pool) e
 		name := poolName(context.Name, pool)
 		if _, err := ceph.GetPoolDetails(context.context, context.ClusterName, name); err != nil {
 			cephConfig.Name = name
-			err := ceph.CreatePoolForApp(context.context, context.ClusterName, cephConfig, appName)
+			// If the ceph config has an EC profile, an EC pool must be created. Otherwise, it's necessary
+			// to create a replicated pool.
+			var err error
+			if isECPool {
+				// An EC pool backing an object store does not need to enable EC overwrites, so the pool is
+				// created with that property disabled to avoid unnecessary performance impact.
+				err = ceph.CreateECPoolForApp(context.context, context.ClusterName, cephConfig, appName, false /* enableECOverwrite */)
+			} else {
+				err = ceph.CreateReplicatedPoolForApp(context.context, context.ClusterName, cephConfig, appName)
+			}
 			if err != nil {
 				return fmt.Errorf("failed to create pool %s for object store %s", name, context.Name)
 			}