Exports Entity for awsemfexporter plugins on PutLogEvent calls

exporter/awsemfexporter/emf_exporter.go

@@ -36,7 +36,7 @@ const (
 )
 
 type emfExporter struct {
-	pusherMap        map[cwlogs.StreamKey]cwlogs.Pusher
+	pusherMap        map[string]cwlogs.Pusher
 	svcStructuredLog *cwlogs.Client
 	config           *Config
 
@@ -86,7 +86,7 @@ func newEmfExporter(config *Config, set exporter.Settings) (*emfExporter, error)
 		metricTranslator:      newMetricTranslator(*config),
 		retryCnt:              *awsConfig.MaxRetries,
 		collectorID:           collectorIdentifier.String(),
-		pusherMap:             map[cwlogs.StreamKey]cwlogs.Pusher{},
+		pusherMap:             map[string]cwlogs.Pusher{},
 		processResourceLabels: func(map[string]string) {},
 	}
 
@@ -179,10 +179,11 @@ func (emf *emfExporter) pushMetricsData(_ context.Context, md pmetric.Metrics) e
 
 func (emf *emfExporter) getPusher(key cwlogs.StreamKey) cwlogs.Pusher {
 	var ok bool
-	if _, ok = emf.pusherMap[key]; !ok {
-		emf.pusherMap[key] = cwlogs.NewPusher(key, emf.retryCnt, *emf.svcStructuredLog, emf.config.logger)
+	hash := key.Hash()
+	if _, ok = emf.pusherMap[hash]; !ok {
+		emf.pusherMap[hash] = cwlogs.NewPusher(key, emf.retryCnt, *emf.svcStructuredLog, emf.config.logger)
 	}
-	return emf.pusherMap[key]
+	return emf.pusherMap[hash]
 }
 
 func (emf *emfExporter) listPushers() []cwlogs.Pusher {

exporter/awsemfexporter/emf_exporter_test.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 
 	"github.com/amazon-contributing/opentelemetry-collector-contrib/extension/awsmiddleware"
+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -23,6 +24,7 @@ import (
 	"go.uber.org/zap/zaptest/observer"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/cwlogs"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/cwlogs/sdk/service/cloudwatchlogs"
 )
 
 const defaultRetryCount = 1
@@ -192,10 +194,15 @@ func TestConsumeMetricsWithLogGroupStreamConfig(t *testing.T) {
 	})
 	require.Error(t, exp.pushMetricsData(ctx, md))
 	require.NoError(t, exp.shutdown(ctx))
-	pusherMap, ok := exp.pusherMap[cwlogs.StreamKey{
+	streamKey := &cwlogs.StreamKey{
 		LogGroupName:  expCfg.LogGroupName,
 		LogStreamName: expCfg.LogStreamName,
-	}]
+		Entity: &cloudwatchlogs.Entity{KeyAttributes: map[string]*string{
+			entityType: aws.String(service),
+		}},
+	}
+	expectedStreamKeyHash := streamKey.Hash()
+	pusherMap, ok := exp.pusherMap[expectedStreamKeyHash]
 	assert.True(t, ok)
 	assert.NotNil(t, pusherMap)
 }
@@ -217,16 +224,29 @@ func TestConsumeMetricsWithLogGroupStreamValidPlaceholder(t *testing.T) {
 		metricNames:  []string{"metric_1", "metric_2"},
 		metricValues: [][]float64{{100}, {4}},
 		resourceAttributeMap: map[string]any{
-			"aws.ecs.cluster.name": "test-cluster-name",
-			"aws.ecs.task.id":      "test-task-id",
+			"aws.ecs.cluster.name":                  "test-cluster-name",
+			"aws.ecs.task.id":                       "test-task-id",
+			keyAttributeEntityServiceName:           "myService",
+			keyAttributeEntityDeploymentEnvironment: "myEnvironment",
+			attributeEntityCluster:                  "test-cluster-name",
 		},
 	})
 	require.Error(t, exp.pushMetricsData(ctx, md))
 	require.NoError(t, exp.shutdown(ctx))
-	pusherMap, ok := exp.pusherMap[cwlogs.StreamKey{
+	streamKey := &cwlogs.StreamKey{
 		LogGroupName:  "/aws/ecs/containerinsights/test-cluster-name/performance",
 		LogStreamName: "test-task-id",
-	}]
+		Entity: &cloudwatchlogs.Entity{
+			Attributes: map[string]*string{
+				"Cluster": aws.String("test-cluster-name"),
+			},
+			KeyAttributes: map[string]*string{
+				"Type":        aws.String(service),
+				"Name":        aws.String("myService"),
+				"Environment": aws.String("myEnvironment"),
+			}},
+	}
+	pusherMap, ok := exp.pusherMap[streamKey.Hash()]
 	assert.True(t, ok)
 	assert.NotNil(t, pusherMap)
 }
@@ -243,21 +263,32 @@ func TestConsumeMetricsWithOnlyLogStreamPlaceholder(t *testing.T) {
 	exp, err := newEmfExporter(expCfg, exportertest.NewNopSettings())
 	assert.NoError(t, err)
 	assert.NotNil(t, exp)
+	var entity = &cloudwatchlogs.Entity{
+		KeyAttributes: map[string]*string{
+			"Type":        aws.String(service),
+			"Name":        aws.String("myService"),
+			"Environment": aws.String("myEnvironment"),
+		},
+	}
 
 	md := generateTestMetrics(testMetric{
 		metricNames:  []string{"metric_1", "metric_2"},
 		metricValues: [][]float64{{100}, {4}},
 		resourceAttributeMap: map[string]any{
-			"aws.ecs.cluster.name": "test-cluster-name",
-			"aws.ecs.task.id":      "test-task-id",
+			"aws.ecs.cluster.name":                  "test-cluster-name",
+			"aws.ecs.task.id":                       "test-task-id",
+			keyAttributeEntityServiceName:           "myService",
+			keyAttributeEntityDeploymentEnvironment: "myEnvironment",
 		},
 	})
 	require.Error(t, exp.pushMetricsData(ctx, md))
 	require.NoError(t, exp.shutdown(ctx))
-	pusherMap, ok := exp.pusherMap[cwlogs.StreamKey{
+	streamKey := cwlogs.StreamKey{
 		LogGroupName:  expCfg.LogGroupName,
 		LogStreamName: "test-task-id",
-	}]
+		Entity:        entity,
+	}
+	pusherMap, ok := exp.pusherMap[streamKey.Hash()]
 	assert.True(t, ok)
 	assert.NotNil(t, pusherMap)
 }
@@ -279,16 +310,26 @@ func TestConsumeMetricsWithWrongPlaceholder(t *testing.T) {
 		metricNames:  []string{"metric_1", "metric_2"},
 		metricValues: [][]float64{{100}, {4}},
 		resourceAttributeMap: map[string]any{
-			"aws.ecs.cluster.name": "test-cluster-name",
-			"aws.ecs.task.id":      "test-task-id",
+			"aws.ecs.cluster.name":                  "test-cluster-name",
+			"aws.ecs.task.id":                       "test-task-id",
+			keyAttributeEntityServiceName:           "myService",
+			keyAttributeEntityDeploymentEnvironment: "myEnvironment",
 		},
 	})
 	require.Error(t, exp.pushMetricsData(ctx, md))
 	require.NoError(t, exp.shutdown(ctx))
-	pusherMap, ok := exp.pusherMap[cwlogs.StreamKey{
+	streamKey := cwlogs.StreamKey{
 		LogGroupName:  expCfg.LogGroupName,
 		LogStreamName: expCfg.LogStreamName,
-	}]
+		Entity: &cloudwatchlogs.Entity{
+			KeyAttributes: map[string]*string{
+				"Type":        aws.String(service),
+				"Name":        aws.String("myService"),
+				"Environment": aws.String("myEnvironment"),
+			},
+		},
+	}
+	pusherMap, ok := exp.pusherMap[streamKey.Hash()]
 	assert.True(t, ok)
 	assert.NotNil(t, pusherMap)
 }
@@ -312,11 +353,17 @@ func TestPushMetricsDataWithErr(t *testing.T) {
 	logPusher.On("ForceFlush", nil).Return("some error").Once()
 	logPusher.On("ForceFlush", nil).Return("").Once()
 	logPusher.On("ForceFlush", nil).Return("some error").Once()
-	exp.pusherMap = map[cwlogs.StreamKey]cwlogs.Pusher{}
-	exp.pusherMap[cwlogs.StreamKey{
+	exp.pusherMap = map[string]cwlogs.Pusher{}
+	streamKey := cwlogs.StreamKey{
 		LogGroupName:  "test-logGroupName",
 		LogStreamName: "test-logStreamName",
-	}] = logPusher
+		Entity: &cloudwatchlogs.Entity{
+			KeyAttributes: map[string]*string{
+				"Type": aws.String(service),
+			},
+		},
+	}
+	exp.pusherMap[streamKey.Hash()] = logPusher
 
 	md := generateTestMetrics(testMetric{
 		metricNames:  []string{"metric_1", "metric_2"},

exporter/awsemfexporter/metric_translator.go

@@ -11,12 +11,15 @@ import (
 	"strings"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws"
+	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/cwlogs"
-	aws "github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/metrics"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/cwlogs/sdk/service/cloudwatchlogs"
+	awsmetrics "github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/metrics"
 )
 
 const (
@@ -32,8 +35,36 @@ const (
 	containerInsightsReceiver = "awscontainerinsight"
 	attributeReceiver         = "receiver"
 	fieldPrometheusMetricType = "prom_metric_type"
+
+	// Entity fields
+	keyAttributeEntityServiceName           = "aws.entity.service.name"
+	serviceName                             = "Name"
+	keyAttributeEntityDeploymentEnvironment = "aws.entity.deployment.environment"
+	deploymentEnvironment                   = "Environment"
+	entityType                              = "Type"
+	service                                 = "Service"
+	attributeEntityCluster                  = "aws.entity.k8s.cluster.name"
+	cluster                                 = "Cluster"
+	attributeEntityNamespace                = "aws.entity.k8s.namespace.name"
+	namespace                               = "Namespace"
+	attributeEntityWorkload                 = "aws.entity.k8s.workload.name"
+	workload                                = "Workload"
+	attributeEntityNode                     = "aws.entity.k8s.node.name"
+	node                                    = "Node"
 )
 
+var keyAttributeEntityToShortNameMap = map[string]string{
+	keyAttributeEntityServiceName:           serviceName,
+	keyAttributeEntityDeploymentEnvironment: deploymentEnvironment,
+}
+
+var attributeEntityToShortNameMap = map[string]string{
+	attributeEntityCluster:   cluster,
+	attributeEntityNamespace: namespace,
+	attributeEntityWorkload:  workload,
+	attributeEntityNode:      node,
+}
+
 var errMissingMetricsForEnhancedContainerInsights = errors.New("nil event detected with EnhancedContainerInsights enabled")
 
 var fieldPrometheusTypes = map[pmetric.MetricType]string{
@@ -79,6 +110,7 @@ type groupedMetricMetadata struct {
 	timestampMs                int64
 	logGroup                   string
 	logStream                  string
+	entity                     *cloudwatchlogs.Entity
 	metricDataType             pmetric.MetricType
 	retainInitialValueForDelta bool
 }
@@ -103,8 +135,8 @@ func newMetricTranslator(config Config) metricTranslator {
 	return metricTranslator{
 		metricDescriptor: mt,
 		calculators: &emfCalculators{
-			delta:   aws.NewFloat64DeltaCalculator(),
-			summary: aws.NewMetricCalculator(calculateSummaryDelta),
+			delta:   awsmetrics.NewFloat64DeltaCalculator(),
+			summary: awsmetrics.NewMetricCalculator(calculateSummaryDelta),
 		},
 	}
 }
@@ -139,6 +171,9 @@ func (mt metricTranslator) translateOTelToGroupedMetric(rm pmetric.ResourceMetri
 		}
 	}
 
+	entity, resourceAttributes := fetchEntityFields(rm.Resource().Attributes())
+	resourceAttributes.CopyTo(rm.Resource().Attributes())
+
 	for j := 0; j < ilms.Len(); j++ {
 		ilm := ilms.At(j)
 		if ilm.Scope().Name() != "" {
@@ -154,6 +189,7 @@ func (mt metricTranslator) translateOTelToGroupedMetric(rm pmetric.ResourceMetri
 					timestampMs:                timestamp,
 					logGroup:                   logGroup,
 					logStream:                  logStream,
+					entity:                     &entity,
 					metricDataType:             metric.Type(),
 					retainInitialValueForDelta: deltaInitialValue,
 				},
@@ -169,6 +205,25 @@ func (mt metricTranslator) translateOTelToGroupedMetric(rm pmetric.ResourceMetri
 	return nil
 }
 
+func fetchEntityFields(resourceAttributes pcommon.Map) (cloudwatchlogs.Entity, pcommon.Map) {
+	// the original resourceAttributes map is immutable, so we need to create a mutable copy
+	// to remove the entity fields from the attributes
+	mutableResourceAttributes := pcommon.NewMap()
+	resourceAttributes.CopyTo(mutableResourceAttributes)
+	serviceKeyAttr := map[string]*string{
+		entityType: aws.String(service),
+	}
+	attributeMap := map[string]*string{}
+
+	processAttributes(keyAttributeEntityToShortNameMap, serviceKeyAttr, mutableResourceAttributes)
+	processAttributes(attributeEntityToShortNameMap, attributeMap, mutableResourceAttributes)
+
+	return cloudwatchlogs.Entity{
+		KeyAttributes: serviceKeyAttr,
+		Attributes:    attributeMap,
+	}, mutableResourceAttributes
+}
+
 // translateGroupedMetricToCWMetric converts Grouped Metric format to CloudWatch Metric format.
 func translateGroupedMetricToCWMetric(groupedMetric *groupedMetric, config *Config) *cWMetrics {
 	labels := groupedMetric.labels
@@ -505,13 +560,15 @@ func translateGroupedMetricToEmf(groupedMetric *groupedMetric, config *Config, d
 
 	logGroup := groupedMetric.metadata.logGroup
 	logStream := groupedMetric.metadata.logStream
+	entity := groupedMetric.metadata.entity
 
 	if logStream == "" {
 		logStream = defaultLogStream
 	}
 
 	event.LogGroupName = logGroup
 	event.LogStreamName = logStream
+	event.Entity = entity
 
 	return event, nil
 }

exporter/awsemfexporter/metric_translator_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -22,6 +23,7 @@ import (
 	"go.uber.org/zap/zaptest/observer"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/cwlogs"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/aws/cwlogs/sdk/service/cloudwatchlogs"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/occonventions"
 )
 
@@ -39,6 +41,7 @@ func createTestResourceMetricsHelper(numMetrics int) pmetric.ResourceMetrics {
 	rm.Resource().Attributes().PutStr("ClusterName", "myCluster")
 	rm.Resource().Attributes().PutStr("PodName", "myPod")
 	rm.Resource().Attributes().PutStr(attributeReceiver, prometheusReceiver)
+	rm.Resource().Attributes().PutStr(keyAttributeEntityServiceName, "myServiceName")
 	sm := rm.ScopeMetrics().AppendEmpty()
 
 	m := sm.Metrics().AppendEmpty()
@@ -2574,6 +2577,35 @@ func TestTranslateOtToGroupedMetricForInitialDeltaValue(t *testing.T) {
 	}
 }
 
+func TestFetchEntityFields(t *testing.T) {
+	resourceMetrics := pmetric.NewResourceMetrics()
+	resourceMetrics.Resource().Attributes().PutStr(keyAttributeEntityDeploymentEnvironment, "my-environment")
+	resourceMetrics.Resource().Attributes().PutStr(keyAttributeEntityServiceName, "my-service")
+	resourceMetrics.Resource().Attributes().PutStr(attributeEntityNode, "my-node")
+	resourceMetrics.Resource().Attributes().PutStr(attributeEntityCluster, "my-cluster")
+	resourceMetrics.Resource().Attributes().PutStr(attributeEntityNamespace, "my-namespace")
+	resourceMetrics.Resource().Attributes().PutStr(attributeEntityWorkload, "my-workload")
+
+	expectedEntity := cloudwatchlogs.Entity{KeyAttributes: map[string]*string{
+		entityType:            aws.String(service),
+		serviceName:           aws.String("my-service"),
+		deploymentEnvironment: aws.String("my-environment"),
+	},
+		Attributes: map[string]*string{
+			node:      aws.String("my-node"),
+			cluster:   aws.String("my-cluster"),
+			namespace: aws.String("my-namespace"),
+			workload:  aws.String("my-workload"),
+		},
+	}
+
+	entity, attrs := fetchEntityFields(resourceMetrics.Resource().Attributes())
+	assert.Equal(t, expectedEntity, entity)
+	attrs.CopyTo(resourceMetrics.Resource().Attributes())
+	assert.Equal(t, 0, resourceMetrics.Resource().Attributes().Len())
+
+}
+
 func generateTestMetrics(tm testMetric) pmetric.Metrics {
 	md := pmetric.NewMetrics()
 	now := time.Now()