storagenode

Documentation

Index

• Constants
• func TestAddLogStreamReplica(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, ...)
• func TestAppend(t *testing.T, tpid types.TopicID, lsid types.LogStreamID, dataBatch [][]byte, ...) []snpb.AppendResult
• func TestGetAdvertiseAddress(t *testing.T, sn *StorageNode) string
• func TestGetStorageNodeID(t *testing.T, sn *StorageNode) types.StorageNodeID
• func TestGetStorageNodeMetadataDescriptor(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, addr string) *snpb.StorageNodeMetadataDescriptor
• func TestGetStorageNodeMetadataDescriptorWithoutAddr(t *testing.T, sn *StorageNode) *snpb.StorageNodeMetadataDescriptor
• func TestNewLogIOClient(t *testing.T, snid types.StorageNodeID, addr string) (*client.LogClient, func())
• func TestNewManagementClient(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, addr string) (*client.ManagementClient, func())
• func TestNewRPCServer(t *testing.T, ctrl *gomock.Controller, snid types.StorageNodeID, ...) *testRPCServer
• func TestSealLogStreamReplica(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, ...) (varlogpb.LogStreamStatus, types.GLSN)
• func TestSubscribe(t *testing.T, tpid types.TopicID, lsid types.LogStreamID, ...) []varlogpb.LogEntry
• func TestSubscribeTo(t *testing.T, tpid types.TopicID, lsid types.LogStreamID, ...) []varlogpb.LogEntry
• func TestSync(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, ...) *snpb.SyncStatus
• func TestTrim(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, ...) map[types.LogStreamID]error
• func TestUnsealLogStreamReplica(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, ...)
• func TestWaitForStartingOfServe(t *testing.T, sn *StorageNode, timeouts ...time.Duration)
• type Option
  • func WithAdvertiseAddress(advertise string) Option
  • func WithAppendPipelineSize(appendPipelineSize int32) Option
  • func WithBallastSize(ballastSize int64) Option
  • func WithClusterID(cid types.ClusterID) Option
  • func WithDefaultLogStreamExecutorOptions(defaultLSEOptions ...logstream.ExecutorOption) Option
  • func WithDefaultStorageOptions(defaultStorageOpts ...storage.Option) Option
  • func WithGRPCServerInitialConnWindowSize(grpcServerInitialConnWindowSize int32) Option
  • func WithGRPCServerInitialWindowSize(grpcServerInitialWindowSize int32) Option
  • func WithGRPCServerMaxRecvMsgSize(grpcServerMaxRecvMsgSize int64) Option
  • func WithGRPCServerReadBufferSize(grpcServerReadBufferSize int64) Option
  • func WithGRPCServerWriteBufferSize(grpcServerWriteBufferSize int64) Option
  • func WithListenAddress(listen string) Option
  • func WithLogger(logger *zap.Logger) Option
  • func WithMaxLogStreamReplicasCount(maxLogStreamReplicasCount int32) Option
  • func WithPProfOptions(pprofOpts ...pprof.Option) Option
  • func WithReplicateClientReadBufferSize(replicateClientReadBufferSize int64) Option
  • func WithReplicateClientWriteBufferSize(replicateClientWriteBufferSize int64) Option
  • func WithStorageNodeID(snid types.StorageNodeID) Option
  • func WithVolumes(volumes ...string) Option
• type StorageNode
  • func NewStorageNode(opts ...Option) (*StorageNode, error)
  • func TestNewSimpleStorageNode(t *testing.T, opts ...Option) *StorageNode
  • func TestNewStorageNode(tb testing.TB, opts ...Option) *StorageNode
  • func (sn *StorageNode) Close() (err error)
  • func (sn *StorageNode) Serve() error

Constants

const (
	DefaultBallastSize                    = "0B"
	DefaultServerReadBufferSize           = 32 << 10
	DefaultServerWriteBufferSize          = 32 << 10
	DefaultServerMaxRecvSize              = 4 << 20
	DefaultReplicateClientReadBufferSize  = 32 << 10
	DefaultReplicateClientWriteBufferSize = 32 << 10
	DefaultMaxLogStreamReplicasCount      = -1

	DefaultAppendPipelineSize = 8
	MinAppendPipelineSize     = 1
	MaxAppendPipelineSize     = 16
)

Functions

func TestAddLogStreamReplica

func TestAddLogStreamReplica(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, tpid types.TopicID, lsid types.LogStreamID, path, addr string)

func TestAppend

func TestAppend(t *testing.T, tpid types.TopicID, lsid types.LogStreamID, dataBatch [][]byte, replicas []varlogpb.LogStreamReplica) []snpb.AppendResult

func TestGetAdvertiseAddress

func TestGetAdvertiseAddress(t *testing.T, sn *StorageNode) string

func TestGetStorageNodeID

func TestGetStorageNodeID(t *testing.T, sn *StorageNode) types.StorageNodeID

func TestGetStorageNodeMetadataDescriptor

func TestGetStorageNodeMetadataDescriptor(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, addr string) *snpb.StorageNodeMetadataDescriptor

func TestGetStorageNodeMetadataDescriptorWithoutAddr

func TestGetStorageNodeMetadataDescriptorWithoutAddr(t *testing.T, sn *StorageNode) *snpb.StorageNodeMetadataDescriptor

func TestNewLogIOClient

func TestNewLogIOClient(t *testing.T, snid types.StorageNodeID, addr string) (*client.LogClient, func())

func TestNewManagementClient

func TestNewManagementClient(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, addr string) (*client.ManagementClient, func())

func TestNewRPCServer

func TestNewRPCServer(t *testing.T, ctrl *gomock.Controller, snid types.StorageNodeID, address ...string) *testRPCServer

func TestSealLogStreamReplica

func TestSealLogStreamReplica(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, tpid types.TopicID, lsid types.LogStreamID, lastCommittedGLSN types.GLSN, addr string) (varlogpb.LogStreamStatus, types.GLSN)

func TestSubscribe

func TestSubscribe(t *testing.T, tpid types.TopicID, lsid types.LogStreamID, begin, end types.GLSN, snid types.StorageNodeID, addr string) []varlogpb.LogEntry

func TestSubscribeTo

func TestSubscribeTo(t *testing.T, tpid types.TopicID, lsid types.LogStreamID, begin, end types.LLSN, snid types.StorageNodeID, addr string) []varlogpb.LogEntry

func TestSync

func TestSync(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, tpid types.TopicID, lsid types.LogStreamID, lastGLSN types.GLSN, addr string, dst varlogpb.StorageNode) *snpb.SyncStatus

func TestTrim

func TestTrim(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, tpid types.TopicID, glsn types.GLSN, addr string) map[types.LogStreamID]error

func TestUnsealLogStreamReplica

func TestUnsealLogStreamReplica(t *testing.T, cid types.ClusterID, snid types.StorageNodeID, tpid types.TopicID, lsid types.LogStreamID, replicas []varlogpb.LogStreamReplica, addr string)

func TestWaitForStartingOfServe

func TestWaitForStartingOfServe(t *testing.T, sn *StorageNode, timeouts ...time.Duration)

Types

type Option

type Option interface {
	// contains filtered or unexported methods
}

func WithAdvertiseAddress

func WithAdvertiseAddress(advertise string) Option

func WithAppendPipelineSize

func WithAppendPipelineSize(appendPipelineSize int32) Option

func WithBallastSize

func WithBallastSize(ballastSize int64) Option

func WithClusterID

func WithClusterID(cid types.ClusterID) Option

func WithDefaultLogStreamExecutorOptions

func WithDefaultLogStreamExecutorOptions(defaultLSEOptions ...logstream.ExecutorOption) Option

func WithDefaultStorageOptions

func WithDefaultStorageOptions(defaultStorageOpts ...storage.Option) Option

func WithGRPCServerInitialConnWindowSize

func WithGRPCServerInitialConnWindowSize(grpcServerInitialConnWindowSize int32) Option

func WithGRPCServerInitialWindowSize

func WithGRPCServerInitialWindowSize(grpcServerInitialWindowSize int32) Option

func WithGRPCServerMaxRecvMsgSize

func WithGRPCServerMaxRecvMsgSize(grpcServerMaxRecvMsgSize int64) Option

func WithGRPCServerReadBufferSize

func WithGRPCServerReadBufferSize(grpcServerReadBufferSize int64) Option

func WithGRPCServerWriteBufferSize

func WithGRPCServerWriteBufferSize(grpcServerWriteBufferSize int64) Option

func WithListenAddress

func WithListenAddress(listen string) Option

func WithLogger

func WithLogger(logger *zap.Logger) Option

func WithMaxLogStreamReplicasCount

func WithMaxLogStreamReplicasCount(maxLogStreamReplicasCount int32) Option

func WithPProfOptions

func WithPProfOptions(pprofOpts ...pprof.Option) Option

func WithReplicateClientReadBufferSize

func WithReplicateClientReadBufferSize(replicateClientReadBufferSize int64) Option

func WithReplicateClientWriteBufferSize

func WithReplicateClientWriteBufferSize(replicateClientWriteBufferSize int64) Option

func WithStorageNodeID

func WithStorageNodeID(snid types.StorageNodeID) Option

func WithVolumes

func WithVolumes(volumes ...string) Option

type StorageNode

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

func NewStorageNode

func NewStorageNode(opts ...Option) (*StorageNode, error)

func TestNewSimpleStorageNode

func TestNewSimpleStorageNode(t *testing.T, opts ...Option) *StorageNode

func TestNewStorageNode

func TestNewStorageNode(tb testing.TB, opts ...Option) *StorageNode

func (*StorageNode) Close

func (sn *StorageNode) Close() (err error)

func (*StorageNode) Serve

func (sn *StorageNode) Serve() error