Function Open

tsdb/agent/db.go:291–369

Open returns a new agent.DB in the given directory.

(l *slog.Logger, reg prometheus.Registerer, rs *remote.Storage, dir string, opts *Options)

// Open returns a new agent.DB in the given directory.
func Open(l *slog.Logger, reg prometheus.Registerer, rs *remote.Storage, dir string, opts *Options) (*DB, error) {
	opts = validateOptions(opts)

	locker, err := tsdbutil.NewDirLocker(dir, "agent", l, reg)
	if err != nil {
		return nil, err
	}
	if !opts.NoLockfile {
		if err := locker.Lock(); err != nil {
			return nil, err
		}
	}

	// remote_write expects WAL to be stored in a "wal" subdirectory of the main storage.
	dir = filepath.Join(dir, "wal")

	w, err := wlog.NewSize(l, reg, dir, opts.WALSegmentSize, opts.WALCompression)
	if err != nil {
		return nil, fmt.Errorf("creating WAL: %w", err)
	}

	db := &DB{
		logger: l,
		opts:   opts,
		rs:     rs,

		wal:    w,
		locker: locker,

		nextRef: atomic.NewUint64(0),
		series:  newStripeSeries(opts.StripeSize),
		deleted: make(map[chunks.HeadSeriesRef]deletedRefMeta),

		donec: make(chan struct{}),
		stopc: make(chan struct{}),

		metrics: newDBMetrics(reg),
	}

	db.bufPool.New = func() any {
		return make([]byte, 0, 1024)
	}

	db.appenderPool.New = func() any {
		return &appender{
			appenderBase: appenderBase{
				DB:                     db,
				pendingSeries:          make([]record.RefSeries, 0, 100),
				pendingSamples:         make([]record.RefSample, 0, 100),
				pendingHistograms:      make([]record.RefHistogramSample, 0, 100),
				pendingFloatHistograms: make([]record.RefFloatHistogramSample, 0, 100),
				pendingExamplars:       make([]record.RefExemplar, 0, 10),
			},
		}
	}
	db.appenderV2Pool.New = func() any {
		return &appenderV2{
			appenderBase: appenderBase{