uploader

Documentation

Index

• Constants
• type Base
  • func (u *Base) MarkAsFinished(filename string)
  • func (u *Base) RemoveFromQueue(filename string)
  • func (u *Base) Start() error
  • func (u *Base) Stat(send func(metric string, value float64))
• type Blacklist
  • func NewBlacklist(ignoredPatterns []string) *Blacklist
  • func (blacklist *Blacklist) Contains(value string, isReverse bool) bool
• type CMap
  • func NewCMap() CMap
  • func (m CMap) Add(key string, value int64)
  • func (m CMap) Clear() int
  • func (m CMap) Count() int
  • func (m CMap) Exists(key string) bool
  • func (m CMap) Expire(ctx context.Context, ttl time.Duration) (int, int64)
  • func (m CMap) ExpireWorker(ctx context.Context, ttl time.Duration, expiredCounter *uint32)
  • func (m CMap) GetShard(key string) *CMapShard
  • func (m CMap) Merge(keys map[string]bool, value int64)
• type CMapShard
• type Config
  • func (cfg *Config) Parse() error
• type DebugCacheDumper
• type Index
  • func NewIndex(base *Base) *Index
  • func (u Index) Reset()
  • func (u Index) Start() error
  • func (u Index) Stat(send func(metric string, value float64))
• type Points
  • func NewPoints(base *Base, reverse bool) *Points
• type Series
  • func NewSeries(base *Base, reverse bool) *Series
  • func (u Series) Reset()
  • func (u Series) Start() error
  • func (u Series) Stat(send func(metric string, value float64))
• type Tagged
  • func NewTagged(base *Base) *Tagged
  • func (u Tagged) Reset()
  • func (u Tagged) Start() error
  • func (u Tagged) Stat(send func(metric string, value float64))
• type Tree
  • func NewTree(base *Base) *Tree
  • func (u Tree) Reset()
  • func (u Tree) Start() error
  • func (u Tree) Stat(send func(metric string, value float64))
• type Uploader
  • func New(path string, name string, config *Config) (Uploader, error)
• type UploaderWithReset

Constants

const DefaultTreeDate = 42 // 1970-02-12

const ReverseLevelOffset = 10000

const ReverseTreeLevelOffset = 30000

const TreeLevelOffset = 20000

Types

type Base

type Base struct {
	stop.Struct
	sync.Mutex
	// contains filtered or unexported fields
}

func (*Base) MarkAsFinished

func (u *Base) MarkAsFinished(filename string)

func (*Base) RemoveFromQueue

func (u *Base) RemoveFromQueue(filename string)

func (*Base) Start

func (u *Base) Start() error

func (*Base) Stat

func (u *Base) Stat(send func(metric string, value float64))

type Blacklist

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

func NewBlacklist

func NewBlacklist(ignoredPatterns []string) *Blacklist

func (*Blacklist) Contains

func (blacklist *Blacklist) Contains(value string, isReverse bool) bool

Contains method determines whether or not this value is blacklisted

type CMap

type CMap []*CMapShard

A "thread" safe map of type string:Anything. To avoid lock bottlenecks this map is dived to several (shardCount) map shards.

func NewCMap

func NewCMap() CMap

Creates a new concurrent map.

func (CMap) Add

func (m CMap) Add(key string, value int64)

Sets the given value under the specified key.

func (CMap) Clear

func (m CMap) Clear() int

func (CMap) Count

func (m CMap) Count() int

Returns the number of elements within the map.

func (CMap) Exists

func (m CMap) Exists(key string) bool

Retrieves an element from map under given key.

func (CMap) Expire

func (m CMap) Expire(ctx context.Context, ttl time.Duration) (int, int64)

func (CMap) ExpireWorker

func (m CMap) ExpireWorker(ctx context.Context, ttl time.Duration, expiredCounter *uint32)

func (CMap) GetShard

func (m CMap) GetShard(key string) *CMapShard

Returns shard under given key

func (CMap) Merge

func (m CMap) Merge(keys map[string]bool, value int64)

type CMapShard

type CMapShard struct {
	sync.RWMutex // Read Write mutex, guards access to internal map.
	// contains filtered or unexported fields
}

A "thread" safe string to anything map.

type Config

type Config struct {
	Type                 string           `toml:"type"`  // points, series, points-reverse, series-reverse
	TableName            string           `toml:"table"` // keep empty for same as key
	Timeout              *config.Duration `toml:"timeout"`
	Date                 string           `toml:"date"` // for tree table
	TreeDate             time.Time        `toml:"-"`
	ZeroTimestamp        bool             `toml:"zero-timestamp"` // for points, points-reverse tables
	Threads              int              `toml:"threads"`
	URL                  string           `toml:"url"`
	CacheTTL             *config.Duration `toml:"cache-ttl"`
	IgnoredPatterns      []string         `toml:"ignored-patterns,omitempty"` // points, points-reverse
	CompressData         bool             `toml:"compress-data"`              //compress data while sending to clickhouse
	IgnoredTaggedMetrics []string         `toml:"ignored-tagged-metrics"`     // for tagged table; create only `__name__` tag for these metrics and ignore others
	Hash                 string           `toml:"hash"`                       // in index uploader store hash in memory instead of full metric
	DisableDailyIndex    bool             `toml:"disable-daily-index"`        // do not calculate and upload daily index to ClickHouse
	// contains filtered or unexported fields
}

func (*Config) Parse

func (cfg *Config) Parse() error

type DebugCacheDumper

type DebugCacheDumper interface {
	CacheDump(io.Writer)
}

type Index

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

func NewIndex

func NewIndex(base *Base) *Index

func (Index) Reset

func (u Index) Reset()

func (Index) Start

func (u Index) Start() error

func (Index) Stat

func (u Index) Stat(send func(metric string, value float64))

type Points

type Points struct {
	*Base
	// contains filtered or unexported fields
}

func NewPoints

func NewPoints(base *Base, reverse bool) *Points

type Series

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

func NewSeries

func NewSeries(base *Base, reverse bool) *Series

func (Series) Reset

func (u Series) Reset()

func (Series) Start

func (u Series) Start() error

func (Series) Stat

func (u Series) Stat(send func(metric string, value float64))

type Tagged

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

func NewTagged

func NewTagged(base *Base) *Tagged

func (Tagged) Reset

func (u Tagged) Reset()

func (Tagged) Start

func (u Tagged) Start() error

func (Tagged) Stat

func (u Tagged) Stat(send func(metric string, value float64))

type Tree

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

func NewTree

func NewTree(base *Base) *Tree

func (Tree) Reset

func (u Tree) Reset()

func (Tree) Start

func (u Tree) Start() error

func (Tree) Stat

func (u Tree) Stat(send func(metric string, value float64))

type Uploader

type Uploader interface {
	Start() error
	Stop()
	Stat(send func(metric string, value float64))
}

func New

func New(path string, name string, config *Config) (Uploader, error)

type UploaderWithReset

type UploaderWithReset interface {
	Reset()
}