CLIENT-3365 Dynamic config changes

Author: mirzakaracic

batch_command_delete.go

@@ -175,7 +175,7 @@ func (cmd *batchCommandDelete) commandType() commandType {
 }
 
 func (cmd *batchCommandDelete) executeSingle(client *Client) Error {
-	policy := cmd.batchDeletePolicy.toWritePolicy(cmd.policy)
+	policy := cmd.batchDeletePolicy.toWritePolicyWithConfig(cmd.policy, client.dynConfig)
 	for i, key := range cmd.keys {
 		res, err := client.Operate(policy, key, DeleteOp())
 		cmd.records[i].setRecord(res)

batch_command_operate.go

@@ -246,16 +246,16 @@ func (cmd *batchCommandOperate) executeSingle(client *Client) Error {
 			} else if len(ops) == 0 {
 				ops = append(ops, GetOp())
 			}
-			res, err = client.Operate(cmd.client.getUsableBatchReadPolicy(br.Policy).toWritePolicy(cmd.policy), br.Key, ops...)
+			res, err = client.Operate(cmd.client.getUsableBatchReadPolicyWithConfig(br.Policy).ToWritePolicyWithConfig(cmd.policy, client.dynConfig), br.Key, ops...)
 		case *BatchWrite:
-			policy := cmd.client.getUsableBatchWritePolicy(br.Policy).toWritePolicy(cmd.policy)
+			policy := cmd.client.getUsableBatchWritePolicyWithConfig(br.Policy).toWritePolicyWithConfig(cmd.policy, client.dynConfig)
 			policy.RespondPerEachOp = true
 			res, err = client.Operate(policy, br.Key, br.Ops...)
 		case *BatchDelete:
-			policy := cmd.client.getUsableBatchDeletePolicy(br.Policy).toWritePolicy(cmd.policy)
+			policy := cmd.client.getUsableBatchDeletePolicyWithConfig(br.Policy).toWritePolicyWithConfig(cmd.policy, client.dynConfig)
 			res, err = client.Operate(policy, br.Key, DeleteOp())
 		case *BatchUDF:
-			policy := cmd.client.getUsableBatchUDFPolicy(br.Policy).toWritePolicy(cmd.policy)
+			policy := cmd.client.getUsableBatchUDFPolicyWithConfig(br.Policy).toWritePolicyWithConfig(cmd.policy, client.dynConfig)
 			policy.RespondPerEachOp = true
 			res, err = client.execute(policy, br.Key, br.PackageName, br.FunctionName, br.FunctionArgs...)
 		}

batch_command_udf.go

@@ -185,7 +185,7 @@ func (cmd *batchCommandUDF) isRead() bool {
 
 func (cmd *batchCommandUDF) executeSingle(client *Client) Error {
 	for i, key := range cmd.keys {
-		policy := cmd.batchUDFPolicy.toWritePolicy(cmd.policy)
+		policy := cmd.batchUDFPolicy.toWritePolicyWithConfig(cmd.policy, client.dynConfig)
 		policy.RespondPerEachOp = true
 		res, err := client.execute(policy, key, cmd.packageName, cmd.functionName, cmd.args...)
 		cmd.records[i].setRecord(res)

batch_delete_policy.go

@@ -14,6 +14,8 @@
 
 package aerospike
 
+import pc "github.com/aerospike/aerospike-client-go/v8/internal/cache"
+
 // BatchDeletePolicy is used in batch delete commands.
 type BatchDeletePolicy struct {
 	// FilterExpression is optional expression filter. If FilterExpression exists and evaluates to false, the specific batch key
@@ -58,6 +60,17 @@ func NewBatchDeletePolicy() *BatchDeletePolicy {
 	}
 }
 
+func NewBatchDeletePolicyOrDefaultFromCache(dynConfig *DynConfig) *BatchDeletePolicy {
+	if dynConfig == nil {
+		return NewBatchDeletePolicy()
+	}
+
+	dynConfig.lock.RLock()
+	defer dynConfig.lock.RUnlock()
+
+	return dynConfig.mappedPolicies.Get(pc.BATCH_DELETE_POLICY).(*BatchDeletePolicy)
+}
+
 func (bdp *BatchDeletePolicy) toWritePolicy(bp *BatchPolicy) *WritePolicy {
 	wp := bp.toWritePolicy()
 
@@ -73,3 +86,115 @@ func (bdp *BatchDeletePolicy) toWritePolicy(bp *BatchPolicy) *WritePolicy {
 	}
 	return wp
 }
+
+func (bdp *BatchDeletePolicy) toWritePolicyWithConfig(bp *BatchPolicy, dynConfig *DynConfig) *WritePolicy {
+	wp := bp.toWritePolicy()
+
+	if bdp != nil {
+		if bdp.FilterExpression != nil {
+			wp.FilterExpression = bdp.FilterExpression
+		}
+		wp.CommitLevel = bdp.CommitLevel
+		wp.GenerationPolicy = bdp.GenerationPolicy
+		wp.Generation = bdp.Generation
+		wp.DurableDelete = bdp.DurableDelete
+		wp.SendKey = bdp.SendKey
+	}
+
+	// In Case dynConfig is not initialized or running return the policy before
+	// merge
+	if dynConfig == nil {
+		return wp
+	}
+
+	dynConfig.lock.RLock()
+	defer dynConfig.lock.RUnlock()
+
+	config := dynConfig.config
+	if config != nil && config.Dynamic.BatchDelete != nil {
+		if config.Dynamic.BatchDelete.DurableDelete != nil {
+			wp.DurableDelete = *config.Dynamic.BatchDelete.DurableDelete
+		}
+		if config.Dynamic.BatchDelete.SendKey != nil {
+			wp.SendKey = *config.Dynamic.BatchDelete.SendKey
+		}
+	}
+
+	return wp
+}
+
+// copyBatchDeletePolicy creates a new BasePolicy instance and copies the values from the source BatchDeletePolicy.
+func copyBatchDeletePolicy(src *BatchDeletePolicy) *BatchDeletePolicy {
+	if src == nil {
+		return nil
+	}
+
+	response := NewBatchDeletePolicy()
+
+	response.FilterExpression = src.FilterExpression
+	response.FilterExpression = src.FilterExpression
+	response.CommitLevel = src.CommitLevel
+	response.GenerationPolicy = src.GenerationPolicy
+	response.Generation = src.Generation
+	response.DurableDelete = src.DurableDelete
+	response.SendKey = src.SendKey
+
+	return response
+}
+
+// applyConfigToBatchDeletePolicy applies the dynamic configuration and generates a new policy
+func applyConfigToBatchDeletePolicy(policy *BatchDeletePolicy, dynConfig *DynConfig) *BatchDeletePolicy {
+	if dynConfig == nil {
+		return policy
+	}
+
+	config := dynConfig.config
+
+	if config == nil && !dynConfig.configInitialized.Load() {
+		// On initial load it is possible that the config is not yet loaded. This will kick things off to make sure
+		// config is loaded.
+		dynConfig.loadConfig()
+		config = dynConfig.config
+	}
+
+	dynConfig.lock.RLock()
+	defer dynConfig.lock.RUnlock()
+
+	if config != nil && config.Dynamic != nil && config.Dynamic.BatchDelete != nil {
+		// Dynamic configuration is exists for policy in question.
+		var responsePolicy *BatchDeletePolicy
+		// User has provided a custom policy. We need to apply the dynamic configuration.
+		if policy != nil {
+			// Copy the existing write policy to preserve any custom settings.
+			responsePolicy = copyBatchDeletePolicy(policy)
+			responsePolicy = mapDynamicBatchDeletePolicy(responsePolicy, dynConfig)
+
+			return responsePolicy
+		} else {
+			// Passed in policy is nil, fetch mapped default policy from cache.
+			responsePolicy = dynConfig.mappedPolicies.Get(pc.BATCH_DELETE_POLICY).(*BatchDeletePolicy)
+
+			// If we have found entry in cache no need to map again return it.
+			return responsePolicy
+		}
+	} else {
+		return policy
+	}
+}
+
+func mapDynamicBatchDeletePolicy(policy *BatchDeletePolicy, dynConfig *DynConfig) *BatchDeletePolicy {
+	if dynConfig.config == nil && dynConfig.config.Dynamic == nil {
+		return policy
+	}
+
+	if dynConfig.config.Dynamic.BatchDelete != nil {
+		if dynConfig.config.Dynamic.BatchDelete.DurableDelete != nil {
+			policy.DurableDelete = *dynConfig.config.Dynamic.BatchDelete.DurableDelete
+		}
+		if dynConfig.config.Dynamic.BatchDelete.SendKey != nil {
+			policy.SendKey = *dynConfig.config.Dynamic.BatchDelete.SendKey
+		}
+	}
+
+	return policy
+}

batch_delete_policy_config_test.go

@@ -0,0 +1,87 @@
+package aerospike
+
+import (
+	dynconfig "github.com/aerospike/aerospike-client-go/v8/config"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("ApplyConfigToBatchDeletePolicy", func() {
+
+	Context("when applying full configuration to batch delete policy", func() {
+		It("should update the policy values based on the dynamic config", func() {
+			// Create the full configuration.
+			config := &DynConfig{
+				config: &dynconfig.Config{
+					Dynamic: &dynconfig.DynamicConfig{
+						BatchDelete: &dynconfig.BatchDelete{
+							DurableDelete: func() *bool {
+								r := true
+								return &r
+							}(),
+							SendKey: func() *bool {
+								r := true
+								return &r
+							}(),
+						},
+					},
+				},
+			}
+
+			// Create an initial BatchReadPolicy.
+			policy := NewBatchDeletePolicy()
+
+			// Verify defaults.
+			Expect(policy).NotTo(BeNil())
+			Expect(policy.DurableDelete).To(BeFalse())
+			Expect(policy.SendKey).To(BeFalse())
+
+			// Apply configuration.
+			updatedPolicy := applyConfigToBatchDeletePolicy(policy, config)
+
+			// Validate applied configuration.
+			Expect(updatedPolicy).NotTo(BeNil())
+			Expect(updatedPolicy.DurableDelete).To(BeTrue())
+			Expect(updatedPolicy.DurableDelete).To(BeTrue())
+		})
+	})
+
+	Context("when applying batch read config to a write policy", func() {
+		It("should update the write policy values based on the batch delete dynamic config", func() {
+			// Create the full configuration.
+			config := &DynConfig{
+				config: &dynconfig.Config{
+					Dynamic: &dynconfig.DynamicConfig{
+						BatchDelete: &dynconfig.BatchDelete{
+							DurableDelete: func() *bool {
+								r := true
+								return &r
+							}(),
+							SendKey: func() *bool {
+								r := true
+								return &r
+							}(),
+						},
+					},
+				},
+			}
+
+			// Create an initial BatchPolicy (used for write operations).
+			batchPolicy := NewBatchPolicy()
+
+			// Verify defaults.
+			Expect(batchPolicy).NotTo(BeNil())
+			Expect(batchPolicy.ReadModeAP).To(Equal(ReadModeAPOne))
+			Expect(batchPolicy.ReadModeSC).To(Equal(ReadModeSCSession))
+			Expect(batchPolicy.ReadTouchTTLPercent).To(Equal(int32(0)))
+
+			batchDeletePolicy := NewBatchDeletePolicy()
+			updatedWritePolicy := batchDeletePolicy.toWritePolicyWithConfig(batchPolicy, config)
+
+			// Validate applied configuration.
+			Expect(updatedWritePolicy).NotTo(BeNil())
+			Expect(updatedWritePolicy.DurableDelete).To(BeTrue())
+			Expect(updatedWritePolicy.SendKey).To(BeTrue())
+		})
+	})
+})