Drop unsupported NaN, Inf, and out of range values. (aws#847)

internal/util/type_conversion.go

@@ -3,43 +3,51 @@
 
 package util
 
-import "github.com/aws/amazon-cloudwatch-agent/metric/distribution"
+import (
+	"fmt"
+	"math"
 
-func ToOtelValue(value interface{}) interface{} {
+	"github.com/aws/amazon-cloudwatch-agent/metric/distribution"
+)
+
+func ToOtelValue(value interface{}) (interface{}, error) {
 	switch v := value.(type) {
 	case int:
-		return int64(v)
+		return int64(v), nil
 	case int8:
-		return int64(v)
+		return int64(v), nil
 	case int16:
-		return int64(v)
+		return int64(v), nil
 	case int32:
-		return int64(v)
+		return int64(v), nil
 	case int64:
-		return v
+		return v, nil
 	case uint:
-		return int64(v)
+		return int64(v), nil
 	case uint8:
-		return int64(v)
+		return int64(v), nil
 	case uint16:
-		return int64(v)
+		return int64(v), nil
 	case uint32:
-		return int64(v)
+		return int64(v), nil
 	case uint64:
-		return int64(v)
+		return int64(v), nil
 	case float32:
-		return float64(v)
+		return float64(v), nil
 	case float64:
-		return v
+		if math.IsNaN(v) || math.IsInf(v, 0) {
+			return nil, fmt.Errorf("unsupported value: %v", v)
+		}
+		return v, nil
 	case bool:
 		if v {
-			return int64(1)
+			return int64(1), nil
 		} else {
-			return int64(0)
+			return int64(0), nil
 		}
 	case distribution.Distribution:
-		return v
+		return v, nil
 	default:
-		return nil
+		return nil, fmt.Errorf("unsupported type: %T", v)
 	}
 }

internal/util/type_conversion_test.go

@@ -0,0 +1,55 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package util
+
+import (
+	"errors"
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/aws/amazon-cloudwatch-agent/metric/distribution/regular"
+)
+
+func TestToOtelValue(t *testing.T) {
+	distribution := regular.NewRegularDistribution()
+	testCases := []struct {
+		input   interface{}
+		want    interface{}
+		wantErr error
+	}{
+		// ints
+		{input: 5, want: int64(5)},
+		{input: int8(5), want: int64(5)},
+		{input: int16(5), want: int64(5)},
+		{input: int32(5), want: int64(5)},
+		{input: int64(5), want: int64(5)},
+		// uints
+		{input: uint(5), want: int64(5)},
+		{input: uint8(5), want: int64(5)},
+		{input: uint16(5), want: int64(5)},
+		{input: uint32(5), want: int64(5)},
+		{input: uint64(5), want: int64(5)},
+		// floats
+		{input: float32(5.5), want: 5.5},
+		{input: 5.5, want: 5.5},
+		// bool
+		{input: false, want: int64(0)},
+		{input: true, want: int64(1)},
+		// distribution
+		{input: distribution, want: distribution},
+		// unsupported floats
+		{input: math.NaN(), want: nil, wantErr: errors.New("unsupported value: NaN")},
+		{input: math.Inf(1), want: nil, wantErr: errors.New("unsupported value: +Inf")},
+		{input: math.Inf(-1), want: nil, wantErr: errors.New("unsupported value: -Inf")},
+		// unsupported types
+		{input: "test", want: nil, wantErr: errors.New("unsupported type: string")},
+	}
+	for _, testCase := range testCases {
+		got, err := ToOtelValue(testCase.input)
+		assert.Equal(t, testCase.wantErr, err)
+		assert.Equal(t, testCase.want, got)
+	}
+}

metric/distribution/distribution.go

@@ -3,7 +3,19 @@
 
 package distribution
 
-import "go.opentelemetry.io/collector/pdata/pmetric"
+import (
+	"errors"
+	"math"
+
+	"go.opentelemetry.io/collector/pdata/pmetric"
+)
+
+var (
+	ErrUnsupportedWeight = errors.New("weight must be larger than 0")
+	ErrUnsupportedValue  = errors.New("value cannot be negative, NaN, Inf, or greater than 2^360")
+	MinValue             = -math.Pow(2, 360)
+	MaxValue             = math.Pow(2, 360)
+)
 
 type Distribution interface {
 	Maximum() float64
@@ -36,3 +48,10 @@ type Distribution interface {
 }
 
 var NewDistribution func() Distribution
+
+// IsSupportedValue checks to see if the metric is between the min value and 2^360 and not a NaN.
+// This matches the accepted range described in the MetricDatum documentation
+// https://docs.aws.amazon.com/AmazonCloudWatch/latest/APIReference/API_MetricDatum.html
+func IsSupportedValue(value, min, max float64) bool {
+	return !math.IsNaN(value) && value >= min && value <= max
+}

metric/distribution/distribution_test.go

@@ -0,0 +1,30 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package distribution
+
+import (
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsAcceptedValue(t *testing.T) {
+	testCases := []struct {
+		input float64
+		want  bool
+	}{
+		{input: MinValue * 1.0001, want: false},
+		{input: MinValue, want: true},
+		{input: MaxValue, want: true},
+		{input: MaxValue * 1.0001, want: false},
+		{input: math.Pow(2, 300), want: true},
+		{input: math.NaN(), want: false},
+		{input: math.Inf(1), want: false},
+		{input: math.Inf(-1), want: false},
+	}
+	for _, testCase := range testCases {
+		assert.Equal(t, testCase.want, IsSupportedValue(testCase.input, MinValue, MaxValue))
+	}
+}

metric/distribution/regular/regular_distribution.go

@@ -4,7 +4,7 @@
 package regular
 
 import (
-	"errors"
+	"fmt"
 	"log"
 	"math"
 
@@ -69,34 +69,33 @@ func (regularDist *RegularDistribution) Size() int {
 
 // weight is 1/samplingRate
 func (regularDist *RegularDistribution) AddEntryWithUnit(value float64, weight float64, unit string) error {
-	if weight > 0 {
-		if value < 0 {
-			return errors.New("negative value")
-		}
-		//sample count
-		regularDist.sampleCount += weight
-		//sum
-		regularDist.sum += value * weight
-		//min
-		if value < regularDist.minimum {
-			regularDist.minimum = value
-		}
-		//max
-		if value > regularDist.maximum {
-			regularDist.maximum = value
-		}
+	if weight <= 0 {
+		return fmt.Errorf("unsupported weight %v: %w", weight, distribution.ErrUnsupportedWeight)
+	}
+	if !distribution.IsSupportedValue(value, 0, distribution.MaxValue) {
+		return fmt.Errorf("unsupported value %v: %w", value, distribution.ErrUnsupportedValue)
+	}
+	//sample count
+	regularDist.sampleCount += weight
+	//sum
+	regularDist.sum += value * weight
+	//min
+	if value < regularDist.minimum {
+		regularDist.minimum = value
+	}
+	//max
+	if value > regularDist.maximum {
+		regularDist.maximum = value
+	}
 
-		//values and counts
-		regularDist.buckets[value] += weight
+	//values and counts
+	regularDist.buckets[value] += weight
 
-		//unit
-		if regularDist.unit == "" {
-			regularDist.unit = unit
-		} else if regularDist.unit != unit && unit != "" {
-			log.Printf("D! Multiple units are detected: %s, %s", regularDist.unit, unit)
-		}
-	} else {
-		log.Printf("D! Weight should be larger than 0: %v", weight)
+	//unit
+	if regularDist.unit == "" {
+		regularDist.unit = unit
+	} else if regularDist.unit != unit && unit != "" {
+		log.Printf("D! Multiple units are detected: %s, %s", regularDist.unit, unit)
 	}
 	return nil
 }

metric/distribution/regular/regular_distribution_test.go

@@ -4,12 +4,15 @@
 package regular
 
 import (
+	"math"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+
+	"github.com/aws/amazon-cloudwatch-agent/metric/distribution"
 )
 
-func TestSEH1Distribution(t *testing.T) {
+func TestRegularDistribution(t *testing.T) {
 	//dist new and add entry
 	dist := NewRegularDistribution()
 
@@ -34,9 +37,9 @@ func TestSEH1Distribution(t *testing.T) {
 	//another dist new and add entry
 	anotherDist := NewRegularDistribution()
 
-	anotherDist.AddEntry(21, 1)
-	anotherDist.AddEntry(22, 1)
-	anotherDist.AddEntry(23, 2)
+	assert.NoError(t, anotherDist.AddEntry(21, 1))
+	assert.NoError(t, anotherDist.AddEntry(22, 1))
+	assert.NoError(t, anotherDist.AddEntry(23, 2))
 
 	assert.Equal(t, 89.0, anotherDist.Sum())
 	assert.Equal(t, 4.0, anotherDist.SampleCount())
@@ -75,6 +78,14 @@ func TestSEH1Distribution(t *testing.T) {
 	//add distClone into another dist
 	anotherDist.AddDistribution(distClone)
 	assert.Equal(t, dist, anotherDist) //the direction of AddDistribution should not matter.
+
+	assert.ErrorIs(t, anotherDist.AddEntry(1, 0), distribution.ErrUnsupportedWeight)
+	assert.ErrorIs(t, anotherDist.AddEntry(-1, 1), distribution.ErrUnsupportedValue)
+	assert.ErrorIs(t, anotherDist.AddEntry(math.NaN(), 1), distribution.ErrUnsupportedValue)
+	assert.ErrorIs(t, anotherDist.AddEntry(math.Inf(1), 1), distribution.ErrUnsupportedValue)
+	assert.ErrorIs(t, anotherDist.AddEntry(math.Inf(-1), 1), distribution.ErrUnsupportedValue)
+	assert.ErrorIs(t, anotherDist.AddEntry(distribution.MaxValue*1.001, 1), distribution.ErrUnsupportedValue)
+	assert.ErrorIs(t, anotherDist.AddEntry(distribution.MinValue*1.001, 1), distribution.ErrUnsupportedValue)
 }
 
 func cloneRegularDistribution(dist *RegularDistribution) *RegularDistribution {

metric/distribution/seh1/seh1_distribution.go

@@ -4,7 +4,7 @@
 package seh1
 
 import (
-	"errors"
+	"fmt"
 	"log"
 	"math"
 
@@ -79,35 +79,34 @@ func (seh1Distribution *SEH1Distribution) Size() int {
 
 // weight is 1/samplingRate
 func (seh1Distribution *SEH1Distribution) AddEntryWithUnit(value float64, weight float64, unit string) error {
-	if weight > 0 {
-		if value < 0 {
-			return errors.New("negative value")
-		}
-		//sample count
-		seh1Distribution.sampleCount += weight
-		//sum
-		seh1Distribution.sum += value * weight
-		//min
-		if value < seh1Distribution.minimum {
-			seh1Distribution.minimum = value
-		}
-		//max
-		if value > seh1Distribution.maximum {
-			seh1Distribution.maximum = value
-		}
+	if weight <= 0 {
+		return fmt.Errorf("unsupported weight %v: %w", weight, distribution.ErrUnsupportedWeight)
+	}
+	if !distribution.IsSupportedValue(value, 0, distribution.MaxValue) {
+		return fmt.Errorf("unsupported value %v: %w", value, distribution.ErrUnsupportedValue)
+	}
+	//sample count
+	seh1Distribution.sampleCount += weight
+	//sum
+	seh1Distribution.sum += value * weight
+	//min
+	if value < seh1Distribution.minimum {
+		seh1Distribution.minimum = value
+	}
+	//max
+	if value > seh1Distribution.maximum {
+		seh1Distribution.maximum = value
+	}
 
-		//seh
-		bucketNumber := bucketNumber(value)
-		seh1Distribution.buckets[bucketNumber] += weight
+	//seh
+	bucketNumber := bucketNumber(value)
+	seh1Distribution.buckets[bucketNumber] += weight
 
-		//unit
-		if seh1Distribution.unit == "" {
-			seh1Distribution.unit = unit
-		} else if seh1Distribution.unit != unit && unit != "" {
-			log.Printf("D! Multiple units are detected: %s, %s", seh1Distribution.unit, unit)
-		}
-	} else {
-		log.Printf("D! Weight should be larger than 0: %v", weight)
+	//unit
+	if seh1Distribution.unit == "" {
+		seh1Distribution.unit = unit
+	} else if seh1Distribution.unit != unit && unit != "" {
+		log.Printf("D! Multiple units are detected: %s, %s", seh1Distribution.unit, unit)
 	}
 	return nil
 }
@@ -151,7 +150,7 @@ func (seh1Distribution *SEH1Distribution) AddDistributionWithWeight(distribution
 		if seh1Distribution.unit == "" {
 			seh1Distribution.unit = distribution.Unit()
 		} else if seh1Distribution.unit != distribution.Unit() && distribution.Unit() != "" {
-			log.Printf("D! Multiple units are dected: %s, %s", seh1Distribution.unit, distribution.Unit())
+			log.Printf("D! Multiple units are detected: %s, %s", seh1Distribution.unit, distribution.Unit())
 		}
 	} else {
 		log.Printf("D! SampleCount * Weight should be larger than 0: %v, %v", distribution.SampleCount(), weight)