deepdb

Documentation

Index

• Constants
• func New(cfg *Config, logger gkLog.Logger) (Reader, Writer, TracepointReader, TracepointWriter, Compactor, error)
• type CompactionBlockSelector
• type Compactor
• type CompactorConfig
• type CompactorOverrides
• type CompactorSharder
• type Config
• type IterateObjectCallback
• type Reader
• type SearchConfig
  • func (c SearchConfig) ApplyToOptions(o *common.SearchOptions)
• type TracepointReader
• type TracepointWriter
• type WriteableBlock
• type Writer

Constants

const (
	DefaultCompactionCycle = 30 * time.Second

	DefaultChunkSizeBytes            = 5 * 1024 * 1024  // 5 MiB
	DefaultFlushSizeBytes     uint32 = 20 * 1024 * 1024 // 20 MiB
	DefaultIteratorBufferSize        = 1000
)

const (
	DefaultBlocklistPoll            = 5 * time.Minute
	DefaultMaxTimePerTenant         = 5 * time.Minute
	DefaultBlocklistPollConcurrency = uint(50)
	DefaultRetentionConcurrency     = uint(10)
	DefaultTenantIndexBuilders      = 2

	DefaultPrefetchSnapshotCount = 1000
	DefaultSearchChunkSizeBytes  = 1_000_000
	DefaultReadBufferCount       = 32
	DefaultReadBufferSize        = 1 * 1024 * 1024
)

const (
	// BlockIDMin is the minimum possible value for a block id as a string
	BlockIDMin = "00000000-0000-0000-0000-000000000000"
	// BlockIDMax is the maximum possible value for a block id as a string
	BlockIDMax = "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF"
)

Functions

func New

func New(cfg *Config, logger gkLog.Logger) (Reader, Writer, TracepointReader, TracepointWriter, Compactor, error)

New creates a new deepdb

Types

type CompactionBlockSelector

type CompactionBlockSelector interface {
	BlocksToCompact() ([]*backend.BlockMeta, string)
}

CompactionBlockSelector is an interface for different algorithms to pick suitable blocks for compaction

type Compactor

type Compactor interface {
	EnableCompaction(ctx context.Context, cfg *CompactorConfig, sharder CompactorSharder, overrides CompactorOverrides)
}

type CompactorConfig

type CompactorConfig struct {
	ChunkSizeBytes          uint32        `yaml:"v2_in_buffer_bytes"`
	FlushSizeBytes          uint32        `yaml:"v2_out_buffer_bytes"`
	IteratorBufferSize      int           `yaml:"v2_prefetch_snapshot_count"`
	MaxCompactionRange      time.Duration `yaml:"compaction_window"`
	MaxCompactionObjects    int           `yaml:"max_compaction_objects"`
	MaxBlockBytes           uint64        `yaml:"max_block_bytes"`
	BlockRetention          time.Duration `yaml:"block_retention"`
	CompactedBlockRetention time.Duration `yaml:"compacted_block_retention"`
	RetentionConcurrency    uint          `yaml:"retention_concurrency"`
	MaxTimePerTenant        time.Duration `yaml:"max_time_per_tenant"`
	CompactionCycle         time.Duration `yaml:"compaction_cycle"`
}

CompactorConfig contains compaction configuration options

type CompactorOverrides

type CompactorOverrides interface {
	BlockRetentionForTenant(tenantID string) time.Duration
	MaxBytesPerSnapshotForTenant(tenantID string) int
}

type CompactorSharder

type CompactorSharder interface {
	Owns(hash string) bool
	RecordDiscardedSnapshots(count int, tenantID string, snapshotID string)
}

type Config

type Config struct {
	Pool   *pool.Config        `yaml:"pool,omitempty"`
	WAL    *wal.Config         `yaml:"wal"`
	Block  *common.BlockConfig `yaml:"block"`
	Search *SearchConfig       `yaml:"search"`

	BlocklistPoll                    time.Duration `yaml:"blocklist_poll"`
	BlocklistPollConcurrency         uint          `yaml:"blocklist_poll_concurrency"`
	BlocklistPollFallback            bool          `yaml:"blocklist_poll_fallback"`
	BlocklistPollTenantIndexBuilders int           `yaml:"blocklist_poll_tenant_index_builders"`
	BlocklistPollStaleTenantIndex    time.Duration `yaml:"blocklist_poll_stale_tenant_index"`
	BlocklistPollJitterMs            int           `yaml:"blocklist_poll_jitter_ms"`

	// backends
	Backend string        `yaml:"backend"`
	Local   *local.Config `yaml:"local"`
	GCS     *gcs.Config   `yaml:"gcs"`
	S3      *s3.Config    `yaml:"s3"`
	Azure   *azure.Config `yaml:"azure"`

	// caches
	Cache                   string                  `yaml:"cache"`
	CacheMinCompactionLevel uint8                   `yaml:"cache_min_compaction_level"`
	CacheMaxBlockAge        time.Duration           `yaml:"cache_max_block_age"`
	BackgroundCache         *cache.BackgroundConfig `yaml:"background_cache"`
	Memcached               *memcached.Config       `yaml:"memcached"`
	Redis                   *redis.Config           `yaml:"redis"`
}

Config holds the entirety of deepdb configuration Defaults are in modules/storage/config.go

type IterateObjectCallback

type IterateObjectCallback func(id common.ID, obj []byte) bool

type Reader

type Reader interface {
	FindSnapshot(ctx context.Context, tenantID string, id common.ID, blockStart string, blockEnd string, timeStart int64, timeEnd int64) (*deepTP.Snapshot, []error, error)

	Search(ctx context.Context, meta *backend.BlockMeta, req *deeppb.SearchRequest, opts common.SearchOptions) (*deeppb.SearchResponse, error)
	Fetch(ctx context.Context, meta *backend.BlockMeta, req deepql.FetchSnapshotRequest, opts common.SearchOptions) (deepql.FetchSnapshotResponse, error)
	BlockMetas(tenantID string) []*backend.BlockMeta
	EnablePolling(sharder blocklist.JobSharder)

	Shutdown()
}

type SearchConfig

type SearchConfig struct {
	// v2 blocks
	ChunkSizeBytes        uint32 `yaml:"chunk_size_bytes"`
	PrefetchSnapshotCount int    `yaml:"prefetch_snapshot_count"`

	// vParquet blocks
	ReadBufferCount     int `yaml:"read_buffer_count"`
	ReadBufferSizeBytes int `yaml:"read_buffer_size_bytes"`
	CacheControl        struct {
		Footer      bool `yaml:"footer"`
		ColumnIndex bool `yaml:"column_index"`
		OffsetIndex bool `yaml:"offset_index"`
	} `yaml:"cache_control"`
}

func (SearchConfig) ApplyToOptions

func (c SearchConfig) ApplyToOptions(o *common.SearchOptions)

type TracepointReader

type TracepointReader interface {
	ReadTracepointBlock(ctx context.Context, tenantID string) (io.ReadCloser, int64, error)
}

type TracepointWriter

type TracepointWriter interface {
	WriteTracepointBlock(ctx context.Context, tenantID string, reader *bytes.Reader, size int64) error
}

type WriteableBlock

type WriteableBlock interface {
	BlockMeta() *backend.BlockMeta
	Write(ctx context.Context, w backend.Writer) error
}

type Writer

type Writer interface {
	WriteBlock(ctx context.Context, block WriteableBlock) error
	CompleteBlock(ctx context.Context, block common.WALBlock) (common.BackendBlock, error)
	CompleteBlockWithBackend(ctx context.Context, block common.WALBlock, r backend.Reader, w backend.Writer) (common.BackendBlock, error)
	WAL() *wal.WAL
}