cnservice

Documentation

Index

• type Config
  • func (c *Config) Validate() error
• type EngineType
• type Option
  • func WithLogger(logger *zap.Logger) Option
  • func WithMessageHandle(...) Option
  • func WithTaskStorageFactory(factory taskservice.TaskStorageFactory) Option
• type Service
  • func NewService(cfg *Config, ctx context.Context, fileService fileservice.FileService, ...) (Service, error)

Types

type Config

type Config struct {
	// UUID cn store uuid
	UUID string `toml:"uuid"`
	// Role cn node role, [AP|TP]
	Role string `toml:"role"`

	// ListenAddress listening address for receiving external requests
	ListenAddress string `toml:"listen-address"`
	// ServiceAddress service address for communication, if this address is not set, use
	// ListenAddress as the communication address.
	ServiceAddress string `toml:"service-address"`
	// SQLAddress service address for receiving external sql clientß
	SQLAddress string `toml:"sql-address"`

	Engine struct {
		Type                EngineType           `toml:"type"`
		Logstore            options.LogstoreType `toml:"logstore"`
		FlushInterval       toml.Duration        `toml:"flush-interval"`
		MinCount            int64                `toml:"min-count"`
		ScanInterval        toml.Duration        `toml:"scan-interval"`
		IncrementalInterval toml.Duration        `toml:"incremental-interval"`
		GlobalInterval      toml.Duration        `toml:"global-interval"`
	}

	// parameters for cn-server related buffer.
	ReadBufferSize  int
	WriteBufferSize int

	// Pipeline configuration
	Pipeline struct {
		// HostSize is the memory limit
		HostSize int64 `toml:"host-size"`
		// GuestSize is the memory limit for one query
		GuestSize int64 `toml:"guest-size"`
		// OperatorSize is the memory limit for one operator
		OperatorSize int64 `toml:"operator-size"`
		// BatchRows is the batch rows limit for one batch
		BatchRows int64 `toml:"batch-rows"`
		// BatchSize is the memory limit for one batch
		BatchSize int64 `toml:"batch-size"`
	}

	// Frontend parameters for the frontend
	Frontend config.FrontendParameters `toml:"frontend"`

	// HAKeeper configuration
	HAKeeper struct {
		// HeatbeatInterval heartbeat interval to send message to hakeeper. Default is 1s
		HeatbeatInterval toml.Duration `toml:"hakeeper-heartbeat-interval"`
		// HeatbeatTimeout heartbeat request timeout. Default is 500ms
		HeatbeatTimeout toml.Duration `toml:"hakeeper-heartbeat-timeout"`
		// DiscoveryTimeout discovery HAKeeper service timeout. Default is 30s
		DiscoveryTimeout toml.Duration `toml:"hakeeper-discovery-timeout"`
		// ClientConfig hakeeper client configuration
		ClientConfig logservice.HAKeeperClientConfig
	}

	// TaskRunner configuration
	TaskRunner struct {
		QueryLimit        int           `toml:"task-query-limit"`
		Parallelism       int           `toml:"task-parallelism"`
		MaxWaitTasks      int           `toml:"task-max-wait-tasks"`
		FetchInterval     toml.Duration `toml:"task-fetch-interval"`
		FetchTimeout      toml.Duration `toml:"task-fetch-timeout"`
		RetryInterval     toml.Duration `toml:"task-retry-interval"`
		HeartbeatInterval toml.Duration `toml:"task-heartbeat-interval"`
	}

	// RPC rpc config used to build txn sender
	RPC rpc.Config `toml:"rpc"`
}

Config cn service

func (*Config) Validate

func (c *Config) Validate() error

type EngineType

type EngineType string

const (
	EngineTAE                  EngineType = "tae"
	EngineDistributedTAE       EngineType = "distributed-tae"
	EngineMemory               EngineType = "memory"
	EngineNonDistributedMemory EngineType = "non-distributed-memory"
)

type Option

type Option func(*service)

Option option to create cn service

func WithLogger

func WithLogger(logger *zap.Logger) Option

WithLogger setup cn service's logger

func WithMessageHandle

func WithMessageHandle(f func(ctx context.Context,
	message morpc.Message,
	cs morpc.ClientSession,
	engine engine.Engine,
	fs fileservice.FileService,
	cli client.TxnClient,
	mAcquirer func() morpc.Message,
	getClusterDetails engine.GetClusterDetailsFunc) error) Option

WithMessageHandle setup message handle

func WithTaskStorageFactory

func WithTaskStorageFactory(factory taskservice.TaskStorageFactory) Option

WithTaskStorageFactory setup the special task strorage factory

type Service

type Service interface {
	Start() error
	Close() error

	GetTaskRunner() taskservice.TaskRunner
	GetTaskService() (taskservice.TaskService, bool)
	WaitSystemInitCompleted(ctx context.Context) error
}

func NewService

func NewService(
	cfg *Config,
	ctx context.Context,
	fileService fileservice.FileService,
	options ...Option,
) (Service, error)