Add network device dimension and option to filter by it

Author: james-bebbington

receiver/hostmetricsreceiver/config_test.go

@@ -71,12 +71,17 @@ func TestLoadConfig(t *testing.T) {
 			loadscraper.TypeStr:       &loadscraper.Config{},
 			filesystemscraper.TypeStr: &filesystemscraper.Config{},
 			memoryscraper.TypeStr:     &memoryscraper.Config{},
-			networkscraper.TypeStr:    &networkscraper.Config{},
-			processesscraper.TypeStr:  &processesscraper.Config{},
-			swapscraper.TypeStr:       &swapscraper.Config{},
+			networkscraper.TypeStr: &networkscraper.Config{
+				Include: networkscraper.MatchConfig{
+					Interfaces: []string{"test1"},
+					Config:     filterset.Config{MatchType: "strict"},
+				},
+			},
+			processesscraper.TypeStr: &processesscraper.Config{},
+			swapscraper.TypeStr:      &swapscraper.Config{},
 			processscraper.TypeStr: &processscraper.Config{
 				Include: processscraper.MatchConfig{
-					Names:  []string{"test1", "test2"},
+					Names:  []string{"test2", "test3"},
 					Config: filterset.Config{MatchType: "regexp"},
 				},
 			},

receiver/hostmetricsreceiver/internal/scraper/networkscraper/config.go

@@ -14,9 +14,24 @@
 
 package networkscraper
 
-import "go.opentelemetry.io/collector/receiver/hostmetricsreceiver/internal"
+import (
+	"go.opentelemetry.io/collector/internal/processor/filterset"
+	"go.opentelemetry.io/collector/receiver/hostmetricsreceiver/internal"
+)
 
 // Config relating to Network Metric Scraper.
 type Config struct {
 	internal.ConfigSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct
+
+	// Include specifies a filter on the network interfaces that should be included from the generated metrics.
+	// Exclude specifies a filter on the network interfaces that should be excluded from the generated metrics.
+	// If neither `include` or `exclude` are set, metrics will be generated for all network interfaces.
+	Include MatchConfig `mapstructure:"include"`
+	Exclude MatchConfig `mapstructure:"exclude"`
+}
+
+type MatchConfig struct {
+	filterset.Config `mapstructure:",squash"`
+
+	Interfaces []string `mapstructure:"interfaces"`
 }

receiver/hostmetricsreceiver/internal/scraper/networkscraper/factory.go

@@ -45,6 +45,10 @@ func (f *Factory) CreateMetricsScraper(
 	logger *zap.Logger,
 	config internal.Config,
 ) (internal.Scraper, error) {
-	cfg := config.(*Config)
-	return obsreportscraper.WrapScraper(newNetworkScraper(ctx, cfg), TypeStr), nil
+	scraper, err := newNetworkScraper(ctx, config.(*Config))
+	if err != nil {
+		return nil, err
+	}
+
+	return obsreportscraper.WrapScraper(scraper, TypeStr), nil
 }

receiver/hostmetricsreceiver/internal/scraper/networkscraper/factory_test.go

@@ -37,3 +37,12 @@ func TestCreateMetricsScraper(t *testing.T) {
 	assert.NoError(t, err)
 	assert.NotNil(t, scraper)
 }
+
+func TestCreateMetricsScraper_Error(t *testing.T) {
+	factory := &Factory{}
+	cfg := &Config{Include: MatchConfig{Interfaces: []string{""}}}
+
+	_, err := factory.CreateMetricsScraper(context.Background(), zap.NewNop(), cfg)
+
+	assert.Error(t, err)
+}

receiver/hostmetricsreceiver/internal/scraper/networkscraper/network_metadata.go

@@ -21,6 +21,7 @@ import (
 // network metric constants
 
 const (
+	interfaceLabelName = "interface"
 	directionLabelName = "direction"
 	stateLabelName     = "state"
 )

receiver/hostmetricsreceiver/internal/scraper/networkscraper/network_scraper.go

@@ -16,19 +16,23 @@ package networkscraper
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/shirou/gopsutil/host"
 	"github.com/shirou/gopsutil/net"
 
 	"go.opentelemetry.io/collector/component/componenterror"
 	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/internal/processor/filterset"
 )
 
 // scraper for Network Metrics
 type scraper struct {
 	config    *Config
 	startTime pdata.TimestampUnixNano
+	includeFS filterset.FilterSet
+	excludeFS filterset.FilterSet
 
 	// for mocking
 	bootTime    func() (uint64, error)
@@ -37,8 +41,26 @@ type scraper struct {
 }
 
 // newNetworkScraper creates a set of Network related metrics
-func newNetworkScraper(_ context.Context, cfg *Config) *scraper {
-	return &scraper{config: cfg, bootTime: host.BootTime, ioCounters: net.IOCounters, connections: net.Connections}
+func newNetworkScraper(_ context.Context, cfg *Config) (*scraper, error) {
+	scraper := &scraper{config: cfg, bootTime: host.BootTime, ioCounters: net.IOCounters, connections: net.Connections}
+
+	var err error
+
+	if len(cfg.Include.Interfaces) > 0 {
+		scraper.includeFS, err = filterset.CreateFilterSet(cfg.Include.Interfaces, &cfg.Include.Config)
+		if err != nil {
+			return nil, fmt.Errorf("error creating network interface include filters: %w", err)
+		}
+	}
+
+	if len(cfg.Exclude.Interfaces) > 0 {
+		scraper.excludeFS, err = filterset.CreateFilterSet(cfg.Exclude.Interfaces, &cfg.Exclude.Config)
+		if err != nil {
+			return nil, fmt.Errorf("error creating network interface exclude filters: %w", err)
+		}
+	}
+
+	return scraper, nil
 }
 
 // Initialize
@@ -82,33 +104,73 @@ func (s *scraper) ScrapeMetrics(_ context.Context) (pdata.MetricSlice, error) {
 
 func (s *scraper) scrapeAndAppendNetworkCounterMetrics(metrics pdata.MetricSlice, startTime pdata.TimestampUnixNano) error {
 	// get total stats only
-	networkStatsSlice, err := s.ioCounters( /*perNetworkInterfaceController=*/ false)
+	ioCounters, err := s.ioCounters( /*perNetworkInterfaceController=*/ true)
 	if err != nil {
 		return err
 	}
 
-	networkStats := networkStatsSlice[0]
+	// filter network interfaces by name
+	ioCounters = s.filterByInterface(ioCounters)
+
+	if len(ioCounters) > 0 {
+		startIdx := metrics.Len()
+		metrics.Resize(startIdx + 4)
+		initializeNetworkPacketsMetric(metrics.At(startIdx+0), networkPacketsDescriptor, startTime, ioCounters)
+		initializeNetworkDroppedPacketsMetric(metrics.At(startIdx+1), networkDroppedPacketsDescriptor, startTime, ioCounters)
+		initializeNetworkErrorsMetric(metrics.At(startIdx+2), networkErrorsDescriptor, startTime, ioCounters)
+		initializeNetworkIOMetric(metrics.At(startIdx+3), networkIODescriptor, startTime, ioCounters)
+	}
 
-	startIdx := metrics.Len()
-	metrics.Resize(startIdx + 4)
-	initializeNetworkMetric(metrics.At(startIdx+0), networkPacketsDescriptor, startTime, networkStats.PacketsSent, networkStats.PacketsRecv)
-	initializeNetworkMetric(metrics.At(startIdx+1), networkDroppedPacketsDescriptor, startTime, networkStats.Dropout, networkStats.Dropin)
-	initializeNetworkMetric(metrics.At(startIdx+2), networkErrorsDescriptor, startTime, networkStats.Errout, networkStats.Errin)
-	initializeNetworkMetric(metrics.At(startIdx+3), networkIODescriptor, startTime, networkStats.BytesSent, networkStats.BytesRecv)
 	return nil
 }
 
-func initializeNetworkMetric(metric pdata.Metric, metricDescriptor pdata.MetricDescriptor, startTime pdata.TimestampUnixNano, transmitValue, receiveValue uint64) {
+func initializeNetworkPacketsMetric(metric pdata.Metric, metricDescriptor pdata.MetricDescriptor, startTime pdata.TimestampUnixNano, ioCountersSlice []net.IOCountersStat) {
 	metricDescriptor.CopyTo(metric.MetricDescriptor())
 
 	idps := metric.Int64DataPoints()
-	idps.Resize(2)
-	initializeNetworkDataPoint(idps.At(0), startTime, transmitDirectionLabelValue, int64(transmitValue))
-	initializeNetworkDataPoint(idps.At(1), startTime, receiveDirectionLabelValue, int64(receiveValue))
+	idps.Resize(2 * len(ioCountersSlice))
+	for idx, ioCounters := range ioCountersSlice {
+		initializeNetworkDataPoint(idps.At(2*idx+0), startTime, ioCounters.Name, transmitDirectionLabelValue, int64(ioCounters.PacketsSent))
+		initializeNetworkDataPoint(idps.At(2*idx+1), startTime, ioCounters.Name, receiveDirectionLabelValue, int64(ioCounters.PacketsRecv))
+	}
 }
 
-func initializeNetworkDataPoint(dataPoint pdata.Int64DataPoint, startTime pdata.TimestampUnixNano, directionLabel string, value int64) {
+func initializeNetworkDroppedPacketsMetric(metric pdata.Metric, metricDescriptor pdata.MetricDescriptor, startTime pdata.TimestampUnixNano, ioCountersSlice []net.IOCountersStat) {
+	metricDescriptor.CopyTo(metric.MetricDescriptor())
+
+	idps := metric.Int64DataPoints()
+	idps.Resize(2 * len(ioCountersSlice))
+	for idx, ioCounters := range ioCountersSlice {
+		initializeNetworkDataPoint(idps.At(2*idx+0), startTime, ioCounters.Name, transmitDirectionLabelValue, int64(ioCounters.Dropout))
+		initializeNetworkDataPoint(idps.At(2*idx+1), startTime, ioCounters.Name, receiveDirectionLabelValue, int64(ioCounters.Dropin))
+	}
+}
+
+func initializeNetworkErrorsMetric(metric pdata.Metric, metricDescriptor pdata.MetricDescriptor, startTime pdata.TimestampUnixNano, ioCountersSlice []net.IOCountersStat) {
+	metricDescriptor.CopyTo(metric.MetricDescriptor())
+
+	idps := metric.Int64DataPoints()
+	idps.Resize(2 * len(ioCountersSlice))
+	for idx, ioCounters := range ioCountersSlice {
+		initializeNetworkDataPoint(idps.At(2*idx+0), startTime, ioCounters.Name, transmitDirectionLabelValue, int64(ioCounters.Errout))
+		initializeNetworkDataPoint(idps.At(2*idx+1), startTime, ioCounters.Name, receiveDirectionLabelValue, int64(ioCounters.Errin))
+	}
+}
+
+func initializeNetworkIOMetric(metric pdata.Metric, metricDescriptor pdata.MetricDescriptor, startTime pdata.TimestampUnixNano, ioCountersSlice []net.IOCountersStat) {
+	metricDescriptor.CopyTo(metric.MetricDescriptor())
+
+	idps := metric.Int64DataPoints()
+	idps.Resize(2 * len(ioCountersSlice))
+	for idx, ioCounters := range ioCountersSlice {
+		initializeNetworkDataPoint(idps.At(2*idx+0), startTime, ioCounters.Name, transmitDirectionLabelValue, int64(ioCounters.BytesSent))
+		initializeNetworkDataPoint(idps.At(2*idx+1), startTime, ioCounters.Name, receiveDirectionLabelValue, int64(ioCounters.BytesRecv))
+	}
+}
+
+func initializeNetworkDataPoint(dataPoint pdata.Int64DataPoint, startTime pdata.TimestampUnixNano, interfaceLabel, directionLabel string, value int64) {
 	labelsMap := dataPoint.LabelsMap()
+	labelsMap.Insert(interfaceLabelName, interfaceLabel)
 	labelsMap.Insert(directionLabelName, directionLabel)
 	dataPoint.SetStartTime(startTime)
 	dataPoint.SetTimestamp(pdata.TimestampUnixNano(uint64(time.Now().UnixNano())))
@@ -170,3 +232,22 @@ func initializeNetworkTCPConnectionsDataPoint(dataPoint pdata.Int64DataPoint, st
 	dataPoint.SetTimestamp(pdata.TimestampUnixNano(uint64(time.Now().UnixNano())))
 	dataPoint.SetValue(value)
 }
+
+func (s *scraper) filterByInterface(ioCounters []net.IOCountersStat) []net.IOCountersStat {
+	if s.includeFS == nil && s.excludeFS == nil {
+		return ioCounters
+	}
+
+	filteredIOCounters := make([]net.IOCountersStat, 0, len(ioCounters))
+	for _, io := range ioCounters {
+		if s.includeInterface(io.Name) {
+			filteredIOCounters = append(filteredIOCounters, io)
+		}
+	}
+	return filteredIOCounters
+}
+
+func (s *scraper) includeInterface(interfaceName string) bool {
+	return (s.includeFS == nil || s.includeFS.Matches(interfaceName)) &&
+		(s.excludeFS == nil || !s.excludeFS.Matches(interfaceName))
+}

receiver/hostmetricsreceiver/internal/scraper/networkscraper/network_scraper_test.go

@@ -24,28 +24,49 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/internal/processor/filterset"
 	"go.opentelemetry.io/collector/receiver/hostmetricsreceiver/internal"
 )
 
 func TestScrapeMetrics(t *testing.T) {
 	type testCase struct {
-		name              string
-		bootTimeFunc      func() (uint64, error)
-		ioCountersFunc    func(bool) ([]net.IOCountersStat, error)
-		connectionsFunc   func(string) ([]net.ConnectionStat, error)
-		expectedStartTime pdata.TimestampUnixNano
-		initializationErr string
-		expectedErr       string
+		name                 string
+		config               Config
+		bootTimeFunc         func() (uint64, error)
+		ioCountersFunc       func(bool) ([]net.IOCountersStat, error)
+		connectionsFunc      func(string) ([]net.ConnectionStat, error)
+		expectNetworkMetrics bool
+		expectedStartTime    pdata.TimestampUnixNano
+		newErrRegex          string
+		initializationErr    string
+		expectedErr          string
 	}
 
 	testCases := []testCase{
 		{
-			name: "Standard",
+			name:                 "Standard",
+			expectNetworkMetrics: true,
 		},
 		{
-			name:              "Validate Start Time",
-			bootTimeFunc:      func() (uint64, error) { return 100, nil },
-			expectedStartTime: 100 * 1e9,
+			name:                 "Validate Start Time",
+			bootTimeFunc:         func() (uint64, error) { return 100, nil },
+			expectNetworkMetrics: true,
+			expectedStartTime:    100 * 1e9,
+		},
+		{
+			name:                 "Include Filter that matches nothing",
+			config:               Config{Include: MatchConfig{filterset.Config{MatchType: "strict"}, []string{"@*^#&*$^#)"}}},
+			expectNetworkMetrics: false,
+		},
+		{
+			name:        "Invalid Include Filter",
+			config:      Config{Include: MatchConfig{Interfaces: []string{"test"}}},
+			newErrRegex: "^error creating network interface include filters:",
+		},
+		{
+			name:        "Invalid Exclude Filter",
+			config:      Config{Exclude: MatchConfig{Interfaces: []string{"test"}}},
+			newErrRegex: "^error creating network interface exclude filters:",
 		},
 		{
 			name:              "Boot Time Error",
@@ -66,7 +87,14 @@ func TestScrapeMetrics(t *testing.T) {
 
 	for _, test := range testCases {
 		t.Run(test.name, func(t *testing.T) {
-			scraper := newNetworkScraper(context.Background(), &Config{})
+			scraper, err := newNetworkScraper(context.Background(), &test.config)
+			if test.newErrRegex != "" {
+				require.Error(t, err)
+				require.Regexp(t, test.newErrRegex, err)
+				return
+			}
+			require.NoError(t, err, "Failed to create network scraper: %v", err)
+
 			if test.bootTimeFunc != nil {
 				scraper.bootTime = test.bootTimeFunc
 			}
@@ -77,7 +105,7 @@ func TestScrapeMetrics(t *testing.T) {
 				scraper.connections = test.connectionsFunc
 			}
 
-			err := scraper.Initialize(context.Background())
+			err = scraper.Initialize(context.Background())
 			if test.initializationErr != "" {
 				assert.EqualError(t, err, test.initializationErr)
 				return
@@ -92,14 +120,21 @@ func TestScrapeMetrics(t *testing.T) {
 			}
 			require.NoError(t, err, "Failed to scrape metrics: %v", err)
 
-			assert.Equal(t, 5, metrics.Len())
-
-			assertNetworkIOMetricValid(t, metrics.At(0), networkPacketsDescriptor, test.expectedStartTime)
-			assertNetworkIOMetricValid(t, metrics.At(1), networkDroppedPacketsDescriptor, test.expectedStartTime)
-			assertNetworkIOMetricValid(t, metrics.At(2), networkErrorsDescriptor, test.expectedStartTime)
-			assertNetworkIOMetricValid(t, metrics.At(3), networkIODescriptor, test.expectedStartTime)
+			expectedMetricCount := 1
+			if test.expectNetworkMetrics {
+				expectedMetricCount += 4
+			}
+			assert.Equal(t, expectedMetricCount, metrics.Len())
 
-			assertNetworkTCPConnectionsMetricValid(t, metrics.At(4))
+			idx := 0
+			if test.expectNetworkMetrics {
+				assertNetworkIOMetricValid(t, metrics.At(idx+0), networkPacketsDescriptor, test.expectedStartTime)
+				assertNetworkIOMetricValid(t, metrics.At(idx+1), networkDroppedPacketsDescriptor, test.expectedStartTime)
+				assertNetworkIOMetricValid(t, metrics.At(idx+2), networkErrorsDescriptor, test.expectedStartTime)
+				assertNetworkIOMetricValid(t, metrics.At(idx+3), networkIODescriptor, test.expectedStartTime)
+				idx += 4
+			}
+			assertNetworkTCPConnectionsMetricValid(t, metrics.At(idx+0))
 		})
 	}
 }
@@ -109,7 +144,8 @@ func assertNetworkIOMetricValid(t *testing.T, metric pdata.Metric, descriptor pd
 	if startTime != 0 {
 		internal.AssertInt64MetricStartTimeEquals(t, metric, startTime)
 	}
-	assert.Equal(t, 2, metric.Int64DataPoints().Len())
+	assert.GreaterOrEqual(t, metric.Int64DataPoints().Len(), 2)
+	internal.AssertInt64MetricLabelExists(t, metric, 0, interfaceLabelName)
 	internal.AssertInt64MetricLabelHasValue(t, metric, 0, directionLabelName, transmitDirectionLabelValue)
 	internal.AssertInt64MetricLabelHasValue(t, metric, 1, directionLabelName, receiveDirectionLabelValue)
 }

receiver/hostmetricsreceiver/testdata/config.yaml

@@ -11,11 +11,14 @@ receivers:
       filesystem:
       memory:
       network:
+        include:
+          interfaces: ["test1"]
+          match_type: "strict"
       processes:
       swap:
       process:
         include:
-          names: ["test1", "test2"]
+          names: ["test2", "test3"]
           match_type: "regexp"
 
 processors: