hanzo/logs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(dataobj): add ability to buffer pending sections to disk (#18780)

Browse Source 
Signed-off-by: Robert Fratto <robertfratto@gmail.com>
This commit is contained in:
Robert Fratto
2025-08-12 11:57:04 -04:00
committed by GitHub
parent 1c87624e1b
commit 2ca8e58b0e
42 changed files with 1249 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/sources/shared/configuration.md
+4
View File
@@ -2457,6 +2457,10 @@ ring:
# the grpc address of the compactor in the form host:port
# CLI flag: -common.compactor-grpc-address
[compactor_grpc_address: <string> | default = ""]
# Experimental: path to use for temporary data, where scratch data is supported.
# CLI flag: -common.scratch-path
[scratch_path: <string> | default = ""]
```
### compactor
pkg/dataobj/builder.go
+22 -26
View File
@@ -1,11 +1,10 @@
package dataobj
import (
"bytes"
"fmt"
"io"
"github.com/grafana/loki/v3/pkg/dataobj/internal/util/bufpool"
"github.com/grafana/loki/v3/pkg/scratch"
)
// A Builder builds data objects from a set of incoming log data. Log data is
@@ -20,8 +19,21 @@ type Builder struct {
// A Builder accumulates data from a set of in-progress sections. A Builder can
// be flushed into a data object by calling [Builder.Flush].
func NewBuilder() *Builder {
return &Builder{encoder: newEncoder()}
//
// If provided, completed sections will be written to sectionScratchPath to
// reduce the peak memory usage of a builder to the peak memory usage of
// in-progress sections.
//
// NewBuilder returns an error if sectionScratchPath specifies an invalid path
// on disk.
func NewBuilder(scratchStore scratch.Store) *Builder {
if scratchStore == nil {
scratchStore = scratch.NewMemory()
}
return &Builder{
encoder: newEncoder(scratchStore),
}
}
// Append flushes a [SectionBuilder], buffering its data and metadata into b.
@@ -67,40 +79,24 @@ func (b *Builder) Bytes() int {
// [Builder.Reset] is called after a successful flush to discard any pending
// data, allowing new data to be appended.
func (b *Builder) Flush() (*Object, io.Closer, error) {
flushSize, err := b.encoder.FlushSize()
if err != nil {
return nil, nil, fmt.Errorf("determining object size: %w", err)
}
buf := bufpool.Get(int(flushSize))
closer := func() error {
bufpool.Put(buf)
return nil
}
sz, err := b.encoder.Flush(buf)
snapshot, err := b.encoder.Flush()
if err != nil {
return nil, nil, fmt.Errorf("flushing object: %w", err)
}
obj, err := FromReaderAt(bytes.NewReader(buf.Bytes()), sz)
obj, err := FromReaderAt(snapshot, snapshot.Size())
if err != nil {
bufpool.Put(buf)
// The snapshot is invalid; in this case, we *don't* want to call
// [snapshot.Close], otherwise it removes all of the in-progress
// sections from our encoder and we can't potentially recover them.
return nil, nil, fmt.Errorf("error building object: %w", err)
}
b.Reset()
return obj, funcIOCloser(closer), nil
return obj, snapshot, nil
}
// Reset discards pending data and resets the builder to an empty state.
func (b *Builder) Reset() {
b.encoder.Reset()
}
type funcIOCloser func() error
func (fc funcIOCloser) Close() error {
return fc()
}
pkg/dataobj/consumer/logsobj/builder.go
+3 -2
View File
@@ -21,6 +21,7 @@ import (
"github.com/grafana/loki/v3/pkg/dataobj/sections/streams"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/logql/syntax"
"github.com/grafana/loki/v3/pkg/scratch"
)
// ErrBuilderFull is returned by [Builder.Append] when the buffer is
@@ -140,7 +141,7 @@ const (
// NewBuilder creates a new [Builder] which stores log-oriented data objects.
//
// NewBuilder returns an error if the provided config is invalid.
func NewBuilder(cfg BuilderConfig) (*Builder, error) {
func NewBuilder(cfg BuilderConfig, scratchStore scratch.Store) (*Builder, error) {
if err := cfg.Validate(); err != nil {
return nil, err
}
@@ -159,7 +160,7 @@ func NewBuilder(cfg BuilderConfig) (*Builder, error) {
labelCache: labelCache,
builder: dataobj.NewBuilder(),
builder: dataobj.NewBuilder(scratchStore),
streams: streams.NewBuilder(metrics.streams, int(cfg.TargetPageSize)),
logs: logs.NewBuilder(metrics.logs, logs.BuilderOptions{
PageSizeHint: int(cfg.TargetPageSize),
pkg/dataobj/consumer/logsobj/builder_test.go
+2 -2
View File
@@ -72,7 +72,7 @@ func TestBuilder(t *testing.T) {
}
t.Run("Build", func(t *testing.T) {
builder, err := NewBuilder(testBuilderConfig)
builder, err := NewBuilder(testBuilderConfig, nil)
require.NoError(t, err)
for _, entry := range testStreams {
@@ -93,7 +93,7 @@ func TestBuilder_Append(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
builder, err := NewBuilder(testBuilderConfig)
builder, err := NewBuilder(testBuilderConfig, nil)
require.NoError(t, err)
for {
pkg/dataobj/consumer/partition_processor.go
+10 -6
View File
@@ -20,6 +20,7 @@ import (
"github.com/grafana/loki/v3/pkg/dataobj/metastore"
"github.com/grafana/loki/v3/pkg/dataobj/uploader"
"github.com/grafana/loki/v3/pkg/kafka"
"github.com/grafana/loki/v3/pkg/scratch"
)
type partitionProcessor struct {
@@ -36,10 +37,11 @@ type partitionProcessor struct {
metastoreUpdater *metastore.Updater
// Builder initialization
builderOnce sync.Once
builderCfg logsobj.BuilderConfig
bucket objstore.Bucket
bufPool *sync.Pool
builderOnce sync.Once
builderCfg logsobj.BuilderConfig
bucket objstore.Bucket
bufPool *sync.Pool
scratchStore scratch.Store
// Idle stream handling
idleFlushTimeout time.Duration
@@ -70,6 +72,7 @@ func newPartitionProcessor(
uploaderCfg uploader.Config,
metastoreCfg metastore.Config,
bucket objstore.Bucket,
scratchStore scratch.Store,
tenantID string,
virtualShard int32,
topic string,
@@ -102,7 +105,7 @@ func newPartitionProcessor(
level.Error(logger).Log("msg", "failed to register uploader metrics", "err", err)
}
metastoreUpdater := metastore.NewUpdater(metastoreCfg, bucket, tenantID, logger)
metastoreUpdater := metastore.NewUpdater(metastoreCfg, bucket, scratchStore, tenantID, logger)
if err := metastoreUpdater.RegisterMetrics(reg); err != nil {
level.Error(logger).Log("msg", "failed to register metastore updater metrics", "err", err)
}
@@ -119,6 +122,7 @@ func newPartitionProcessor(
reg: reg,
builderCfg: builderCfg,
bucket: bucket,
scratchStore: scratchStore,
tenantID: []byte(tenantID),
metrics: metrics,
uploader: uploader,
@@ -186,7 +190,7 @@ func (p *partitionProcessor) initBuilder() error {
var initErr error
p.builderOnce.Do(func() {
// Dataobj builder
builder, err := logsobj.NewBuilder(p.builderCfg)
builder, err := logsobj.NewBuilder(p.builderCfg, p.scratchStore)
if err != nil {
initErr = err
return
pkg/dataobj/consumer/partition_processor_test.go
+1
View File
@@ -118,6 +118,7 @@ func newTestPartitionProcessor(_ *testing.T, clock quartz.Clock) *partitionProce
uploader.Config{},
metastore.Config{},
newMockBucket(),
nil,
"test-tenant",
0,
"test-topic",
pkg/dataobj/consumer/service.go
+9 -6
View File
@@ -21,6 +21,7 @@ import (
"github.com/grafana/loki/v3/pkg/kafka"
"github.com/grafana/loki/v3/pkg/kafka/client"
"github.com/grafana/loki/v3/pkg/kafka/partitionring/consumer"
"github.com/grafana/loki/v3/pkg/scratch"
)
const (
@@ -37,10 +38,11 @@ type Service struct {
eventsProducerClient *kgo.Client
eventConsumerClient *kgo.Client
cfg Config
mCfg metastore.Config
bucket objstore.Bucket
codec distributor.TenantPrefixCodec
cfg Config
mCfg metastore.Config
bucket objstore.Bucket
scratchStore scratch.Store
codec distributor.TenantPrefixCodec
// Partition management
partitionMtx sync.RWMutex
@@ -49,12 +51,13 @@ type Service struct {
bufPool *sync.Pool
}
func New(kafkaCfg kafka.Config, cfg Config, mCfg metastore.Config, topicPrefix string, bucket objstore.Bucket, instanceID string, partitionRing ring.PartitionRingReader, reg prometheus.Registerer, logger log.Logger) *Service {
func New(kafkaCfg kafka.Config, cfg Config, mCfg metastore.Config, topicPrefix string, bucket objstore.Bucket, scratchStore scratch.Store, instanceID string, partitionRing ring.PartitionRingReader, reg prometheus.Registerer, logger log.Logger) *Service {
s := &Service{
logger: log.With(logger, "component", groupName),
cfg: cfg,
mCfg: mCfg,
bucket: bucket,
scratchStore: scratchStore,
codec: distributor.TenantPrefixCodec(topicPrefix),
partitionHandlers: make(map[string]map[int32]*partitionProcessor),
reg: reg,
@@ -117,7 +120,7 @@ func (s *Service) handlePartitionsAssigned(ctx context.Context, client *kgo.Clie
}
for _, partition := range parts {
processor := newPartitionProcessor(ctx, client, s.cfg.BuilderConfig, s.cfg.UploaderConfig, s.mCfg, s.bucket, tenant, virtualShard, topic, partition, s.logger, s.reg, s.bufPool, s.cfg.IdleFlushTimeout, s.eventsProducerClient)
processor := newPartitionProcessor(ctx, client, s.cfg.BuilderConfig, s.cfg.UploaderConfig, s.mCfg, s.bucket, s.scratchStore, tenant, virtualShard, topic, partition, s.logger, s.reg, s.bufPool, s.cfg.IdleFlushTimeout, s.eventsProducerClient)
s.partitionHandlers[topic][partition] = processor
processor.start()
}
pkg/dataobj/encoder.go
+89 -117
View File
@@ -4,13 +4,14 @@ import (
"bytes"
"encoding/binary"
"fmt"
"slices"
"github.com/gogo/protobuf/proto"
"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/filemd"
"github.com/grafana/loki/v3/pkg/dataobj/internal/streamio"
"github.com/grafana/loki/v3/pkg/dataobj/internal/util/bufpool"
"github.com/grafana/loki/v3/pkg/dataobj/internal/util/protocodec"
"github.com/grafana/loki/v3/pkg/scratch"
)
var (
@@ -29,82 +30,50 @@ var (
legacySectionTypeLogs = SectionType{"github.com/grafana/loki", "logs"}
)
// TODO(rfratto): the memory footprint of [Encoder] can very slowly grow in
// memory as [bufpool] is filled with buffers with increasing capacity:
// each encoding pass has a different number of elements, shuffling which
// elements of the hierarchy get which pooled buffers.
//
// This means that elements that require more bytes will grow the capacity of
// the buffer and put the buffer back into the pool. Even if further encoding
// passes don't need that many bytes, the buffer is kept alive with its larger
// footprint. Given enough time, all buffers in the pool will have a large
// capacity.
//
// The bufpool package provides bucketed pools as a solution to, but this
// requires knowing how many bytes are needed.
//
// Encoder can eventually be moved to the bucketed pools by calculating a
// rolling maximum of encoding size used per element across usages of an
// Encoder instance. This would then allow larger buffers to be eventually
// reclaimed regardless of how often encoding is done.
// encoder encodes a data object. Data objects are hierarchical, split into
// distinct sections that contain their own hierarchy.
//
// To support hierarchical encoding, a set of Open* methods are provided to
// open a child element. Only one child element may be open at a given time;
// call Commit or Discard on a child element to close it.
type encoder struct {
startOffset int // Byte offset in the file where data starts after the header.
totalBytes int // Total bytes buffered in store so far.
sections []*filemd.SectionInfo
store scratch.Store
sections []sectionInfo // Sections buffered in store.
typesReady bool
dictionary []string
dictionaryLookup map[string]uint32
rawTypes []*filemd.SectionType
typeRefLookup map[SectionType]uint32
data *bytes.Buffer
}
// newEncoder creates a new Encoder which writes a data object to the provided
// writer.
func newEncoder() *encoder {
return &encoder{startOffset: len(magic)}
func newEncoder(store scratch.Store) *encoder {
return &encoder{
startOffset: len(magic),
store: store,
}
}
// AppendSection appends a section to the data object. AppendSection panics if
// typ is not SectionTypeLogs or SectionTypeStreams.
func (enc *encoder) AppendSection(typ SectionType, data, metadata []byte) {
if enc.data == nil {
// Lazily initialize enc.data. This allows an Encoder to persist for the
// lifetime of a dataobj.Builder without holding onto memory when no data
// is present.
enc.data = bufpool.GetUnsized()
}
var (
dataHandle = enc.store.Put(data)
metadataHandle = enc.store.Put(metadata)
)
info := &filemd.SectionInfo{
TypeRef: enc.getTypeRef(typ),
Layout: &filemd.SectionLayout{
Data: &filemd.Region{
Offset: uint64(enc.startOffset + enc.data.Len()),
Length: uint64(len(data)),
},
enc.sections = append(enc.sections, sectionInfo{
Type: typ,
Metadata: &filemd.Region{
Offset: uint64(enc.startOffset + enc.data.Len() + len(data)),
Length: uint64(len(metadata)),
},
},
}
Data: dataHandle,
Metadata: metadataHandle,
// bytes.Buffer.Write never fails.
enc.data.Grow(len(data) + len(metadata))
_, _ = enc.data.Write(data)
_, _ = enc.data.Write(metadata)
DataSize: len(data),
MetadataSize: len(metadata),
})
enc.sections = append(enc.sections, info)
enc.totalBytes += len(data) + len(metadata)
}
// getTypeRef returns the type reference for the given type or creates a new
@@ -174,61 +143,52 @@ func (enc *encoder) getDictionaryKey(text string) uint32 {
return key
}
// MetadataSize returns an estimate of the current size of the metadata for the
// data object. MetadataSize does not include the size of data appended or the
// currently open stream.
func (enc *encoder) MetadataSize() int { return proto.Size(enc.Metadata()) }
func (enc *encoder) Metadata() (proto.Message, error) {
sections := make([]*filemd.SectionInfo, len(enc.sections))
offset := enc.startOffset
for i, info := range enc.sections {
dataOffset := offset // Data starts at the current total offset
metaOffset := dataOffset + info.DataSize // Metadata starts right after data
sections[i] = &filemd.SectionInfo{
TypeRef: enc.getTypeRef(info.Type),
Layout: &filemd.SectionLayout{
Data: &filemd.Region{
Offset: uint64(dataOffset),
Length: uint64(info.DataSize),
},
Metadata: &filemd.Region{
Offset: uint64(metaOffset),
Length: uint64(info.MetadataSize),
},
},
}
offset += info.DataSize + info.MetadataSize
}
func (enc *encoder) Metadata() proto.Message {
sections := enc.sections[:len(enc.sections):cap(enc.sections)]
return &filemd.Metadata{
Sections: sections,
Sections: sections,
Dictionary: enc.dictionary,
Types: enc.rawTypes,
}
}, nil
}
func (enc *encoder) Bytes() int {
if enc.data == nil {
return 0
}
return enc.data.Len()
}
// Bytes returns the total number of bytes appended to the data object.
func (enc *encoder) Bytes() int { return enc.startOffset + enc.totalBytes }
// FlushSize returns the exact number of bytes that would be written upon
// calling [encoder.Flush].
func (enc *encoder) FlushSize() (int64, error) {
return enc.flush(streamio.Discard)
}
// Flush flushes any buffered data to the underlying writer. After flushing,
// enc is reset.
func (enc *encoder) Flush(w streamio.Writer) (int64, error) {
size, err := enc.flush(w)
if err != nil {
return size, err
}
enc.Reset()
return size, nil
}
func (enc *encoder) flush(w streamio.Writer) (int64, error) {
cw := countingWriter{w: w}
if enc.data == nil {
return cw.count, fmt.Errorf("empty Encoder")
}
metadataBuffer := bufpool.GetUnsized()
defer bufpool.PutUnsized(metadataBuffer)
// The file metadata should start with the version.
if err := streamio.WriteUvarint(metadataBuffer, fileFormatVersion); err != nil {
return cw.count, err
} else if err := protocodec.Encode(metadataBuffer, enc.Metadata()); err != nil {
return cw.count, err
// Flush converts all accumulated data into a [snapshot], allowing for streaming
// encoded bytes. [snapshot.Close] should be called when the snapshot is no
// longer needed to release sections.
//
// Callers must manually call [encoder.Reset] after calling Flush to prepare for
// the next encoding session. This is done to allow callers to ensure that the
// returned snapshot is complete before the encoder is reset.
func (enc *encoder) Flush() (*snapshot, error) {
if len(enc.sections) == 0 {
return nil, fmt.Errorf("empty Encoder")
}
// The overall structure is:
@@ -236,33 +196,48 @@ func (enc *encoder) flush(w streamio.Writer) (int64, error) {
// header:
// [magic]
// body:
// [data]
// [metadata]
// [section data + metadata]
// tailer:
// [file metadata version]
// [file metadata]
// [file metadata size (32 bits)]
// [magic]
//
// The file metadata size *must not* be varint since we need the last 8 bytes
// of the file to consistently retrieve the tailer.
if _, err := cw.Write(magic); err != nil {
return cw.count, fmt.Errorf("writing magic header: %w", err)
} else if _, err := cw.Write(enc.data.Bytes()); err != nil {
return cw.count, fmt.Errorf("writing data: %w", err)
} else if _, err := cw.Write(metadataBuffer.Bytes()); err != nil {
return cw.count, fmt.Errorf("writing metadata: %w", err)
} else if err := binary.Write(&cw, binary.LittleEndian, uint32(metadataBuffer.Len())); err != nil {
return cw.count, fmt.Errorf("writing metadata size: %w", err)
} else if _, err := cw.Write(magic); err != nil {
return cw.count, fmt.Errorf("writing magic tailer: %w", err)
var tailerBuffer bytes.Buffer
metadata, err := enc.Metadata()
if err != nil {
return nil, fmt.Errorf("building metadata: %w", err)
} else if err := streamio.WriteUvarint(&tailerBuffer, fileFormatVersion); err != nil {
return nil, err
} else if err := protocodec.Encode(&tailerBuffer, metadata); err != nil {
return nil, err
} else if err := binary.Write(&tailerBuffer, binary.LittleEndian, uint32(tailerBuffer.Len())); err != nil {
return nil, fmt.Errorf("writing metadata size: %w", err)
} else if _, err := tailerBuffer.Write(magic); err != nil {
return nil, fmt.Errorf("writing magic tailer: %w", err)
}
return cw.count, nil
snapshot, err := newSnapshot(
enc.store,
magic, // header
slices.Clone(enc.sections),
tailerBuffer.Bytes(), // tailer
)
if err != nil {
return nil, fmt.Errorf("creating snapshot: %w", err)
}
return snapshot, nil
}
// Reset resets the Encoder to a fresh state.
func (enc *encoder) Reset() {
enc.startOffset = len(magic)
enc.totalBytes = 0
enc.sections = nil
@@ -270,9 +245,6 @@ func (enc *encoder) Reset() {
enc.dictionary = nil
enc.rawTypes = nil
enc.typeRefLookup = nil
bufpool.PutUnsized(enc.data)
enc.data = nil
}
type countingWriter struct {
pkg/dataobj/encoder_test.go
+2 -1
View File
@@ -6,6 +6,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/grafana/loki/v3/pkg/dataobj/internal/metadata/filemd"
"github.com/grafana/loki/v3/pkg/scratch"
)
func Test_encoder_typeRefs(t *testing.T) {
@@ -79,7 +80,7 @@ func Test_encoder_typeRefs(t *testing.T) {
},
}
enc := newEncoder()
enc := newEncoder(scratch.NewMemory())
// Test are run sequentially so we can check the behaviour of streaming types
// in.
pkg/dataobj/index/builder.go
+7 -4
View File
@@ -27,6 +27,7 @@ import (
"github.com/grafana/loki/v3/pkg/dataobj/metastore"
"github.com/grafana/loki/v3/pkg/kafka"
"github.com/grafana/loki/v3/pkg/kafka/client"
"github.com/grafana/loki/v3/pkg/scratch"
)
type Config struct {
@@ -71,8 +72,9 @@ type Builder struct {
bufferedEvents map[string][]metastore.ObjectWrittenEvent
// Builder initialization
builderCfg indexobj.BuilderConfig
bucket objstore.Bucket
builderCfg indexobj.BuilderConfig
bucket objstore.Bucket
scratchStore scratch.Store
// Metrics
metrics *indexBuilderMetrics
@@ -91,6 +93,7 @@ func NewIndexBuilder(
logger log.Logger,
instanceID string,
bucket objstore.Bucket,
scratchStore scratch.Store,
reg prometheus.Registerer,
) (*Builder, error) {
kafkaCfg.AutoCreateTopicEnabled = true
@@ -121,7 +124,7 @@ func NewIndexBuilder(
return nil, fmt.Errorf("failed to register metrics for index builder: %w", err)
}
builder, err := indexobj.NewBuilder(cfg.BuilderConfig)
builder, err := indexobj.NewBuilder(cfg.BuilderConfig, scratchStore)
if err != nil {
return nil, fmt.Errorf("failed to create index builder: %w", err)
}
@@ -320,7 +323,7 @@ func (p *Builder) buildIndex(events []metastore.ObjectWrittenEvent) error {
return fmt.Errorf("failed to upload index: %w", err)
}
metastoreUpdater := metastore.NewUpdater(p.mCfg, indexStorageBucket, events[0].Tenant, p.logger)
metastoreUpdater := metastore.NewUpdater(p.mCfg, indexStorageBucket, p.scratchStore, events[0].Tenant, p.logger)
if minTime.IsZero() || maxTime.IsZero() {
return errors.New("failed to get min/max timestamps")
}
pkg/dataobj/index/builder_test.go
+2 -1
View File
@@ -43,7 +43,7 @@ func buildLogObject(t *testing.T, app string, path string, bucket objstore.Bucke
BufferSize: 4 * 1024 * 1024,
SectionStripeMergeLimit: 2,
})
}, nil)
require.NoError(t, err)
for i := 0; i < 10; i++ {
@@ -104,6 +104,7 @@ func TestIndexBuilder(t *testing.T) {
log.NewNopLogger(),
"instance-id",
bucket,
nil,
prometheus.NewRegistry(),
)
require.NoError(t, err)
pkg/dataobj/index/calculate_test.go
+2 -2
View File
@@ -38,7 +38,7 @@ func createTestLogObject(t *testing.T) *dataobj.Object {
TargetSectionSize: 1 << 21,
BufferSize: 2048 * 8,
SectionStripeMergeLimit: 2,
})
}, nil)
require.NoError(t, err)
// Add test streams with structured metadata
@@ -100,7 +100,7 @@ func createTestLogObject(t *testing.T) *dataobj.Object {
func TestCalculator_Calculate(t *testing.T) {
t.Run("successful calculation", func(t *testing.T) {
indexBuilder, err := indexobj.NewBuilder(testCalculatorConfig)
indexBuilder, err := indexobj.NewBuilder(testCalculatorConfig, nil)
require.NoError(t, err)
calculator := NewCalculator(indexBuilder)
pkg/dataobj/index/indexobj/builder.go
+3 -2
View File
@@ -18,6 +18,7 @@ import (
"github.com/grafana/loki/v3/pkg/dataobj/sections/indexpointers"
"github.com/grafana/loki/v3/pkg/dataobj/sections/pointers"
"github.com/grafana/loki/v3/pkg/dataobj/sections/streams"
"github.com/grafana/loki/v3/pkg/scratch"
)
// ErrBuilderFull is returned by [Builder.Append] when the buffer is
@@ -138,7 +139,7 @@ const (
// NewBuilder creates a new [Builder] which stores log-oriented data objects.
//
// NewBuilder returns an error if the provided config is invalid.
func NewBuilder(cfg BuilderConfig) (*Builder, error) {
func NewBuilder(cfg BuilderConfig, scratchStore scratch.Store) (*Builder, error) {
if err := cfg.Validate(); err != nil {
return nil, err
}
@@ -157,7 +158,7 @@ func NewBuilder(cfg BuilderConfig) (*Builder, error) {
labelCache: labelCache,
builder: dataobj.NewBuilder(),
builder: dataobj.NewBuilder(scratchStore),
streams: streams.NewBuilder(metrics.streams, int(cfg.TargetPageSize)),
pointers: pointers.NewBuilder(metrics.pointers, int(cfg.TargetPageSize)),
indexPointers: indexpointers.NewBuilder(metrics.indexPointers, int(cfg.TargetPageSize)),
pkg/dataobj/index/indexobj/builder_test.go
+3 -3
View File
@@ -62,7 +62,7 @@ func TestBuilder(t *testing.T) {
}
t.Run("Build", func(t *testing.T) {
builder, err := NewBuilder(testBuilderConfig)
builder, err := NewBuilder(testBuilderConfig, nil)
require.NoError(t, err)
for _, stream := range testStreams {
@@ -90,7 +90,7 @@ func TestBuilder_Append(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
builder, err := NewBuilder(testBuilderConfig)
builder, err := NewBuilder(testBuilderConfig, nil)
require.NoError(t, err)
i := 0
@@ -120,7 +120,7 @@ func TestBuilder_AppendIndexPointer(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
builder, err := NewBuilder(testBuilderConfig)
builder, err := NewBuilder(testBuilderConfig, nil)
require.NoError(t, err)
i := 0
pkg/dataobj/metastore/metastore_test.go
+3 -3
View File
@@ -35,7 +35,7 @@ func BenchmarkWriteMetastores(b *testing.B) {
Updater: UpdaterConfig{
StorageFormat: bm.format,
},
}, bucket, tenantID, log.NewNopLogger())
}, bucket, nil, tenantID, log.NewNopLogger())
// Set limits for the test
m.backoff = backoff.New(context.TODO(), backoff.Config{
@@ -87,7 +87,7 @@ func TestWriteMetastores(t *testing.T) {
Updater: UpdaterConfig{
StorageFormat: tt.format,
},
}, bucket, tenantID, log.NewNopLogger())
}, bucket, nil, tenantID, log.NewNopLogger())
// Set limits for the test
m.backoff = backoff.New(context.TODO(), backoff.Config{
@@ -239,7 +239,7 @@ func TestDataObjectsPathsV1(t *testing.T) {
IndexStoragePrefix: tt.prefix,
EnabledTenantIDs: tt.enabledTenantIDs,
},
}, bucket, tenantID, log.NewNopLogger())
}, bucket, nil, tenantID, log.NewNopLogger())
// Set limits for the test
m.backoff = backoff.New(context.TODO(), backoff.Config{
pkg/dataobj/metastore/object_test.go
+4 -4
View File
@@ -253,7 +253,7 @@ func TestSectionsForStreamMatchers(t *testing.T) {
TargetSectionSize: 128,
BufferSize: 1024 * 1024,
SectionStripeMergeLimit: 2,
})
}, nil)
require.NoError(t, err)
for i, ts := range testStreams {
@@ -286,7 +286,7 @@ func TestSectionsForStreamMatchers(t *testing.T) {
path, err := uploader.Upload(context.Background(), obj)
require.NoError(t, err)
metastoreUpdater := NewUpdater(Config{}, bucket, tenantID, log.NewNopLogger())
metastoreUpdater := NewUpdater(Config{}, bucket, nil, tenantID, log.NewNopLogger())
err = metastoreUpdater.Update(context.Background(), path, minTime, maxTime)
require.NoError(t, err)
@@ -362,13 +362,13 @@ func newTestDataBuilder(t *testing.T, tenantID string) *testDataBuilder {
TargetSectionSize: 1024 * 1024, // 1MB
BufferSize: 1024 * 1024, // 1MB
SectionStripeMergeLimit: 2,
})
}, nil)
require.NoError(t, err)
logger := log.NewLogfmtLogger(os.Stdout)
logger = log.With(logger, "test", t.Name())
meta := NewUpdater(Config{}, bucket, tenantID, logger)
meta := NewUpdater(Config{}, bucket, nil, tenantID, logger)
require.NoError(t, meta.RegisterMetrics(prometheus.NewPedanticRegistry()))
uploader := uploader.New(uploader.Config{SHAPrefixSize: 2}, bucket, tenantID, logger)
pkg/dataobj/metastore/updater.go
+12 -8
View File
@@ -23,6 +23,7 @@ import (
"github.com/grafana/loki/v3/pkg/dataobj/sections/indexpointers"
"github.com/grafana/loki/v3/pkg/dataobj/sections/streams"
"github.com/grafana/loki/v3/pkg/logproto"
"github.com/grafana/loki/v3/pkg/scratch"
)
const (
@@ -58,6 +59,7 @@ type Updater struct {
tenantID string
metrics *metastoreMetrics
bucket objstore.Bucket
scratchStore scratch.Store
logger log.Logger
backoff *backoff.Backoff
buf *bytes.Buffer
@@ -65,15 +67,16 @@ type Updater struct {
builderOnce sync.Once
}
func NewUpdater(cfg Config, bucket objstore.Bucket, tenantID string, logger log.Logger) *Updater {
func NewUpdater(cfg Config, bucket objstore.Bucket, scratchStore scratch.Store, tenantID string, logger log.Logger) *Updater {
metrics := newMetastoreMetrics()
return &Updater{
cfg: cfg,
bucket: bucket,
metrics: metrics,
logger: logger,
tenantID: tenantID,
cfg: cfg,
bucket: bucket,
scratchStore: scratchStore,
metrics: metrics,
logger: logger,
tenantID: tenantID,
backoff: backoff.New(context.TODO(), backoff.Config{
MinBackoff: 50 * time.Millisecond,
MaxBackoff: 10 * time.Second,
@@ -93,7 +96,7 @@ func (m *Updater) UnregisterMetrics(reg prometheus.Registerer) {
func (m *Updater) initBuilder() error {
var initErr error
m.builderOnce.Do(func() {
metastoreBuilder, err := logsobj.NewBuilder(metastoreBuilderCfg)
metastoreBuilder, err := logsobj.NewBuilder(metastoreBuilderCfg, m.scratchStore)
if err != nil {
initErr = err
return
@@ -107,7 +110,7 @@ func (m *Updater) initBuilder() error {
BufferSize: metastoreBuilderCfg.BufferSize,
TargetSectionSize: metastoreBuilderCfg.TargetSectionSize,
SectionStripeMergeLimit: metastoreBuilderCfg.SectionStripeMergeLimit,
})
}, m.scratchStore)
if err != nil {
initErr = err
return
@@ -221,6 +224,7 @@ func (m *Updater) Update(ctx context.Context, dataobjPath string, minTimestamp,
m.metrics.incMetastoreWrites(statusFailure)
m.backoff.Wait()
}
// Reset at the end too so we don't leave our memory hanging around between calls.
m.metastoreBuilder.Reset()
}
pkg/dataobj/metastore/updater_test.go
+3 -3
View File
@@ -25,7 +25,7 @@ import (
func TestUpdater(t *testing.T) {
t.Run("append new top-level object to deprecated metastore v1", func(t *testing.T) {
tenantID := "test"
builder, err := logsobj.NewBuilder(metastoreBuilderCfg)
builder, err := logsobj.NewBuilder(metastoreBuilderCfg, nil)
require.NoError(t, err)
ls := labels.New(
@@ -60,7 +60,7 @@ func TestUpdater(t *testing.T) {
TargetSectionSize: metastoreBuilderCfg.TargetSectionSize,
BufferSize: metastoreBuilderCfg.BufferSize,
SectionStripeMergeLimit: metastoreBuilderCfg.SectionStripeMergeLimit,
})
}, nil)
require.NoError(t, err)
err = builder.AppendIndexPointer("testdata/metastore.obj", unixTime(10), unixTime(20))
@@ -86,7 +86,7 @@ func TestUpdater(t *testing.T) {
TargetSectionSize: metastoreBuilderCfg.TargetSectionSize,
BufferSize: metastoreBuilderCfg.BufferSize,
SectionStripeMergeLimit: metastoreBuilderCfg.SectionStripeMergeLimit,
})
}, nil)
require.NoError(t, err)
bucket := newInMemoryBucket(t, tenantID, unixTime(0), nil)
pkg/dataobj/querier/store_test.go
+2 -2
View File
@@ -530,10 +530,10 @@ func newTestDataBuilder(t *testing.T, tenantID string) *testDataBuilder {
BufferSize: 1024 * 1024, // 1MB
SectionStripeMergeLimit: 2,
})
}, nil)
require.NoError(t, err)
meta := metastore.NewUpdater(metastore.Config{}, bucket, tenantID, log.NewNopLogger())
meta := metastore.NewUpdater(metastore.Config{}, bucket, nil, tenantID, log.NewNopLogger())
require.NoError(t, meta.RegisterMetrics(prometheus.NewRegistry()))
uploader := uploader.New(uploader.Config{SHAPrefixSize: 2}, bucket, tenantID, log.NewNopLogger())
pkg/dataobj/sections/indexpointers/builder_test.go
+2 -3
View File
@@ -27,9 +27,8 @@ func TestBuilder(t *testing.T) {
ib.Append(p.path, p.start, p.end)
}
b := dataobj.NewBuilder()
err := b.Append(ib)
require.NoError(t, err)
b := dataobj.NewBuilder(nil)
require.NoError(t, b.Append(ib))
obj, closer, err := b.Flush()
require.NoError(t, err)
pkg/dataobj/sections/indexpointers/row_reader_test.go
+1 -1
View File
@@ -36,7 +36,7 @@ func buildIndexPointersDecoder(t *testing.T, pageSize int) *Section {
s.Append(d.Path, d.StartTs, d.EndTs)
}
builder := dataobj.NewBuilder()
builder := dataobj.NewBuilder(nil)
require.NoError(t, builder.Append(s))
obj, closer, err := builder.Flush()
pkg/dataobj/sections/logs/builder_test.go
+1 -1
View File
@@ -83,7 +83,7 @@ func Test(t *testing.T) {
}
func buildObject(lt *logs.Builder) (*dataobj.Object, io.Closer, error) {
builder := dataobj.NewBuilder()
builder := dataobj.NewBuilder(nil)
if err := builder.Append(lt); err != nil {
return nil, nil, err
}
pkg/dataobj/sections/logs/reader_test.go
+1 -1
View File
@@ -99,7 +99,7 @@ func buildSection(t *testing.T, recs []logs.Record) *logs.Section {
sectionBuilder.Append(rec)
}
objectBuilder := dataobj.NewBuilder()
objectBuilder := dataobj.NewBuilder(nil)
require.NoError(t, objectBuilder.Append(sectionBuilder))
obj, closer, err := objectBuilder.Flush()
pkg/dataobj/sections/logs/row_reader_test.go
+2 -3
View File
@@ -49,9 +49,8 @@ func buildSection(t *testing.T) *Section {
Line: []byte("test2"),
})
b := dataobj.NewBuilder()
err := b.Append(logsBuilder)
require.NoError(t, err)
b := dataobj.NewBuilder(nil)
require.NoError(t, b.Append(logsBuilder))
obj, closer, err := b.Flush()
require.NoError(t, err)
pkg/dataobj/sections/pointers/builder_test.go
+1 -1
View File
@@ -149,7 +149,7 @@ func TestAddingColumnIndexes(t *testing.T) {
}
func buildObject(st *Builder) (*dataobj.Object, io.Closer, error) {
builder := dataobj.NewBuilder()
builder := dataobj.NewBuilder(nil)
if err := builder.Append(st); err != nil {
return nil, nil, err
}
pkg/dataobj/sections/pointers/row_reader_test.go
+1 -1
View File
@@ -43,7 +43,7 @@ func buildPointersDecoder(t *testing.T, pageSize int) *Section {
}
}
builder := dataobj.NewBuilder()
builder := dataobj.NewBuilder(nil)
require.NoError(t, builder.Append(s))
obj, closer, err := builder.Flush()
pkg/dataobj/sections/streams/builder_test.go
+1 -1
View File
@@ -79,7 +79,7 @@ func copyLabels(in labels.Labels) labels.Labels {
}
func buildObject(st *streams.Builder) (*dataobj.Object, io.Closer, error) {
builder := dataobj.NewBuilder()
builder := dataobj.NewBuilder(nil)
if err := builder.Append(st); err != nil {
return nil, nil, err
}
pkg/dataobj/sections/streams/row_reader_test.go
+1 -1
View File
@@ -88,7 +88,7 @@ func buildStreamsSection(t *testing.T, pageSize int) *streams.Section {
s.Record(d.Labels, d.Timestamp, d.UncompressedSize)
}
builder := dataobj.NewBuilder()
builder := dataobj.NewBuilder(nil)
require.NoError(t, builder.Append(s))
obj, closer, err := builder.Flush()
pkg/dataobj/snapshot.go
+160
View File
@@ -0,0 +1,160 @@
package dataobj
import (
"bytes"
"errors"
"fmt"
"io"
"sort"
"github.com/grafana/loki/v3/pkg/scratch"
)
// A snapshot represents a complete data object where all sections are stored in
// a [scratch.Store].
//
// snapshot allows for reading the data object without materializing the entire
// object into memory. When using a disk-basked scratch store, reads are served by
// opening and reading the underlying files on demand.
type snapshot struct {
store scratch.Store
// Regions to read from (sorted by offset).
regions []snapshotRegion
header []byte
sections []sectionInfo
tailer []byte
}
type sectionInfo struct {
Type SectionType
Data, Metadata scratch.Handle
DataSize, MetadataSize int
}
// newSnapshot creates a new snapshot for the given data. [sectionHandle]s
// passed to the newSnapshot are owned by the snapshot, and are deleted
// by the snapshot when calling [snapshot.Close].
func newSnapshot(store scratch.Store, header []byte, sections []sectionInfo, tailer []byte) (*snapshot, error) {
s := &snapshot{
store: store,
header: header,
sections: sections,
tailer: tailer,
}
if err := s.initRegions(); err != nil {
return nil, err
}
return s, nil
}
func (s *snapshot) initRegions() error {
s.addRegion(int64(len(s.header)), func() (io.ReadSeekCloser, error) {
return nopReadSeekerCloser{bytes.NewReader(s.header)}, nil
})
for _, section := range s.sections {
s.addRegion(int64(section.DataSize), func() (io.ReadSeekCloser, error) { return s.store.Read(section.Data) })
s.addRegion(int64(section.MetadataSize), func() (io.ReadSeekCloser, error) { return s.store.Read(section.Metadata) })
}
s.addRegion(int64(len(s.tailer)), func() (io.ReadSeekCloser, error) {
return nopReadSeekerCloser{bytes.NewReader(s.tailer)}, nil
})
return nil
}
type nopReadSeekerCloser struct{ io.ReadSeeker }
func (nopReadSeekerCloser) Close() error { return nil }
func (s *snapshot) addRegion(size int64, newReader func() (io.ReadSeekCloser, error)) {
s.regions = append(s.regions, snapshotRegion{
offset: s.Size(), // Use current size as offset
length: size,
NewReader: newReader,
})
}
// ReadAt implements [io.ReaderAt], returning bytes over the entire encoded
// data object.
func (s *snapshot) ReadAt(p []byte, off int64) (n int, err error) {
if off < 0 {
return 0, fmt.Errorf("invalid offset: %d", off)
}
for len(p) > 0 {
absoluteOffset := off + int64(n)
// Binary search to find the first region that ends after absoluteOffset.
regionIdx := sort.Search(len(s.regions), func(i int) bool {
return s.regions[i].offset+s.regions[i].length > absoluteOffset
})
if regionIdx == len(s.regions) {
return n, io.EOF
}
region := s.regions[regionIdx]
localOffset := absoluteOffset - region.offset
// Open our region at the correct offset.
reader, err := region.NewReader()
if err != nil {
return n, fmt.Errorf("creating region reader: %w", err)
} else if _, err := reader.Seek(localOffset, io.SeekStart); err != nil {
return n, fmt.Errorf("seeking region reader: %w", err)
}
readSize := min(len(p), int(region.length-localOffset))
readCount, err := readFullAndClose(reader, p[:readSize])
n += readCount
if err != nil {
return n, fmt.Errorf("reading region: %w", err)
}
p = p[readCount:]
}
return n, nil
}
func readFullAndClose(r io.ReadCloser, dst []byte) (int, error) {
defer r.Close()
return io.ReadFull(r, dst)
}
// Size returns the total size of the snapshot in bytes.
func (s *snapshot) Size() int64 {
if len(s.regions) == 0 {
return 0
}
lastRegion := s.regions[len(s.regions)-1]
return lastRegion.offset + lastRegion.length
}
// Close releases resources associated with the snapsshot, removing all section
// handles from the backing scratch store.
func (s *snapshot) Close() error {
var errs []error
for _, section := range s.sections {
errs = append(errs, s.store.Remove(section.Data))
errs = append(errs, s.store.Remove(section.Metadata))
}
return errors.Join(errs...)
}
// snapshotRegion is a contiguous region of a snapshot's data.
type snapshotRegion struct {
offset int64
length int64
// NewReader attempts to open a reader for the region's data.
NewReader func() (io.ReadSeekCloser, error)
}
pkg/dataobj/snapshot_test.go
+194
View File
@@ -0,0 +1,194 @@
package dataobj
import (
"io"
"testing"
"github.com/stretchr/testify/require"
"github.com/grafana/loki/v3/pkg/scratch"
)
func Test_snapshot(t *testing.T) {
store := scratch.NewMemory()
sections := []sectionInfo{
{
Type: SectionType{},
Data: store.Put([]byte("example data 1")),
Metadata: store.Put([]byte("example metadata 1")),
DataSize: len("example data 1"),
MetadataSize: len("example metadata 1"),
},
{
Type: SectionType{},
Data: store.Put([]byte("example data 2")),
Metadata: store.Put([]byte("example metadata 2")),
DataSize: len("example data 2"),
MetadataSize: len("example metadata 2"),
},
{
Type: SectionType{},
Data: store.Put([]byte("example data 3")),
Metadata: store.Put([]byte("example metadata 3")),
DataSize: len("example data 3"),
MetadataSize: len("example metadata 3"),
},
}
var (
header = []byte("example header")
tailer = []byte("example tailer")
)
snapshot, err := newSnapshot(store, header, sections, tailer)
require.NoError(t, err)
t.Run("full region", func(t *testing.T) {
headerData := make([]byte, len(header))
n, err := snapshot.ReadAt(headerData, 0)
require.NoError(t, err)
require.Equal(t, len(headerData), n)
require.Equal(t, header, headerData)
// Test reading the first section's data region completely
sectionData := make([]byte, 14) // "example data 1" is 14 bytes
n, err = snapshot.ReadAt(sectionData, 14) // offset after header
require.NoError(t, err)
require.Equal(t, 14, n)
require.Equal(t, "example data 1", string(sectionData))
})
t.Run("full section", func(t *testing.T) {
fullSectionData := make([]byte, sections[0].DataSize)
n, err := snapshot.ReadAt(fullSectionData, int64(sections[0].DataSize))
require.NoError(t, err)
require.Equal(t, sections[0].DataSize, n)
require.Equal(t, "example data 1", string(fullSectionData))
})
t.Run("partial region (start)", func(t *testing.T) {
expect := "example"
partialData := make([]byte, len(expect))
n, err := snapshot.ReadAt(partialData, 0)
require.NoError(t, err)
require.Equal(t, len(expect), n)
require.Equal(t, expect, string(partialData))
})
t.Run("partial region (middle)", func(t *testing.T) {
expect := "data"
// Read "data" from "example data 1"
data := make([]byte, len(expect))
n, err := snapshot.ReadAt(data, int64(len(header)+len("example ")))
require.NoError(t, err)
require.Equal(t, len(expect), n)
require.Equal(t, expect, string(data))
})
t.Run("multiple regions", func(t *testing.T) {
expected := "example data 1example metadata 1"
// Test reading two entire sections.
data := make([]byte, len(expected))
n, err := snapshot.ReadAt(data, int64(len(header)))
require.NoError(t, err)
require.Equal(t, len(expected), n)
require.Equal(t, expected, string(data))
})
t.Run("cross region", func(t *testing.T) {
// Test reading across multiple regions with ReadAt, where we start
// partially in the starting region and end partially in the ending
// region.
// Read from middle of header through start of section 1 data
expect := " headerexamp" // " header" + "examp" from "example data 1"
data := make([]byte, len(expect))
n, err := snapshot.ReadAt(data, int64(len("example"))) // Start after "example" from header
require.NoError(t, err)
require.Equal(t, len(expect), n)
require.Equal(t, expect, string(data))
})
t.Run("full range", func(t *testing.T) {
expect := "example header" +
"example data 1" + "example metadata 1" +
"example data 2" + "example metadata 2" +
"example data 3" + "example metadata 3" +
"example tailer"
// Test reading the entire range with ReadAt.
totalSize := snapshot.Size()
require.Equal(t, len(expect), int(totalSize))
data := make([]byte, len(expect))
n, err := snapshot.ReadAt(data, 0)
require.NoError(t, err)
require.Equal(t, len(expect), n)
require.Equal(t, expect, string(data))
})
t.Run("full range (io.SectionReader)", func(t *testing.T) {
// This tests that our implementation of [io.ReaderAt] is compatible
// with [io.SectionReader]. If it fails, we likely have a bug fulfilling
// the contract of [io.ReaderAt].
expect := "example header" +
"example data 1" + "example metadata 1" +
"example data 2" + "example metadata 2" +
"example data 3" + "example metadata 3" +
"example tailer"
totalSize := snapshot.Size()
sr := io.NewSectionReader(snapshot, 0, totalSize)
bb, err := io.ReadAll(sr)
require.NoError(t, err)
require.Equal(t, expect, string(bb))
})
t.Run("at EOF", func(t *testing.T) {
totalSize := snapshot.Size()
// Reading at EOF should produce no bytes and an EOF error.
n, err := snapshot.ReadAt(make([]byte, 1), totalSize)
require.Zero(t, n, "reading at EOF should produce no bytes")
require.ErrorIs(t, err, io.EOF, "reading at EOF should produce an EOF error")
})
t.Run("beyond EOF", func(t *testing.T) {
totalSize := snapshot.Size()
// Reading fully beyond EOF should produce no bytes and an EOF error.
n, err := snapshot.ReadAt(make([]byte, 10), totalSize+10)
require.Equal(t, 0, n, "reading beyond EOF should produce no bytes")
require.ErrorIs(t, err, io.EOF, "reading beyond EOF should produce an EOF error")
})
t.Run("through EOF", func(t *testing.T) {
totalSize := snapshot.Size()
expect := "tailer" // End of "example tailer"
// Request more bytes than available should return available data and
// EOF error.
data := make([]byte, 20)
n, err := snapshot.ReadAt(data, totalSize-int64(len(expect)))
require.Equal(t, len(expect), n, "reading through EOF should return available data")
require.ErrorIs(t, err, io.EOF, "reading through EOF should still return EOF")
require.Equal(t, expect, string(data[:n]))
})
require.NoError(t, snapshot.Close())
for _, section := range sections {
require.ErrorAs(t, store.Remove(section.Data), new(scratch.HandleNotFoundError), "section should be deleted after closing a snapshot")
require.ErrorAs(t, store.Remove(section.Metadata), new(scratch.HandleNotFoundError), "section should be deleted after closing a snapshot")
}
}
pkg/engine/executor/dataobjscan_test.go
+1 -1
View File
@@ -349,7 +349,7 @@ func buildDataobj(t testing.TB, streams []logproto.Stream) *dataobj.Object {
TargetSectionSize: 32_000,
BufferSize: 8_000,
SectionStripeMergeLimit: 2,
})
}, nil)
require.NoError(t, err)
for _, stream := range streams {
pkg/engine/executor/streams_view_test.go
+1 -1
View File
@@ -144,7 +144,7 @@ func buildStreamsSection(t *testing.T, streamLabels []labels.Labels) *streams.Se
_ = streamsBuilder.Record(stream, time.Now().UTC(), 0)
}
objBuilder := dataobj.NewBuilder()
objBuilder := dataobj.NewBuilder(nil)
require.NoError(t, objBuilder.Append(streamsBuilder), "failed to append streams section")
obj, closer, err := objBuilder.Flush()
pkg/logql/bench/store_dataobj.go
+4 -4
View File
@@ -79,18 +79,18 @@ func NewDataObjStore(dir, tenantID string) (*DataObjStore, error) {
BufferSize: 16 * 1024 * 1024, // 16MB
SectionStripeMergeLimit: 2,
})
}, nil)
if err != nil {
return nil, fmt.Errorf("failed to create builder: %w", err)
}
logger := level.NewFilter(log.NewLogfmtLogger(os.Stdout), level.AllowWarn())
meta := metastore.NewUpdater(metastore.Config{}, bucket, tenantID, logger)
meta := metastore.NewUpdater(metastore.Config{}, bucket, nil, tenantID, logger)
uploader := uploader.New(uploader.Config{SHAPrefixSize: 2}, bucket, tenantID, logger)
// Create prefixed bucket & metastore for indexes
indexWriterBucket := objstore.NewPrefixedBucket(bucket, indexDirPrefix)
indexMetastore := metastore.NewUpdater(metastore.Config{}, indexWriterBucket, tenantID, logger)
indexMetastore := metastore.NewUpdater(metastore.Config{}, indexWriterBucket, nil, tenantID, logger)
return &DataObjStore{
dir: storeDir,
@@ -216,7 +216,7 @@ func (s *DataObjStore) buildIndex() error {
BufferSize: 16 * 1024 * 1024, // 16MB
SectionStripeMergeLimit: 2,
})
}, nil)
if err != nil {
return fmt.Errorf("failed to create index builder: %w", err)
}
pkg/loki/common/common.go
+8
View File
@@ -49,6 +49,13 @@ type Config struct {
// CompactorAddress is the grpc address of the compactor in the form host:port
CompactorGRPCAddress string `yaml:"compactor_grpc_address"`
// ScratchPath is a path to a directory where Loki can store temporary data.
// All files in this path with the extension ".lokiscratch" will be deleted
// on startup.
//
// If ScratchPath is not set, Loki will buffer temporary data in memory.
ScratchPath string `yaml:"scratch_path" experimental:"true"`
}
func (c *Config) RegisterFlags(f *flag.FlagSet) {
@@ -67,6 +74,7 @@ func (c *Config) RegisterFlags(f *flag.FlagSet) {
f.StringVar(&c.CompactorAddress, "common.compactor-address", "", "the http address of the compactor in the form http://host:port")
f.StringVar(&c.CompactorGRPCAddress, "common.compactor-grpc-address", "", "the grpc address of the compactor in the form host:port")
f.StringVar(&c.PathPrefix, "common.path-prefix", "", "prefix for the path")
f.StringVar(&c.ScratchPath, "common.scratch-path", "", "Experimental: path to use for temporary data, where scratch data is supported.")
}
type Storage struct {
pkg/loki/loki.go
+6 -2
View File
@@ -62,6 +62,7 @@ import (
"github.com/grafana/loki/v3/pkg/ruler/rulestore"
"github.com/grafana/loki/v3/pkg/runtime"
"github.com/grafana/loki/v3/pkg/scheduler"
"github.com/grafana/loki/v3/pkg/scratch"
internalserver "github.com/grafana/loki/v3/pkg/server"
"github.com/grafana/loki/v3/pkg/storage"
"github.com/grafana/loki/v3/pkg/storage/config"
@@ -455,6 +456,7 @@ type Loki struct {
blockScheduler *blockscheduler.BlockScheduler
dataObjConsumer *consumer.Service
dataObjIndexBuilder *dataobjindex.Builder
scratchStore scratch.Store
ClientMetrics storage.ClientMetrics
deleteClientMetrics *deletion.DeleteRequestClientMetrics
@@ -798,6 +800,7 @@ func (t *Loki) setupModuleManager() error {
mm.RegisterModule(UI, t.initUI)
mm.RegisterModule(DataObjConsumer, t.initDataObjConsumer)
mm.RegisterModule(DataObjIndexBuilder, t.initDataObjIndexBuilder)
mm.RegisterModule(ScratchStore, t.initScratchStore)
mm.RegisterModule(All, nil)
mm.RegisterModule(Read, nil)
@@ -843,8 +846,9 @@ func (t *Loki) setupModuleManager() error {
BlockBuilder: {PartitionRing, Store, Server, UI},
BlockScheduler: {Server, UI},
DataObjExplorer: {Server, UI},
DataObjConsumer: {PartitionRing, Server, UI},
DataObjIndexBuilder: {Server, UI},
DataObjConsumer: {ScratchStore, PartitionRing, Server, UI},
DataObjIndexBuilder: {ScratchStore, Server, UI},
ScratchStore: {},
Read: {QueryFrontend, Querier},
Write: {Ingester, Distributor, PatternIngester},
pkg/loki/modules.go
+20
View File
@@ -82,6 +82,7 @@ import (
"github.com/grafana/loki/v3/pkg/runtime"
"github.com/grafana/loki/v3/pkg/scheduler"
"github.com/grafana/loki/v3/pkg/scheduler/schedulerpb"
"github.com/grafana/loki/v3/pkg/scratch"
"github.com/grafana/loki/v3/pkg/storage"
"github.com/grafana/loki/v3/pkg/storage/bucket"
"github.com/grafana/loki/v3/pkg/storage/chunk/cache"
@@ -160,6 +161,7 @@ const (
DataObjExplorer = "dataobj-explorer"
DataObjConsumer = "dataobj-consumer"
DataObjIndexBuilder = "dataobj-index-builder"
ScratchStore = "scratch-store"
UI = "ui"
All = "all"
Read = "read"
@@ -2203,6 +2205,7 @@ func (t *Loki) initDataObjConsumer() (services.Service, error) {
t.Cfg.DataObj.Metastore,
t.Cfg.Distributor.TenantTopic.TopicPrefix,
store,
t.scratchStore,
t.Cfg.Ingester.LifecyclerConfig.ID,
t.partitionRing,
prometheus.DefaultRegisterer,
@@ -2229,12 +2232,29 @@ func (t *Loki) initDataObjIndexBuilder() (services.Service, error) {
util_log.Logger,
t.Cfg.Ingester.LifecyclerConfig.ID,
store,
t.scratchStore,
prometheus.DefaultRegisterer,
)
return t.dataObjIndexBuilder, err
}
func (t *Loki) initScratchStore() (services.Service, error) {
logger := log.With(util_log.Logger, "module", "scratch-store")
store, err := scratch.Open(logger, t.Cfg.Common.ScratchPath)
if err != nil {
return nil, err
}
metrics := scratch.NewMetrics()
if err := metrics.Register(prometheus.DefaultRegisterer); err != nil {
return nil, err
}
t.scratchStore = scratch.ObserveStore(metrics, store)
return services.NewIdleService(nil, nil), nil
}
func (t *Loki) createDataObjBucket(clientName string) (objstore.Bucket, error) {
schema, err := t.Cfg.SchemaConfig.SchemaForTime(model.Now())
if err != nil {
pkg/scratch/filesystem.go
+194
View File
@@ -0,0 +1,194 @@
package scratch
import (
"fmt"
"io"
"io/fs"
"os"
"path/filepath"
"sync"
"github.com/go-kit/log"
"github.com/go-kit/log/level"
)
// filesystemExt is the extension used for scratch files stored on disk.
const filesystemExt = ".lokiscratch"
// Filesystem is an implementation of [Store] that retains data on the
// filesystem.
//
// Filesystem supports a memory fallback mechanism if writing data fails.
type Filesystem struct {
logger log.Logger
path string // Path on disk to store data.
fallback *Memory
mut sync.RWMutex
nextHandle *uint64 // nextHandle is a pointer to allow multiple stores to synchronize handles
files map[Handle]string // Relative path to file on disk.
}
var _ Store = (*Filesystem)(nil)
// NewFilesystem returns a new Filesystem store. Stored data is persisted to
// disk with a random filename and an extension ".lokiscratch".
//
// To avoid leaking scratch files between process iterations, NewFilesystem will
// remove any existing ".lokiscratch" files in path upon startup.
//
// NewFilesystem creates the path if it does not exist.
func NewFilesystem(logger log.Logger, path string) (*Filesystem, error) {
if _, err := os.Stat(path); err != nil {
return nil, fmt.Errorf("failed to stat scratch path %q: %w", path, err)
}
// Use absolute paths for better logging quality.
path, err := filepath.Abs(path)
if err != nil {
return nil, fmt.Errorf("failed to get absolute path for scratch path %q: %w", path, err)
}
return newUncheckedFilesystem(logger, path), nil
}
// newUncheckedFilesystem creates a new Filesystem store without checking if the
// path exists.
func newUncheckedFilesystem(logger log.Logger, path string) *Filesystem {
if logger == nil {
logger = log.NewNopLogger()
}
if err := cleanScratchDirectory(logger, path); err != nil {
// Cleaning the scratch directory is best-effort. Log a warning but
// otherwise continue.
level.Warn(logger).Log("msg", "failed to clean scratch directory", "err", err)
}
nextHandle := uint64(1) // Handles start at 1
return &Filesystem{
logger: logger,
path: path,
// Our store and fallback must share the same atomic nextHandle to avoid
// returning the same handle value twice.
fallback: newMemoryWithNextHandle(&nextHandle),
nextHandle: &nextHandle,
files: make(map[Handle]string),
}
}
// cleanScratchDirectory removes any existing ".lokiscratch" files in the given
// directory. It does not traverse into subdirectories.
func cleanScratchDirectory(logger log.Logger, scratchPath string) error {
return filepath.WalkDir(scratchPath, func(path string, d fs.DirEntry, _ error) error {
if path == scratchPath {
// Ignore root.
return nil
} else if d == nil {
// Ignore errors (d == nil only if there's an error); process what
// we can.
return nil
} else if d.IsDir() {
// Ignore subdirectories.
return fs.SkipDir
} else if filepath.Ext(path) != filesystemExt {
// Ignore non-scratch files.
return nil
}
if err := os.Remove(path); err != nil {
level.Warn(logger).Log("msg", "failed to remove old scratch file", "path", path, "err", err)
} else {
level.Debug(logger).Log("msg", "removed old scratch file", "path", path)
}
return nil
})
}
// Put stores the contents of p into scratch space.
//
// Put will be written with a random filename and the extension ".lokiscratch".
// If writing to the filesystem fails, p will instead be stored in-memory.
func (fs *Filesystem) Put(p []byte) Handle {
handle, err := fs.tryPut(p)
if err != nil {
level.Warn(fs.logger).Log("msg", "failed to store scratch data on filesystem, falling back to in-memory storage", "err", err)
return fs.fallback.Put(p)
}
return handle
}
func (fs *Filesystem) tryPut(p []byte) (Handle, error) {
f, err := os.CreateTemp(fs.path, "*"+filesystemExt)
if err != nil {
return InvalidHandle, err
}
defer f.Close()
if _, err := f.Write(p); err != nil {
// If we couldn't write the file, we'll opportunistically remove it
// (since it will never be used).
if err := os.Remove(f.Name()); err != nil {
level.Warn(fs.logger).Log("msg", "failed to remove unused scratch file", "path", f.Name(), "err", err)
}
return InvalidHandle, err
}
fs.mut.Lock()
defer fs.mut.Unlock()
handle := fs.getNextHandle()
fs.files[handle] = filepath.Base(f.Name())
return handle, err
}
// getNextHandle returns the next unique handle. getNextHandle must be called
// while holding m.mut.
func (fs *Filesystem) getNextHandle() Handle {
handle := Handle(*fs.nextHandle)
*fs.nextHandle++
return handle
}
// Read returns a reader for the file identifier by h. Read returns
// [HandleNotFoundError] if h doesn't exist either on disk or in-memory.
//
// Callers must close the reader when done to release resources. Reading may
// fail if the handle is removed while reading data.
func (fs *Filesystem) Read(h Handle) (io.ReadSeekCloser, error) {
fs.mut.RLock()
name, ok := fs.files[h]
fs.mut.RUnlock()
if !ok {
return fs.fallback.Read(h)
}
return os.Open(filepath.Join(fs.path, name))
}
// Remove removes the handle identified by h. If h is stored on disk, its
// corresponding file will be removed. Remove returns [HandleNotFoundError] if h
// doesn't exist on disk or in-memory.
func (fs *Filesystem) Remove(h Handle) error {
fs.mut.Lock()
defer fs.mut.Unlock()
name, ok := fs.files[h]
if !ok {
return fs.fallback.Remove(h)
}
fullPath := filepath.Join(fs.path, name)
// Opportunistically try to remove the file, but we don't want to fail the
// entire Remove call if there's a disk error.
if err := os.Remove(fullPath); err != nil && !os.IsNotExist(err) {
level.Warn(fs.logger).Log("msg", "failed to remove cache file", "path", fullPath, "err", err)
}
delete(fs.files, h)
return nil
}
pkg/scratch/filesystem_test.go
+70
View File
@@ -0,0 +1,70 @@
package scratch
import (
"io"
"path/filepath"
"testing"
"github.com/go-kit/log"
"github.com/stretchr/testify/require"
)
// TestFilesystem_Remove_deletes_files ensures that calling
// [Filesystem.Remove] will cause the files on disk to be deleted.
func TestFilesystem_Remove_deletes_files(t *testing.T) {
var (
logger = log.NewNopLogger()
tempDir = t.TempDir()
testData = []byte("test data content")
)
store, err := NewFilesystem(logger, tempDir)
require.NoError(t, err)
// Create a handle and ensure that its file exists on disk.
handle := store.Put(testData)
name := store.files[handle]
require.FileExists(t, filepath.Join(tempDir, name), "data file should exist on disk")
// Remove the handle, and then ensure the files have been deleted.
err = store.Remove(handle)
require.NoError(t, err)
require.NoFileExists(t, filepath.Join(tempDir, name), "data file should be deleted from disk")
}
func TestFilesystem_fallback(t *testing.T) {
logger := log.NewNopLogger()
// Create a Filesystem with an invalid path that will cause file
// operations to fail.
var (
store = newUncheckedFilesystem(logger, "/invalid/nonexistent/path")
testData = []byte("test data for fallback")
)
handle := store.Put(testData)
_, existsInDisk := store.files[handle]
require.False(t, existsInDisk, "handle should not exist in disk store maps since fallback was used")
// [diskScratchStore] should pass all calls down to the fallback store.
require.Equal(t, testData, readData(t, store, handle))
// Removing should remove the section from the fallback store.
require.NoError(t, store.Remove(handle))
_, err := store.Read(handle)
require.ErrorAs(t, err, new(HandleNotFoundError))
require.ErrorAs(t, store.Remove(handle), new(HandleNotFoundError))
}
func readData(t *testing.T, store Store, handle Handle) []byte {
dataReader, err := store.Read(handle)
require.NoError(t, err)
defer dataReader.Close()
actualData, err := io.ReadAll(dataReader)
require.NoError(t, err)
return actualData
}
pkg/scratch/memory.go
+95
View File
@@ -0,0 +1,95 @@
package scratch
import (
"bytes"
"io"
"sync"
)
// Memory is an implementation of [Store] that retains data in memory.
type Memory struct {
mut sync.RWMutex
nextHandle *uint64 // nextHandle is a pointer to allow multiple stores to synchronize handles
data map[Handle]*bytes.Buffer
}
var _ Store = (*Memory)(nil)
// NewMemory returns a new Memory store.
func NewMemory() *Memory {
// Handles start at 1.
nextHandle := uint64(1)
return newMemoryWithNextHandle(&nextHandle)
}
// newMemoryWithNextHandle creates a new Memory store with a custom nextHandle.
func newMemoryWithNextHandle(nextHandle *uint64) *Memory {
return &Memory{
nextHandle: nextHandle,
data: make(map[Handle]*bytes.Buffer),
}
}
// Put stores the contents of p into memory.
func (m *Memory) Put(p []byte) Handle {
buf := bytes.NewBuffer(make([]byte, 0, len(p)))
if _, err := buf.Write(p); err != nil {
// buf.Write doesn't fail unless writing more than [math.MaxInt].
panic(err)
}
m.mut.Lock()
defer m.mut.Unlock()
handle := m.getNextHandle()
m.data[handle] = buf
return handle
}
// getNextHandle returns the next unique handle. getNextHandle must be called
// while holding m.mut.
func (m *Memory) getNextHandle() Handle {
handle := Handle(*m.nextHandle)
*m.nextHandle++
return handle
}
// Read returns a reader for h. Read returns [HandleNotFoundError] if h doesn't
// exist in memory.
func (m *Memory) Read(h Handle) (io.ReadSeekCloser, error) {
buf, err := m.getBuffer(h)
if err != nil {
return nil, err
}
return nopReadSeekCloser{bytes.NewReader(buf.Bytes())}, nil
}
func (m *Memory) getBuffer(h Handle) (*bytes.Buffer, error) {
m.mut.RLock()
defer m.mut.RUnlock()
if _, ok := m.data[h]; !ok {
return nil, HandleNotFoundError(h)
}
return m.data[h], nil
}
// Remove removes h. Remove returns [HandleNotFoundError] if h doesn't exist in memory.
func (m *Memory) Remove(h Handle) error {
m.mut.Lock()
defer m.mut.Unlock()
if _, ok := m.data[h]; !ok {
return HandleNotFoundError(h)
}
delete(m.data, h)
return nil
}
type nopReadSeekCloser struct {
io.ReadSeeker
}
func (n nopReadSeekCloser) Close() error { return nil }
pkg/scratch/metrics.go
+128
View File
@@ -0,0 +1,128 @@
package scratch
import (
"errors"
"io"
"sync"
"time"
"github.com/prometheus/client_golang/prometheus"
)
// Metrics is a set of metrics for monitoring a [Store].
type Metrics struct {
handles prometheus.Gauge
bytes prometheus.Gauge
requestsSeconds *prometheus.HistogramVec
}
// NewMetrics returns a new Metrics. Returned Metrics must be registered to a
// registerer using [Metrics.Register].
func NewMetrics() *Metrics {
return &Metrics{
handles: prometheus.NewGauge(prometheus.GaugeOpts{
Name: "loki_scratch_store_handles",
Help: "Current number of handles in the scratch store",
}),
bytes: prometheus.NewGauge(prometheus.GaugeOpts{
Name: "loki_scratch_store_bytes",
Help: "Current number of bytes stored in the scratch store",
}),
requestsSeconds: prometheus.NewHistogramVec(prometheus.HistogramOpts{
Name: "loki_scratch_store_requests_seconds",
Help: "Time taken to perform operations on the scratch store",
Buckets: prometheus.DefBuckets,
NativeHistogramBucketFactor: 1.1,
NativeHistogramMaxBucketNumber: 100,
NativeHistogramMinResetDuration: 0,
}, []string{"operation", "result"}),
}
}
// Register registers metrics to report to reg.
func (m *Metrics) Register(reg prometheus.Registerer) error {
var errs []error
errs = append(errs, reg.Register(m.handles))
errs = append(errs, reg.Register(m.bytes))
errs = append(errs, reg.Register(m.requestsSeconds))
return errors.Join(errs...)
}
// Unregister unregisters metrics from reg.
func (m *Metrics) Unregister(reg prometheus.Registerer) {
reg.Unregister(m.handles)
reg.Unregister(m.bytes)
reg.Unregister(m.requestsSeconds)
}
type observedStore struct {
metrics *Metrics
inner Store
handleSizes sync.Map // map[Handle]uint (uint is byte size of data in handle)
}
var _ Store = (*observedStore)(nil)
// ObserveStore wraps a store and accumulates statistics into the provided
// Metrics. If metrics is nil, ObserveStore performs no wrapping.
func ObserveStore(metrics *Metrics, store Store) Store {
if metrics == nil {
return store // nothing to do
}
return &observedStore{
metrics: metrics,
inner: store,
}
}
func (os *observedStore) Put(p []byte) Handle {
start := time.Now()
handle := os.inner.Put(p)
duration := time.Since(start)
os.handleSizes.Store(handle, uint(len(p)))
os.metrics.handles.Add(1)
os.metrics.bytes.Add(float64(len(p)))
os.metrics.requestsSeconds.WithLabelValues("Put", "success").Observe(duration.Seconds())
return handle
}
func (os *observedStore) Read(h Handle) (io.ReadSeekCloser, error) {
start := time.Now()
reader, err := os.inner.Read(h)
duration := time.Since(start)
os.metrics.requestsSeconds.WithLabelValues("Read", errorToResult(err)).Observe(duration.Seconds())
return reader, err
}
func errorToResult(err error) string {
if err == nil {
return "success"
}
return "error"
}
func (os *observedStore) Remove(h Handle) error {
start := time.Now()
err := os.inner.Remove(h)
duration := time.Since(start)
if value, existed := os.handleSizes.LoadAndDelete(h); existed {
os.metrics.handles.Sub(1)
os.metrics.bytes.Sub(float64(value.(uint)))
}
os.metrics.requestsSeconds.WithLabelValues("Remove", errorToResult(err)).Observe(duration.Seconds())
return err
}
pkg/scratch/scratch.go
+84
View File
@@ -0,0 +1,84 @@
// Package scratch provides an abstraction for scratch space.
package scratch
import (
"fmt"
"io"
"github.com/go-kit/log"
)
// HandleNotFoundError is returned when performing an operation on a handle that
// doesn't exist.
type HandleNotFoundError Handle
// Error returns a string representation of e.
func (e HandleNotFoundError) Error() string {
return fmt.Sprintf("handle %d not found", uint64(e))
}
// Handle is a identifier for data placed into scratch space. Handles are unique
// per store. Handles are immutable once created and can only be read or
// deleted.
//
// The zero value for handle is invalid.
type Handle uint64
// InvalidHandle is the zero value for Handle.
const InvalidHandle = Handle(0)
// Store is an abstraction for temporarily storing data. Store is ephemeral and
// starts fresh every time it is created.
//
// Store implementations must be safe for concurrent access.
type Store interface {
// Ideally Store would support some kind of streaming
//
// PutReader(r io.Reader) (Handle, error)
//
// method, but it's tricky to implement when considering [Filesystem]s
// fallback mechanism, which naively would require buffering the entire data
// into memory anyway (for passing to the fallback if the write to disk
// fails).
//
// It may be possible to implement efficiently if we buffered the most
// recent written chunk; then falling back could be implemented by replaying
// what _did_ get written to disk, followed by the chunk that failed, and
// then all other data. However, this is more complicated than it's worth
// right now.
//
// This can be revisited if there's a growing need for streaming data to the
// Store without needing to buffer it all into memory first.
// Put stores the contents of p into scratch space.
Put(p []byte) Handle
// Read returns a reader for the file identifier by h. Read returns
// [InvalidHandleError] if h doesn't exist in the Store.
//
// Callers must close the reader when done to release resources. Reading may
// fail if the handle is removed while reading data.
Read(h Handle) (io.ReadSeekCloser, error)
// Remove removes the handle identified by h. Remove returns
// [InvalidHandleError] if h doesn't exist in the Store.
Remove(h Handle) error
}
// Open returns the default [Store] implementation based on whether path is set.
//
// If path is empty, Open returns a [Memory] store. If path is non-empty, Open
// returns a [Filesystem] store.
//
// Open returns an error if the path is non-empty but refers to a non-existent
// directory.
//
// To observe metrics on the returned store, wrap the result with
// [ObserveStore].
func Open(logger log.Logger, path string) (Store, error) {
if path == "" {
return NewMemory(), nil
}
return NewFilesystem(logger, path)
}
pkg/scratch/scratch_test.go
+90
View File
@@ -0,0 +1,90 @@
package scratch_test
import (
"io"
"math"
"testing"
"github.com/go-kit/log"
"github.com/stretchr/testify/require"
"github.com/grafana/loki/v3/pkg/scratch"
)
func TestStore(t *testing.T) {
t.Run("impl=Memory", func(t *testing.T) {
testStore(t, func() scratch.Store { return scratch.NewMemory() })
})
t.Run("impl=Filesystem", func(t *testing.T) {
testStore(t, func() scratch.Store {
logger := log.NewNopLogger()
store, err := scratch.NewFilesystem(logger, t.TempDir())
require.NoError(t, err)
return store
})
})
t.Run("impl=Observed", func(t *testing.T) {
testStore(t, func() scratch.Store {
return scratch.ObserveStore(scratch.NewMetrics(), scratch.NewMemory())
})
})
}
func testStore(t *testing.T, makeStore func() scratch.Store) {
exampleData := []byte("test data content")
t.Run("Put", func(t *testing.T) {
store := makeStore()
require.NotPanics(t, func() {
_ = store.Put(exampleData)
})
})
t.Run("Read", func(t *testing.T) {
store := makeStore()
handle := store.Put(exampleData)
reader, err := store.Read(handle)
require.NoError(t, err)
defer reader.Close()
actualData, err := io.ReadAll(reader)
require.NoError(t, err)
require.Equal(t, exampleData, actualData)
})
t.Run("Read (invalid handle)", func(t *testing.T) {
store := makeStore()
invalidHandle := scratch.Handle(math.MaxUint64)
_, err := store.Read(invalidHandle)
require.ErrorAs(t, err, new(scratch.HandleNotFoundError))
})
t.Run("Remove", func(t *testing.T) {
store := makeStore()
handle := store.Put(exampleData)
// Check that the handle exists.
closer, err := store.Read(handle)
require.NoError(t, err)
require.NoError(t, closer.Close())
require.NoError(t, store.Remove(handle))
_, err = store.Read(handle)
require.ErrorAs(t, err, new(scratch.HandleNotFoundError))
require.ErrorAs(t, store.Remove(handle), new(scratch.HandleNotFoundError))
})
t.Run("Remove (invalid handle)", func(t *testing.T) {
store := makeStore()
invalidHandle := scratch.Handle(math.MaxUint64)
require.ErrorAs(t, store.Remove(invalidHandle), new(scratch.HandleNotFoundError))
})
}