collector

README

go-bitflow-collector

go-bitflow-collector is a Go (Golang) tool for collecting time-series data from various sources in high frequency intervals. It uses the github.com/bitflow-stream/go-bitflow library for generating and providing bitflow.Sample instances. The bitflow-collector sub-package provides an executable with the same name. The data collection and other configuration options can be configured through numerous command line flags.

Run bitflow-collector --help for a list of command line flags.

The main source of data is the /proc filesystem on the local Linux machine (although data collection should also work on other platforms in general). Other implemented data sources include the remote API provided by libvirt and the OVSDB protocol offered by Open vSwitch.

Installation:

• Install packages: libvirt-dev libpcap-dev
• Install git and go (at least version 1.8).
• Make sure $GOPATH is set to some existing directory.
• Get and install this tool:

go get github.com/bitflow-stream/go-bitflow-collector/bitflow-collector

• The binary executable bitflow-collector will be compiled to $GOPATH/bin.
• Add that directory to your $PATH, or copy the executable to a different location.

Installation without PCAP and LIBVIRT

To avoid installing these dependencies, follow above instructions, but change the go get command to the following:

go get -tags "nopcap nolibvirt" github.com/bitflow-stream/go-bitflow-collector/bitflow-collector

Documentation

Index

• Constants
• Variables
• type AbstractCollector
  • func RootCollector(name string) AbstractCollector
  • func (col *AbstractCollector) Child(name string) AbstractCollector
  • func (col *AbstractCollector) Depends() []Collector
  • func (col *AbstractCollector) Init() ([]Collector, error)
  • func (col *AbstractCollector) Metrics() MetricReaderMap
  • func (col *AbstractCollector) MetricsChanged() error
  • func (col *AbstractCollector) String() string
  • func (col *AbstractCollector) Update() error
• type Collector
• type LogbackValue
• type Metric
  • func (metric *Metric) Update()
• type MetricReader
• type MetricReaderMap
• type MetricSlice
  • func (s MetricSlice) ConstructSample(source *SampleSource) ([]string, func() []bitflow.Value)
  • func (s MetricSlice) Len() int
  • func (s MetricSlice) Less(i, j int) bool
  • func (s MetricSlice) Swap(i, j int)
  • func (s MetricSlice) UpdateAll()
• type SampleSource
  • func (source *SampleSource) Close()
  • func (source *SampleSource) CurrentMetrics() []string
  • func (source *SampleSource) PrintGraph(file string, fullGraph bool) error
  • func (source *SampleSource) PrintGraphDot(file string, fullGraph bool) error
  • func (source *SampleSource) PrintMetrics() error
  • func (source *SampleSource) Start(wg *sync.WaitGroup) golib.StopChan
  • func (source *SampleSource) String() string
• type StoredValue
  • func (val StoredValue) AddValue(incoming LogbackValue) LogbackValue
  • func (val StoredValue) DiffValue(logback LogbackValue, interval time.Duration) bitflow.Value
  • func (val StoredValue) GetValue() bitflow.Value
• type TimedValue
  • func (val TimedValue) String() string
• type ValueRing
  • func (ring *ValueRing) Add(val LogbackValue)
  • func (ring *ValueRing) AddToHead(val LogbackValue)
  • func (ring *ValueRing) AddValue(val bitflow.Value)
  • func (ring *ValueRing) AddValueToHead(val bitflow.Value)
  • func (ring *ValueRing) FlushHead()
  • func (ring *ValueRing) GetDiff() bitflow.Value
  • func (ring *ValueRing) GetHead() LogbackValue
  • func (ring *ValueRing) GetHeadValue() bitflow.Value
  • func (ring *ValueRing) Increment(val LogbackValue)
  • func (ring *ValueRing) IncrementValue(val bitflow.Value)
• type ValueRingFactory
  • func (factory *ValueRingFactory) NewValueRing() *ValueRing

Constants

const (
	ToleratedUpdateFailures = 2
)

Variables

var MetricsChanged = errors.New("Metrics of this collector have changed")

Types

type AbstractCollector

type AbstractCollector struct {
	Parent *AbstractCollector
	Name   string
}

================================= Abstract Collector =================================

func RootCollector

func RootCollector(name string) AbstractCollector

func (*AbstractCollector) Child

func (col *AbstractCollector) Child(name string) AbstractCollector

func (*AbstractCollector) Depends

func (col *AbstractCollector) Depends() []Collector

func (*AbstractCollector) Init

func (col *AbstractCollector) Init() ([]Collector, error)

func (*AbstractCollector) Metrics

func (col *AbstractCollector) Metrics() MetricReaderMap

func (*AbstractCollector) MetricsChanged

func (col *AbstractCollector) MetricsChanged() error

func (*AbstractCollector) String

func (col *AbstractCollector) String() string

func (*AbstractCollector) Update

func (col *AbstractCollector) Update() error

type Collector

type Collector interface {

	// Init prepares this collector for collecting metrics and instantiates sub-collectors.
	// If there is no error, the sub-collectors will also be initialized, until there are
	// no more sub-nodes. The metrics in the MetricReaderMap are all stored in one flat list,
	// the keys must be globally unique.
	Init() (subCollectors []Collector, err error)

	// Metrics will only be called after Init() returned successfully. It returns the metrics
	// that are provided by this collector.
	Metrics() MetricReaderMap

	// Depends returns a slice of collectors whose Update() this collector depends on.
	// This means that this collector needs data from those other collectors to perform
	// its Update() routine correctly. Therefore, Update() will be called on those other
	// collectors first. The Depends() methods build up an acyclic dependency graph, whose
	// topological order gives the order of Update() calls.
	Depends() []Collector

	// All collectors are updated in the order they were initialized: from the root node, down the tree.
	// An error stops descending down the tree. After a collector has been updated,
	// the metrics associated with that collector will be read. Collectors with only excluded metrics
	// will not be updated.
	Update() error

	// MetricsChanged should check if the collector can produce a different set of metrics, and if so,
	// the MetricsChanged error instance. Many collectors have a fixed set of metrics, so nil should
	// be returned here (as in AbstractCollector). Collectors that potentially return MetricsChanged from
	// Update(), should use Update() as implementation for MetricsChanged().
	MetricsChanged() error

	// String returns a short but unique label for the collector.
	String() string
}

Collector forms a tree-structure of objects that are able to provide regularly updated metric values. A collector is first initialized, which can optionally return a new list of Collectors that will also be considered. The new collectors will also be initialized, until the tree exhausted. Individual collectors can fail the initialization, which will not influence the non-failed collectors. After the Init() sequence, the Metrics() method is queried to retrieve a list of metrics that are delivered by every collector. It may return an empty slice in case of collectors that are only there to satisfy dependencies of other collectors. Then, the Depends() method is used to build up a dependency graph between the collectors. Typically, each collector will returns its parent-collector as sole dependency, but it can also return an empty slice or multiple dependencies. All collectors returned from any Depends() method must already have been initialized in the Init() sequence.

type LogbackValue

type LogbackValue interface {
	DiffValue(previousValue LogbackValue, interval time.Duration) bitflow.Value
	AddValue(val LogbackValue) LogbackValue
	GetValue() bitflow.Value
}

type Metric

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

==================== Metric ====================

func (*Metric) Update

func (metric *Metric) Update()

type MetricReader

type MetricReader func() bitflow.Value

type MetricReaderMap

type MetricReaderMap map[string]MetricReader

type MetricSlice

type MetricSlice []*Metric

==================== Metric Slice ====================

func (MetricSlice) ConstructSample

func (s MetricSlice) ConstructSample(source *SampleSource) ([]string, func() []bitflow.Value)

func (MetricSlice) Len

func (s MetricSlice) Len() int

func (MetricSlice) Less

func (s MetricSlice) Less(i, j int) bool

func (MetricSlice) Swap

func (s MetricSlice) Swap(i, j int)

func (MetricSlice) UpdateAll

func (s MetricSlice) UpdateAll()

type SampleSource

type SampleSource struct {
	bitflow.AbstractSampleSource

	RootCollectors     []Collector
	CollectInterval    time.Duration
	UpdateFrequencies  map[*regexp.Regexp]time.Duration
	SinkInterval       time.Duration
	ExcludeMetrics     []*regexp.Regexp
	IncludeMetrics     []*regexp.Regexp
	DisabledCollectors []string

	FailedCollectorCheckInterval   time.Duration
	FilteredCollectorCheckInterval time.Duration
	// contains filtered or unexported fields
}

func (*SampleSource) Close

func (source *SampleSource) Close()

func (*SampleSource) CurrentMetrics

func (source *SampleSource) CurrentMetrics() []string

func (*SampleSource) PrintGraph

func (source *SampleSource) PrintGraph(file string, fullGraph bool) error

func (*SampleSource) PrintGraphDot

func (source *SampleSource) PrintGraphDot(file string, fullGraph bool) error

func (*SampleSource) PrintMetrics

func (source *SampleSource) PrintMetrics() error

func (*SampleSource) Start

func (source *SampleSource) Start(wg *sync.WaitGroup) golib.StopChan

func (*SampleSource) String

func (source *SampleSource) String() string

type StoredValue

type StoredValue bitflow.Value

func (StoredValue) AddValue

func (val StoredValue) AddValue(incoming LogbackValue) LogbackValue

func (StoredValue) DiffValue

func (val StoredValue) DiffValue(logback LogbackValue, interval time.Duration) bitflow.Value

func (StoredValue) GetValue

func (val StoredValue) GetValue() bitflow.Value

type TimedValue

type TimedValue struct {
	time.Time // Timestamp of recording
	// contains filtered or unexported fields
}

func (TimedValue) String

func (val TimedValue) String() string

type ValueRing

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

func (*ValueRing) Add

func (ring *ValueRing) Add(val LogbackValue)

func (*ValueRing) AddToHead

func (ring *ValueRing) AddToHead(val LogbackValue)

func (*ValueRing) AddValue

func (ring *ValueRing) AddValue(val bitflow.Value)

func (*ValueRing) AddValueToHead

func (ring *ValueRing) AddValueToHead(val bitflow.Value)

func (*ValueRing) FlushHead

func (ring *ValueRing) FlushHead()

func (*ValueRing) GetDiff

func (ring *ValueRing) GetDiff() bitflow.Value

func (*ValueRing) GetHead

func (ring *ValueRing) GetHead() LogbackValue

May return nil in case of an empty ring

func (*ValueRing) GetHeadValue

func (ring *ValueRing) GetHeadValue() bitflow.Value

func (*ValueRing) Increment

func (ring *ValueRing) Increment(val LogbackValue)

func (*ValueRing) IncrementValue

func (ring *ValueRing) IncrementValue(val bitflow.Value)

type ValueRingFactory

type ValueRingFactory struct {
	Length   int
	Interval time.Duration
}

func (*ValueRingFactory) NewValueRing

func (factory *ValueRingFactory) NewValueRing() *ValueRing