Merge branch 'aws-cwa-dev' into mpmann/more-efa-metrics

Author: the-mann

exporter/awsemfexporter/emf_exporter.go

@@ -231,7 +231,7 @@ func (emf *emfExporter) start(_ context.Context, host component.Host) error {
 	// Below are optimizatons to minimize amoount of
 	// metrics processing. We have two scearios
 	// 1. AppSignal - Only run Process function for AppSignal related useragent
-	// 2. Enhanced Container Insights - Only run ProcessMetrics function for CI EBS related useragent
+	// 2. Enhanced Container Insights - Only run ProcessMetrics function for CI related useragent
 	if emf.config.IsAppSignalsEnabled() || emf.config.IsEnhancedContainerInsights() {
 		userAgent := useragent.NewUserAgent()
 		emf.svcStructuredLog.Handlers().Build.PushFrontNamed(userAgent.Handler())

exporter/awsemfexporter/internal/useragent/useragent.go

@@ -28,11 +28,16 @@ const (
 
 	// TODO: Available in semconv/v1.21.0+. Replace after collector dependency is v0.91.0+.
 	attributeTelemetryDistroVersion = "telemetry.distro.version"
-
-	attributeEBS    = "ci_ebs"
-	ebsMetricPrefix = "node_diskio_ebs"
+	attributeEBS                    = "ci_ebs"
+	attributeLocalInstanceStore     = "ci_lis"
 )
 
+// Map of NVMe feature attributes to their corresponding metric prefixes
+var featureMetricPrefixes = map[string]string{
+	attributeEBS:                "node_diskio_ebs",
+	attributeLocalInstanceStore: "node_diskio_instance_store",
+}
+
 type UserAgent struct {
 	mu          sync.RWMutex
 	prebuiltStr string
@@ -95,8 +100,15 @@ func (ua *UserAgent) Process(labels map[string]string) {
 
 // ProcessMetrics checks metric names for specific patterns and updates user agent accordingly
 func (ua *UserAgent) ProcessMetrics(metrics pmetric.Metrics) {
-	// Check if we've already detected NVME
-	if _, exists := ua.featureList[attributeEBS]; exists {
+	// Check if all NVME features are already detected
+	allFeaturesFound := true
+	for feature := range featureMetricPrefixes {
+		if _, exists := ua.featureList[feature]; !exists {
+			allFeaturesFound = false
+			break
+		}
+	}
+	if allFeaturesFound {
 		return
 	}
 
@@ -107,14 +119,17 @@ func (ua *UserAgent) ProcessMetrics(metrics pmetric.Metrics) {
 			ms := ilms.At(j).Metrics()
 			for k := 0; k < ms.Len(); k++ {
 				metric := ms.At(k)
-				if strings.HasPrefix(metric.Name(), ebsMetricPrefix) {
-					ua.featureList[attributeEBS] = struct{}{}
-					ua.build()
-					return
+				for feature, prefix := range featureMetricPrefixes {
+					if strings.HasPrefix(metric.Name(), prefix) {
+						if _, exists := ua.featureList[feature]; !exists {
+							ua.featureList[feature] = struct{}{}
+						}
+					}
 				}
 			}
 		}
 	}
+	ua.build()
 }
 
 // build the user agent string from the items in the cache. Format is telemetry-sdk (<lang1>/<ver1>;<lang2>/<ver2>).

exporter/awsemfexporter/internal/useragent/useragent_test.go

@@ -75,6 +75,10 @@ func TestUserAgent(t *testing.T) {
 			metrics: []string{"node_diskio_ebs_something"},
 			want:    "feature:(ci_ebs)",
 		},
+		"WithLocalInstanceStoreMetrics": {
+			metrics: []string{"node_diskio_instance_store_something"},
+			want:    "feature:(ci_lis)",
+		},
 		"WithBothTelemetryAndEBS": {
 			labelSets: []map[string]string{
 				{
@@ -85,14 +89,32 @@ func TestUserAgent(t *testing.T) {
 			metrics: []string{"node_diskio_ebs_something"},
 			want:    "telemetry-sdk (test/1.0) feature:(ci_ebs)",
 		},
+		"WithBothTelemetryAndLocalInstanceStore": {
+			labelSets: []map[string]string{
+				{
+					semconv.AttributeTelemetrySDKLanguage: "test",
+					attributeTelemetryDistroVersion:       "1.0",
+				},
+			},
+			metrics: []string{"node_diskio_instance_store_something"},
+			want:    "telemetry-sdk (test/1.0) feature:(ci_lis)",
+		},
 		"WithNonEBSMetrics": {
 			metrics: []string{"some_other_metric"},
 			want:    "",
 		},
-		"WithMultipleFeatures": {
+		"WithMultipleEBSMetrics": {
 			metrics: []string{"node_diskio_ebs_something", "node_diskio_ebs_something_else"},
 			want:    "feature:(ci_ebs)",
 		},
+		"WithMultipleLocalInstanceStoreMetrics": {
+			metrics: []string{"node_diskio_instance_store_something", "node_diskio_instance_store_something_else"},
+			want:    "feature:(ci_lis)",
+		},
+		"WithMixedEBSAndInstanceStoreMetrics": {
+			metrics: []string{"node_diskio_ebs_something", "node_diskio_ebs_something", "node_diskio_instance_store_something", "node_diskio_instance_store_something"},
+			want:    "feature:(ci_ebs ci_lis)", // Both features should be detected and sorted alphabetically
+		},
 	}
 	for name, testCase := range testCases {
 		t.Run(name, func(t *testing.T) {

exporter/awsemfexporter/metric_translator.go

@@ -147,7 +147,8 @@ func (mt metricTranslator) translateOTelToGroupedMetric(rm pmetric.ResourceMetri
 			strings.HasPrefix(serviceName.Str(), "containerInsightsDCGMExporterScraper") ||
 			strings.HasPrefix(serviceName.Str(), "containerInsightsNeuronMonitorScraper") ||
 			strings.HasPrefix(serviceName.Str(), "containerInsightsKueueMetricsScraper") ||
-			strings.HasPrefix(serviceName.Str(), "containerInsightsNVMeExporterScraper") {
+			strings.HasPrefix(serviceName.Str(), "containerInsightsNVMeEBSScraper") ||
+			strings.HasPrefix(serviceName.Str(), "containerInsightsNVMeLISScraper") {
 			// the prometheus metrics that come from the container insight receiver need to be clearly tagged as coming from container insights
 			metricReceiver = containerInsightsReceiver
 		}

exporter/awsemfexporter/metric_translator_test.go

@@ -279,7 +279,9 @@ func TestTranslateOtToGroupedMetric(t *testing.T) {
 	kueueMetric := createTestResourceMetricsHelper(defaultNumberOfTestMetrics + 1)
 	kueueMetric.Resource().Attributes().PutStr(conventions.AttributeServiceName, "containerInsightsKueueMetricsScraper")
 	nvmeMetric := createTestResourceMetricsHelper(defaultNumberOfTestMetrics + 1)
-	nvmeMetric.Resource().Attributes().PutStr(conventions.AttributeServiceName, "containerInsightsNVMeExporterScraper")
+	nvmeMetric.Resource().Attributes().PutStr(conventions.AttributeServiceName, "containerInsightsNVMeEBSScraper")
+	nvmeLisMetric := createTestResourceMetricsHelper(defaultNumberOfTestMetrics + 1)
+	nvmeLisMetric.Resource().Attributes().PutStr(conventions.AttributeServiceName, "containerInsightsNVMeLISScraper")
 
 	counterSumMetrics := map[string]*metricInfo{
 		"spanCounter": {
@@ -407,7 +409,20 @@ func TestTranslateOtToGroupedMetric(t *testing.T) {
 			map[string]string{
 				"spanName": "testSpan",
 			},
-			"myServiceNS/containerInsightsNVMeExporterScraper",
+			"myServiceNS/containerInsightsNVMeEBSScraper",
+			containerInsightsReceiver,
+		},
+		{
+			"nvme lis receiver",
+			nvmeLisMetric,
+			map[string]string{
+				"isItAnError": "false",
+				"spanName":    "testSpan",
+			},
+			map[string]string{
+				"spanName": "testSpan",
+			},
+			"myServiceNS/containerInsightsNVMeLISScraper",
 			containerInsightsReceiver,
 		},
 	}

exporter/awsxrayexporter/internal/translator/segment.go

@@ -334,8 +334,8 @@ func MakeSegment(span ptrace.Span, resource pcommon.Resource, indexedAttrs []str
 	attributes := span.Attributes()
 
 	var (
-		startTime                                          = timestampToFloatSeconds(span.StartTimestamp())
-		endTime                                            = timestampToFloatSeconds(span.EndTimestamp())
+		startTimePtr                                       = awsP.Float64(timestampToFloatSeconds(span.StartTimestamp()))
+		endTimePtr, inProgress                             = makeEndTimeAndInProgress(span, attributes)
 		httpfiltered, http                                 = makeHTTP(span)
 		isError, isFault, isThrottle, causefiltered, cause = makeCause(span, httpfiltered, resource)
 		origin                                             = determineAwsOrigin(resource)
@@ -456,8 +456,9 @@ func MakeSegment(span ptrace.Span, resource pcommon.Resource, indexedAttrs []str
 		ID:          awsxray.String(traceutil.SpanIDToHexOrEmptyString(span.SpanID())),
 		TraceID:     awsxray.String(traceID),
 		Name:        awsxray.String(name),
-		StartTime:   awsP.Float64(startTime),
-		EndTime:     awsP.Float64(endTime),
+		StartTime:   startTimePtr,
+		EndTime:     endTimePtr,
+		InProgress:  inProgress,
 		ParentID:    awsxray.String(traceutil.SpanIDToHexOrEmptyString(span.ParentSpanID())),
 		Fault:       awsP.Bool(isFault),
 		Error:       awsP.Bool(isError),
@@ -758,6 +759,17 @@ func fixAnnotationKey(key string) string {
 	}, key)
 }
 
+func makeEndTimeAndInProgress(span ptrace.Span, attributes pcommon.Map) (*float64, *bool) {
+	if inProgressAttr, ok := attributes.Get(awsxray.AWSXRayInProgressAttribute); ok && inProgressAttr.Type() == pcommon.ValueTypeBool {
+		inProgressVal := inProgressAttr.Bool()
+		attributes.Remove(awsxray.AWSXRayInProgressAttribute)
+		if inProgressVal {
+			return nil, &inProgressVal
+		}
+	}
+	return awsP.Float64(timestampToFloatSeconds(span.EndTimestamp())), nil
+}
+
 func trimAwsSdkPrefix(name string, span ptrace.Span) string {
 	if isAwsSdkSpan(span) {
 		if strings.HasPrefix(name, "AWS.SDK.") {

exporter/awsxrayexporter/internal/translator/segment_test.go

@@ -1444,6 +1444,11 @@ func TestLocalRootConsumer(t *testing.T) {
 	// Checks these values are the same for both
 	assert.Equal(t, segments[0].StartTime, segments[1].StartTime)
 	assert.Equal(t, segments[0].EndTime, segments[1].EndTime)
+
+	// Check that segment EndTime matches span's EndTime
+	expectedEndTime := float64(span.EndTimestamp()) / float64(time.Second)
+	assert.Equal(t, expectedEndTime, *segments[0].EndTime)
+	assert.Equal(t, expectedEndTime, *segments[1].EndTime)
 }
 
 func TestNonLocalRootConsumerProcess(t *testing.T) {
@@ -1966,6 +1971,8 @@ func constructSpanAttributes(attributes map[string]any) pcommon.Map {
 			attrs.PutInt(key, int64(cast))
 		} else if cast, ok := value.(int64); ok {
 			attrs.PutInt(key, cast)
+		} else if cast, ok := value.(bool); ok {
+			attrs.PutBool(key, cast)
 		} else if cast, ok := value.([]string); ok {
 			slice := attrs.PutEmptySlice(key)
 			for _, v := range cast {
@@ -2041,6 +2048,40 @@ func constructTimedEventsWithSentMessageEvent(tm pcommon.Timestamp) ptrace.SpanE
 }
 
 // newTraceID generates a new valid X-Ray TraceID
+func TestSpanWithInProgressTrue(t *testing.T) {
+	spanName := "/api/test"
+	parentSpanID := newSegmentID()
+	attributes := make(map[string]any)
+	attributes[awsxray.AWSXRayInProgressAttribute] = true
+	resource := constructDefaultResource()
+	span := constructServerSpan(parentSpanID, spanName, ptrace.StatusCodeOk, "OK", attributes)
+
+	segment, _ := MakeSegment(span, resource, nil, false, nil, false)
+
+	assert.NotNil(t, segment)
+	assert.NotNil(t, segment.InProgress)
+	assert.True(t, *segment.InProgress)
+	assert.Nil(t, segment.EndTime)
+	assert.Nil(t, segment.Metadata["default"][awsxray.AWSXRayInProgressAttribute])
+}
+
+func TestSpanWithInProgressFalse(t *testing.T) {
+	spanName := "/api/test"
+	parentSpanID := newSegmentID()
+	attributes := make(map[string]any)
+	attributes[awsxray.AWSXRayInProgressAttribute] = false
+	resource := constructDefaultResource()
+	span := constructServerSpan(parentSpanID, spanName, ptrace.StatusCodeOk, "OK", attributes)
+
+	segment, _ := MakeSegment(span, resource, nil, false, nil, false)
+
+	assert.NotNil(t, segment)
+	assert.Nil(t, segment.InProgress)
+	assert.NotNil(t, segment.EndTime)
+	assert.Empty(t, segment.Annotations)
+	assert.Nil(t, segment.Metadata["default"][awsxray.AWSXRayInProgressAttribute])
+}
+
 func newTraceID() pcommon.TraceID {
 	var r [16]byte
 	epoch := time.Now().Unix()

receiver/awscontainerinsightreceiver/internal/nvme/metric_unit.go

@@ -7,7 +7,7 @@ import (
 )
 
 const (
-	// Original Metric Names
+	// EBS Original Metric Names
 	ebsReadOpsTotal        = "aws_ebs_csi_read_ops_total"
 	ebsWriteOpsTotal       = "aws_ebs_csi_write_ops_total"
 	ebsReadBytesTotal      = "aws_ebs_csi_read_bytes_total"
@@ -19,9 +19,21 @@ const (
 	ebsExceededEC2IOPSTime = "aws_ebs_csi_ec2_exceeded_iops_seconds_total"
 	ebsExceededEC2TPTime   = "aws_ebs_csi_ec2_exceeded_tp_seconds_total"
 	ebsVolumeQueueLength   = "aws_ebs_csi_volume_queue_length"
+
+	// LIS Original Metric Names
+	lisReadOpsTotal      = "aws_ec2_instance_store_csi_read_ops_total"
+	lisWriteOpsTotal     = "aws_ec2_instance_store_csi_write_ops_total"
+	lisReadBytesTotal    = "aws_ec2_instance_store_csi_read_bytes_total"
+	lisWriteBytesTotal   = "aws_ec2_instance_store_csi_write_bytes_total"
+	lisReadTime          = "aws_ec2_instance_store_csi_read_seconds_total"
+	lisWriteTime         = "aws_ec2_instance_store_csi_write_seconds_total"
+	lisExceededIOPSTime  = "aws_ec2_instance_store_csi_ec2_exceeded_iops_seconds_total"
+	lisExceededTPTime    = "aws_ec2_instance_store_csi_ec2_exceeded_tp_seconds_total"
+	lisVolumeQueueLength = "aws_ec2_instance_store_csi_volume_queue_length"
 )
 
 var MetricToUnit = map[string]string{
+	// EBS metrics
 	ebsReadOpsTotal:        containerinsight.UnitCount,
 	ebsWriteOpsTotal:       containerinsight.UnitCount,
 	ebsReadBytesTotal:      containerinsight.UnitBytes,
@@ -33,4 +45,15 @@ var MetricToUnit = map[string]string{
 	ebsExceededEC2IOPSTime: containerinsight.UnitSecond,
 	ebsExceededEC2TPTime:   containerinsight.UnitSecond,
 	ebsVolumeQueueLength:   containerinsight.UnitCount,
+
+	// LIS metrics
+	lisReadOpsTotal:      containerinsight.UnitCount,
+	lisWriteOpsTotal:     containerinsight.UnitCount,
+	lisReadBytesTotal:    containerinsight.UnitBytes,
+	lisWriteBytesTotal:   containerinsight.UnitBytes,
+	lisReadTime:          containerinsight.UnitSecond,
+	lisWriteTime:         containerinsight.UnitSecond,
+	lisExceededIOPSTime:  containerinsight.UnitSecond,
+	lisExceededTPTime:    containerinsight.UnitSecond,
+	lisVolumeQueueLength: containerinsight.UnitCount,
 }

…iver/internal/nvme/nvmescraper_config.go …internal/nvme/nvme_ebs_scraper_config.goreceiver/awscontainerinsightreceiver/internal/nvme/nvmescraper_config.go renamed to receiver/awscontainerinsightreceiver/internal/nvme/nvme_ebs_scraper_config.go receiver/awscontainerinsightreceiver/internal/nvme/nvmescraper_config.go renamed to receiver/awscontainerinsightreceiver/internal/nvme/nvme_ebs_scraper_config.go

@@ -18,7 +18,7 @@ import (
 
 const (
 	collectionInterval        = 60 * time.Second
-	jobName                   = "containerInsightsNVMeExporterScraper"
+	jobName                   = "containerInsightsNVMeEBSScraper"
 	scraperMetricsPath        = "/metrics"
 	scraperK8sServiceSelector = "app=ebs-csi-node"
 )
@@ -29,7 +29,7 @@ type hostInfoProvider interface {
 	GetInstanceType() string
 }
 
-func GetScraperConfig(hostInfoProvider hostInfoProvider) *config.ScrapeConfig {
+func GetEbsScraperConfig(hostInfoProvider hostInfoProvider) *config.ScrapeConfig {
 	return &config.ScrapeConfig{
 		ScrapeInterval:         model.Duration(collectionInterval),
 		ScrapeTimeout:          model.Duration(collectionInterval),

…ceiver/internal/nvme/nvmescraper_test.go …r/internal/nvme/nvme_ebs_scraper_test.goreceiver/awscontainerinsightreceiver/internal/nvme/nvmescraper_test.go renamed to receiver/awscontainerinsightreceiver/internal/nvme/nvme_ebs_scraper_test.go receiver/awscontainerinsightreceiver/internal/nvme/nvmescraper_test.go renamed to receiver/awscontainerinsightreceiver/internal/nvme/nvme_ebs_scraper_test.go

@@ -141,7 +141,7 @@ func TestNewNVMEScraperEndToEnd(t *testing.T) {
 		Consumer:          mConsumer,
 		Host:              componenttest.NewNopHost(),
 		HostInfoProvider:  mockHostInfoProvider{},
-		ScraperConfigs:    GetScraperConfig(mockHostInfoProvider{}),
+		ScraperConfigs:    GetEbsScraperConfig(mockHostInfoProvider{}),
 		Logger:            settings.Logger,
 	})
 	assert.NoError(t, err)
@@ -218,5 +218,5 @@ func TestNewNVMEScraperEndToEnd(t *testing.T) {
 
 func TestNvmeScraperJobName(t *testing.T) {
 	// needs to start with containerInsights
-	assert.True(t, strings.HasPrefix(jobName, "containerInsightsNVMeExporterScraper"))
+	assert.True(t, strings.HasPrefix(jobName, "containerInsightsNVMeEBSScraper"))
 }