refactor(fracmanager): move name generation and sealing logic to FractionProvider

Author: eguguchkin

fracmanager/fracmanager.go

@@ -3,14 +3,11 @@ package fracmanager
 import (
 	"context"
 	"errors"
-	"io"
-	"math/rand"
 	"os"
 	"path/filepath"
 	"sync"
 	"time"
 
-	"github.com/oklog/ulid/v2"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"go.uber.org/atomic"
@@ -44,6 +41,7 @@ type FracManager struct {
 	remoteFracs []*frac.Remote
 	active      activeRef
 
+	indexer      *frac.ActiveIndexer
 	fracProvider *fractionProvider
 
 	oldestCTLocal  atomic.Uint64
@@ -56,8 +54,6 @@ type FracManager struct {
 	cacheWG *sync.WaitGroup
 
 	s3cli *s3.Client
-
-	ulidEntropy io.Reader
 }
 
 type fracRef struct {
@@ -100,25 +96,24 @@ func NewFracManager(ctx context.Context, cfg *Config, s3cli *s3.Client) *FracMan
 		ChangeGenerations: metric.CacheChangeGenerations,
 	})
 
+	readLimiter := storage.NewReadLimiter(config.ReaderWorkers, storeBytesRead)
+	indexer := frac.NewActiveIndexer(config.IndexWorkers, config.IndexWorkers)
+	indexer.Start()
+
 	fracManager := &FracManager{
 		config:          cfg,
 		ctx:             ctx,
 		s3cli:           s3cli,
 		mature:          atomic.Bool{},
 		cacheMaintainer: cacheMaintainer,
-		fracProvider:    newFractionProvider(&cfg.Fraction, s3cli, cacheMaintainer, config.ReaderWorkers, config.IndexWorkers),
-		ulidEntropy:     ulid.Monotonic(rand.New(rand.NewSource(time.Now().UnixNano())), 0),
+		indexer:         indexer,
+		fracProvider:    newFractionProvider(cfg, s3cli, cacheMaintainer, readLimiter, indexer),
 		fracCache:       NewSealedFracCache(filepath.Join(cfg.DataDir, consts.FracCacheFileSuffix)),
 	}
 
 	return fracManager
 }
 
-// This method is not thread safe. Use consciously to avoid race
-func (fm *FracManager) nextFractionID() string {
-	return ulid.MustNew(ulid.Timestamp(time.Now()), fm.ulidEntropy).String()
-}
-
 func (fm *FracManager) maintenance(sealWg, cleanupWg *sync.WaitGroup) {
 	logger.Debug("maintenance started")
 
@@ -289,7 +284,7 @@ func (fm *FracManager) cleanupFractions(cleanupWg *sync.WaitGroup) {
 			}
 
 			offloadStart := time.Now()
-			mustBeOffloaded, err := outsider.Offload(fm.ctx, s3.NewUploader(fm.s3cli))
+			remote, err := fm.fracProvider.Offload(fm.ctx, outsider)
 			if err != nil {
 				metric.OffloadingTotal.WithLabelValues("failure").Inc()
 				metric.OffloadingDurationSeconds.Observe(float64(time.Since(offloadStart).Seconds()))
@@ -307,7 +302,7 @@ func (fm *FracManager) cleanupFractions(cleanupWg *sync.WaitGroup) {
 				return
 			}
 
-			if !mustBeOffloaded {
+			if remote == nil {
 				fm.fracCache.RemoveFraction(info.Name())
 				outsider.Suicide()
 				return
@@ -322,8 +317,6 @@ func (fm *FracManager) cleanupFractions(cleanupWg *sync.WaitGroup) {
 				zap.String("took", time.Since(offloadStart).String()),
 			)
 
-			remote := fm.fracProvider.NewRemote(fm.ctx, info.Path, info)
-
 			fm.fracMu.Lock()
 			// FIXME(dkharms): We had previously shifted fraction from local fracs list (in [fm.determineOutsiders] via [fm.shiftFirstFrac])
 			// and therefore excluded it from search queries.
@@ -565,7 +558,7 @@ func (fm *FracManager) seal(activeRef activeRef) {
 		sealsDoneSeconds.Observe(time.Since(now).Seconds())
 	}()
 
-	sealed, err := activeRef.frac.Seal(fm.config.SealParams)
+	sealed, err := activeRef.frac.Seal()
 	if err != nil {
 		if errors.Is(err, ErrSealingFractionSuicided) {
 			// the faction is suicided, this means that it has already pushed out of the list of factions,
@@ -584,18 +577,16 @@ func (fm *FracManager) seal(activeRef activeRef) {
 }
 
 func (fm *FracManager) rotate() activeRef {
-	filePath := fileBasePattern + fm.nextFractionID()
-	baseFilePath := filepath.Join(fm.config.DataDir, filePath)
-	logger.Info("creating new fraction", zap.String("filepath", baseFilePath))
-
-	next := fm.newActiveRef(fm.fracProvider.NewActive(baseFilePath))
+	next := fm.newActiveRef(fm.fracProvider.CreateActive())
 
 	fm.fracMu.Lock()
 	prev := fm.active
 	fm.active = next
 	fm.localFracs = append(fm.localFracs, fm.active.ref)
 	fm.fracMu.Unlock()
 
+	logger.Info("new fraction created", zap.String("filepath", next.frac.active.BaseFileName))
+
 	return prev
 }
 
@@ -604,7 +595,7 @@ func (fm *FracManager) minFracSizeToSeal() uint64 {
 }
 
 func (fm *FracManager) Stop() {
-	fm.fracProvider.Stop()
+	fm.indexer.Stop()
 	fm.stopFn()
 
 	fm.statWG.Wait()

fracmanager/fracmanager_test.go

@@ -110,7 +110,7 @@ func TestMatureMode(t *testing.T) {
 
 		checkFn(fm)
 
-		fm.fracProvider.Stop()
+		fm.indexer.Stop()
 	}
 
 	id := 1
@@ -151,15 +151,6 @@ func TestMatureMode(t *testing.T) {
 
 }
 
-func TestNewULID(t *testing.T) {
-	fm := NewFracManager(context.Background(), &Config{}, nil)
-	ulid1 := fm.nextFractionID()
-	ulid2 := fm.nextFractionID()
-	assert.NotEqual(t, ulid1, ulid2, "ULIDs should be different")
-	assert.Equal(t, 26, len(ulid1), "ULID should have length 26")
-	assert.Greater(t, ulid2, ulid1)
-}
-
 func TestOldestCT(t *testing.T) {
 	const fracCount = 10
 

fracmanager/fraction_provider.go

@@ -2,13 +2,19 @@ package fracmanager
 
 import (
 	"context"
+	"io"
+	"math/rand"
+	"path/filepath"
+	"time"
 
+	"github.com/oklog/ulid/v2"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 
 	"github.com/ozontech/seq-db/frac"
 	"github.com/ozontech/seq-db/frac/common"
 	"github.com/ozontech/seq-db/frac/sealed"
+	"github.com/ozontech/seq-db/frac/sealed/sealing"
 	"github.com/ozontech/seq-db/storage"
 	"github.com/ozontech/seq-db/storage/s3"
 )
@@ -19,27 +25,28 @@ var storeBytesRead = promauto.NewCounter(prometheus.CounterOpts{
 	Name:      "bytes_read",
 })
 
+// fractionProvider is a factory for creating different types of fractions
+// Contains all necessary dependencies for creating and managing fractions
 type fractionProvider struct {
-	s3cli         *s3.Client
-	config        *frac.Config
-	cacheProvider *CacheMaintainer
-	activeIndexer *frac.ActiveIndexer
-	readLimiter   *storage.ReadLimiter
+	s3cli         *s3.Client           // Client for S3 storage operations
+	config        *Config              // Fraction manager configuration
+	cacheProvider *CacheMaintainer     // Cache provider for data access optimization
+	activeIndexer *frac.ActiveIndexer  // Indexer for active fractions
+	readLimiter   *storage.ReadLimiter // Read rate limiter
+	ulidEntropy   io.Reader            // Entropy source for ULID generation
 }
 
 func newFractionProvider(
-	c *frac.Config, s3cli *s3.Client, cp *CacheMaintainer,
-	readerWorkers, indexWorkers int,
+	cfg *Config, s3cli *s3.Client, cp *CacheMaintainer,
+	readLimiter *storage.ReadLimiter, indexer *frac.ActiveIndexer,
 ) *fractionProvider {
-	ai := frac.NewActiveIndexer(indexWorkers, indexWorkers)
-	ai.Start() // first start indexWorkers to allow active frac replaying
-
 	return &fractionProvider{
 		s3cli:         s3cli,
-		config:        c,
+		config:        cfg,
 		cacheProvider: cp,
-		activeIndexer: ai,
-		readLimiter:   storage.NewReadLimiter(readerWorkers, storeBytesRead),
+		activeIndexer: indexer,
+		readLimiter:   readLimiter,
+		ulidEntropy:   ulid.Monotonic(rand.New(rand.NewSource(time.Now().UnixNano())), 0),
 	}
 }
 
@@ -50,7 +57,7 @@ func (fp *fractionProvider) NewActive(name string) *frac.Active {
 		fp.readLimiter,
 		fp.cacheProvider.CreateDocBlockCache(),
 		fp.cacheProvider.CreateSortDocsCache(),
-		fp.config,
+		&fp.config.Fraction,
 	)
 }
 
@@ -60,37 +67,75 @@ func (fp *fractionProvider) NewSealed(name string, cachedInfo *common.Info) *fra
 		fp.readLimiter,
 		fp.cacheProvider.CreateIndexCache(),
 		fp.cacheProvider.CreateDocBlockCache(),
-		cachedInfo,
-		fp.config,
+		cachedInfo, // Preloaded meta information
+		&fp.config.Fraction,
 	)
 }
 
 func (fp *fractionProvider) NewSealedPreloaded(name string, preloadedData *sealed.PreloadedData) *frac.Sealed {
 	return frac.NewSealedPreloaded(
 		name,
-		preloadedData,
+		preloadedData, // Data already loaded into memory
 		fp.readLimiter,
 		fp.cacheProvider.CreateIndexCache(),
 		fp.cacheProvider.CreateDocBlockCache(),
-		fp.config,
+		&fp.config.Fraction,
 	)
 }
 
-func (fp *fractionProvider) NewRemote(
-	ctx context.Context, name string, cachedInfo *common.Info,
-) *frac.Remote {
+func (fp *fractionProvider) NewRemote(ctx context.Context, name string, cachedInfo *common.Info) *frac.Remote {
 	return frac.NewRemote(
 		ctx,
 		name,
 		fp.readLimiter,
 		fp.cacheProvider.CreateIndexCache(),
 		fp.cacheProvider.CreateDocBlockCache(),
 		cachedInfo,
-		fp.config,
+		&fp.config.Fraction,
 		fp.s3cli,
 	)
 }
 
-func (fp *fractionProvider) Stop() {
-	fp.activeIndexer.Stop()
+// nextFractionID generates a unique identifier for a new fraction
+// IMPORTANT: This method is not thread-safe. When used in concurrent environments,
+// external synchronization must be provided to avoid ID collisions
+func (fp *fractionProvider) nextFractionID() string {
+	return ulid.MustNew(ulid.Timestamp(time.Now()), fp.ulidEntropy).String()
+}
+
+// CreateActive creates a new active fraction with auto-generated filename
+// Filename pattern: base_pattern + ULID
+func (fp *fractionProvider) CreateActive() *frac.Active {
+	filePath := fileBasePattern + fp.nextFractionID()
+	baseFilePath := filepath.Join(fp.config.DataDir, filePath)
+	return fp.NewActive(baseFilePath)
+}
+
+// Seal converts an active fraction to a sealed one
+// Process includes sorting, indexing, and data optimization for reading
+func (fp *fractionProvider) Seal(active *frac.Active) (*frac.Sealed, error) {
+	src, err := frac.NewActiveSealingSource(active, fp.config.SealParams)
+	if err != nil {
+		return nil, err
+	}
+	preloaded, err := sealing.Seal(src, fp.config.SealParams)
+	if err != nil {
+		return nil, err
+	}
+
+	return fp.NewSealedPreloaded(active.BaseFileName, preloaded), nil
+}
+
+// Offload uploads fraction to S3 storage and returns a remote fraction
+// IMPORTANT: context controls timeouts and operation cancellation
+func (fp *fractionProvider) Offload(ctx context.Context, f frac.Fraction) (*frac.Remote, error) {
+	mustBeOffloaded, err := f.Offload(ctx, s3.NewUploader(fp.s3cli))
+	if err != nil {
+		return nil, err
+	}
+	if !mustBeOffloaded {
+		return nil, nil
+	}
+	info := f.Info()
+	return fp.NewRemote(ctx, info.Path, info), nil
 }

fracmanager/fraction_provider_test.go

@@ -0,0 +1,16 @@
+package fracmanager
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFractionID(t *testing.T) {
+	fp := newFractionProvider(nil, nil, nil, nil, nil)
+	ulid1 := fp.nextFractionID()
+	ulid2 := fp.nextFractionID()
+	assert.NotEqual(t, ulid1, ulid2, "ULIDs should be different")
+	assert.Equal(t, 26, len(ulid1), "ULID should have length 26")
+	assert.Greater(t, ulid2, ulid1)
+}

fracmanager/proxy_frac.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/ozontech/seq-db/frac"
 	"github.com/ozontech/seq-db/frac/common"
-	"github.com/ozontech/seq-db/frac/sealed/sealing"
 	"github.com/ozontech/seq-db/logger"
 	"github.com/ozontech/seq-db/metric"
 	"github.com/ozontech/seq-db/seq"
@@ -118,7 +117,7 @@ func (f *proxyFrac) WaitWriteIdle() {
 	logger.Info("write is stopped", zap.String("name", f.name), zap.Float64("time_wait_s", waitTime))
 }
 
-func (f *proxyFrac) Seal(params common.SealParams) (*frac.Sealed, error) {
+func (f *proxyFrac) Seal() (*frac.Sealed, error) {
 	f.useMu.Lock()
 	if f.isSuicidedState() {
 		f.useMu.Unlock()
@@ -135,16 +134,10 @@ func (f *proxyFrac) Seal(params common.SealParams) (*frac.Sealed, error) {
 
 	f.WaitWriteIdle()
 
-	src, err := frac.NewActiveSealingSource(active, params)
+	sealed, err := f.fp.Seal(active)
 	if err != nil {
 		return nil, err
 	}
-	preloaded, err := sealing.Seal(src, params)
-	if err != nil {
-		return nil, err
-	}
-
-	sealed := f.fp.NewSealedPreloaded(active.BaseFileName, preloaded)
 
 	f.useMu.Lock()
 	f.sealed = sealed

fracmanager/sealer_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/ozontech/seq-db/frac/sealed"
 	"github.com/ozontech/seq-db/frac/sealed/sealing"
 	"github.com/ozontech/seq-db/seq"
+	"github.com/ozontech/seq-db/storage"
 	testscommon "github.com/ozontech/seq-db/tests/common"
 )
 
@@ -107,15 +108,19 @@ func BenchmarkSealing_WithSort(b *testing.B) {
 
 func runSealingBench(b *testing.B, cfg *frac.Config) {
 	cm := NewCacheMaintainer(uint64(units.MiB)*64, uint64(units.MiB)*64, nil)
-	fp := newFractionProvider(cfg, nil, cm, 1, 1)
+
+	idx := frac.NewActiveIndexer(1, 1)
+	rl := storage.NewReadLimiter(1, storeBytesRead)
+	fp := newFractionProvider(&Config{Fraction: *cfg}, nil, cm, rl, idx)
+	idx.Start()
 
 	dataDir := filepath.Join(b.TempDir(), "BenchmarkSealing")
 	testscommon.RecreateDir(dataDir)
 
 	active := fp.NewActive(filepath.Join(dataDir, "test"))
 	err := fillActiveFraction(active)
 	assert.NoError(b, err)
-	fp.Stop()
+	idx.Stop()
 
 	seal := func(active *frac.Active, params common.SealParams) (*sealed.PreloadedData, error) {
 		src, err := frac.NewActiveSealingSource(active, params)