aggregator

Documentation

Overview

Receives stats messages from components, aggregates them pushes to the a stats reporter.

Example Usage:

import "github.com/Nextdoor/pg-bifrost.git/stats/Aggregator"

// Setup
in_stats := make(chan Stat, 1000)
statsAggregator = Aggregator.New(in_stats, time.Now)
go statsAggregator.StartTransporting()
out_stats := statsAggregator.outputChan

// send a Stat
Stat = NewStatCount("rplclient",
		"bufferedMessages",
		1,
		timeNow)
in_stats <- Stat

// receive aggregated stats
aggregated_stat := <-out_stats

Index

• type Aggregator
  • func New(shutdownHandler shutdown.ShutdownHandler, inputChan <-chan stats.Stat) Aggregator
  • func NewConfigureAggregates(shutdownHandler shutdown.ShutdownHandler, muAggregates *sync.Mutex, ...) Aggregator
  • func (a *Aggregator) GetOutputChan() chan stats.Stat
  • func (a *Aggregator) Start()

Types

type Aggregator

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

Aggregator type that holds send/receive channels and map of Stat aggregates

func New

func New(shutdownHandler shutdown.ShutdownHandler, inputChan <-chan stats.Stat) Aggregator

New returns an Aggregator object with defaults.

func NewConfigureAggregates

func NewConfigureAggregates(shutdownHandler shutdown.ShutdownHandler,
	muAggregates *sync.Mutex,
	inputChan <-chan stats.Stat,
	outputChan chan stats.Stat,
	timeNow func() time.Time,
	aggregateTimeNano int64,
	aggregateMaxBuckets uint,
	aggregates map[int64]map[string]*aggregate) Aggregator

NewConfigureAggregates returns an Aggregator object primarily for use in tests and lets us configure Aggregator parameters and provide a time.Now function for time manipulation.

func (*Aggregator) GetOutputChan

func (a *Aggregator) GetOutputChan() chan stats.Stat

GetOutputChan returns the channel of aggregates for use by a reporter.

func (*Aggregator) Start

func (a *Aggregator) Start()

StartTransporting loops on the input channel and aggregates stats from it.