copyer

Documentation

Index

• type CopyOpt
  • func (opt *CopyOpt) BlockRange(split int64) chan [2]int64
  • func (opt *CopyOpt) Copy(mint, maxt int64) *CopyOpt
  • func (opt *CopyOpt) Range() chan [2]int64
  • func (opt *CopyOpt) ShowTime() (string, string)
• type Copyer
  • func MustNewCopyer(c *Copyer, err error) *Copyer
  • func NewLocalCopyer(originDir, blockId string) (*Copyer, error)
  • func NewRemoteCopyer(uri string, lbApi *url.URL) (*Copyer, error)
  • func (c *Copyer) CopyPerBlock(opt *CopyOpt, ch chan<- *TenantResult) (err error)
  • func (c *Copyer) GetDbTimes() (int64, int64)
• type MultiTenantWriter
  • func NewMultiTenantWriter(logger log.Logger, dir string, blockSize int64, al labels.Labels, ...) *MultiTenantWriter
  • func (mw *MultiTenantWriter) Count() int
  • func (mw *MultiTenantWriter) Get(tenant string) (*Writer, error)
  • func (mw *MultiTenantWriter) Range(fn func(string, *Writer) bool)
• type Queue
  • func (q *Queue) Pull() any
  • func (q *Queue) PullNoWait() any
  • func (q *Queue) Push(val any)
• type TenantResult
  • func (r *TenantResult) New(tenant string) *TenantResult
  • func (r *TenantResult) Show() string
• type TenantResults
  • func (t TenantResults) ToMap() map[string][]string
• type Writer
  • func NewWriter(logger log.Logger, dir string, blockSize int64, al labels.Labels, ...) (*Writer, error)
  • func (w *Writer) AppendFn(fn *getSeries)
  • func (w *Writer) FlushJob(wg *sync.WaitGroup, mint, maxt int64, tenant string, ch chan<- *TenantResult) func()
  • func (w *Writer) WriteJob()

Types

type CopyOpt

type CopyOpt struct {
	ManualGC      bool
	WriteThread   int
	Mint          int64
	Cent          int64
	Maxt          int64
	BlockSplit    int64
	TimeSplit     int64
	CommitCount   uint64
	TargetDir     string
	TenantLabel   string
	DefaultTenant string
	LabelAppends  labels.Labels
	LabelMatchers []*labels.Matcher
	JobPool       *ants.Pool
	FlushPool     *ants.Pool
	FlushGroup    *sync.WaitGroup
}

func (*CopyOpt) BlockRange

func (opt *CopyOpt) BlockRange(split int64) chan [2]int64

func (*CopyOpt) Copy

func (opt *CopyOpt) Copy(mint, maxt int64) *CopyOpt

func (*CopyOpt) Range

func (opt *CopyOpt) Range() chan [2]int64

func (*CopyOpt) ShowTime

func (opt *CopyOpt) ShowTime() (string, string)

type Copyer

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

func MustNewCopyer

func MustNewCopyer(c *Copyer, err error) *Copyer

func NewLocalCopyer

func NewLocalCopyer(originDir, blockId string) (*Copyer, error)

func NewRemoteCopyer

func NewRemoteCopyer(uri string, lbApi *url.URL) (*Copyer, error)

func (*Copyer) CopyPerBlock

func (c *Copyer) CopyPerBlock(opt *CopyOpt, ch chan<- *TenantResult) (err error)

func (*Copyer) GetDbTimes

func (c *Copyer) GetDbTimes() (int64, int64)

type MultiTenantWriter

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

func NewMultiTenantWriter

func NewMultiTenantWriter(logger log.Logger, dir string, blockSize int64, al labels.Labels, commitCount uint64, gc bool) *MultiTenantWriter

func (*MultiTenantWriter) Count

func (mw *MultiTenantWriter) Count() int

func (*MultiTenantWriter) Get

func (mw *MultiTenantWriter) Get(tenant string) (*Writer, error)

func (*MultiTenantWriter) Range

func (mw *MultiTenantWriter) Range(fn func(string, *Writer) bool)

type Queue

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

func (*Queue) Pull

func (q *Queue) Pull() any

func (*Queue) PullNoWait

func (q *Queue) PullNoWait() any

func (*Queue) Push

func (q *Queue) Push(val any)

type TenantResult

type TenantResult struct {
	Mint, Maxt int64
	Ulid       string
	Tenant     string
	SamCount   uint64
}

func (*TenantResult) New

func (r *TenantResult) New(tenant string) *TenantResult

func (*TenantResult) Show

func (r *TenantResult) Show() string

type TenantResults

type TenantResults []*TenantResult

func (TenantResults) ToMap

func (t TenantResults) ToMap() map[string][]string

type Writer

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

func NewWriter

func NewWriter(logger log.Logger, dir string, blockSize int64, al labels.Labels, commitCount uint64, gc bool) (*Writer, error)

func (*Writer) AppendFn

func (w *Writer) AppendFn(fn *getSeries)

func (*Writer) FlushJob

func (w *Writer) FlushJob(wg *sync.WaitGroup, mint, maxt int64, tenant string, ch chan<- *TenantResult) func()

func (*Writer) WriteJob

func (w *Writer) WriteJob()