Skip to content

Commit

Permalink
fixed stdev metric and added test
Browse files Browse the repository at this point in the history
  • Loading branch information
toni-moreno committed Dec 14, 2016
1 parent d631f9a commit 3c7e9ac
Show file tree
Hide file tree
Showing 2 changed files with 169 additions and 21 deletions.
39 changes: 18 additions & 21 deletions plugins/aggregators/basicstats/basicstats.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
package basicstats

import (
"math"

"github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/plugins/aggregators"
)
Expand All @@ -26,8 +28,8 @@ type basicstats struct {
min float64
max float64
mean float64
M2 float64 //intermeia
stdev float64
M2 float64 //intermemedia value for variance/stdev
//stdev float64
}

var sampleConfig = `
Expand Down Expand Up @@ -64,7 +66,6 @@ func (m *BasicStats) Add(in telegraf.Metric) {
max: fv,
mean: fv,
M2: 0.0,
stdev: 0.0,
}
}
}
Expand All @@ -80,38 +81,33 @@ func (m *BasicStats) Add(in telegraf.Metric) {
max: fv,
mean: fv,
M2: 0.0,
stdev: 0.0,
}
continue
}

tmp := m.cache[id].fields[k]
//https://en.m.wikipedia.org/wiki/Algorithms_for_calculating_variance
//counter
//variable initialization
x := fv
mean := tmp.mean
M2 := tmp.M2
//counter compute
n := tmp.count + 1
//mean

tmp.count = n
//mean compute
delta := x - mean
mean = mean + delta/n
M2 = M2 + delta*(x-mean)

tmp.mean = mean
//variance/stdev compute
M2 = M2 + delta*(x-mean)
tmp.M2 = M2
tmp.count = n

//max/min compute
if fv < tmp.min {
//tmp := m.cache[id].fields[k]
tmp.min = fv
//m.cache[id].fields[k] = tmp
} else if fv > tmp.max {
//tmp := m.cache[id].fields[k]
tmp.max = fv
//m.cache[id].fields[k] = tmp
}

//store final data
m.cache[id].fields[k] = tmp
}
}
Expand All @@ -126,12 +122,13 @@ func (m *BasicStats) Push(acc telegraf.Accumulator) {
fields[k+"_min"] = v.min
fields[k+"_max"] = v.max
fields[k+"_mean"] = v.mean
if v.count == 1 {
fields[k+"_stdev"] = 0.0
} else {
fields[k+"_stdev"] = v.M2 / (v.count - 1)
//v.count always >=1
if v.count > 1 {
variance := v.M2 / (v.count - 1)
fields[k+"_s2"] = variance
fields[k+"_stdev"] = math.Sqrt(variance)
}

//if count == 1 StdDev = infinite => so I won't send data
}
acc.AddFields(aggregate.name, fields, aggregate.tags)
}
Expand Down
151 changes: 151 additions & 0 deletions plugins/aggregators/basicstats/basicstats_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,151 @@
package basicstats

import (
"math"
"testing"
"time"

"github.com/influxdata/telegraf/metric"
"github.com/influxdata/telegraf/testutil"
)

var m1, _ = metric.New("m1",
map[string]string{"foo": "bar"},
map[string]interface{}{
"a": int64(1),
"b": int64(1),
"c": float64(2),
"d": float64(2),
},
time.Now(),
)
var m2, _ = metric.New("m1",
map[string]string{"foo": "bar"},
map[string]interface{}{
"a": int64(1),
"b": int64(3),
"c": float64(4),
"d": float64(6),
"e": float64(200),
"ignoreme": "string",
"andme": true,
},
time.Now(),
)

func BenchmarkApply(b *testing.B) {
minmax := NewBasicStats()

for n := 0; n < b.N; n++ {
minmax.Add(m1)
minmax.Add(m2)
}
}

// Test two metrics getting added.
func TestBasicStatsWithPeriod(t *testing.T) {
acc := testutil.Accumulator{}
minmax := NewBasicStats()

minmax.Add(m1)
minmax.Add(m2)
minmax.Push(&acc)

expectedFields := map[string]interface{}{
"a_count": float64(2), //a
"a_max": float64(1),
"a_min": float64(1),
"a_mean": float64(1),
"a_stdev": float64(0),
"a_s2": float64(0),
"b_count": float64(2), //b
"b_max": float64(3),
"b_min": float64(1),
"b_mean": float64(2),
"b_s2": float64(2),
"b_stdev": math.Sqrt(2),
"c_count": float64(2), //c
"c_max": float64(4),
"c_min": float64(2),
"c_mean": float64(3),
"c_s2": float64(2),
"c_stdev": math.Sqrt(2),
"d_count": float64(2), //d
"d_max": float64(6),
"d_min": float64(2),
"d_mean": float64(4),
"d_s2": float64(8),
"d_stdev": math.Sqrt(8),
"e_count": float64(1), //e
"e_max": float64(200),
"e_min": float64(200),
"e_mean": float64(200),
}
expectedTags := map[string]string{
"foo": "bar",
}
acc.AssertContainsTaggedFields(t, "m1", expectedFields, expectedTags)
}

// Test two metrics getting added with a push/reset in between (simulates
// getting added in different periods.)
func TestBasicStatsDifferentPeriods(t *testing.T) {
acc := testutil.Accumulator{}
minmax := NewBasicStats()

minmax.Add(m1)
minmax.Push(&acc)
expectedFields := map[string]interface{}{
"a_count": float64(1), //a
"a_max": float64(1),
"a_min": float64(1),
"a_mean": float64(1),
"b_count": float64(1), //b
"b_max": float64(1),
"b_min": float64(1),
"b_mean": float64(1),
"c_count": float64(1), //c
"c_max": float64(2),
"c_min": float64(2),
"c_mean": float64(2),
"d_count": float64(1), //d
"d_max": float64(2),
"d_min": float64(2),
"d_mean": float64(2),
}
expectedTags := map[string]string{
"foo": "bar",
}
acc.AssertContainsTaggedFields(t, "m1", expectedFields, expectedTags)

acc.ClearMetrics()
minmax.Reset()
minmax.Add(m2)
minmax.Push(&acc)
expectedFields = map[string]interface{}{
"a_count": float64(1), //a
"a_max": float64(1),
"a_min": float64(1),
"a_mean": float64(1),
"b_count": float64(1), //b
"b_max": float64(3),
"b_min": float64(3),
"b_mean": float64(3),
"c_count": float64(1), //c
"c_max": float64(4),
"c_min": float64(4),
"c_mean": float64(4),
"d_count": float64(1), //d
"d_max": float64(6),
"d_min": float64(6),
"d_mean": float64(6),
"e_count": float64(1), //e
"e_max": float64(200),
"e_min": float64(200),
"e_mean": float64(200),
}
expectedTags = map[string]string{
"foo": "bar",
}
acc.AssertContainsTaggedFields(t, "m1", expectedFields, expectedTags)
}

0 comments on commit 3c7e9ac

Please sign in to comment.