Change minmax aggregator to store float64

This commit is contained in:
Cameron Sparr 2016-09-20 14:15:23 +01:00
parent 64a71263a1
commit ef885eda62
5 changed files with 303 additions and 75 deletions

View File

@ -887,8 +887,7 @@ func buildAggregator(name string, tbl *ast.Table) (*models.AggregatorConfig, err
// buildProcessor TODO doc // buildProcessor TODO doc
func buildProcessor(name string, tbl *ast.Table) (*models.ProcessorConfig, error) { func buildProcessor(name string, tbl *ast.Table) (*models.ProcessorConfig, error) {
conf := &models.ProcessorConfig{Name: name} conf := &models.ProcessorConfig{Name: name}
unsupportedFields := []string{"pass", "fieldpass", "drop", "fielddrop", unsupportedFields := []string{"tagexclude", "taginclude"}
"tagexclude", "taginclude"}
for _, field := range unsupportedFields { for _, field := range unsupportedFields {
if _, ok := tbl.Fields[field]; ok { if _, ok := tbl.Fields[field]; ok {
// TODO raise error because field is not supported // TODO raise error because field is not supported

View File

@ -26,8 +26,8 @@ type MinMax struct {
} }
type minmax struct { type minmax struct {
min interface{} min float64
max interface{} max float64
} }
var sampleConfig = ` var sampleConfig = `
@ -55,28 +55,40 @@ func (m *MinMax) apply(in telegraf.Metric) {
m.tagCache[id] = in.Tags() m.tagCache[id] = in.Tags()
m.fieldCache[id] = make(map[string]minmax) m.fieldCache[id] = make(map[string]minmax)
for k, v := range in.Fields() { for k, v := range in.Fields() {
if fv, ok := convert(v); ok {
m.fieldCache[id][k] = minmax{ m.fieldCache[id][k] = minmax{
min: v, min: fv,
max: v, max: fv,
}
} }
} }
} else { } else {
for k, v := range in.Fields() { for k, v := range in.Fields() {
cmpmin := compare(m.fieldCache[id][k].min, v) if fv, ok := convert(v); ok {
cmpmax := compare(m.fieldCache[id][k].max, v) if _, ok := m.fieldCache[id][k]; !ok {
// hit an uncached field of a cached metric
m.fieldCache[id][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 { if cmpmin == 1 {
tmp := m.fieldCache[id][k] tmp := m.fieldCache[id][k]
tmp.min = v tmp.min = fv
m.fieldCache[id][k] = tmp m.fieldCache[id][k] = tmp
} }
if cmpmax == -1 { if cmpmax == -1 {
tmp := m.fieldCache[id][k] tmp := m.fieldCache[id][k]
tmp.max = v tmp.max = fv
m.fieldCache[id][k] = tmp m.fieldCache[id][k] = tmp
} }
} }
} }
} }
}
func (m *MinMax) Start(acc telegraf.Accumulator) error { func (m *MinMax) Start(acc telegraf.Accumulator) error {
m.metrics = make(chan telegraf.Metric, 10) m.metrics = make(chan telegraf.Metric, 10)
@ -156,32 +168,23 @@ func (m *MinMax) continuousHandler() {
} }
} }
func compare(a, b interface{}) int { func compare(a, b float64) int {
switch at := a.(type) { if a < b {
case int64:
if bt, ok := b.(int64); ok {
if at < bt {
return -1 return -1
} else if at > bt { } else if a > b {
return 1 return 1
} }
return 0 return 0
} else {
return 0
} }
func convert(in interface{}) (float64, bool) {
switch v := in.(type) {
case float64: case float64:
if bt, ok := b.(float64); ok { return v, true
if at < bt { case int64:
return -1 return float64(v), true
} else if at > bt {
return 1
}
return 0
} else {
return 0
}
default: default:
return 0 return 0, false
} }
} }

View File

@ -5,13 +5,13 @@ import (
"time" "time"
"github.com/influxdata/telegraf" "github.com/influxdata/telegraf"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/assert"
) )
func BenchmarkApply(b *testing.B) { var m1, _ = telegraf.NewMetric("m1",
minmax := MinMax{}
minmax.clearCache()
m1, _ := telegraf.NewMetric("m1",
map[string]string{"foo": "bar"}, map[string]string{"foo": "bar"},
map[string]interface{}{ map[string]interface{}{
"a": int64(1), "a": int64(1),
@ -27,10 +27,10 @@ func BenchmarkApply(b *testing.B) {
}, },
time.Now(), time.Now(),
) )
m2, _ := telegraf.NewMetric("m1", var m2, _ = telegraf.NewMetric("m1",
map[string]string{"foo": "bar"}, map[string]string{"foo": "bar"},
map[string]interface{}{ map[string]interface{}{
"a": int64(3), "a": int64(1),
"b": int64(3), "b": int64(3),
"c": int64(3), "c": int64(3),
"d": int64(3), "d": int64(3),
@ -40,12 +40,230 @@ func BenchmarkApply(b *testing.B) {
"h": float64(1), "h": float64(1),
"i": float64(1), "i": float64(1),
"j": float64(1), "j": float64(1),
"k": float64(200),
"ignoreme": "string",
"andme": true,
}, },
time.Now(), time.Now(),
) )
func BenchmarkApply(b *testing.B) {
minmax := MinMax{}
minmax.clearCache()
for n := 0; n < b.N; n++ { for n := 0; n < b.N; n++ {
minmax.apply(m1) minmax.apply(m1)
minmax.apply(m2) minmax.apply(m2)
} }
} }
// Test two metrics getting added, when running with a period, and the metrics
// are added in the same period.
func TestMinMaxWithPeriod(t *testing.T) {
acc := testutil.Accumulator{}
minmax := MinMax{
Period: internal.Duration{Duration: time.Millisecond * 500},
}
assert.NoError(t, minmax.Start(&acc))
defer minmax.Stop()
minmax.Apply(m1)
minmax.Apply(m2)
for {
if acc.NMetrics() > 0 {
break
}
time.Sleep(time.Millisecond)
}
expectedFields := map[string]interface{}{
"a_max": float64(1),
"a_min": float64(1),
"b_max": float64(3),
"b_min": float64(1),
"c_max": float64(3),
"c_min": float64(1),
"d_max": float64(3),
"d_min": float64(1),
"e_max": float64(3),
"e_min": float64(1),
"f_max": float64(2),
"f_min": float64(1),
"g_max": float64(2),
"g_min": float64(1),
"h_max": float64(2),
"h_min": float64(1),
"i_max": float64(2),
"i_min": float64(1),
"j_max": float64(3),
"j_min": float64(1),
"k_max": float64(200),
"k_min": float64(200),
}
expectedTags := map[string]string{
"foo": "bar",
}
acc.AssertContainsTaggedFields(t, "m1", expectedFields, expectedTags)
}
// Test two metrics getting added, when running with a period, and the metrics
// are added in two different periods.
func TestMinMaxDifferentPeriods(t *testing.T) {
acc := testutil.Accumulator{}
minmax := MinMax{
Period: internal.Duration{Duration: time.Millisecond * 100},
}
assert.NoError(t, minmax.Start(&acc))
defer minmax.Stop()
minmax.Apply(m1)
for {
if acc.NMetrics() > 0 {
break
}
time.Sleep(time.Millisecond)
}
expectedFields := map[string]interface{}{
"a_max": float64(1),
"a_min": float64(1),
"b_max": float64(1),
"b_min": float64(1),
"c_max": float64(1),
"c_min": float64(1),
"d_max": float64(1),
"d_min": float64(1),
"e_max": float64(1),
"e_min": float64(1),
"f_max": float64(2),
"f_min": float64(2),
"g_max": float64(2),
"g_min": float64(2),
"h_max": float64(2),
"h_min": float64(2),
"i_max": float64(2),
"i_min": float64(2),
"j_max": float64(3),
"j_min": float64(3),
}
expectedTags := map[string]string{
"foo": "bar",
}
acc.AssertContainsTaggedFields(t, "m1", expectedFields, expectedTags)
acc.ClearMetrics()
minmax.Apply(m2)
for {
if acc.NMetrics() > 0 {
break
}
time.Sleep(time.Millisecond)
}
expectedFields = map[string]interface{}{
"a_max": float64(1),
"a_min": float64(1),
"b_max": float64(3),
"b_min": float64(3),
"c_max": float64(3),
"c_min": float64(3),
"d_max": float64(3),
"d_min": float64(3),
"e_max": float64(3),
"e_min": float64(3),
"f_max": float64(1),
"f_min": float64(1),
"g_max": float64(1),
"g_min": float64(1),
"h_max": float64(1),
"h_min": float64(1),
"i_max": float64(1),
"i_min": float64(1),
"j_max": float64(1),
"j_min": float64(1),
"k_max": float64(200),
"k_min": float64(200),
}
expectedTags = map[string]string{
"foo": "bar",
}
acc.AssertContainsTaggedFields(t, "m1", expectedFields, expectedTags)
}
// Test two metrics getting added, when running without a period.
func TestMinMaxWithoutPeriod(t *testing.T) {
acc := testutil.Accumulator{}
minmax := MinMax{}
assert.NoError(t, minmax.Start(&acc))
defer minmax.Stop()
minmax.Apply(m1)
for {
if acc.NMetrics() > 0 {
break
}
time.Sleep(time.Millisecond)
}
expectedFields := map[string]interface{}{
"a_max": float64(1),
"a_min": float64(1),
"b_max": float64(1),
"b_min": float64(1),
"c_max": float64(1),
"c_min": float64(1),
"d_max": float64(1),
"d_min": float64(1),
"e_max": float64(1),
"e_min": float64(1),
"f_max": float64(2),
"f_min": float64(2),
"g_max": float64(2),
"g_min": float64(2),
"h_max": float64(2),
"h_min": float64(2),
"i_max": float64(2),
"i_min": float64(2),
"j_max": float64(3),
"j_min": float64(3),
}
expectedTags := map[string]string{
"foo": "bar",
}
acc.AssertContainsTaggedFields(t, "m1", expectedFields, expectedTags)
acc.ClearMetrics()
minmax.Apply(m2)
for {
if acc.NMetrics() > 0 {
break
}
time.Sleep(time.Millisecond)
}
expectedFields = map[string]interface{}{
"a_max": float64(1),
"a_min": float64(1),
"b_max": float64(3),
"b_min": float64(1),
"c_max": float64(3),
"c_min": float64(1),
"d_max": float64(3),
"d_min": float64(1),
"e_max": float64(3),
"e_min": float64(1),
"f_max": float64(2),
"f_min": float64(1),
"g_max": float64(2),
"g_min": float64(1),
"h_max": float64(2),
"h_min": float64(1),
"i_max": float64(2),
"i_min": float64(1),
"j_max": float64(3),
"j_min": float64(1),
"k_max": float64(200),
"k_min": float64(200),
}
expectedTags = map[string]string{
"foo": "bar",
}
acc.AssertContainsTaggedFields(t, "m1", expectedFields, expectedTags)
}

View File

@ -0,0 +1 @@
package printer

View File

@ -39,6 +39,13 @@ func (a *Accumulator) NMetrics() uint64 {
return atomic.LoadUint64(&a.nMetrics) return atomic.LoadUint64(&a.nMetrics)
} }
func (a *Accumulator) ClearMetrics() {
atomic.StoreUint64(&a.nMetrics, 0)
a.Lock()
defer a.Unlock()
a.Metrics = make([]*Metric, 0)
}
// AddFields adds a measurement point with a specified timestamp. // AddFields adds a measurement point with a specified timestamp.
func (a *Accumulator) AddFields( func (a *Accumulator) AddFields(
measurement string, measurement string,