From d4bed35e7ead8210c07d058b4210cd1c5d9c8ceb Mon Sep 17 00:00:00 2001 From: Cameron Sparr Date: Mon, 7 Mar 2016 15:46:23 +0100 Subject: [PATCH] Prevent Inf and NaN from being added, and unit test Accumulator closes #803 --- agent/accumulator.go | 7 +- agent/accumulator_test.go | 302 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 308 insertions(+), 1 deletion(-) create mode 100644 agent/accumulator_test.go diff --git a/agent/accumulator.go b/agent/accumulator.go index b04ff2b53..7ec22cd7f 100644 --- a/agent/accumulator.go +++ b/agent/accumulator.go @@ -105,7 +105,6 @@ func (ac *accumulator) AddFields( continue } } - result[k] = v // Validate uint64 and float64 fields switch val := v.(type) { @@ -116,6 +115,7 @@ func (ac *accumulator) AddFields( } else { result[k] = int64(9223372036854775807) } + continue case float64: // NaNs are invalid values in influxdb, skip measurement if math.IsNaN(val) || math.IsInf(val, 0) { @@ -127,6 +127,8 @@ func (ac *accumulator) AddFields( continue } } + + result[k] = v } fields = nil if len(result) == 0 { @@ -168,5 +170,8 @@ func (ac *accumulator) setDefaultTags(tags map[string]string) { } func (ac *accumulator) addDefaultTag(key, value string) { + if ac.defaultTags == nil { + ac.defaultTags = make(map[string]string) + } ac.defaultTags[key] = value } diff --git a/agent/accumulator_test.go b/agent/accumulator_test.go new file mode 100644 index 000000000..05f9b02aa --- /dev/null +++ b/agent/accumulator_test.go @@ -0,0 +1,302 @@ +package agent + +import ( + "fmt" + "math" + "testing" + "time" + + "github.com/influxdata/telegraf" + "github.com/influxdata/telegraf/internal/models" + + "github.com/stretchr/testify/assert" +) + +func TestAdd(t *testing.T) { + a := accumulator{} + now := time.Now() + a.metrics = make(chan telegraf.Metric, 10) + defer close(a.metrics) + a.inputConfig = &internal_models.InputConfig{} + + a.Add("acctest", float64(101), map[string]string{}) + a.Add("acctest", float64(101), map[string]string{"acc": "test"}) + a.Add("acctest", float64(101), map[string]string{"acc": "test"}, now) + + testm := <-a.metrics + actual := testm.String() + assert.Contains(t, actual, "acctest value=101") + + testm = <-a.metrics + actual = testm.String() + assert.Contains(t, actual, "acctest,acc=test value=101") + + testm = <-a.metrics + actual = testm.String() + assert.Equal(t, + fmt.Sprintf("acctest,acc=test value=101 %d", now.UnixNano()), + actual) +} + +func TestAddDefaultTags(t *testing.T) { + a := accumulator{} + a.addDefaultTag("default", "tag") + now := time.Now() + a.metrics = make(chan telegraf.Metric, 10) + defer close(a.metrics) + a.inputConfig = &internal_models.InputConfig{} + + a.Add("acctest", float64(101), map[string]string{}) + a.Add("acctest", float64(101), map[string]string{"acc": "test"}) + a.Add("acctest", float64(101), map[string]string{"acc": "test"}, now) + + testm := <-a.metrics + actual := testm.String() + assert.Contains(t, actual, "acctest,default=tag value=101") + + testm = <-a.metrics + actual = testm.String() + assert.Contains(t, actual, "acctest,acc=test,default=tag value=101") + + testm = <-a.metrics + actual = testm.String() + assert.Equal(t, + fmt.Sprintf("acctest,acc=test,default=tag value=101 %d", now.UnixNano()), + actual) +} + +func TestAddFields(t *testing.T) { + a := accumulator{} + now := time.Now() + a.metrics = make(chan telegraf.Metric, 10) + defer close(a.metrics) + a.inputConfig = &internal_models.InputConfig{} + + fields := map[string]interface{}{ + "usage": float64(99), + } + a.AddFields("acctest", fields, map[string]string{}) + a.AddFields("acctest", fields, map[string]string{"acc": "test"}) + a.AddFields("acctest", fields, map[string]string{"acc": "test"}, now) + + testm := <-a.metrics + actual := testm.String() + assert.Contains(t, actual, "acctest usage=99") + + testm = <-a.metrics + actual = testm.String() + assert.Contains(t, actual, "acctest,acc=test usage=99") + + testm = <-a.metrics + actual = testm.String() + assert.Equal(t, + fmt.Sprintf("acctest,acc=test usage=99 %d", now.UnixNano()), + actual) +} + +// Test that all Inf fields get dropped, and not added to metrics channel +func TestAddInfFields(t *testing.T) { + inf := math.Inf(1) + ninf := math.Inf(-1) + + a := accumulator{} + now := time.Now() + a.metrics = make(chan telegraf.Metric, 10) + defer close(a.metrics) + a.inputConfig = &internal_models.InputConfig{} + + fields := map[string]interface{}{ + "usage": inf, + "nusage": ninf, + } + a.AddFields("acctest", fields, map[string]string{}) + a.AddFields("acctest", fields, map[string]string{"acc": "test"}) + a.AddFields("acctest", fields, map[string]string{"acc": "test"}, now) + + assert.Len(t, a.metrics, 0) + + // test that non-inf fields are kept and not dropped + fields["notinf"] = float64(100) + a.AddFields("acctest", fields, map[string]string{}) + testm := <-a.metrics + actual := testm.String() + assert.Contains(t, actual, "acctest notinf=100") +} + +// Test that nan fields are dropped and not added +func TestAddNaNFields(t *testing.T) { + nan := math.NaN() + + a := accumulator{} + now := time.Now() + a.metrics = make(chan telegraf.Metric, 10) + defer close(a.metrics) + a.inputConfig = &internal_models.InputConfig{} + + fields := map[string]interface{}{ + "usage": nan, + } + a.AddFields("acctest", fields, map[string]string{}) + a.AddFields("acctest", fields, map[string]string{"acc": "test"}) + a.AddFields("acctest", fields, map[string]string{"acc": "test"}, now) + + assert.Len(t, a.metrics, 0) + + // test that non-nan fields are kept and not dropped + fields["notnan"] = float64(100) + a.AddFields("acctest", fields, map[string]string{}) + testm := <-a.metrics + actual := testm.String() + assert.Contains(t, actual, "acctest notnan=100") +} + +func TestAddUint64Fields(t *testing.T) { + a := accumulator{} + now := time.Now() + a.metrics = make(chan telegraf.Metric, 10) + defer close(a.metrics) + a.inputConfig = &internal_models.InputConfig{} + + fields := map[string]interface{}{ + "usage": uint64(99), + } + a.AddFields("acctest", fields, map[string]string{}) + a.AddFields("acctest", fields, map[string]string{"acc": "test"}) + a.AddFields("acctest", fields, map[string]string{"acc": "test"}, now) + + testm := <-a.metrics + actual := testm.String() + assert.Contains(t, actual, "acctest usage=99i") + + testm = <-a.metrics + actual = testm.String() + assert.Contains(t, actual, "acctest,acc=test usage=99i") + + testm = <-a.metrics + actual = testm.String() + assert.Equal(t, + fmt.Sprintf("acctest,acc=test usage=99i %d", now.UnixNano()), + actual) +} + +func TestAddUint64Overflow(t *testing.T) { + a := accumulator{} + now := time.Now() + a.metrics = make(chan telegraf.Metric, 10) + defer close(a.metrics) + a.inputConfig = &internal_models.InputConfig{} + + fields := map[string]interface{}{ + "usage": uint64(9223372036854775808), + } + a.AddFields("acctest", fields, map[string]string{}) + a.AddFields("acctest", fields, map[string]string{"acc": "test"}) + a.AddFields("acctest", fields, map[string]string{"acc": "test"}, now) + + testm := <-a.metrics + actual := testm.String() + assert.Contains(t, actual, "acctest usage=9223372036854775807i") + + testm = <-a.metrics + actual = testm.String() + assert.Contains(t, actual, "acctest,acc=test usage=9223372036854775807i") + + testm = <-a.metrics + actual = testm.String() + assert.Equal(t, + fmt.Sprintf("acctest,acc=test usage=9223372036854775807i %d", now.UnixNano()), + actual) +} + +func TestAddInts(t *testing.T) { + a := accumulator{} + a.addDefaultTag("default", "tag") + now := time.Now() + a.metrics = make(chan telegraf.Metric, 10) + defer close(a.metrics) + a.inputConfig = &internal_models.InputConfig{} + + a.Add("acctest", int(101), map[string]string{}) + a.Add("acctest", int32(101), map[string]string{"acc": "test"}) + a.Add("acctest", int64(101), map[string]string{"acc": "test"}, now) + + testm := <-a.metrics + actual := testm.String() + assert.Contains(t, actual, "acctest,default=tag value=101i") + + testm = <-a.metrics + actual = testm.String() + assert.Contains(t, actual, "acctest,acc=test,default=tag value=101i") + + testm = <-a.metrics + actual = testm.String() + assert.Equal(t, + fmt.Sprintf("acctest,acc=test,default=tag value=101i %d", now.UnixNano()), + actual) +} + +func TestAddFloats(t *testing.T) { + a := accumulator{} + a.addDefaultTag("default", "tag") + now := time.Now() + a.metrics = make(chan telegraf.Metric, 10) + defer close(a.metrics) + a.inputConfig = &internal_models.InputConfig{} + + a.Add("acctest", float32(101), map[string]string{"acc": "test"}) + a.Add("acctest", float64(101), map[string]string{"acc": "test"}, now) + + testm := <-a.metrics + actual := testm.String() + assert.Contains(t, actual, "acctest,acc=test,default=tag value=101") + + testm = <-a.metrics + actual = testm.String() + assert.Equal(t, + fmt.Sprintf("acctest,acc=test,default=tag value=101 %d", now.UnixNano()), + actual) +} + +func TestAddStrings(t *testing.T) { + a := accumulator{} + a.addDefaultTag("default", "tag") + now := time.Now() + a.metrics = make(chan telegraf.Metric, 10) + defer close(a.metrics) + a.inputConfig = &internal_models.InputConfig{} + + a.Add("acctest", "test", map[string]string{"acc": "test"}) + a.Add("acctest", "foo", map[string]string{"acc": "test"}, now) + + testm := <-a.metrics + actual := testm.String() + assert.Contains(t, actual, "acctest,acc=test,default=tag value=\"test\"") + + testm = <-a.metrics + actual = testm.String() + assert.Equal(t, + fmt.Sprintf("acctest,acc=test,default=tag value=\"foo\" %d", now.UnixNano()), + actual) +} + +func TestAddBools(t *testing.T) { + a := accumulator{} + a.addDefaultTag("default", "tag") + now := time.Now() + a.metrics = make(chan telegraf.Metric, 10) + defer close(a.metrics) + a.inputConfig = &internal_models.InputConfig{} + + a.Add("acctest", true, map[string]string{"acc": "test"}) + a.Add("acctest", false, map[string]string{"acc": "test"}, now) + + testm := <-a.metrics + actual := testm.String() + assert.Contains(t, actual, "acctest,acc=test,default=tag value=true") + + testm = <-a.metrics + actual = testm.String() + assert.Equal(t, + fmt.Sprintf("acctest,acc=test,default=tag value=false %d", now.UnixNano()), + actual) +}