bulk_load

Documentation

Index

• Constants
• Variables
• func RegisterHandler(notifHandler NotifyHandler)
• type BatchProcessor
• type BulkLoad
• type HistoryItem
• type LoadRunner
  • func (r *LoadRunner) HasEndedPrematurely() bool
  • func (r *LoadRunner) Init(defaultBatchSize int)
  • func (r *LoadRunner) Run(load BulkLoad) int
  • func (r *LoadRunner) SetPrematureEnd(reason string)
  • func (r *LoadRunner) Validate()
• type NotifyHandler
• type NotifyReceiver
  • func (t *NotifyReceiver) Notify(args *int, reply *int) error
• type Scanner
• type SimpleRegression
  • func (sr *SimpleRegression) Intercept() float64
  • func (sr *SimpleRegression) Slope() float64
  • func (sr *SimpleRegression) Update(x, y float64)
• type Stat
  • func (s *Stat) Init(label []byte, value float64)
• type StatGroup
  • func (s *StatGroup) Push(n float64)
  • func (s *StatGroup) String() string
• type TimedStatGroup
  • func NewTimedStatGroup(maxDuration time.Duration, maxTrendSamples int) *TimedStatGroup
  • func (m *TimedStatGroup) Avg() float64
  • func (m *TimedStatGroup) Median() float64
  • func (m *TimedStatGroup) Push(timestamp time.Time, value float64)
  • func (m *TimedStatGroup) Rate() float64
  • func (m *TimedStatGroup) UpdateAvg(now time.Time, workers int) (float64, float64)
• type TrendStat
  • func NewTrendStat(size int, skipFirst bool) *TrendStat
  • func (ls *TrendStat) Add(y float64)

Constants

const ValuesPerMeasurement = 9.63636 // dashboard use-case, original value was: 11.2222

TODO: distinguish per use case

Variables

var Runner = &LoadRunner{}

Functions

func RegisterHandler

func RegisterHandler(notifHandler NotifyHandler)

Types

type BatchProcessor

type BatchProcessor interface {
	PrepareProcess(i int)
	RunProcess(i int, waitGroup *sync.WaitGroup, telemetryPoints chan *report.Point, reportTags [][2]string) error
	AfterRunProcess(i int)
	EmptyBatchChanel()
}

type BulkLoad

type BulkLoad interface {
	Init()
	Validate()
	CreateDb()
	PrepareWorkers()
	GetBatchProcessor() BatchProcessor
	GetScanner() Scanner
	SyncEnd()
	CleanUp()
	UpdateReport(params *report.LoadReportParams) (reportTags [][2]string, extraVals []report.ExtraVal)
}

type HistoryItem

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

type LoadRunner

type LoadRunner struct {
	DbName    string
	Workers   int
	ItemLimit int64
	BatchSize int
	TimeLimit time.Duration

	DoLoad         bool
	DoDBCreate     bool
	DoAbortOnExist bool

	DoJson bool

	StatPool sync.Pool
	StatChan chan *Stat

	Json map[string]interface{}
	// contains filtered or unexported fields
}

func (*LoadRunner) HasEndedPrematurely

func (r *LoadRunner) HasEndedPrematurely() bool

func (*LoadRunner) Init

func (r *LoadRunner) Init(defaultBatchSize int)

func (*LoadRunner) Run

func (r *LoadRunner) Run(load BulkLoad) int

func (*LoadRunner) SetPrematureEnd

func (r *LoadRunner) SetPrematureEnd(reason string)

func (*LoadRunner) Validate

func (r *LoadRunner) Validate()

type NotifyHandler

type NotifyHandler func(arg int) (int, error)

type NotifyReceiver

type NotifyReceiver struct {
}

func (*NotifyReceiver) Notify

func (t *NotifyReceiver) Notify(args *int, reply *int) error

type Scanner

type Scanner interface {
	RunScanner(r io.Reader, syncChanDone chan int)
	IsScanFinished() bool
	GetReadStatistics() (itemsRead, bytesRead, valuesRead int64)
}

type SimpleRegression

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

func (*SimpleRegression) Intercept

func (sr *SimpleRegression) Intercept() float64

func (*SimpleRegression) Slope

func (sr *SimpleRegression) Slope() float64

func (*SimpleRegression) Update

func (sr *SimpleRegression) Update(x, y float64)

type Stat

type Stat struct {
	Label []byte
	Value float64
}

Stat represents one statistical measurement.

func (*Stat) Init

func (s *Stat) Init(label []byte, value float64)

Init safely initializes a stat while minimizing heap allocations.

type StatGroup

type StatGroup struct {
	Min  float64
	Max  float64
	Mean float64
	Sum  float64

	Count int64
}

StatGroup collects simple streaming statistics.

func (*StatGroup) Push

func (s *StatGroup) Push(n float64)

Push updates a StatGroup with a new value.

func (*StatGroup) String

func (s *StatGroup) String() string

String makes a simple description of a StatGroup.

type TimedStatGroup

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

func NewTimedStatGroup

func NewTimedStatGroup(maxDuration time.Duration, maxTrendSamples int) *TimedStatGroup

func (*TimedStatGroup) Avg

func (m *TimedStatGroup) Avg() float64

func (*TimedStatGroup) Median

func (m *TimedStatGroup) Median() float64

func (*TimedStatGroup) Push

func (m *TimedStatGroup) Push(timestamp time.Time, value float64)

func (*TimedStatGroup) Rate

func (m *TimedStatGroup) Rate() float64

func (*TimedStatGroup) UpdateAvg

func (m *TimedStatGroup) UpdateAvg(now time.Time, workers int) (float64, float64)

type TrendStat

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

func NewTrendStat

func NewTrendStat(size int, skipFirst bool) *TrendStat

func (*TrendStat) Add

func (ls *TrendStat) Add(y float64)