Update state updater and Lsm holder to use conditional put using etag (…

…#348)
spirit-labs · Jan 4, 2025 · 8703a18 · 8703a18
1 parent d183c4e
commit 8703a18
Show file tree

Hide file tree

Showing 16 changed files with 329 additions and 515 deletions.
diff --git a/agent/agent.go b/agent/agent.go
@@ -47,7 +47,7 @@ type Agent struct {
 	partitionLeaders         map[string]map[int]map[int]int32
 	clusterMembershipFactory ClusterMembershipFactory
 	tableGetter              sst.TableGetter
-	authCaches *auth.UserAuthCaches
+	authCaches               *auth.UserAuthCaches
 }
 
 func NewAgent(cfg Conf, objStore objstore.Client) (*Agent, error) {

diff --git a/agent/conf.go b/agent/conf.go
@@ -109,13 +109,8 @@ func CreateConfFromCommandConf(commandConf CommandConf) (Conf, error) {
 	cfg.ClusterMembershipConfig.EvictionInterval = evictionInterval
 	// configure controller
 	cfg.ControllerConf.SSTableBucketName = dataBucketName
-	cfg.ControllerConf.ControllerMetaDataKeyPrefix = "meta-data"
+	cfg.ControllerConf.ControllerMetaDataKey = "meta-data"
 	cfg.ControllerConf.ControllerMetaDataBucketName = dataBucketName
-	// We put the controller state machine in a different bucket - as this will likely be configured with an expiry
-	// TODO move to Dynamo based state machine
-	controllerStateMachineBucketName := commandConf.ClusterName + "-controller-sm"
-	cfg.ControllerConf.ControllerStateUpdaterBucketName = controllerStateMachineBucketName
-	cfg.ControllerConf.ControllerStateUpdaterKeyPrefix = "meta-state"
 	cfg.ControllerConf.AzInfo = commandConf.Location
 	cfg.ControllerConf.LsmConf.SSTableBucketName = dataBucketName
 	// configure compaction workers

diff --git a/cluster/clustered_data.go b/cluster/clustered_data.go
@@ -45,9 +45,9 @@ type ClusteredData struct {
 	dataKeyPrefix  string
 	stateMachine   *StateUpdater
 	objStoreClient objstore.Client
-	epoch      uint64
-	opts       ClusteredDataConf
-	readyState clusteredDataState
+	epoch          uint64
+	opts           ClusteredDataConf
+	readyState     clusteredDataState
 	stopping       atomic.Bool
 	logPrefix      string
 }
@@ -66,8 +66,8 @@ func NewClusteredData(stateMachineBucketName string, stateMachineKeyPrefix strin
 		objStoreClient: objStoreClient,
 		dataBucketName: dataBucketName,
 		dataKeyPrefix:  dataKeyPrefix,
-		stateMachine: NewStateUpdator(stateMachineBucketName, stateMachineKeyPrefix, objStoreClient,
-			StateUpdatorOpts{}),
+		stateMachine: NewStateUpdater(stateMachineBucketName, stateMachineKeyPrefix, objStoreClient,
+			StateUpdaterOpts{}),
 		opts:       opts,
 		readyState: clusteredDataStateReady,
 		logPrefix:  fmt.Sprintf("clustered data(%s / %s) -", dataBucketName, stateMachineBucketName),

diff --git a/cluster/membership.go b/cluster/membership.go
@@ -51,7 +51,7 @@ func NewMembership(cfg MembershipConf, data []byte, objStoreClient objstore.Clie
 	return &Membership{
 		id:                   -1,
 		data:                 data,
-		stateUpdater:         NewStateUpdator(cfg.BucketName, cfg.KeyPrefix, objStoreClient, StateUpdatorOpts{}),
+		stateUpdater:         NewStateUpdater(cfg.BucketName, cfg.KeyPrefix, objStoreClient, StateUpdaterOpts{}),
 		updateInterval:       cfg.UpdateInterval,
 		evictionDuration:     cfg.EvictionInterval,
 		stateChangedCallback: stateChangedCallback,