writer

Documentation

Index

• func InitMetrics(registry *prometheus.Registry)
• func NewBlackHoleWriter() *blackHoleWriter
• type FileWriterConfig
• type LogWriter
  • func NewLogWriter(ctx context.Context, cfg *LogWriterConfig) (*LogWriter, error)
  • func (l *LogWriter) Close() error
  • func (l *LogWriter) DeleteAllLogs(ctx context.Context) error
  • func (l *LogWriter) EmitCheckpointTs(ctx context.Context, ts uint64) error
  • func (l *LogWriter) EmitResolvedTs(ctx context.Context, ts uint64) error
  • func (l *LogWriter) FlushLog(ctx context.Context, tableID int64, ts uint64) error
  • func (l *LogWriter) GetCurrentResolvedTs(ctx context.Context, tableIDs []int64) (map[int64]uint64, error)
  • func (l *LogWriter) SendDDL(ctx context.Context, ddl *model.RedoDDLEvent) error
  • func (l *LogWriter) WriteLog(ctx context.Context, tableID int64, rows []*model.RedoRowChangedEvent) (uint64, error)
• type LogWriterConfig
  • func (cfg LogWriterConfig) String() string
• type MockRedoLogWriter
  • func (_m *MockRedoLogWriter) Close() error
  • func (_m *MockRedoLogWriter) EmitCheckpointTs(ctx context.Context, ts uint64) error
  • func (_m *MockRedoLogWriter) EmitResolvedTs(ctx context.Context, ts uint64) error
  • func (_m *MockRedoLogWriter) FlushLog(ctx context.Context, tableID int64, ts uint64) error
  • func (_m *MockRedoLogWriter) GetCurrentResolvedTs(ctx context.Context, tableIDs []int64) (map[int64]uint64, error)
  • func (_m *MockRedoLogWriter) SendDDL(ctx context.Context, ddl *model.RedoDDLEvent) error
  • func (_m *MockRedoLogWriter) WriteLog(ctx context.Context, tableID int64, rows []*model.RedoRowChangedEvent) (uint64, error)
• type Option
  • func WithLogFileName(f func() string) Option
• type RedoLogWriter
• type Writer
  • func NewWriter(ctx context.Context, cfg *FileWriterConfig, opts ...Option) (*Writer, error)
  • func (w *Writer) AdvanceTs(commitTs uint64)
  • func (w *Writer) Close() error
  • func (w *Writer) Flush() error
  • func (w *Writer) GC(checkPointTs uint64) error
  • func (w *Writer) IsRunning() bool
  • func (w *Writer) Write(rawData []byte) (int, error)

Functions

func InitMetrics

func InitMetrics(registry *prometheus.Registry)

InitMetrics registers all metrics in this file

func NewBlackHoleWriter

func NewBlackHoleWriter() *blackHoleWriter

NewBlackHoleWriter creates a blackHole writer

Types

type FileWriterConfig

type FileWriterConfig struct {
	Dir          string
	ChangeFeedID model.ChangeFeedID
	CaptureID    string
	FileType     string
	CreateTime   time.Time
	// MaxLogSize is the maximum size of log in megabyte, defaults to defaultMaxLogSize.
	MaxLogSize        int64
	FlushIntervalInMs int64
	S3Storage         bool
	S3URI             url.URL
}

FileWriterConfig is the configuration used by a Writer.

type LogWriter

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

LogWriter implement the RedoLogWriter interface

func NewLogWriter

func NewLogWriter(ctx context.Context, cfg *LogWriterConfig) (*LogWriter, error)

NewLogWriter creates a LogWriter instance. It is guaranteed only one LogWriter per changefeed

func (*LogWriter) Close

func (l *LogWriter) Close() error

Close implements RedoLogWriter.Close.

func (*LogWriter) DeleteAllLogs

func (l *LogWriter) DeleteAllLogs(ctx context.Context) error

DeleteAllLogs implement DeleteAllLogs api

func (*LogWriter) EmitCheckpointTs

func (l *LogWriter) EmitCheckpointTs(ctx context.Context, ts uint64) error

EmitCheckpointTs implement EmitCheckpointTs api

func (*LogWriter) EmitResolvedTs

func (l *LogWriter) EmitResolvedTs(ctx context.Context, ts uint64) error

EmitResolvedTs implement EmitResolvedTs api

func (*LogWriter) FlushLog

func (l *LogWriter) FlushLog(ctx context.Context, tableID int64, ts uint64) error

FlushLog implement FlushLog api

func (*LogWriter) GetCurrentResolvedTs

func (l *LogWriter) GetCurrentResolvedTs(ctx context.Context, tableIDs []int64) (map[int64]uint64, error)

GetCurrentResolvedTs implement GetCurrentResolvedTs api

func (*LogWriter) SendDDL

func (l *LogWriter) SendDDL(ctx context.Context, ddl *model.RedoDDLEvent) error

SendDDL implement SendDDL api

func (*LogWriter) WriteLog

func (l *LogWriter) WriteLog(ctx context.Context, tableID int64, rows []*model.RedoRowChangedEvent) (uint64, error)

WriteLog implement WriteLog api

type LogWriterConfig

type LogWriterConfig struct {
	Dir          string
	ChangeFeedID model.ChangeFeedID
	CaptureID    string
	CreateTime   time.Time
	// MaxLogSize is the maximum size of log in megabyte, defaults to defaultMaxLogSize.
	MaxLogSize        int64
	FlushIntervalInMs int64
	S3Storage         bool
	// S3URI should be like S3URI="s3://logbucket/test-changefeed?endpoint=http://$S3_ENDPOINT/"
	S3URI url.URL
}

LogWriterConfig is the configuration used by a Writer.

func (LogWriterConfig) String

func (cfg LogWriterConfig) String() string

type MockRedoLogWriter

type MockRedoLogWriter struct {
	mock.Mock
}

MockRedoLogWriter is an autogenerated mock type for the RedoLogWriter type

func (*MockRedoLogWriter) Close

func (_m *MockRedoLogWriter) Close() error

Close provides a mock function with given fields:

func (*MockRedoLogWriter) EmitCheckpointTs

func (_m *MockRedoLogWriter) EmitCheckpointTs(ctx context.Context, ts uint64) error

EmitCheckpointTs provides a mock function with given fields: ctx, ts

func (*MockRedoLogWriter) EmitResolvedTs

func (_m *MockRedoLogWriter) EmitResolvedTs(ctx context.Context, ts uint64) error

EmitResolvedTs provides a mock function with given fields: ctx, ts

func (*MockRedoLogWriter) FlushLog

func (_m *MockRedoLogWriter) FlushLog(ctx context.Context, tableID int64, ts uint64) error

FlushLog provides a mock function with given fields: ctx, tableID, ts

func (*MockRedoLogWriter) GetCurrentResolvedTs

func (_m *MockRedoLogWriter) GetCurrentResolvedTs(ctx context.Context, tableIDs []int64) (map[int64]uint64, error)

GetCurrentResolvedTs provides a mock function with given fields: ctx, tableIDs

func (*MockRedoLogWriter) SendDDL

func (_m *MockRedoLogWriter) SendDDL(ctx context.Context, ddl *model.RedoDDLEvent) error

SendDDL provides a mock function with given fields: ctx, ddl

func (*MockRedoLogWriter) WriteLog

func (_m *MockRedoLogWriter) WriteLog(ctx context.Context, tableID int64, rows []*model.RedoRowChangedEvent) (uint64, error)

WriteLog provides a mock function with given fields: ctx, tableID, rows

type Option

type Option func(writer *writerOptions)

Option define the writerOptions

func WithLogFileName

func WithLogFileName(f func() string) Option

WithLogFileName provide the Option for fileName

type RedoLogWriter

type RedoLogWriter interface {
	io.Closer

	// WriteLog writer RedoRowChangedEvent to row log file
	WriteLog(ctx context.Context, tableID int64, rows []*model.RedoRowChangedEvent) (resolvedTs uint64, err error)

	// SendDDL EmitCheckpointTs and EmitResolvedTs are called from owner only
	// SendDDL writer RedoDDLEvent to ddl log file
	SendDDL(ctx context.Context, ddl *model.RedoDDLEvent) error

	// FlushLog sends resolved-ts from table pipeline to log writer, it is
	// essential to flush when a table doesn't have any row change event for
	// some time, and the resolved ts of this table should be moved forward.
	FlushLog(ctx context.Context, tableID int64, ts uint64) error

	// EmitCheckpointTs write CheckpointTs to meta file
	EmitCheckpointTs(ctx context.Context, ts uint64) error

	// EmitResolvedTs write ResolvedTs to meta file
	EmitResolvedTs(ctx context.Context, ts uint64) error

	// GetCurrentResolvedTs return all the ResolvedTs list for given tableIDs
	GetCurrentResolvedTs(ctx context.Context, tableIDs []int64) (resolvedTsList map[int64]uint64, err error)

	// DeleteAllLogs delete all log files related to the changefeed, called from owner only when delete changefeed
	DeleteAllLogs(ctx context.Context) error
}

RedoLogWriter defines the interfaces used to write redo log, all operations are thread-safe

type Writer

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

Writer is a redo log event Writer which writes redo log events to a file.

func NewWriter

func NewWriter(ctx context.Context, cfg *FileWriterConfig, opts ...Option) (*Writer, error)

NewWriter return a file rotated writer, TODO: extract to a common rotate Writer

func (*Writer) AdvanceTs

func (w *Writer) AdvanceTs(commitTs uint64)

AdvanceTs implement Advance interface

func (*Writer) Close

func (w *Writer) Close() error

Close implements fileWriter.Close.

func (*Writer) Flush

func (w *Writer) Flush() error

Flush implement Flush interface

func (*Writer) GC

func (w *Writer) GC(checkPointTs uint64) error

GC implement GC interface

func (*Writer) IsRunning

func (w *Writer) IsRunning() bool

IsRunning implement IsRunning interface

func (*Writer) Write

func (w *Writer) Write(rawData []byte) (int, error)

Write implement write interface TODO: more general api with fileName generated by caller