Expose cassandra consistency and serial consistency settings in configuration (#533)

mastermanu · web-flow · commit 89c51ccd5841 · 2020-07-13T15:51:35.000-07:00
This change enables an end-user to configure consistency and serial consistency for their Cassandra Datastores.

The behavior is as follows:
If the user chooses not to set this at all, we default to local_quorum and local_serial respectively.
Otherwise, the user can choose to set a "default" that applies to ALL stores.

We expect the string values as understood by the gocql library and will validate this on server startup.
diff --git a/common/payloads/payloads.go b/common/payloads/payloads.go
@@ -65,7 +65,7 @@ func Decode(ps *commonpb.Payloads, valuePtr ...interface{}) error {
 func ToString(ps *commonpb.Payloads) string {
 	str, err := toString(ps)
 	if err != nil {
-		return fmt.Sprintf("Unable to decode heartbeat details. %+v", err)
+		return fmt.Sprintf("Unable to decode payloads. %+v", err)
 	}
 	return str
 }
diff --git a/common/persistence/cassandra/cassandraClusterMetadata.go b/common/persistence/cassandra/cassandraClusterMetadata.go
@@ -29,7 +29,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/gocql/gocql"
 	"github.com/pborman/uuid"
 	"go.temporal.io/api/serviceerror"
 
@@ -85,12 +84,12 @@ type (
 
 var _ p.ClusterMetadataStore = (*cassandraClusterMetadata)(nil)
 
-// newMetadataPersistenceV2 is used to create an instance of HistoryManager implementation
+// newClusterMetadataInstance is used to create an instance of ClusterMetadataStore implementation
 func newClusterMetadataInstance(cfg config.Cassandra, logger log.Logger) (p.ClusterMetadataStore, error) {
 	cluster := cassandra.NewCassandraCluster(cfg)
 	cluster.ProtoVersion = cassandraProtoVersion
-	cluster.Consistency = gocql.LocalQuorum
-	cluster.SerialConsistency = gocql.LocalSerial
+	cluster.Consistency = cfg.Consistency.GetConsistency()
+	cluster.SerialConsistency = cfg.Consistency.GetSerialConsistency()
 	cluster.Timeout = defaultSessionTimeout
 
 	session, err := cluster.CreateSession()
diff --git a/common/persistence/cassandra/cassandraHistoryPersistence.go b/common/persistence/cassandra/cassandraHistoryPersistence.go
@@ -79,16 +79,16 @@ func NewHistoryV2PersistenceFromSession(
 	return &cassandraHistoryV2Persistence{cassandraStore: cassandraStore{session: session, logger: logger}}
 }
 
-// newHistoryPersistence is used to create an instance of HistoryManager implementation
+// newHistoryV2Persistence is used to create an instance of HistoryManager implementation
 func newHistoryV2Persistence(
 	cfg config.Cassandra,
 	logger log.Logger,
 ) (p.HistoryStore, error) {
 
 	cluster := cassandra.NewCassandraCluster(cfg)
 	cluster.ProtoVersion = cassandraProtoVersion
-	cluster.Consistency = gocql.LocalQuorum
-	cluster.SerialConsistency = gocql.LocalSerial
+	cluster.Consistency = cfg.Consistency.GetConsistency()
+	cluster.SerialConsistency = cfg.Consistency.GetSerialConsistency()
 	cluster.Timeout = defaultSessionTimeout
 	session, err := cluster.CreateSession()
 	if err != nil {
diff --git a/common/persistence/cassandra/cassandraMetadataPersistenceV2.go b/common/persistence/cassandra/cassandraMetadataPersistenceV2.go
@@ -99,12 +99,12 @@ type (
 	}
 )
 
-// newMetadataPersistenceV2 is used to create an instance of HistoryManager implementation
+// newMetadataPersistenceV2 is used to create an instance of the Namespace MetadataStore implementation
 func newMetadataPersistenceV2(cfg config.Cassandra, currentClusterName string, logger log.Logger) (p.MetadataStore, error) {
 	cluster := cassandra.NewCassandraCluster(cfg)
 	cluster.ProtoVersion = cassandraProtoVersion
-	cluster.Consistency = gocql.LocalQuorum
-	cluster.SerialConsistency = gocql.LocalSerial
+	cluster.Consistency = cfg.Consistency.GetConsistency()
+	cluster.SerialConsistency = cfg.Consistency.GetSerialConsistency()
 	cluster.Timeout = defaultSessionTimeout
 
 	session, err := cluster.CreateSession()
diff --git a/common/persistence/cassandra/cassandraPersistence.go b/common/persistence/cassandra/cassandraPersistence.go
@@ -726,8 +726,8 @@ var _ p.ExecutionStore = (*cassandraPersistence)(nil)
 func newShardPersistence(cfg config.Cassandra, clusterName string, logger log.Logger) (p.ShardStore, error) {
 	cluster := cassandra.NewCassandraCluster(cfg)
 	cluster.ProtoVersion = cassandraProtoVersion
-	cluster.Consistency = gocql.LocalQuorum
-	cluster.SerialConsistency = gocql.LocalSerial
+	cluster.Consistency = cfg.Consistency.GetConsistency()
+	cluster.SerialConsistency = cfg.Consistency.GetSerialConsistency()
 	cluster.Timeout = defaultSessionTimeout
 
 	session, err := cluster.CreateSession()
@@ -755,8 +755,8 @@ func NewWorkflowExecutionPersistence(
 func newTaskPersistence(cfg config.Cassandra, logger log.Logger) (p.TaskStore, error) {
 	cluster := cassandra.NewCassandraCluster(cfg)
 	cluster.ProtoVersion = cassandraProtoVersion
-	cluster.Consistency = gocql.LocalQuorum
-	cluster.SerialConsistency = gocql.LocalSerial
+	cluster.Consistency = cfg.Consistency.GetConsistency()
+	cluster.SerialConsistency = cfg.Consistency.GetSerialConsistency()
 	cluster.Timeout = defaultSessionTimeout
 	session, err := cluster.CreateSession()
 	if err != nil {
@@ -1970,7 +1970,7 @@ func (d *cassandraPersistence) GetTransferTask(request *p.GetTransferTaskRequest
 
 func (d *cassandraPersistence) GetTransferTasks(request *p.GetTransferTasksRequest) (*p.GetTransferTasksResponse, error) {
 
-	// Reading transfer tasks need to be quorum level consistent, otherwise we could loose task
+	// Reading transfer tasks need to be quorum level consistent, otherwise we could lose task
 	query := d.session.Query(templateGetTransferTasksQuery,
 		d.shardID,
 		rowTypeTransferTask,
@@ -2041,7 +2041,7 @@ func (d *cassandraPersistence) GetReplicationTasks(
 	request *p.GetReplicationTasksRequest,
 ) (*p.GetReplicationTasksResponse, error) {
 
-	// Reading replication tasks need to be quorum level consistent, otherwise we could loose task
+	// Reading replication tasks need to be quorum level consistent, otherwise we could lose task
 	query := d.session.Query(templateGetReplicationTasksQuery,
 		d.shardID,
 		rowTypeReplicationTask,
@@ -2540,7 +2540,7 @@ func (d *cassandraPersistence) GetTasks(request *p.GetTasksRequest) (*p.GetTasks
 		return &p.GetTasksResponse{}, nil
 	}
 
-	// Reading taskqueue tasks need to be quorum level consistent, otherwise we could loose task
+	// Reading taskqueue tasks need to be quorum level consistent, otherwise we could lose tasks
 	query := d.session.Query(templateGetTasksQuery,
 		request.NamespaceID,
 		request.TaskQueue,
@@ -2671,7 +2671,7 @@ func (d *cassandraPersistence) GetTimerTask(request *p.GetTimerTaskRequest) (*p.
 
 func (d *cassandraPersistence) GetTimerIndexTasks(request *p.GetTimerIndexTasksRequest) (*p.GetTimerIndexTasksResponse,
 	error) {
-	// Reading timer tasks need to be quorum level consistent, otherwise we could loose task
+	// Reading timer tasks need to be quorum level consistent, otherwise we could lose tasks
 	minTimestamp := p.UnixNanoToDBTimestamp(request.MinTimestamp.UnixNano())
 	maxTimestamp := p.UnixNanoToDBTimestamp(request.MaxTimestamp.UnixNano())
 	query := d.session.Query(templateGetTimerTasksQuery,
@@ -2743,7 +2743,7 @@ func (d *cassandraPersistence) PutReplicationTaskToDLQ(request *p.PutReplication
 func (d *cassandraPersistence) GetReplicationTasksFromDLQ(
 	request *p.GetReplicationTasksFromDLQRequest,
 ) (*p.GetReplicationTasksFromDLQResponse, error) {
-	// Reading replication tasks need to be quorum level consistent, otherwise we could loose task
+	// Reading replication tasks need to be quorum level consistent, otherwise we could lose tasks
 	query := d.session.Query(templateGetReplicationTasksQuery,
 		d.shardID,
 		rowTypeDLQ,
diff --git a/common/persistence/cassandra/cassandraQueue.go b/common/persistence/cassandra/cassandraQueue.go
@@ -80,8 +80,8 @@ func newQueue(
 ) (persistence.Queue, error) {
 	cluster := cassandra.NewCassandraCluster(cfg)
 	cluster.ProtoVersion = cassandraProtoVersion
-	cluster.Consistency = gocql.LocalQuorum
-	cluster.SerialConsistency = gocql.LocalSerial
+	cluster.Consistency = cfg.Consistency.GetConsistency()
+	cluster.SerialConsistency = cfg.Consistency.GetSerialConsistency()
 	cluster.Timeout = defaultSessionTimeout
 
 	session, err := cluster.CreateSession()
@@ -200,7 +200,7 @@ func (q *cassandraQueue) ReadMessages(
 	lastMessageID int64,
 	maxCount int,
 ) ([]*persistence.QueueMessage, error) {
-	// Reading replication tasks need to be quorum level consistent, otherwise we could loose task
+	// Reading replication tasks need to be quorum level consistent, otherwise we could lose tasks
 	query := q.session.Query(templateGetMessagesQuery,
 		q.queueType,
 		lastMessageID,
@@ -234,7 +234,7 @@ func (q *cassandraQueue) ReadMessagesFromDLQ(
 	pageSize int,
 	pageToken []byte,
 ) ([]*persistence.QueueMessage, []byte, error) {
-	// Reading replication tasks need to be quorum level consistent, otherwise we could loose task
+	// Reading replication tasks need to be quorum level consistent, otherwise we could lose tasks
 	// Use negative queue type as the dlq type
 	query := q.session.Query(templateGetMessagesFromDLQQuery,
 		q.getDLQTypeFromQueueType(),
diff --git a/common/persistence/cassandra/cassandraVisibilityPersistence.go b/common/persistence/cassandra/cassandraVisibilityPersistence.go
@@ -152,8 +152,8 @@ type (
 func newVisibilityPersistence(cfg config.Cassandra, logger log.Logger) (p.VisibilityStore, error) {
 	cluster := cassandra.NewCassandraCluster(cfg)
 	cluster.ProtoVersion = cassandraProtoVersion
-	cluster.Consistency = gocql.LocalQuorum
-	cluster.SerialConsistency = gocql.LocalSerial
+	cluster.Consistency = cfg.Consistency.GetConsistency()
+	cluster.SerialConsistency = cfg.Consistency.GetSerialConsistency()
 	cluster.Timeout = defaultSessionTimeout
 
 	session, err := cluster.CreateSession()
diff --git a/common/persistence/cassandra/cassandraVisibilityPersistenceV2.go b/common/persistence/cassandra/cassandraVisibilityPersistenceV2.go
@@ -81,8 +81,8 @@ type (
 func NewVisibilityPersistenceV2(persistence p.VisibilityStore, cfg *config.Cassandra, logger log.Logger) (p.VisibilityStore, error) {
 	cluster := cassandra.NewCassandraCluster(*cfg)
 	cluster.ProtoVersion = cassandraProtoVersion
-	cluster.Consistency = gocql.LocalQuorum
-	cluster.SerialConsistency = gocql.LocalSerial
+	cluster.Consistency = cfg.Consistency.GetConsistency()
+	cluster.SerialConsistency = cfg.Consistency.GetSerialConsistency()
 	cluster.Timeout = defaultSessionTimeout
 
 	session, err := cluster.CreateSession()
diff --git a/common/persistence/cassandra/factory.go b/common/persistence/cassandra/factory.go
@@ -138,8 +138,8 @@ func (f *Factory) executionStoreFactory() (*executionStoreFactory, error) {
 func newExecutionStoreFactory(cfg config.Cassandra, logger log.Logger) (*executionStoreFactory, error) {
 	cluster := cassandra.NewCassandraCluster(cfg)
 	cluster.ProtoVersion = cassandraProtoVersion
-	cluster.Consistency = gocql.LocalQuorum
-	cluster.SerialConsistency = gocql.LocalSerial
+	cluster.Consistency = cfg.Consistency.GetConsistency()
+	cluster.SerialConsistency = cfg.Consistency.GetSerialConsistency()
 	cluster.Timeout = defaultSessionTimeout
 	session, err := cluster.CreateSession()
 	if err != nil {
diff --git a/common/service/config/config.go b/common/service/config/config.go
@@ -238,6 +238,23 @@ type (
 		MaxConns int `yaml:"maxConns"`
 		// TLS configuration
 		TLS *auth.TLS `yaml:"tls"`
+		// Consistency configuration (defaults to LOCAL_QUORUM / LOCAL_SERIAL for all stores if this field not set)
+		Consistency *CassandraStoreConsistency `yaml:"consistency"`
+	}
+
+	// CassandraStoreConsistency enables you to set the consistency settings for each Cassandra Persistence Store for Temporal
+	CassandraStoreConsistency struct {
+		// Default defines the consistency level for ALL stores.
+		// Defaults to LOCAL_QUORUM and LOCAL_SERIAL if not set
+		Default *CassandraConsistencySettings `yaml:"default"`
+	}
+
+	// CassandraConsistencySettings sets the default consistency level for regular & serial queries to Cassandra.
+	CassandraConsistencySettings struct {
+		// Consistency sets the default consistency level. Values identical to gocql Consistency values. (defaults to LOCAL_QUORUM if not set).
+		Consistency string `yaml:"consistency"`
+		// SerialConsistency sets the consistency for the serial prtion of queries. Values identical to gocql SerialConsistency values. (defaults to LOCAL_SERIAL if not set)
+		SerialConsistency string `yaml:"serialConsistency"`
 	}
 
 	// SQL is the configuration for connecting to a SQL backed datastore
diff --git a/common/service/config/persistence.go b/common/service/config/persistence.go
@@ -24,7 +24,13 @@
 
 package config
 
-import "fmt"
+import (
+	"fmt"
+	"reflect"
+	"strings"
+
+	"github.com/gocql/gocql"
+)
 
 const (
 	// StoreTypeSQL refers to sql based storage as persistence store
@@ -58,6 +64,11 @@ func (c *Persistence) Validate() error {
 		if ds.SQL != nil && ds.SQL.NumShards == 0 {
 			ds.SQL.NumShards = 1
 		}
+		if ds.Cassandra != nil {
+			if err := ds.Cassandra.validate(); err != nil {
+				return err
+			}
+		}
 	}
 	return nil
 }
@@ -66,3 +77,89 @@ func (c *Persistence) Validate() error {
 func (c *Persistence) IsAdvancedVisibilityConfigExist() bool {
 	return len(c.AdvancedVisibilityStore) != 0
 }
+
+// GetConsistency returns the gosql.Consistency setting from the configuration for the given store type
+func (c *CassandraStoreConsistency) GetConsistency() gocql.Consistency {
+	return gocql.ParseConsistency(c.getConsistencySettings().Consistency)
+}
+
+// GetSerialConsistency returns the gosql.SerialConsistency setting from the configuration for the store
+func (c *CassandraStoreConsistency) GetSerialConsistency() gocql.SerialConsistency {
+	res, err := parseSerialConsistency(c.getConsistencySettings().SerialConsistency)
+	if err != nil {
+		panic(fmt.Sprintf("unable to decode cassandra serial consistency: %v", err))
+	}
+	return res
+}
+
+func (c *CassandraStoreConsistency) getConsistencySettings() *CassandraConsistencySettings {
+	return ensureStoreConsistencyNotNil(c).Default
+}
+
+func ensureStoreConsistencyNotNil(c *CassandraStoreConsistency) *CassandraStoreConsistency {
+	if c == nil {
+		c = &CassandraStoreConsistency{}
+	}
+	if c.Default == nil {
+		c.Default = &CassandraConsistencySettings{}
+	}
+	if c.Default.Consistency == "" {
+		c.Default.Consistency = "LOCAL_QUORUM"
+	}
+	if c.Default.SerialConsistency == "" {
+		c.Default.SerialConsistency = "LOCAL_SERIAL"
+	}
+
+	return c
+}
+
+func (c *Cassandra) validate() error {
+	return c.Consistency.validate()
+}
+
+func (c *CassandraStoreConsistency) validate() error {
+	if c == nil {
+		return nil
+	}
+
+	v := reflect.ValueOf(*c)
+
+	for i := 0; i < v.NumField(); i++ {
+		s, ok := v.Field(i).Interface().(*CassandraConsistencySettings)
+		if ok {
+			if err := s.validate(); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}
+
+func (c *CassandraConsistencySettings) validate() error {
+	if c == nil {
+		return nil
+	}
+
+	if c.Consistency != "" {
+		_, err := gocql.ParseConsistencyWrapper(c.Consistency)
+		if err != nil {
+			return fmt.Errorf("bad cassandra consistency: %v", err)
+		}
+	}
+
+	if c.SerialConsistency != "" {
+		_, err := parseSerialConsistency(c.SerialConsistency)
+		if err != nil {
+			return fmt.Errorf("bad cassandra serial consistency: %v", err)
+		}
+	}
+
+	return nil
+}
+
+func parseSerialConsistency(serialConsistency string) (gocql.SerialConsistency, error) {
+	var s gocql.SerialConsistency
+	err := s.UnmarshalText([]byte(strings.ToUpper(serialConsistency)))
+	return s, err
+}
diff --git a/common/service/config/persistence_test.go b/common/service/config/persistence_test.go
diff --git a/host/testcluster.go b/host/testcluster.go
diff --git a/service/history/shardController.go b/service/history/shardController.go
diff --git a/service/history/timerQueueActiveTaskExecutor.go b/service/history/timerQueueActiveTaskExecutor.go

Original file line number	Diff line number	Diff line change
`@@ -65,7 +65,7 @@ func Decode(ps *commonpb.Payloads, valuePtr ...interface{}) error {`
`65`	`65`	`func ToString(ps *commonpb.Payloads) string {`
`66`	`66`	`str, err := toString(ps)`
`67`	`67`	`if err != nil {`
`68`		`- return fmt.Sprintf("Unable to decode heartbeat details. %+v", err)`
	`68`	`+ return fmt.Sprintf("Unable to decode payloads. %+v", err)`
`69`	`69`	`}`
`70`	`70`	`return str`
`71`	`71`	`}`
Original file line number	Diff line number	Diff line change
`@@ -99,12 +99,12 @@ type (`
`99`	`99`	`}`
`100`	`100`	`)`
`101`	`101`
`102`		`-// newMetadataPersistenceV2 is used to create an instance of HistoryManager implementation`
	`102`	`+// newMetadataPersistenceV2 is used to create an instance of the Namespace MetadataStore implementation`
`103`	`103`	`func newMetadataPersistenceV2(cfg config.Cassandra, currentClusterName string, logger log.Logger) (p.MetadataStore, error) {`
`104`	`104`	`cluster := cassandra.NewCassandraCluster(cfg)`
`105`	`105`	`cluster.ProtoVersion = cassandraProtoVersion`
`106`		`- cluster.Consistency = gocql.LocalQuorum`
`107`		`- cluster.SerialConsistency = gocql.LocalSerial`
	`106`	`+ cluster.Consistency = cfg.Consistency.GetConsistency()`
	`107`	`+ cluster.SerialConsistency = cfg.Consistency.GetSerialConsistency()`
`108`	`108`	`cluster.Timeout = defaultSessionTimeout`
`109`	`109`
`110`	`110`	`session, err := cluster.CreateSession()`