cassandra

Documentation

Index

• Constants
• func Fixtures() []testutils.Fixture
• func NewTableClient(_ context.Context, cfg Config, registerer prometheus.Registerer) (index.TableClient, error)
• type Config
  • func (cfg *Config) RegisterFlags(f *flag.FlagSet)
  • func (cfg *Config) Validate() error
• type CustomPasswordAuthenticator
  • func (p CustomPasswordAuthenticator) Challenge(req []byte) ([]byte, gocql.Authenticator, error)
  • func (p CustomPasswordAuthenticator) Success(_ []byte) error
• type ObjectClient
  • func NewObjectClient(cfg Config, schemaCfg config.SchemaConfig, registerer prometheus.Registerer, ...) (*ObjectClient, error)
  • func (s *ObjectClient) DeleteChunk(ctx context.Context, userID, chunkID string) error
  • func (s *ObjectClient) GetChunks(ctx context.Context, input []chunk.Chunk) ([]chunk.Chunk, error)
  • func (s *ObjectClient) IsChunkNotFoundErr(_ error) bool
  • func (s *ObjectClient) IsRetryableErr(_ error) bool
  • func (s *ObjectClient) PutChunks(ctx context.Context, chunks []chunk.Chunk) error
  • func (s *ObjectClient) Stop()
• type StorageClient
  • func NewStorageClient(cfg Config, schemaCfg config.SchemaConfig, registerer prometheus.Registerer) (*StorageClient, error)
  • func (s *StorageClient) BatchWrite(ctx context.Context, batch index.WriteBatch) error
  • func (s *StorageClient) GetReadSession() *gocql.Session
  • func (s *StorageClient) NewWriteBatch() index.WriteBatch
  • func (s *StorageClient) QueryPages(ctx context.Context, queries []index.Query, callback index.QueryPagesCallback) error
  • func (s *StorageClient) Stop()

Constants

const (
	HostPolicyRoundRobin = "round-robin"
	HostPolicyTokenAware = "token-aware"
)

Functions

func Fixtures

func Fixtures() []testutils.Fixture

Fixtures for unit testing Cassandra integration.

func NewTableClient

func NewTableClient(_ context.Context, cfg Config, registerer prometheus.Registerer) (index.TableClient, error)

NewTableClient returns a new TableClient.

Types

type Config

type Config struct {
	Addresses                string              `yaml:"addresses"`
	Port                     int                 `yaml:"port"`
	Keyspace                 string              `yaml:"keyspace"`
	Consistency              string              `yaml:"consistency"`
	ReplicationFactor        int                 `yaml:"replication_factor"`
	DisableInitialHostLookup bool                `yaml:"disable_initial_host_lookup"`
	SSL                      bool                `yaml:"SSL"`
	HostVerification         bool                `yaml:"host_verification"`
	HostSelectionPolicy      string              `yaml:"host_selection_policy"`
	CAPath                   string              `yaml:"CA_path"`
	CertPath                 string              `yaml:"tls_cert_path"`
	KeyPath                  string              `yaml:"tls_key_path"`
	Auth                     bool                `yaml:"auth"`
	Username                 string              `yaml:"username"`
	Password                 flagext.Secret      `yaml:"password"`
	PasswordFile             string              `yaml:"password_file"`
	CustomAuthenticators     flagext.StringSlice `yaml:"custom_authenticators"`
	Timeout                  time.Duration       `yaml:"timeout"`
	ConnectTimeout           time.Duration       `yaml:"connect_timeout"`
	ReconnectInterval        time.Duration       `yaml:"reconnect_interval"`
	Retries                  int                 `yaml:"max_retries"`
	MaxBackoff               time.Duration       `yaml:"retry_max_backoff"`
	MinBackoff               time.Duration       `yaml:"retry_min_backoff"`
	QueryConcurrency         int                 `yaml:"query_concurrency"`
	NumConnections           int                 `yaml:"num_connections"`
	ConvictHosts             bool                `yaml:"convict_hosts_on_failure"`
	TableOptions             string              `yaml:"table_options"`
}

Config for a StorageClient

func (*Config) RegisterFlags

func (cfg *Config) RegisterFlags(f *flag.FlagSet)

RegisterFlags adds the flags required to config this to the given FlagSet

func (*Config) Validate

func (cfg *Config) Validate() error

nolint: revive

type CustomPasswordAuthenticator

type CustomPasswordAuthenticator struct {
	ApprovedAuthenticators []string
	Username               string
	Password               string
}

CustomPasswordAuthenticator provides the default behaviour for Username/Password authentication with Cassandra while allowing users to specify a non-default Authenticator to accept.

func (CustomPasswordAuthenticator) Challenge

func (p CustomPasswordAuthenticator) Challenge(req []byte) ([]byte, gocql.Authenticator, error)

Challenge verifies the name of the authenticator and formats the provided username and password into a response

func (CustomPasswordAuthenticator) Success

func (p CustomPasswordAuthenticator) Success(_ []byte) error

Success returns nil by default, identical to the default PasswordAuthenticator

type ObjectClient

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

ObjectClient implements chunk.ObjectClient for Cassandra.

func NewObjectClient

func NewObjectClient(cfg Config, schemaCfg config.SchemaConfig, registerer prometheus.Registerer, maxGetParallel int) (*ObjectClient, error)

NewObjectClient returns a new ObjectClient.

func (*ObjectClient) DeleteChunk

func (s *ObjectClient) DeleteChunk(ctx context.Context, userID, chunkID string) error

func (*ObjectClient) GetChunks

func (s *ObjectClient) GetChunks(ctx context.Context, input []chunk.Chunk) ([]chunk.Chunk, error)

GetChunks implements chunk.ObjectClient.

func (*ObjectClient) IsChunkNotFoundErr

func (s *ObjectClient) IsChunkNotFoundErr(_ error) bool

func (*ObjectClient) IsRetryableErr

func (s *ObjectClient) IsRetryableErr(_ error) bool

func (*ObjectClient) PutChunks

func (s *ObjectClient) PutChunks(ctx context.Context, chunks []chunk.Chunk) error

PutChunks implements chunk.ObjectClient.

func (*ObjectClient) Stop

func (s *ObjectClient) Stop()

Stop implement chunk.ObjectClient.

type StorageClient

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

StorageClient implements chunk.IndexClient and chunk.ObjectClient for Cassandra.

func NewStorageClient

func NewStorageClient(cfg Config, schemaCfg config.SchemaConfig, registerer prometheus.Registerer) (*StorageClient, error)

NewStorageClient returns a new StorageClient.

func (*StorageClient) BatchWrite

func (s *StorageClient) BatchWrite(ctx context.Context, batch index.WriteBatch) error

BatchWrite implement chunk.IndexClient.

func (*StorageClient) GetReadSession

func (s *StorageClient) GetReadSession() *gocql.Session

Allow other packages to interact with Cassandra directly

func (*StorageClient) NewWriteBatch

func (s *StorageClient) NewWriteBatch() index.WriteBatch

NewWriteBatch implement chunk.IndexClient.

func (*StorageClient) QueryPages

func (s *StorageClient) QueryPages(ctx context.Context, queries []index.Query, callback index.QueryPagesCallback) error

QueryPages implement chunk.IndexClient.

func (*StorageClient) Stop

func (s *StorageClient) Stop()

Stop implement chunk.IndexClient.