storage

package
v0.30.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 15, 2025 License: Apache-2.0 Imports: 25 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	DefaultL0CompactionFileThreshold   = 500
	DefaultL0CompactionThreshold       = 4
	DefaultL0StopWritesThreshold       = 12
	DefaultL0TargetFileSize            = 2 << 20
	DefaultFlushSplitBytes             = 0
	DefaultLBaseMaxBytes               = 64 << 20
	DefaultMaxOpenFiles                = 1000
	DefaultMemTableSize                = 4 << 20
	DefaultMemTableStopWritesThreshold = 2
	DefaultMaxConcurrentCompactions    = 1
	DefaultWALBytesPerSync             = 0
	DefaultSSTBytesPerSync             = 512 << 10
)
View Source 
const (
	DefaultCacheSize = 128 << 20
)
View Source 
const (
	DefaultMetricsLogInterval = time.Duration(0)
)

Variables

View Source 
var (
	ErrNoLogEntry                   = errors.New("storage: no log entry")
	ErrNoCommitContext              = errors.New("storage: no commit context")
	ErrInconsistentWriteCommitState = errors.New("storage: inconsistent write and commit")
)

Functions

func NewOTELMetricRecorder added in v0.28.0 

func NewOTELMetricRecorder(histogramSet *opentelemetry.Int64HistogramSet[int64], lsid types.LogStreamID, storeKind StoreKind) *otelMetricRecorder

func TestAppendLogEntryWithoutCommitContext added in v0.5.0 

func TestAppendLogEntryWithoutCommitContext(tb testing.TB, stg *Storage, llsn types.LLSN, glsn types.GLSN, data []byte)

TestAppendLogEntryWithoutCommitContext stores log entries without commit context.

func TestDeleteCommitContext added in v0.7.0 

func TestDeleteCommitContext(tb testing.TB, stg *Storage)

func TestDeleteLogEntry added in v0.7.0 

func TestDeleteLogEntry(tb testing.TB, stg *Storage, lsn varlogpb.LogSequenceNumber)

func TestGetUnderlyingStore added in v0.28.0 

func TestGetUnderlyingStore(tb testing.TB, stg *Storage) (dataStore, commitStore *pebble.DB)

TestGetUnderlyingStore returns a pebble that is an internal database in the storage.

func TestSetCommitContext added in v0.5.0 

func TestSetCommitContext(tb testing.TB, stg *Storage, cc CommitContext)

TestSetCommitContext stores only commit context.

func TestWriteLogEntry added in v0.15.0 

func TestWriteLogEntry(tb testing.TB, stg *Storage, llsn types.LLSN, data []byte)

TestWriteLogEntry stores data located by the llsn. The data is not committed because it does not store commits.

Types

type AppendBatch added in v0.7.0 

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

AppendBatch is a batch to put one or more log entries.

func (*AppendBatch) Apply added in v0.7.0 

func (ab *AppendBatch) Apply() error

Apply saves a batch of appended log entries to the storage.

func (*AppendBatch) Close added in v0.7.0 

func (ab *AppendBatch) Close() error

Close releases an AppendBatch.

func (*AppendBatch) SetCommitContext added in v0.7.0 

func (ab *AppendBatch) SetCommitContext(cc CommitContext) error

SetCommitContext inserts a commit context.

func (*AppendBatch) SetLogEntry added in v0.7.0 

func (ab *AppendBatch) SetLogEntry(llsn types.LLSN, glsn types.GLSN, data []byte) error

SetLogEntry inserts a log entry.

type BatchCommitDurationKind added in v0.28.0 

type BatchCommitDurationKind int8
const (
	BatchCommitDurationKindTotal BatchCommitDurationKind = iota
	BatchCommitDurationKindSemaphoreWait
	BatchCommitDurationKindWALQueueWait
	BatchCommitDurationKindMemTableWriteStall
	BatchCommitDurationKindL0ReadAmpWriteStall
	BatchCommitDurationKindWALRotation
	BatchCommitDurationKindCommitWait
)

type BatchCommitStats added in v0.28.0 

type BatchCommitStats struct {
	pebble.BatchCommitStats
}

BatchCommitStats wraps pebble.BatchCommitStats.

type Cache added in v0.21.0 

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

func NewCache added in v0.21.0 

func NewCache(size int64) *Cache

func (*Cache) Unref added in v0.21.0 

func (cache *Cache) Unref()

type CommitBatch 

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

func (*CommitBatch) Apply 

func (cb *CommitBatch) Apply() error

func (*CommitBatch) Close 

func (cb *CommitBatch) Close() error

func (*CommitBatch) Set 

func (cb *CommitBatch) Set(llsn types.LLSN, glsn types.GLSN) error

type CommitContext 

type CommitContext struct {
	// Version is the unique identifier of the commit generated by the metadata
	// repository.
	Version types.Version
	// HighWatermark is the maximum GLSN in the versioned commit.
	HighWatermark types.GLSN
	// CommittedGLSNBegin is the start GLSN of this commit.
	CommittedGLSNBegin types.GLSN
	// CommittedGLSNEnd is the end GLSN of this commit exclusively. If none of
	// the log entries are committed, it equals the CommittedGLSNBegin.
	// The GLSN of the local high watermark in a log stream can be inferred by
	// CommittedGLSNEnd minus one. Even if the last commit does not contain any
	// log entries, CommitedGLSNEnd - 1 should be the GLSN of the local high
	// watermark.
	CommittedGLSNEnd types.GLSN
	// CommittedLLSNBegin is the starting LLSN of this commit.
	// The CommittedLLSNEnd can be inferred by the length of committed log
	// entries computed by the CommittedGLSNBegin and CommittedGLSNEnd and
	// CommittedLLSNBegin. Similarly, the CommittedLLSNEnd - 1 is the LLSN of
	// the local high watermark since the LLSNs are sequential.
	CommittedLLSNBegin types.LLSN
}

CommitContext is metadata that represents the environment in which the commit is issued; for instance, the commit's version, the largest GLSN having the same commit version, and the commit range of this log stream.

When a storage node restarts and, thus, its log stream replicas also restart, the log stream executor which runs the log stream replica should recover the previous status - internal/storagenode/logstream.(logStreamContext). To regain their prior status, they utilize the CommitContext.

Here are examples of the CommitContext. You can see github.com/kakao/varlog/proto/snpb.LogStreamCommitResult to refer to related commit messages. 

// LogStream 1 has not committed any log entries, and LogStream 2 has
// committed ten.
CommitContext{
	// LogStream 1
    Version: 1,
    HighWatermark: 10,
    CommittedGLSNBegin: 1,
    CommittedGLSNEnd: 1,
    CommittedLLSNBegin: 1,
}
CommitContext{
	// LogStream 2
    Version: 1,
    HighWatermark: 10,
    CommittedGLSNBegin: 1,
    CommittedGLSNEnd: 11,
    CommittedLLSNBegin: 1,
}

// LogStream 1 has not committed any log entries again, and LogStream 2 has
// committed ten.
// LogStream 1 can recover that there was no log entry committed by loading
// this commit context during restarting.
CommitContext{
	// LogStream 1
    Version: 2,
    HighWatermark: 20,
    CommittedGLSNBegin: 1,
    CommittedGLSNEnd: 1,
    CommittedLLSNBegin: 1,
}
CommitContext{
	// LogStream 2
    Version: 2,
    HighWatermark: 20,
    CommittedGLSNBegin: 11,
    CommittedGLSNEnd: 21,
    CommittedLLSNBegin: 11,
}

// LogStream 1 has committed ten log entries, and LogStream 2 has committed
// ten.
CommitContext{
	// LogStream 1
	Version: 3,
	HighWatermark: 40,
	CommittedGLSNBegin: 21,
	CommittedGLSNEnd: 31,
	CommittedLLSNBegin: 1,
}
CommitContext{
	// LogStream 2
	Version: 3,
	HighWatermark: 40,
	CommittedGLSNBegin: 31,
	CommittedGLSNEnd: 41,
	CommittedLLSNBegin: 21,
}

// LogStream 1 has committed ten log entries, and LogStream 2 has not.
// LogStream 2 can recover the GLSN and LLSN of the last committed log entry
// by loading this commit context when restarting itself.
CommitContext{
	// LogStream 1
	Version: 4,
	HighWatermark: 50,
	CommittedGLSNBegin: 41,
	CommittedGLSNEnd: 51,
	CommittedLLSNBegin: 11,
}
CommitContext{
	// LogStream 2
	Version: 4,
	HighWatermark: 50,
	CommittedGLSNBegin: 41,
	CommittedGLSNEnd: 41,
	CommittedLLSNBegin: 31,
}

func (CommitContext) Empty 

func (cc CommitContext) Empty() bool

func (CommitContext) Equal 

func (cc CommitContext) Equal(other CommitContext) bool

func (CommitContext) MarshalLogObject added in v0.26.0 

func (cc CommitContext) MarshalLogObject(enc zapcore.ObjectEncoder) error

type MetricRecorder added in v0.28.0 

type MetricRecorder interface {
	// RecordBatchCommitStats records the batch commit statistics.
	RecordBatchCommitStats(context.Context, BatchCommitStats)
}

MetricRecorder is an interface for recording statistics for store.

type Option 

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

func WithCommitStoreOptions added in v0.28.0 

func WithCommitStoreOptions(commitStoreOpts ...StoreOption) Option

WithCommitStoreOptions adds options for the commit store. Options set earlier are overridden by options set later.

func WithLogger 

func WithLogger(logger *zap.Logger) Option

func WithMetricsLogInterval added in v0.15.0 

func WithMetricsLogInterval(metricsLogInterval time.Duration) Option

func WithPath 

func WithPath(path string) Option

func WithValueStoreOptions added in v0.28.0 

func WithValueStoreOptions(valueStoreOpts ...StoreOption) Option

WithValueStoreOptions adds options for the value store. Options set earlier are overridden by options set later.

type ReadOption 

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

func AtGLSN 

func AtGLSN(glsn types.GLSN) ReadOption

func AtLLSN 

func AtLLSN(llsn types.LLSN) ReadOption

type RecoveryPoints 

type RecoveryPoints struct {
	LastCommitContext *CommitContext
	CommittedLogEntry struct {
		First *varlogpb.LogSequenceNumber
		Last  *varlogpb.LogSequenceNumber
	}
	UncommittedLLSN struct {
		Begin types.LLSN
		End   types.LLSN
	}
}

func (RecoveryPoints) MarshalLogObject added in v0.26.0 

func (rp RecoveryPoints) MarshalLogObject(enc zapcore.ObjectEncoder) error

type ScanOption 

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

func WithGLSN 

func WithGLSN(begin, end types.GLSN) ScanOption

func WithLLSN 

func WithLLSN(begin, end types.LLSN) ScanOption

type Scanner 

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

func (*Scanner) Close 

func (s *Scanner) Close() (err error)

func (*Scanner) Next 

func (s *Scanner) Next() bool

func (*Scanner) Valid 

func (s *Scanner) Valid() bool

func (*Scanner) Value 

func (s *Scanner) Value() (le varlogpb.LogEntry, err error)

type Storage 

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

func New 

func New(opts ...Option) (*Storage, error)

New creates a new storage.

func TestNewStorage 

func TestNewStorage(tb testing.TB, opts ...Option) *Storage

func (*Storage) Close 

func (s *Storage) Close() (err error)

Close closes the storage.

func (*Storage) DiskUsage 

func (s *Storage) DiskUsage() uint64

func (*Storage) NewAppendBatch added in v0.7.0 

func (s *Storage) NewAppendBatch() *AppendBatch

NewAppendBatch creates a batch for appending log entries. It does not put commit context.

func (*Storage) NewCommitBatch 

func (s *Storage) NewCommitBatch(cc CommitContext) (*CommitBatch, error)

NewCommitBatch creates a batch for commit operations.

func (*Storage) NewScanner 

func (s *Storage) NewScanner(opts ...ScanOption) (scanner *Scanner, err error)

NewScanner creates a scanner for the given key range.

func (*Storage) NewWriteBatch 

func (s *Storage) NewWriteBatch() *WriteBatch

NewWriteBatch creates a batch for write operations.

func (*Storage) Path 

func (s *Storage) Path() string

Path returns the path to the storage.

func (*Storage) Read 

func (s *Storage) Read(opts ...ReadOption) (le varlogpb.LogEntry, err error)

Read reads the log entry at the glsn.

func (*Storage) ReadCommitContext added in v0.5.0 

func (s *Storage) ReadCommitContext() (cc CommitContext, err error)

func (*Storage) ReadRecoveryPoints 

func (s *Storage) ReadRecoveryPoints() (rp RecoveryPoints, err error)

ReadRecoveryPoints reads data necessary to restore the status of a log stream replica - the first and last log entries and commit context. It is okay when the commit context is not matched with the last log entry, resolved through synchronization between replicas later. The first and last log entries can be nil if there is no log entry or they can't be read due to inconsistency between data and commit. However, if there is a fatal error, it returns an error.

func (*Storage) Trim 

func (s *Storage) Trim(glsn types.GLSN) error

Trim deletes log entries whose GLSNs are less than or equal to the argument glsn. Internally, it removes records for both data and commits but does not remove the commit context. It returns the ErrNoLogEntry if there are no logs to delete.

type StoreKind added in v0.28.0 

type StoreKind int8
const (
	StoreKindValue StoreKind = iota
	StoreKindCommit
)

type StoreOption added in v0.28.0 

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

func EnableTelemetry added in v0.28.0 

func EnableTelemetry(enableTelemetry bool) StoreOption

EnableTelemetry enables or disables telemetry for the store. When disabled, even if WithMetricRecorder is set, no metrics will be recorded. It is disabled by default.

func ReadOnly added in v0.5.0 

func ReadOnly(readOnly bool) StoreOption

ReadOnly sets the store to read-only mode. This is mainly useful for testing. Most users do not need to call this option.

func WithCache added in v0.21.0 

func WithCache(cache *Cache) StoreOption

WithCache sets the cache for the store. Users can share the same cache across multiple stores. If not set, each store uses its own 8 MB cache by default.

func WithFlushSplitBytes added in v0.13.0 

func WithFlushSplitBytes(flushSplitBytes int64) StoreOption

func WithL0CompactionFileThreshold added in v0.13.0 

func WithL0CompactionFileThreshold(l0CompactionFileThreshold int) StoreOption

func WithL0CompactionThreshold 

func WithL0CompactionThreshold(l0CompactionThreshold int) StoreOption

func WithL0StopWritesThreshold 

func WithL0StopWritesThreshold(l0StopWritesThreshold int) StoreOption

func WithL0TargetFileSize added in v0.13.0 

func WithL0TargetFileSize(l0TargetFileSize int64) StoreOption

func WithLBaseMaxBytes 

func WithLBaseMaxBytes(lbaseMaxBytes int64) StoreOption

func WithMaxConcurrentCompaction 

func WithMaxConcurrentCompaction(maxConcurrentCompaction int) StoreOption

func WithMaxOpenFiles 

func WithMaxOpenFiles(maxOpenFiles int) StoreOption

func WithMemTableSize 

func WithMemTableSize(memTableSize int) StoreOption

func WithMemTableStopWritesThreshold 

func WithMemTableStopWritesThreshold(memTableStopWritesThreshold int) StoreOption

func WithMetricRecorder added in v0.28.0 

func WithMetricRecorder(metricRecorder MetricRecorder) StoreOption

WithMetricRecorder sets the MetricRecorder for the store. If telemetry is disabled via EnableTelemetry(false), the provided MetricRecorder will not record metrics. This option allows integration with custom metrics backends.

func WithSSTBytesperSync added in v0.28.0 

func WithSSTBytesperSync(sstBytesPerSync int) StoreOption

WithSSTBytesperSync sets the number of bytes to write to SSTables before triggering a background sync. This helps avoid latency spikes caused by the OS flushing large amounts of dirty data at once. This option only affects SSTable syncs and maps to Pebble's BytesPerSync. The default value is 512KB.

func WithSyncWAL added in v0.28.0 

func WithSyncWAL(syncWAL bool) StoreOption

WithSyncWAL sets whether write operations are synchronized to disk by waiting for the Pebble WAL (Write-Ahead Log) to be flushed to durable storage.

Enabling this option ensures each write is durable, but may reduce write performance. Disabling it can improve performance but risks recent writes being lost if a process or machine crash occurs, as data may only be buffered in memory.

func WithTrimDelay added in v0.15.0 

func WithTrimDelay(trimDelay time.Duration) StoreOption

WithTrimDelay sets the delay before the store removes data. If set to zero, Trim will delay removal until additional data is written.

func WithTrimRateByte added in v0.15.0 

func WithTrimRateByte(trimRateByte int) StoreOption

WithTrimRateByte sets the Trim deletion speed in bytes per second. If set to zero, Trim removes data without throttling.

func WithVerbose added in v0.28.0 

func WithVerbose(verbose bool) StoreOption

WithVerbose enables or disables verbose mode for the store. When enabled, the store logs compaction events.

func WithWAL added in v0.26.0 

func WithWAL(wal bool) StoreOption

func WithWALBytesPerSync added in v0.28.0 

func WithWALBytesPerSync(walBytesPerSync int) StoreOption

WithWALBytesPerSync sets the number of bytes to write to the WAL (Write-Ahead Log) before triggering a background sync. This helps smooth out disk write latencies and avoids large, sudden disk writes. This option maps to Pebble's WALBytesPerSync. The default value is 0 (no background syncing).

type WriteBatch 

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

WriteBatch is a batch of writes to storage.

func (*WriteBatch) Apply 

func (wb *WriteBatch) Apply() error

Apply applies the batch to the underlying storage.

func (*WriteBatch) Close 

func (wb *WriteBatch) Close() error

Close releases acquired resources.

func (*WriteBatch) Set 

func (wb *WriteBatch) Set(llsn types.LLSN, data []byte) error

Set writes the given LLSN and data to the batch.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.