cassandra

Documentation

Index

• Variables
• func CreateCassandraKeyspace(s gocql.Session, keyspace string, replicas int, overwrite bool, ...) (err error)
• func DropCassandraKeyspace(s gocql.Session, keyspace string, logger log.Logger) (err error)
• func GetTaskTTL(task *persistencespb.TaskInfo) int64
• func NewHistoryV2PersistenceFromSession(session gocql.Session, logger log.Logger) p.HistoryStore
• func NewWorkflowExecutionPersistence(shardID int32, session gocql.Session, logger log.Logger) (p.ExecutionStore, error)
• func VerifyCompatibleVersion(cfg config.Persistence, r resolver.ServiceResolver, checkVisibility bool) error
• type Factory
  • func NewFactory(cfg config.Cassandra, r resolver.ServiceResolver, clusterName string, ...) *Factory
  • func (f *Factory) Close()
  • func (f *Factory) NewClusterMetadataStore() (p.ClusterMetadataStore, error)
  • func (f *Factory) NewExecutionStore(shardID int32) (p.ExecutionStore, error)
  • func (f *Factory) NewHistoryStore() (p.HistoryStore, error)
  • func (f *Factory) NewMetadataStore() (p.MetadataStore, error)
  • func (f *Factory) NewQueue(queueType p.QueueType) (p.Queue, error)
  • func (f *Factory) NewShardStore() (p.ShardStore, error)
  • func (f *Factory) NewTaskStore() (p.TaskStore, error)
  • func (f *Factory) NewVisibilityStore() (p.VisibilityStore, error)
• type FieldNotFoundError
  • func (f FieldNotFoundError) Error() string
• type PersistedTypeMismatchError
  • func (f PersistedTypeMismatchError) Error() string
• type SchemaVersionReader
  • func NewSchemaVersionReader(session gocql.Session) *SchemaVersionReader
  • func (svr *SchemaVersionReader) ReadSchemaVersion(keyspace string) (string, error)
• type TestCluster
  • func NewTestCluster(keyspace, username, password, host string, port int, schemaDir string, ...) *TestCluster
  • func (s *TestCluster) Config() config.Persistence
  • func (s *TestCluster) CreateDatabase()
  • func (s *TestCluster) CreateSession(keyspace string)
  • func (s *TestCluster) DatabaseName() string
  • func (s *TestCluster) DropDatabase()
  • func (s *TestCluster) LoadSchema(schemaFile string)
  • func (s *TestCluster) SetupTestDatabase()
  • func (s *TestCluster) TearDownTestDatabase()

Variables

var (
	ErrGetSchemaVersion = errors.New("failed to get current schema version from cassandra")
)

Functions

func CreateCassandraKeyspace

func CreateCassandraKeyspace(s gocql.Session, keyspace string, replicas int, overwrite bool, logger log.Logger) (err error)

CreateCassandraKeyspace creates the keyspace using this session for given replica count

func DropCassandraKeyspace

func DropCassandraKeyspace(s gocql.Session, keyspace string, logger log.Logger) (err error)

DropCassandraKeyspace drops the given keyspace, if it exists

func GetTaskTTL

func GetTaskTTL(task *persistencespb.TaskInfo) int64

func NewHistoryV2PersistenceFromSession

func NewHistoryV2PersistenceFromSession(
	session gocql.Session,
	logger log.Logger,
) p.HistoryStore

NewHistoryV2PersistenceFromSession returns new HistoryStore

func NewWorkflowExecutionPersistence

func NewWorkflowExecutionPersistence(
	shardID int32,
	session gocql.Session,
	logger log.Logger,
) (p.ExecutionStore, error)

NewWorkflowExecutionPersistence is used to create an instance of workflowExecutionManager implementation

func VerifyCompatibleVersion

func VerifyCompatibleVersion(
	cfg config.Persistence,
	r resolver.ServiceResolver,
	checkVisibility bool,
) error

VerifyCompatibleVersion ensures that the installed version of temporal and visibility keyspaces is greater than or equal to the expected version. In most cases, the versions should match. However if after a schema upgrade there is a code rollback, the code version (expected version) would fall lower than the actual version in cassandra.

Types

type Factory

type Factory struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Factory vends datastore implementations backed by cassandra

func NewFactory

func NewFactory(
	cfg config.Cassandra,
	r resolver.ServiceResolver,
	clusterName string,
	logger log.Logger,
) *Factory

NewFactory returns an instance of a factory object which can be used to create datastores that are backed by cassandra

func (*Factory) Close

func (f *Factory) Close()

Close closes the factory

func (*Factory) NewClusterMetadataStore

func (f *Factory) NewClusterMetadataStore() (p.ClusterMetadataStore, error)

NewClusterMetadataStore returns a metadata store

func (*Factory) NewExecutionStore

func (f *Factory) NewExecutionStore(shardID int32) (p.ExecutionStore, error)

NewExecutionStore returns an ExecutionStore for a given shardID

func (*Factory) NewHistoryStore

func (f *Factory) NewHistoryStore() (p.HistoryStore, error)

NewHistoryStore returns a new history store

func (*Factory) NewMetadataStore

func (f *Factory) NewMetadataStore() (p.MetadataStore, error)

NewMetadataStore returns a metadata store that understands only v2

func (*Factory) NewQueue

func (f *Factory) NewQueue(queueType p.QueueType) (p.Queue, error)

NewQueue returns a new queue backed by cassandra

func (*Factory) NewShardStore

func (f *Factory) NewShardStore() (p.ShardStore, error)

NewShardStore returns a new shard store

func (*Factory) NewTaskStore

func (f *Factory) NewTaskStore() (p.TaskStore, error)

NewTaskStore returns a new task store

func (*Factory) NewVisibilityStore

func (f *Factory) NewVisibilityStore() (p.VisibilityStore, error)

NewVisibilityStore returns a visibility store

type FieldNotFoundError

type FieldNotFoundError struct {
	Msg string
}

FieldNotFoundError is an error type returned when an untyped query return does not contain the expected fields.

func (FieldNotFoundError) Error

func (f FieldNotFoundError) Error() string

type PersistedTypeMismatchError

type PersistedTypeMismatchError struct {
	Msg string
}

PersistedTypeMismatchError is an error type returned when a persisted cassandra value does not match the expected type.

func (PersistedTypeMismatchError) Error

func (f PersistedTypeMismatchError) Error() string

type SchemaVersionReader

type SchemaVersionReader struct {
	// contains filtered or unexported fields
}

func NewSchemaVersionReader

func NewSchemaVersionReader(session gocql.Session) *SchemaVersionReader

func (*SchemaVersionReader) ReadSchemaVersion

func (svr *SchemaVersionReader) ReadSchemaVersion(keyspace string) (string, error)

ReadSchemaVersion returns the current schema version for the Keyspace

type TestCluster

type TestCluster struct {
	// contains filtered or unexported fields
}

TestCluster allows executing cassandra operations in testing.

func NewTestCluster

func NewTestCluster(keyspace, username, password, host string, port int, schemaDir string, logger log.Logger) *TestCluster

NewTestCluster returns a new cassandra test cluster

func (*TestCluster) Config

func (s *TestCluster) Config() config.Persistence

Config returns the persistence config for connecting to this test cluster

func (*TestCluster) CreateDatabase

func (s *TestCluster) CreateDatabase()

CreateDatabase from PersistenceTestCluster interface

func (*TestCluster) CreateSession

func (s *TestCluster) CreateSession(
	keyspace string,
)

CreateSession from PersistenceTestCluster interface

func (*TestCluster) DatabaseName

func (s *TestCluster) DatabaseName() string

DatabaseName from PersistenceTestCluster interface

func (*TestCluster) DropDatabase

func (s *TestCluster) DropDatabase()

DropDatabase from PersistenceTestCluster interface

func (*TestCluster) LoadSchema

func (s *TestCluster) LoadSchema(schemaFile string)

LoadSchema from PersistenceTestCluster interface

func (*TestCluster) SetupTestDatabase

func (s *TestCluster) SetupTestDatabase()

SetupTestDatabase from PersistenceTestCluster interface

func (*TestCluster) TearDownTestDatabase

func (s *TestCluster) TearDownTestDatabase()

TearDownTestDatabase from PersistenceTestCluster interface