Fixup some code based on feedback from @dgnorton

agent/agent.go

@@ -300,6 +300,8 @@ func (a *Agent) flusher(shutdown chan struct{}, metricC chan telegraf.Metric) er
  internal.RandomSleep(a.Config.Agent.FlushJitter.Duration, shutdown)
  a.flush()
  case metric := <-metricC:
+ // NOTE potential bottleneck here as we put each metric through the
+ // processors serially.
  mS := []telegraf.Metric{metric}
  for _, processor := range a.Config.Processors {
  mS = processor.Apply(mS...)
@@ -321,7 +323,7 @@ func (a *Agent) Run(shutdown chan struct{}) error {
  a.Config.Agent.Hostname, a.Config.Agent.FlushInterval.Duration)
 
  // channel shared between all input threads for accumulating metrics
- metricC := make(chan telegraf.Metric, 10000)
+ metricC := make(chan telegraf.Metric, 100)
 
  // Start all ServicePlugins
  for _, input := range a.Config.Inputs {

internal/config/config.go

@@ -841,7 +841,11 @@ func buildAggregator(name string, tbl *ast.Table) (*models.AggregatorConfig, err
  }
  }
 
- conf := &models.AggregatorConfig{Name: name}
+ conf := &models.AggregatorConfig{
+ Name: name,
+ Delay: time.Millisecond * 100,
+ Period: time.Second * 30,
+ }
 
  if node, ok := tbl.Fields["period"]; ok {
  if kv, ok := node.(*ast.KeyValue); ok {

internal/models/makemetric.go

@@ -19,6 +19,7 @@ import (
 // applyFilter: if false, the above filter is not applied to each metric.
 // This is used by Aggregators, because aggregators use filters
 // on incoming metrics instead of on created metrics.
+// TODO refactor this to not have such a huge func signature.
 func makemetric(
  measurement string,
  fields map[string]interface{},

internal/models/running_aggregator.go

@@ -109,13 +109,6 @@ func (r *RunningAggregator) Run(
  acc telegraf.Accumulator,
  shutdown chan struct{},
 ) {
- if r.Config.Delay == 0 {
- r.Config.Delay = time.Millisecond * 100
- }
- if r.Config.Period == 0 {
- r.Config.Period = time.Second * 30
- }
-
  time.Sleep(r.Config.Delay)
  periodT := time.NewTicker(r.Config.Period)
  defer periodT.Stop()

internal/models/running_aggregator_test.go

@@ -20,6 +20,7 @@ func TestAdd(t *testing.T) {
  Filter: Filter{
  NamePass: []string{"*"},
  },
+ Period: time.Millisecond * 500,
  })
  assert.NoError(t, ra.Config.Filter.Compile())
  acc := testutil.Accumulator{}

metric.go

@@ -35,6 +35,7 @@ type Metric interface {
  UnixNano() int64
 
  // HashID returns a non-cryptographic hash of the metric (name + tags)
+ // NOTE: do not persist & depend on this value to disk.
  HashID() uint64
 
  // Fields returns the fields for the metric

plugins/aggregators/minmax/minmax.go

@@ -6,10 +6,7 @@ import (
 )
 
 type MinMax struct {
- // caches for metric fields, names, and tags
- fieldCache map[uint64]map[string]minmax
- nameCache map[uint64]string
- tagCache map[uint64]map[string]string
+ cache map[uint64]aggregate
 }
 
 func NewMinMax() telegraf.Aggregator {
@@ -18,6 +15,12 @@ func NewMinMax() telegraf.Aggregator {
  return mm
 }
 
+type aggregate struct {
+ fields map[string]minmax
+ name string
+ tags map[string]string
+}
+
 type minmax struct {
  min float64
  max float64
@@ -42,71 +45,60 @@ func (m *MinMax) Description() string {
 
 func (m *MinMax) Add(in telegraf.Metric) {
  id := in.HashID()
- if _, ok := m.nameCache[id]; !ok {
+ if _, ok := m.cache[id]; !ok {
  // hit an uncached metric, create caches for first time:
- m.nameCache[id] = in.Name()
- m.tagCache[id] = in.Tags()
- m.fieldCache[id] = make(map[string]minmax)
+ a := aggregate{
+ name: in.Name(),
+ tags: in.Tags(),
+ fields: make(map[string]minmax),
+ }
  for k, v := range in.Fields() {
  if fv, ok := convert(v); ok {
- m.fieldCache[id][k] = minmax{
+ a.fields[k] = minmax{
  min: fv,
  max: fv,
  }
  }
  }
+ m.cache[id] = a
  } else {
  for k, v := range in.Fields() {
  if fv, ok := convert(v); ok {
- if _, ok := m.fieldCache[id][k]; !ok {
+ if _, ok := m.cache[id].fields[k]; !ok {
  // hit an uncached field of a cached metric
- m.fieldCache[id][k] = minmax{
+ m.cache[id].fields[k] = minmax{
  min: fv,
  max: fv,
  }
  continue
  }
- cmpmin := compare(m.fieldCache[id][k].min, fv)
- cmpmax := compare(m.fieldCache[id][k].max, fv)
- if cmpmin == 1 {
- tmp := m.fieldCache[id][k]
+ if fv < m.cache[id].fields[k].min {
+ tmp := m.cache[id].fields[k]
  tmp.min = fv
- m.fieldCache[id][k] = tmp
- }
- if cmpmax == -1 {
- tmp := m.fieldCache[id][k]
+ m.cache[id].fields[k] = tmp
+ } else if fv > m.cache[id].fields[k].max {
+ tmp := m.cache[id].fields[k]
  tmp.max = fv
- m.fieldCache[id][k] = tmp
+ m.cache[id].fields[k] = tmp
  }
  }
  }
  }
 }
 
 func (m *MinMax) Push(acc telegraf.Accumulator) {
- for id, _ := range m.nameCache {
+ for _, aggregate := range m.cache {
  fields := map[string]interface{}{}
- for k, v := range m.fieldCache[id] {
+ for k, v := range aggregate.fields {
  fields[k+"_min"] = v.min
  fields[k+"_max"] = v.max
  }
- acc.AddFields(m.nameCache[id], fields, m.tagCache[id])
+ acc.AddFields(aggregate.name, fields, aggregate.tags)
  }
 }
 
 func (m *MinMax) Reset() {
- m.fieldCache = make(map[uint64]map[string]minmax)
- m.nameCache = make(map[uint64]string)
- m.tagCache = make(map[uint64]map[string]string)
-}
-
-func compare(a, b float64) int {
- if a < b {
- return -1
- } else if a > b {
- return 1
- }
- return 0
+ m.cache = make(map[uint64]aggregate)
 }
 
 func convert(in interface{}) (float64, bool) {