add extractors for diskio/fs/net metrics

receiver/awscontainerinsightreceiver/internal/cadvisor/extractors/diskio_extractor.go

@@ -0,0 +1,102 @@
+// Copyright  OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package extractors
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	cInfo "github.com/google/cadvisor/info/v1"
+	"go.uber.org/zap"
+
+	ci "github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/containerinsight"
+	awsmetrics "github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/metrics"
+)
+
+type DiskIOMetricExtractor struct {
+	logger         *zap.Logger
+	rateCalculator awsmetrics.MetricCalculator
+}
+
+func (d *DiskIOMetricExtractor) HasValue(info *cInfo.ContainerInfo) bool {
+	return info.Spec.HasDiskIo
+}
+
+func (d *DiskIOMetricExtractor) GetValue(info *cInfo.ContainerInfo, _ CPUMemInfoProvider, containerType string) []*CAdvisorMetric {
+	var metrics []*CAdvisorMetric
+	if containerType != ci.TypeNode && containerType != ci.TypeInstance {
+		return metrics
+	}
+
+	curStats := GetStats(info)
+	metrics = append(metrics, d.extractIoMetrics(curStats.DiskIo.IoServiceBytes, ci.DiskIOServiceBytesPrefix, containerType, info.Name, curStats.Timestamp)...)
+	metrics = append(metrics, d.extractIoMetrics(curStats.DiskIo.IoServiced, ci.DiskIOServicedPrefix, containerType, info.Name, curStats.Timestamp)...)
+	return metrics
+}
+
+func (d *DiskIOMetricExtractor) extractIoMetrics(curStatsSet []cInfo.PerDiskStats, namePrefix string, containerType string, infoName string, curTime time.Time) []*CAdvisorMetric {
+	var metrics []*CAdvisorMetric
+	expectedKey := []string{ci.DiskIOAsync, ci.DiskIOSync, ci.DiskIORead, ci.DiskIOWrite, ci.DiskIOTotal}
+	for _, cur := range curStatsSet {
+		curDevName := devName(cur)
+		metric := newCadvisorMetric(getDiskIOMetricType(containerType, d.logger), d.logger)
+		metric.tags[ci.DiskDev] = curDevName
+		for _, key := range expectedKey {
+			if curVal, curOk := cur.Stats[key]; curOk {
+				mname := ci.MetricName(containerType, ioMetricName(namePrefix, key))
+				assignRateValueToField(&d.rateCalculator, metric.fields, mname, infoName, float64(curVal), curTime, float64(time.Second))
+			}
+		}
+		if len(metric.fields) > 0 {
+			metrics = append(metrics, metric)
+		}
+	}
+	return metrics
+}
+
+func ioMetricName(prefix, key string) string {
+	return prefix + strings.ToLower(key)
+}
+
+func devName(dStats cInfo.PerDiskStats) string {
+	devName := dStats.Device
+	if devName == "" {
+		devName = fmt.Sprintf("%d:%d", dStats.Major, dStats.Minor)
+	}
+	return devName
+}
+
+func NewDiskIOMetricExtractor(logger *zap.Logger) *DiskIOMetricExtractor {
+	return &DiskIOMetricExtractor{
+		logger:         logger,
+		rateCalculator: newFloat64RateCalculator(),
+	}
+}
+
+func getDiskIOMetricType(containerType string, logger *zap.Logger) string {
+	metricType := ""
+	switch containerType {
+	case ci.TypeNode:
+		metricType = ci.TypeNodeDiskIO
+	case ci.TypeInstance:
+		metricType = ci.TypeInstanceDiskIO
+	case ci.TypeContainer:
+		metricType = ci.TypeContainerDiskIO
+	default:
+		logger.Warn("diskio_extractor: diskIO metric extractor is parsing unexpected containerType", zap.String("containerType", containerType))
+	}
+	return metricType
+}

receiver/awscontainerinsightreceiver/internal/cadvisor/extractors/diskio_extractor_test.go

@@ -0,0 +1,110 @@
+// Copyright  OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package extractors
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	. "github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/containerinsight"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/awscontainerinsightreceiver/internal/cadvisor/testutils"
+)
+
+func TestDiskIOStats(t *testing.T) {
+
+	result := testutils.LoadContainerInfo(t, "./testdata/PreInfoContainer.json")
+	result2 := testutils.LoadContainerInfo(t, "./testdata/CurInfoContainer.json")
+	//for eks node-level metrics
+	containerType := TypeNode
+	extractor := NewDiskIOMetricExtractor(nil)
+
+	var cMetrics []*CAdvisorMetric
+	if extractor.HasValue(result[0]) {
+		cMetrics = extractor.GetValue(result[0], nil, containerType)
+	}
+
+	if extractor.HasValue(result2[0]) {
+		cMetrics = extractor.GetValue(result2[0], nil, containerType)
+	}
+
+	expectedFieldsService := map[string]interface{}{
+		"node_diskio_io_service_bytes_write": float64(10000),
+		"node_diskio_io_service_bytes_total": float64(10010),
+		"node_diskio_io_service_bytes_async": float64(10000),
+		"node_diskio_io_service_bytes_sync":  float64(10000),
+		"node_diskio_io_service_bytes_read":  float64(10),
+	}
+	expectedFieldsServiced := map[string]interface{}{
+		"node_diskio_io_serviced_async": float64(10),
+		"node_diskio_io_serviced_sync":  float64(10),
+		"node_diskio_io_serviced_read":  float64(10),
+		"node_diskio_io_serviced_write": float64(10),
+		"node_diskio_io_serviced_total": float64(20),
+	}
+	expectedTags := map[string]string{
+		"device": "/dev/xvda",
+		"Type":   "NodeDiskIO",
+	}
+	AssertContainsTaggedField(t, cMetrics[0], expectedFieldsService, expectedTags)
+	AssertContainsTaggedField(t, cMetrics[1], expectedFieldsServiced, expectedTags)
+
+	//for ecs node-level metrics
+	containerType = TypeInstance
+	extractor = NewDiskIOMetricExtractor(nil)
+
+	if extractor.HasValue(result[0]) {
+		cMetrics = extractor.GetValue(result[0], nil, containerType)
+	}
+
+	if extractor.HasValue(result2[0]) {
+		cMetrics = extractor.GetValue(result2[0], nil, containerType)
+	}
+
+	expectedFieldsService = map[string]interface{}{
+		"instance_diskio_io_service_bytes_write": float64(10000),
+		"instance_diskio_io_service_bytes_total": float64(10010),
+		"instance_diskio_io_service_bytes_async": float64(10000),
+		"instance_diskio_io_service_bytes_sync":  float64(10000),
+		"instance_diskio_io_service_bytes_read":  float64(10),
+	}
+	expectedFieldsServiced = map[string]interface{}{
+		"instance_diskio_io_serviced_async": float64(10),
+		"instance_diskio_io_serviced_sync":  float64(10),
+		"instance_diskio_io_serviced_read":  float64(10),
+		"instance_diskio_io_serviced_write": float64(10),
+		"instance_diskio_io_serviced_total": float64(20),
+	}
+	expectedTags = map[string]string{
+		"device": "/dev/xvda",
+		"Type":   "InstanceDiskIO",
+	}
+	AssertContainsTaggedField(t, cMetrics[0], expectedFieldsService, expectedTags)
+	AssertContainsTaggedField(t, cMetrics[1], expectedFieldsServiced, expectedTags)
+
+	//for non supported type
+	containerType = TypeContainerDiskIO
+	extractor = NewDiskIOMetricExtractor(nil)
+
+	if extractor.HasValue(result[0]) {
+		cMetrics = extractor.GetValue(result[0], nil, containerType)
+	}
+
+	if extractor.HasValue(result2[0]) {
+		cMetrics = extractor.GetValue(result2[0], nil, containerType)
+	}
+
+	assert.Equal(t, len(cMetrics), 0)
+}

receiver/awscontainerinsightreceiver/internal/cadvisor/extractors/fs_extractor.go

@@ -0,0 +1,104 @@
+// Copyright  OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package extractors
+
+import (
+	"regexp"
+
+	cinfo "github.com/google/cadvisor/info/v1"
+	"go.uber.org/zap"
+
+	ci "github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/containerinsight"
+)
+
+const (
+	allowList = "^tmpfs$|^/dev/|^overlay$"
+)
+
+type FileSystemMetricExtractor struct {
+	allowListRegexP *regexp.Regexp
+	logger          *zap.Logger
+}
+
+func (f *FileSystemMetricExtractor) HasValue(info *cinfo.ContainerInfo) bool {
+	return info.Spec.HasFilesystem
+}
+
+func (f *FileSystemMetricExtractor) GetValue(info *cinfo.ContainerInfo, _ CPUMemInfoProvider, containerType string) []*CAdvisorMetric {
+	var metrics []*CAdvisorMetric
+	if containerType == ci.TypePod || info.Spec.Labels[containerNameLable] == infraContainerName {
+		return metrics
+	}
+
+	containerType = getFSMetricType(containerType, f.logger)
+	stats := GetStats(info)
+
+	for _, v := range stats.Filesystem {
+		metric := newCadvisorMetric(containerType, f.logger)
+		if v.Device == "" {
+			continue
+		}
+		if f.allowListRegexP != nil && !f.allowListRegexP.MatchString(v.Device) {
+			continue
+		}
+
+		metric.tags[ci.DiskDev] = v.Device
+		metric.tags[ci.FSType] = v.Type
+
+		metric.fields[ci.MetricName(containerType, ci.FSUsage)] = v.Usage
+		metric.fields[ci.MetricName(containerType, ci.FSCapacity)] = v.Limit
+		metric.fields[ci.MetricName(containerType, ci.FSAvailable)] = v.Available
+
+		if v.Limit != 0 {
+			metric.fields[ci.MetricName(containerType, ci.FSUtilization)] = float64(v.Usage) / float64(v.Limit) * 100
+		}
+
+		if v.HasInodes {
+			metric.fields[ci.MetricName(containerType, ci.FSInodes)] = v.Inodes
+			metric.fields[ci.MetricName(containerType, ci.FSInodesfree)] = v.InodesFree
+		}
+
+		metrics = append(metrics, metric)
+	}
+	return metrics
+}
+
+func NewFileSystemMetricExtractor(logger *zap.Logger) *FileSystemMetricExtractor {
+	fse := &FileSystemMetricExtractor{
+		logger: logger,
+	}
+	if p, err := regexp.Compile(allowList); err == nil {
+		fse.allowListRegexP = p
+	} else {
+		logger.Error("NewFileSystemMetricExtractor set regex failed", zap.Error(err))
+	}
+
+	return fse
+}
+
+func getFSMetricType(containerType string, logger *zap.Logger) string {
+	metricType := ""
+	switch containerType {
+	case ci.TypeNode:
+		metricType = ci.TypeNodeFS
+	case ci.TypeInstance:
+		metricType = ci.TypeInstanceFS
+	case ci.TypeContainer:
+		metricType = ci.TypeContainerFS
+	default:
+		logger.Warn("fs_extractor: fs metric extractor is parsing unexpected containerType", zap.String("containerType", containerType))
+	}
+	return metricType
+}