Support filters and pdata being emitted from monitors

internal/signalfx-agent/go.mod

@@ -88,6 +88,7 @@ require (
 	github.com/influxdata/telegraf v1.30.1
 	github.com/microsoft/go-mssqldb v1.7.2
 	github.com/smartystreets/goconvey v1.8.1
+	go.opentelemetry.io/collector/pdata v1.15.0
 )
 
 require (

internal/signalfx-agent/go.sum

@@ -574,6 +574,8 @@ github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN
 github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0=
 github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA=
 github.com/zenazn/goji v0.9.0/go.mod h1:7S9M489iMyHBNxwZnk9/EHS098H4/F6TATF2mIxtB1Q=
+go.opentelemetry.io/collector/pdata v1.15.0 h1:q/T1sFpRKJnjDrUsHdJ6mq4uSqViR/f92yvGwDby/gY=
+go.opentelemetry.io/collector/pdata v1.15.0/go.mod h1:2wcsTIiLAJSbqBq/XUUYbi+cP+N87d0jEJzmb9nT19U=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.55.0 h1:ZIg3ZT/aQ7AfKqdwp7ECpOK6vHqquXXuyTjIO8ZdmPs=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.55.0/go.mod h1:DQAwmETtZV00skUwgD6+0U89g80NKsJE3DCKeLLPQMI=
 go.opentelemetry.io/otel v1.30.0 h1:F2t8sK4qf1fAmY9ua4ohFS/K+FUuOPemHUIXHtktrts=

internal/signalfx-agent/pkg/core/dpfilters/filter.go

@@ -5,11 +5,14 @@ package dpfilters
 
 import (
 	"github.com/signalfx/golib/v3/datapoint"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 )
 
 // DatapointFilter can be used to filter out datapoints
 type DatapointFilter interface {
 	// Matches takes a datapoint and returns whether it is matched by the
 	// filter
 	Matches(dp *datapoint.Datapoint) bool
+
+	MatchesMetricDataPoint(metricName string, dimensions pcommon.Map) bool
 }

internal/signalfx-agent/pkg/core/dpfilters/negated.go

@@ -2,6 +2,7 @@ package dpfilters
 
 import (
 	"github.com/signalfx/golib/v3/datapoint"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 )
 
 // NegatedDatapointFilter is a datapoint filter whose Matches method is made
@@ -16,6 +17,12 @@ func (n *NegatedDatapointFilter) Matches(dp *datapoint.Datapoint) bool {
 	return !n.DatapointFilter.Matches(dp)
 }
 
+// MatchesMetricDataPoint returns the opposite of what the original filter would have
+// returned.
+func (n *NegatedDatapointFilter) MatchesMetricDataPoint(metricName string, dimensions pcommon.Map) bool {
+	return !n.DatapointFilter.MatchesMetricDataPoint(metricName, dimensions)
+}
+
 // Negate returns the supplied filter negated such Matches returns the
 // opposite.
 func Negate(f DatapointFilter) DatapointFilter {

internal/signalfx-agent/pkg/core/dpfilters/overridable.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/signalfx/golib/v3/datapoint"
 	"github.com/signalfx/signalfx-agent/pkg/utils/filter"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 )
 
 type overridableDatapointFilter struct {
@@ -47,3 +48,13 @@ func (f *overridableDatapointFilter) Matches(dp *datapoint.Datapoint) bool {
 	return (f.metricFilter == nil || f.metricFilter.Matches(dp.Metric)) &&
 		(f.dimFilter == nil || f.dimFilter.Matches(dp.Dimensions))
 }
+
+func (f *overridableDatapointFilter) MatchesMetricDataPoint(metricName string, dimensions pcommon.Map) bool {
+	if f.metricFilter != nil && !f.metricFilter.Matches(metricName) {
+		return false
+	}
+	if f.dimFilter != nil {
+		return f.dimFilter.MatchesMap(dimensions)
+	}
+	return true
+}

internal/signalfx-agent/pkg/core/dpfilters/overridable_test.go

@@ -3,210 +3,198 @@ package dpfilters
 import (
 	"testing"
 
-	"github.com/signalfx/golib/v3/datapoint"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+
 	"github.com/stretchr/testify/assert"
 )
 
+var cpu pmetric.Metric
+var memory pmetric.Metric
+var disk pmetric.Metric
+
+func init() {
+	cpu = pmetric.NewMetric()
+	cpu.SetName("cpu.utilization")
+	cpu.SetEmptyGauge()
+	memory = pmetric.NewMetric()
+	memory.SetName("memory.utilization")
+	memory.SetEmptyGauge()
+	disk = pmetric.NewMetric()
+	disk.SetName("disk.utilization")
+	disk.SetEmptyGauge()
+}
 func TestOverridableFilters(t *testing.T) {
 	t.Run("Exclude based on simple metric name", func(t *testing.T) {
 		f, _ := NewOverridable([]string{"cpu.utilization"}, nil)
-		assert.True(t, f.Matches(&datapoint.Datapoint{Metric: "cpu.utilization"}))
-		assert.False(t, f.Matches(&datapoint.Datapoint{Metric: "memory.utilization"}))
+		assert.True(t, f.MatchesMetricDataPoint(cpu.Name(), pcommon.NewMap()))
+		assert.False(t, f.MatchesMetricDataPoint(memory.Name(), pcommon.NewMap()))
 	})
 
 	t.Run("Excludes based on multiple metric names", func(t *testing.T) {
 		f, _ := NewOverridable([]string{"cpu.utilization", "memory.utilization"}, nil)
-		assert.True(t, f.Matches(&datapoint.Datapoint{Metric: "cpu.utilization"}))
-
-		assert.True(t, f.Matches(&datapoint.Datapoint{Metric: "memory.utilization"}))
 
-		assert.False(t, f.Matches(&datapoint.Datapoint{Metric: "disk.utilization"}))
+		assert.True(t, f.MatchesMetricDataPoint(cpu.Name(), pcommon.NewMap()))
+		assert.True(t, f.MatchesMetricDataPoint(memory.Name(), pcommon.NewMap()))
+		assert.False(t, f.MatchesMetricDataPoint(disk.Name(), pcommon.NewMap()))
 	})
 
 	t.Run("Excludes based on regex metric name", func(t *testing.T) {
 		f, _ := NewOverridable([]string{`/cpu\..*/`}, nil)
-		assert.True(t, f.Matches(&datapoint.Datapoint{Metric: "cpu.utilization"}))
+		assert.True(t, f.MatchesMetricDataPoint(cpu.Name(), pcommon.NewMap()))
 
-		assert.False(t, f.Matches(&datapoint.Datapoint{Metric: "disk.utilization"}))
+		assert.False(t, f.MatchesMetricDataPoint(disk.Name(), pcommon.NewMap()))
 	})
 
 	t.Run("Excludes based on glob metric name", func(t *testing.T) {
 		f, _ := NewOverridable([]string{`cpu.util*`, "memor*"}, nil)
-		assert.True(t, f.Matches(&datapoint.Datapoint{Metric: "cpu.utilization"}))
-		assert.True(t, f.Matches(&datapoint.Datapoint{Metric: "memory.utilization"}))
+		assert.True(t, f.MatchesMetricDataPoint(cpu.Name(), pcommon.NewMap()))
+		assert.True(t, f.MatchesMetricDataPoint(memory.Name(), pcommon.NewMap()))
 
-		assert.False(t, f.Matches(&datapoint.Datapoint{Metric: "disk.utilization"}))
+		assert.False(t, f.MatchesMetricDataPoint(disk.Name(), pcommon.NewMap()))
 	})
 
 	t.Run("Excludes based on dimension name", func(t *testing.T) {
 		f, _ := NewOverridable(nil, map[string][]string{
 			"container_name": {"PO"},
 		})
 
-		assert.True(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"container_name": "PO",
-			},
-		}))
-
-		assert.False(t, f.Matches(&datapoint.Datapoint{
-			Metric: "disk.utilization",
-			Dimensions: map[string]string{
-				"container_name": "test",
-			},
-		}))
+		m := pmetric.NewMetric()
+		m.SetName("cpu.utilization")
+		m.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "PO")
+		assert.True(t, f.MatchesMetricDataPoint(m.Name(), m.Gauge().DataPoints().At(0).Attributes()))
+		m2 := pmetric.NewMetric()
+		m2.SetName("disk.utilization")
+		m2.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "test")
+		assert.False(t, f.MatchesMetricDataPoint(m2.Name(), m2.Gauge().DataPoints().At(0).Attributes()))
 	})
 
 	t.Run("Excludes based on dimension name regex", func(t *testing.T) {
 		f, _ := NewOverridable(nil, map[string][]string{
 			"container_name": {`/^[A-Z][A-Z]$/`},
 		})
 
-		assert.True(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"container_name": "PO",
-			},
-		}))
-
-		assert.False(t, f.Matches(&datapoint.Datapoint{
-			Metric: "disk.utilization",
-			Dimensions: map[string]string{
-				"container_name": "test",
-			},
-		}))
+		m := pmetric.NewMetric()
+		m.SetName("cpu.utilization")
+		m.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "PO")
+		assert.True(t, f.MatchesMetricDataPoint(m.Name(), m.Gauge().DataPoints().At(0).Attributes()))
+		m2 := pmetric.NewMetric()
+		m2.SetName("disk.utilization")
+		m2.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "test")
+		assert.False(t, f.MatchesMetricDataPoint(m2.Name(), m2.Gauge().DataPoints().At(0).Attributes()))
 	})
 
 	t.Run("Excludes based on dimension presence", func(t *testing.T) {
 		f, _ := NewOverridable(nil, map[string][]string{
 			"container_name": {`/.+/`},
 		})
 
-		assert.True(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"container_name": "test",
-			},
-		}))
-
-		assert.False(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"host": "localhost",
-			},
-		}))
+		m := pmetric.NewMetric()
+		m.SetName("cpu.utilization")
+		m.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "test")
+		assert.True(t, f.MatchesMetricDataPoint(m.Name(), m.Gauge().DataPoints().At(0).Attributes()))
+		m2 := pmetric.NewMetric()
+		m2.SetName("cpu.utilization")
+		m2.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("host", "localhost")
+		assert.False(t, f.MatchesMetricDataPoint(m2.Name(), m2.Gauge().DataPoints().At(0).Attributes()))
 	})
 
 	t.Run("Excludes based on dimension name glob", func(t *testing.T) {
 		f, _ := NewOverridable(nil, map[string][]string{
 			"container_name": {`*O*`},
 		})
+		m := pmetric.NewMetric()
+		m.SetName("cpu.utilization")
+		m.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "POD")
+
+		assert.True(t, f.MatchesMetricDataPoint(m.Name(), m.Gauge().DataPoints().At(0).Attributes()))
+
+		m2 := pmetric.NewMetric()
+		m2.SetName("cpu.utilization")
+		m2.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "POD123")
+		assert.True(t, f.MatchesMetricDataPoint(m2.Name(), m2.Gauge().DataPoints().At(0).Attributes()))
+
+		m3 := pmetric.NewMetric()
+		m3.SetName("disk.utilization")
+		m3.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "test")
 
-		assert.True(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"container_name": "POD",
-			},
-		}))
-
-		assert.True(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"container_name": "POD123",
-			},
-		}))
-
-		assert.False(t, f.Matches(&datapoint.Datapoint{
-			Metric: "disk.utilization",
-			Dimensions: map[string]string{
-				"container_name": "test",
-			},
-		}))
+		assert.False(t, f.MatchesMetricDataPoint(m3.Name(), m3.Gauge().DataPoints().At(0).Attributes()))
 	})
 
 	t.Run("Excludes based on conjunction of both dimensions and metric name", func(t *testing.T) {
 		f, _ := NewOverridable([]string{"*.utilization"}, map[string][]string{
 			"container_name": {"test"},
 		})
 
-		assert.False(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"container_name": "not matching",
-			},
-		}))
-
-		assert.True(t, f.Matches(&datapoint.Datapoint{
-			Metric: "disk.utilization",
-			Dimensions: map[string]string{
-				"container_name": "test",
-			},
-		}))
-
-		assert.False(t, f.Matches(&datapoint.Datapoint{
-			Metric: "disk.usage",
-			Dimensions: map[string]string{
-				"container_name": "test",
-			},
-		}))
+		m := pmetric.NewMetric()
+		m.SetName("cpu.utilization")
+		m.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "not_matching")
+
+		assert.False(t, f.MatchesMetricDataPoint(m.Name(), m.Gauge().DataPoints().At(0).Attributes()))
+
+		m2 := pmetric.NewMetric()
+		m2.SetName("disk.utilization")
+		m2.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "test")
+
+		assert.True(t, f.MatchesMetricDataPoint(m2.Name(), m2.Gauge().DataPoints().At(0).Attributes()))
+
+		m3 := pmetric.NewMetric()
+		m3.SetName("disk.usage")
+		m3.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "test")
+
+		assert.False(t, f.MatchesMetricDataPoint(m3.Name(), m3.Gauge().DataPoints().At(0).Attributes()))
 	})
 
 	t.Run("Doesn't match if no dimension filter specified", func(t *testing.T) {
 		f, _ := NewOverridable([]string{"cpu.utilization"}, nil)
-		assert.False(t, f.Matches(&datapoint.Datapoint{
-			Metric: "disk.utilization",
-			Dimensions: map[string]string{
-				"container_name": "test",
-			},
-		}))
+		m := pmetric.NewMetric()
+		m.SetName("disk.utilization")
+		m.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "test")
+		assert.False(t, f.MatchesMetricDataPoint(m.Name(), m.Gauge().DataPoints().At(0).Attributes()))
 	})
 
 	t.Run("Doesn't match if no metric name filter specified", func(t *testing.T) {
 		f, _ := NewOverridable(nil, map[string][]string{
 			"container_name": {"mycontainer"},
 		})
-		assert.False(t, f.Matches(&datapoint.Datapoint{Metric: "cpu.utilization"}))
+		assert.False(t, f.MatchesMetricDataPoint(cpu.Name(), pcommon.NewMap()))
 	})
 
 	t.Run("Matches against all dimension pairs", func(t *testing.T) {
 		f, _ := NewOverridable(nil, map[string][]string{
 			"host":   {"localhost"},
 			"system": {"r4"},
 		})
-		assert.False(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"host": "localhost",
-			}}))
-		assert.True(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"host":   "localhost",
-				"system": "r4",
-			}}))
+		m := pmetric.NewMetric()
+		m.SetName("cpu.utilization")
+		m.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("host", "localhost")
+		assert.False(t, f.MatchesMetricDataPoint(m.Name(), m.Gauge().DataPoints().At(0).Attributes()))
+		m2 := pmetric.NewMetric()
+		m2.SetName("cpu.utilization")
+		attrs := m2.SetEmptyGauge().DataPoints().AppendEmpty().Attributes()
+		attrs.PutStr("host", "localhost")
+		attrs.PutStr("system", "r4")
+
+		assert.True(t, f.MatchesMetricDataPoint(m2.Name(), m2.Gauge().DataPoints().At(0).Attributes()))
 	})
 
 	t.Run("Negated dim values take precedent", func(t *testing.T) {
 		f, _ := NewOverridable(nil, map[string][]string{
 			"container_name": {"*", "!pause", "!/.*idle/"},
 		})
 		// Shouldn't match when dimension isn't even present
-		assert.False(t, f.Matches(&datapoint.Datapoint{Metric: "cpu.utilization"}))
-		assert.False(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"container_name": "pause",
-			}}))
-		assert.False(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"container_name": "is_idle",
-			}}))
-		assert.True(t, f.Matches(&datapoint.Datapoint{
-			Metric: "cpu.utilization",
-			Dimensions: map[string]string{
-				"container_name": "mycontainer",
-			}}))
+		assert.False(t, f.MatchesMetricDataPoint(cpu.Name(), pcommon.NewMap()))
+		m := pmetric.NewMetric()
+		m.SetName("cpu.utilization")
+		m.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "pause")
+		assert.False(t, f.MatchesMetricDataPoint(m.Name(), m.Gauge().DataPoints().At(0).Attributes()))
+		m2 := pmetric.NewMetric()
+		m2.SetName("cpu.utilization")
+		m2.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "is_idle")
+		assert.False(t, f.MatchesMetricDataPoint(m2.Name(), m2.Gauge().DataPoints().At(0).Attributes()))
+		m3 := pmetric.NewMetric()
+		m3.SetName("cpu.utilization")
+		m3.SetEmptyGauge().DataPoints().AppendEmpty().Attributes().PutStr("container_name", "mycontainer")
+		assert.True(t, f.MatchesMetricDataPoint(m3.Name(), m3.Gauge().DataPoints().At(0).Attributes()))
 	})
 }