Working write policy

Author: mirzakaracic

aerospike_suite_test.go

@@ -41,7 +41,7 @@ import (
 var (
 	hosts                = flag.String("hosts", "", "Comma separated Aerospike server seed hostnames or IP addresses and ports. eg: s1:3000,s2:3000,s3:3000")
 	host                 = flag.String("h", "127.0.0.1", "Aerospike server seed hostnames or IP addresses")
-	port                 = flag.Int("p", 3000, "Aerospike server seed hostname or IP address port number.")
+	port                 = flag.Int("p", 3100, "Aerospike server seed hostname or IP address port number.")
 	user                 = flag.String("U", "", "Username.")
 	password             = flag.String("P", "", "Password.")
 	authMode             = flag.String("A", "internal", "Authentication mode: internal | external")

client.go

@@ -328,7 +328,7 @@ func (clnt *Client) PutPayload(policy *WritePolicy, key *Key, payload []byte) Er
 // handled when the record already exists.
 // If the policy is nil, the default relevant policy will be used.
 func (clnt *Client) Put(policy *WritePolicy, key *Key, binMap BinMap) Error {
-	policy = clnt.getUsableWritePolicy(policy)
+	policy = clnt.getUsableWritePolicyWithConfig(policy, applyConfigToWritePolicy)
 
 	if policy.Txn != nil {
 		if err := txnMonitor.addKey(clnt.cluster, policy, key); err != nil {
@@ -350,7 +350,7 @@ func (clnt *Client) Put(policy *WritePolicy, key *Key, binMap BinMap) Error {
 // This method avoids using the BinMap allocation and iteration and is lighter on GC.
 // If the policy is nil, the default relevant policy will be used.
 func (clnt *Client) PutBins(policy *WritePolicy, key *Key, bins ...*Bin) Error {
-	policy = clnt.getUsableWritePolicy(policy)
+	policy = clnt.getUsableWritePolicyWithConfig(policy, applyConfigToWritePolicy)
 
 	if policy.Txn != nil {
 		if err := txnMonitor.addKey(clnt.cluster, policy, key); err != nil {
@@ -2140,7 +2140,6 @@ func (clnt *Client) getUsableBatchUDFPolicy(policy *BatchUDFPolicy) *BatchUDFPol
 }
 
 func (clnt *Client) getUsableWritePolicy(policy *WritePolicy) *WritePolicy {
-
 	if policy == nil {
 		if clnt.DefaultWritePolicy != nil {
 			return clnt.DefaultWritePolicy
@@ -2150,6 +2149,10 @@ func (clnt *Client) getUsableWritePolicy(policy *WritePolicy) *WritePolicy {
 	return policy
 }
 
+func (clnt *Client) getUsableWritePolicyWithConfig(policy *WritePolicy, fn func(*WritePolicy, *DynConfig) *WritePolicy) *WritePolicy {
+	return fn(policy, clnt.dynConfig)
+}
+
 func (clnt *Client) getUsableScanPolicy(policy *ScanPolicy) *ScanPolicy {
 	if policy == nil {
 		if clnt.DefaultScanPolicy != nil {

config.go

@@ -46,8 +46,8 @@ func register(provider *dynconfig.ConfigProvider) {
 }
 
 func (dc *DynConfig) loadConfig() error {
-	configProviderMu.RLock()
-	defer configProviderMu.RUnlock()
+	configProviderMu.Lock()
+	defer configProviderMu.Unlock()
 
 	if !dc.configInitialized.Load() && configProvider != nil {
 		logger.Logger.Debug("Initializing configuration...")
@@ -70,12 +70,13 @@ func (dc *DynConfig) providerLoadConfig() {
 }
 
 func (dc *DynConfig) initConfig() {
+	dc.lock.RLock()
+	defer dc.lock.RUnlock()
+
 	loadedConfig := (*configProvider).LoadConfig()
-	dc.lock.Lock()
 	if loadedConfig != nil {
 		dc.config = loadedConfig
 	}
-	dc.lock.Unlock()
 }
 
 func (dc *DynConfig) watchConfig() {

config/dynconfig.go

@@ -1,5 +1,7 @@
 package dynconfig
 
+import "time"
+
 type ConfigProvider interface {
 	LoadConfig() *Config
 }
@@ -60,36 +62,37 @@ type Client struct {
 type Read struct {
 	ReadModeAp          *ReadModeAp `yaml:"read_mode_ap"`
 	ReadModeSc          *ReadModeSc `yaml:"read_mode_sc"`
-	ConnectTimeout      *int        `yaml:"connect_timeout"`
+	ConnectTimeout      *Duration   `yaml:"connect_timeout"`
 	FailOnFilteredOut   *bool       `yaml:"fail_on_filtered_out"`
 	Replica             *Replica    `yaml:"replica"`
-	SleepBetweenRetries *int        `yaml:"sleep_between_retries"`
-	SocketTimeout       *int        `yaml:"socket_timeout"`
+	SleepBetweenRetries *Duration   `yaml:"sleep_between_retries"`
+	SocketTimeout       *Duration   `yaml:"socket_timeout"`
 	TimeoutDelay        *int        `yaml:"timeout_delay"`
-	TotalTimeout        *int        `yaml:"total_timeout"`
+	TotalTimeout        *Duration   `yaml:"total_timeout"`
 	MaxRetries          *int        `yaml:"max_retries"`
 }
 
 type Write struct {
-	ConnectTimeout      *int     `yaml:"connect_timeout"`
-	FailOnFilteredOut   *bool    `yaml:"fail_on_filtered_out"`
-	Replica             *Replica `yaml:"replica"`
-	SendKey             *bool    `yaml:"send_key"`
-	SleepBetweenRetries *int     `yaml:"sleep_between_retries"`
-	SocketTimeout       *int     `yaml:"socket_timeout"`
-	MaxRetries          *int     `yaml:"max_retries"`
-	DurableDelete       *bool    `yaml:"durable_delete"`
+	ConnectTimeout      *Duration `yaml:"connect_timeout"`
+	FailOnFilteredOut   *bool     `yaml:"fail_on_filtered_out"`
+	Replica             *Replica  `yaml:"replica"`
+	SendKey             *bool     `yaml:"send_key"`
+	SleepBetweenRetries *Duration `yaml:"sleep_between_retries"`
+	SocketTimeout       *Duration `yaml:"socket_timeout"`
+	TotalTimeout        *Duration `yaml:"total_timeout"`
+	MaxRetries          *int      `yaml:"max_retries"`
+	DurableDelete       *bool     `yaml:"durable_delete"`
 }
 
 type Query struct {
 	ReadModeAp          *ReadModeAp `yaml:"read_mode_ap"`
 	ReadModeSc          *ReadModeSc `yaml:"read_mode_sc"`
 	ConnectTimeout      *int        `yaml:"connect_timeout"`
 	Replica             *Replica    `yaml:"replica"`
-	SleepBetweenRetries *int        `yaml:"sleep_between_retries"`
-	SocketTimeout       *int        `yaml:"socket_timeout"`
+	SleepBetweenRetries *Duration   `yaml:"sleep_between_retries"`
+	SocketTimeout       *Duration   `yaml:"socket_timeout"`
 	TimeoutDelay        *int        `yaml:"timeout_delay"`
-	TotalTimeout        *int        `yaml:"total_timeout"`
+	TotalTimeout        *Duration   `yaml:"total_timeout"`
 	MaxRetries          *int        `yaml:"max_retries"`
 	IncludeBinData      *bool       `yaml:"include_bin_data"`
 	InfoTimeout         *int        `yaml:"info_timeout"`
@@ -197,57 +200,59 @@ type Metrics struct {
 type ReadModeAp int
 
 const (
-	One ReadModeAp = iota
+	ONE ReadModeAp = iota
 	All
 )
 
 var readModeAp = map[ReadModeAp]string{
-	One: "one",
-	All: "all",
+	ONE: "ONE",
+	All: "ALL",
 }
 
 type ReadModeSc int
 
 const (
-	Session ReadModeSc = iota
-	Linearize
-	AllowReplica
-	AllowUnavailable
+	SESSION ReadModeSc = iota
+	LINEARIZE
+	ALLOWREPLICA
+	ALLOWUNAVAILABLE
 )
 
 var readModeSc = map[ReadModeSc]string{
-	Session:          "session",
-	Linearize:        "linearize",
-	AllowReplica:     "allow_replica",
-	AllowUnavailable: "allow_unavailable",
+	SESSION:          "SESSION",
+	LINEARIZE:        "LINEARIZE",
+	ALLOWREPLICA:     "ALLOW_REPLICA",
+	ALLOWUNAVAILABLE: "ALLOW_UNAVAILABLE",
 }
 
 type Replica int
 
 const (
-	Master Replica = iota
-	MasterProles
-	Sequence
-	PreferRack
+	MASTER Replica = iota
+	MASTER_PROLES
+	SEQUENCE
+	PREFER_RACK
 )
 
 var replica = map[Replica]string{
-	Master:       "master",
-	MasterProles: "master_proles",
-	Sequence:     "sequence",
-	PreferRack:   "prefer_rack",
+	MASTER:        "MASTER",
+	MASTER_PROLES: "MASTER_PROLES",
+	SEQUENCE:      "SEQUENCE",
+	PREFER_RACK:   "PREFER_RACK",
 }
 
-type Duration int
+type ExpectedDuration int
 
 const (
-	Long Duration = iota
-	Short
-	LongRelaxAp
+	LONG ExpectedDuration = iota
+	SHORT
+	LONG_RELAX_AP
 )
 
-var duration = map[Duration]string{
-	Long:        "long",
-	Short:       "short",
-	LongRelaxAp: "long_relax_ap",
+var duration = map[ExpectedDuration]string{
+	LONG:          "LONG",
+	SHORT:         "SHORT",
+	LONG_RELAX_AP: "LONG_RELAX_AP",
 }
+
+type Duration time.Duration

config/provider/yaml-config-provider.go

@@ -1,15 +1,16 @@
 package provider
 
 import (
-	"log"
+	"fmt"
 	"os"
 	"time"
 
 	dynconfig "github.com/aerospike/aerospike-client-go/v8/config"
-	"github.com/stretchr/testify/assert/yaml"
+	"github.com/aerospike/aerospike-client-go/v8/logger"
+	"gopkg.in/yaml.v3"
 )
 
-const defaultFilePath = "~/aerospikeconfig.yaml"
+const defaultFilePath = "aerospikeconfig.yaml"
 
 type YamlConfigProvider struct {
 	configFilePath string
@@ -36,25 +37,39 @@ func (yc *YamlConfigProvider) LoadConfig() *dynconfig.Config {
 	if err != nil {
 		// handle error
 	}
+	if info == nil {
+		logger.Logger.Debug("File does not exist %s . Nothing to do...", defaultFilePath)
+		return nil
+	}
 
 	modTime := info.ModTime()
 	// Compare to previously stored modTime
 	if modTime.After(yc.oldModTime) {
-		// file changed
 		yc.oldModTime = modTime
-		// re-unmarshal your struct
 
-		data, err := os.ReadFile("config.yaml")
+		data, err := os.ReadFile(defaultFilePath)
 		if err != nil {
-			log.Fatalf("failed to read config file: %v", err)
+			logger.Logger.Error("Failed to read file %s. Error: %v", defaultFilePath, err)
 		}
 		var config dynconfig.Config
 		if err := yaml.Unmarshal(data, &config); err != nil {
-			log.Fatalf("failed to unmarshal yaml: %v", err)
+			fmt.Printf("Failed to serialize file %s to object. Error: %v", defaultFilePath, err)
+			logger.Logger.Error("Failed to serialize file %s to object. Error: %v", defaultFilePath, err)
 		}
 
 		return &config
 	}
 
 	return nil
 }
+
+type Duration time.Duration
+
+func (d *Duration) UnmarshalYAML(b []byte) error {
+	var value int64
+	if err := yaml.Unmarshal(b, &value); err != nil {
+		return err
+	}
+	*d = Duration(time.Duration(value))
+	return nil
+}

go.mod

@@ -5,18 +5,20 @@ go 1.23.0
 require (
 	github.com/onsi/ginkgo/v2 v2.22.2
 	github.com/onsi/gomega v1.36.2
+	github.com/stretchr/testify v1.10.0
 	github.com/wadey/gocovmerge v0.0.0-20160331181800-b5bfa59ec0ad
 	github.com/yuin/gopher-lua v1.1.1
 	golang.org/x/sync v0.12.0
 )
 
 require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/google/pprof v0.0.0-20241210010833-40e02aabc2ad // indirect
 	github.com/kr/pretty v0.3.1 // indirect
-	github.com/stretchr/testify v1.10.0 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
 	golang.org/x/net v0.37.0 // indirect
 	golang.org/x/sys v0.31.0 // indirect
 	golang.org/x/text v0.23.0 // indirect

policy.go

@@ -16,6 +16,8 @@ package aerospike
 
 import (
 	"time"
+
+	dynconfig "github.com/aerospike/aerospike-client-go/v8/config"
 )
 
 // Policy Interface
@@ -205,3 +207,64 @@ func (p *BasePolicy) deadline() time.Time {
 func (p *BasePolicy) compress() bool {
 	return p.UseCompression
 }
+
+// copyBasePolicy creates a new BasePolicy instance and copies the values from the source BasePolicy.
+func copyBasePolicy(src *BasePolicy) *BasePolicy {
+	response := NewPolicy()
+	response.Txn = src.Txn
+	response.FilterExpression = src.FilterExpression
+	response.ReadModeAP = src.ReadModeAP
+	response.ReadModeSC = src.ReadModeSC
+	response.TotalTimeout = src.TotalTimeout
+	response.SocketTimeout = src.SocketTimeout
+	response.MaxRetries = src.MaxRetries
+	response.ReadTouchTTLPercent = src.ReadTouchTTLPercent
+	response.SleepBetweenRetries = src.SleepBetweenRetries
+	response.SleepMultiplier = src.SleepMultiplier
+	response.ExitFastOnExhaustedConnectionPool = src.ExitFastOnExhaustedConnectionPool
+	response.SendKey = src.SendKey
+	response.UseCompression = src.UseCompression
+	response.ReplicaPolicy = src.ReplicaPolicy
+
+	return response
+}
+
+// applyConfigToBasePolicy applies the dynamic configuration and generates a new policy. This function
+// will NOT override any custom settings in the BasePolicy.
+func applyConfigToBasePolicy(policy *BasePolicy, config *dynconfig.Config) *BasePolicy {
+	if config != nil && config.Dynamic != nil && config.Dynamic.Read != nil {
+		var responsePolicy *BasePolicy
+		if policy != nil {
+			// Copy the existing write policy to preserve any custom settings.
+			responsePolicy = copyBasePolicy(policy)
+		} else {
+			responsePolicy = NewPolicy()
+		}
+
+		if config.Dynamic.Read.ReadModeAp != nil {
+			responsePolicy.ReadModeAP = mapReadModeAPToReadModeAP(*config.Dynamic.Read.ReadModeAp)
+		}
+		if config.Dynamic.Read.ReadModeSc != nil {
+			responsePolicy.ReadModeSC = mapReadModeSCToReadModeSC(*config.Dynamic.Read.ReadModeSc)
+		}
+		if config.Dynamic.Read.TotalTimeout != nil {
+			responsePolicy.TotalTimeout = time.Duration(*config.Dynamic.Read.TotalTimeout)
+		}
+		if config.Dynamic.Read.SocketTimeout != nil {
+			responsePolicy.SocketTimeout = time.Duration(*config.Dynamic.Read.SocketTimeout)
+		}
+		if config.Dynamic.Read.MaxRetries != nil {
+			responsePolicy.MaxRetries = *config.Dynamic.Read.MaxRetries
+		}
+		if config.Dynamic.Read.SleepBetweenRetries != nil {
+			responsePolicy.SleepBetweenRetries = time.Duration(*config.Dynamic.Read.SleepBetweenRetries)
+		}
+		if config.Dynamic.Read.Replica != nil {
+			responsePolicy.ReplicaPolicy = mapReplicaToReplicaPolicy(*config.Dynamic.Read.Replica)
+		}
+
+		return responsePolicy
+	} else {
+		return policy
+	}
+}