Add Histogram support aggregation

Godeps

@@ -1,5 +1,6 @@
 github.com/Shopify/sarama 8aadb476e66ca998f2f6bb3c993e9a2daa3666b9
 github.com/Sirupsen/logrus 219c8cb75c258c552e999735be6df753ffc7afdc
+github.com/VividCortex/gohistogram da38b6e56f2f7dc1999a037141441e50d6213f5d
 github.com/amir/raidman 53c1b967405155bfc8758557863bf2e14f814687
 github.com/aws/aws-sdk-go 13a12060f716145019378a10e2806c174356b857
 github.com/beorn7/perks 3ac7bf7a47d159a033b107610db8a1b6575507a4

agent/agent.go

@@ -242,28 +242,43 @@ func (a *Agent) flush() {
  }
  }(o)
  }
-
  wg.Wait()
 }
 
 // flusher monitors the metrics input channel and flushes on the minimum interval
 func (a *Agent) flusher(shutdown chan struct{}, metricC chan telegraf.Metric) error {
+ var metricStream chan telegraf.Metric
+ stopFilter := make(chan struct{}, len(a.Config.Filters))
  // Inelegant, but this sleep is to allow the Gather threads to run, so that
  // the flusher will flush after metrics are collected.
  time.Sleep(time.Millisecond * 200)
-
  ticker := time.NewTicker(a.Config.Agent.FlushInterval.Duration)
+ metricStream = metricC
+ if len(a.Config.Filters) != 0 {
+ for _, name := range a.Config.FiltersOrder {
+ filter := a.Config.Filters[name]
+ log.Printf("Filter %s is enabled", name)
+ metricStream = filter.Pipe(metricStream)
+ go func(f telegraf.Filter) {
+ f.Start(stopFilter)
+ }(filter)
+ }
+ }
 
  for {
  select {
  case <-shutdown:
+ //sending shutdown signal for all filters
+ for range a.Config.Filters {
+ stopFilter <- struct{}{}
+ }
  log.Println("Hang on, flushing any cached metrics before shutdown")
  a.flush()
  return nil
  case <-ticker.C:
  internal.RandomSleep(a.Config.Agent.FlushJitter.Duration, shutdown)
  a.flush()
- case m := <-metricC:
+ case m := <-metricStream:
  for _, o := range a.Config.Outputs {
  o.AddMetric(m)
  }

cmd/telegraf/telegraf.go

@@ -11,6 +11,7 @@ import (
 
  "github.com/influxdata/telegraf/agent"
  "github.com/influxdata/telegraf/internal/config"
+ _ "github.com/influxdata/telegraf/plugins/filters/all"
  "github.com/influxdata/telegraf/plugins/inputs"
  _ "github.com/influxdata/telegraf/plugins/inputs/all"
  "github.com/influxdata/telegraf/plugins/outputs"

filter.go

@@ -0,0 +1,15 @@
+package telegraf
+
+type Filter interface {
+ // SampleConfig returns the default configuration of the Input
+ SampleConfig() string
+
+ // Description returns a one-sentence description on the Input
+ Description() string
+
+ //create pipe for filter
+ Pipe(in chan Metric) chan Metric
+
+ // start the filter
+ Start(shutdown chan struct{})
+}

internal/config/config.go

@@ -16,6 +16,7 @@ import (
  "github.com/influxdata/telegraf"
  "github.com/influxdata/telegraf/internal"
  "github.com/influxdata/telegraf/internal/models"
+ "github.com/influxdata/telegraf/plugins/filters"
  "github.com/influxdata/telegraf/plugins/inputs"
  "github.com/influxdata/telegraf/plugins/outputs"
  "github.com/influxdata/telegraf/plugins/parsers"
@@ -46,9 +47,11 @@ type Config struct {
  InputFilters []string
  OutputFilters []string
 
- Agent *AgentConfig
- Inputs []*internal_models.RunningInput
- Outputs []*internal_models.RunningOutput
+ Agent *AgentConfig
+ Inputs []*internal_models.RunningInput
+ Outputs []*internal_models.RunningOutput
+ Filters map[string]telegraf.Filter
+ FiltersOrder []string
 }
 
 func NewConfig() *Config {
@@ -63,6 +66,8 @@ func NewConfig() *Config {
  Tags: make(map[string]string),
  Inputs: make([]*internal_models.RunningInput, 0),
  Outputs: make([]*internal_models.RunningOutput, 0),
+ Filters: make(map[string]telegraf.Filter),
+ FiltersOrder: make([]string, 0),
  InputFilters: make([]string, 0),
  OutputFilters: make([]string, 0),
  }
@@ -516,6 +521,25 @@ func (c *Config) LoadConfig(path string) error {
  pluginName, path)
  }
  }
+ case "filter":
+ for pluginName, pluginVal := range subTable.Fields {
+ switch pluginSubTable := pluginVal.(type) {
+ case *ast.Table:
+ if err = c.addFilter(pluginName, pluginSubTable); err != nil {
+ return fmt.Errorf("Error parsing %s, %s", path, err)
+ }
+ case []*ast.Table:
+ for _, t := range pluginSubTable {
+ if err = c.addFilter(pluginName, t); err != nil {
+ return fmt.Errorf("Error parsing %s, %s", path, err)
+ }
+ }
+ default:
+ return fmt.Errorf("Unsupported config format: %s, file %s",
+ pluginName, path)
+ }
+ }
+
  // Assume it's an input input for legacy config file support if no other
  // identifiers are present
  default:
@@ -583,6 +607,25 @@ func (c *Config) addOutput(name string, table *ast.Table) error {
  return nil
 }
 
+func (c *Config) addFilter(name string, table *ast.Table) error {
+ creator, ok := filters.Filters[name]
+ if !ok {
+ return fmt.Errorf("Undefined but requested filter: %s", name)
+ }
+ filter := creator()
+
+ if err := config.UnmarshalTable(table, filter); err != nil {
+ return err
+ }
+
+ if _, ok = c.Filters[name]; ok {
+ return fmt.Errorf("Filter already defined %s", name)
+ }
+ c.Filters[name] = filter
+ c.FiltersOrder = append(c.FiltersOrder, name)
+ return nil
+}
+
 func (c *Config) addInput(name string, table *ast.Table) error {
  if len(c.InputFilters) > 0 && !sliceContains(name, c.InputFilters) {
  return nil

plugins/filters/all/all.go

@@ -0,0 +1,5 @@
+package all
+
+import (
+ _ "github.com/influxdata/telegraf/plugins/filters/histogram"
+)

plugins/filters/histogram/aggregate.go

@@ -0,0 +1,65 @@
+package histogram
+
+import (
+ "github.com/VividCortex/gohistogram"
+ "math"
+)
+
+type Aggregate struct {
+ hist *gohistogram.NumericHistogram
+ sum float64
+ max float64
+ min float64
+}
+
+func (a *Aggregate) Add(n float64) {
+ a.sum += n
+ if a.max < n {
+ a.max = n
+ }
+ if a.min > n {
+ a.min = n
+ }
+ a.hist.Add(n)
+}
+
+func (a *Aggregate) Quantile(q float64) float64 {
+ return a.hist.Quantile(q)
+}
+
+func (a *Aggregate) Sum() float64 {
+ return a.sum
+}
+
+func (a *Aggregate) CDF(x float64) float64 {
+ return a.hist.CDF(x)
+}
+
+func (a *Aggregate) Mean() float64 {
+ return a.hist.Mean()
+}
+
+func (a *Aggregate) Variance() float64 {
+ return a.hist.Variance()
+}
+
+func (a *Aggregate) Count() float64 {
+ return a.hist.Count()
+}
+
+func (a *Aggregate) Max() float64 {
+ return a.max
+}
+
+func (a *Aggregate) Min() float64 {
+ return a.min
+}
+
+func NewAggregate(n int) *Aggregate {
+ return &Aggregate{
+ hist: gohistogram.NewHistogram(n),
+ max: math.SmallestNonzeroFloat64,
+ min: math.MaxFloat64,
+ sum: 0,
+ }
+}