[processor/awsentity] Add awsentity processor into EMF pipeline (#1482)

aws · Jan 6, 2025 · dfd2070 · dfd2070
1 parent ef15aa4
commit dfd2070
Show file tree

Hide file tree

Showing 15 changed files with 80 additions and 7 deletions.
diff --git a/plugins/processors/awsentity/factory.go b/plugins/processors/awsentity/factory.go
@@ -26,7 +26,9 @@ func NewFactory() processor.Factory {
 	return processor.NewFactory(
 		TypeStr,
 		createDefaultConfig,
-		processor.WithMetrics(createMetricsProcessor, stability))
+		processor.WithMetrics(createMetricsProcessor, stability),
+		processor.WithLogs(createLogsProcessor, stability),
+	)
 }
 
 func createDefaultConfig() component.Config {
@@ -53,3 +55,23 @@ func createMetricsProcessor(
 		metricsProcessor.processMetrics,
 		processorhelper.WithCapabilities(processorCapabilities))
 }
+
+func createLogsProcessor(
+	ctx context.Context,
+	set processor.Settings,
+	cfg component.Config,
+	nextConsumer consumer.Logs,
+) (processor.Logs, error) {
+	processorConfig, ok := cfg.(*Config)
+	if !ok {
+		return nil, errors.New("configuration parsing error")
+	}
+	logProcessor := newAwsEntityProcessor(processorConfig, set.Logger)
+	return processorhelper.NewLogsProcessor(
+		ctx,
+		set,
+		cfg,
+		nextConsumer,
+		logProcessor.processLogs,
+		processorhelper.WithCapabilities(processorCapabilities))
+}
diff --git a/plugins/processors/awsentity/factory_test.go b/plugins/processors/awsentity/factory_test.go
@@ -40,6 +40,6 @@ func TestCreateProcessor(t *testing.T) {
 	assert.NotNil(t, mProcessor)
 
 	lProcessor, err := factory.CreateLogsProcessor(context.Background(), setting, cfg, consumertest.NewNop())
-	assert.Equal(t, err, component.ErrDataTypeIsNotSupported)
-	assert.Nil(t, lProcessor)
+	assert.NoError(t, err)
+	assert.NotNil(t, lProcessor)
 }
diff --git a/plugins/processors/awsentity/processor.go b/plugins/processors/awsentity/processor.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/go-playground/validator/v10"
 	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	semconv "go.opentelemetry.io/collector/semconv/v1.22.0"
 	"go.uber.org/zap"
@@ -110,6 +111,10 @@ func newAwsEntityProcessor(config *Config, logger *zap.Logger) *awsEntityProcess
 	}
 }
 
+func (p *awsEntityProcessor) processLogs(_ context.Context, ld plog.Logs) (plog.Logs, error) {
+	return ld, nil
+}
+
 func (p *awsEntityProcessor) processMetrics(_ context.Context, md pmetric.Metrics) (pmetric.Metrics, error) {
 	// Get the following metric attributes from the EntityStore: PlatformType, EC2.InstanceId, EC2.AutoScalingGroup
 

diff --git a/translator/tocwconfig/sampleConfig/base_container_insights_config.yaml b/translator/tocwconfig/sampleConfig/base_container_insights_config.yaml
@@ -159,6 +159,9 @@ extensions:
         mode: ec2
         region: us-east-1
 processors:
+    awsentity/service/emf:
+        entity_type: Service
+        platform: ec2
     batch/containerinsights:
         metadata_cardinality_limit: 1000
         send_batch_max_size: 0
@@ -235,6 +238,7 @@ service:
             exporters:
                 - awscloudwatchlogs/emf_logs
             processors:
+                - awsentity/service/emf
                 - batch/emf_logs
             receivers:
                 - tcplog/emf_logs

diff --git a/translator/tocwconfig/sampleConfig/complete_darwin_config.yaml b/translator/tocwconfig/sampleConfig/complete_darwin_config.yaml
@@ -105,6 +105,9 @@ processors:
     awsentity/resource:
         entity_type: Resource
         platform: ec2
+    awsentity/service/emf:
+        entity_type: Service
+        platform: ec2
     awsentity/service/telegraf:
         entity_type: Service
         platform: ec2
@@ -273,6 +276,7 @@ service:
             exporters:
                 - awscloudwatchlogs/emf_logs
             processors:
+                - awsentity/service/emf
                 - batch/emf_logs
             receivers:
                 - udplog/emf_logs

diff --git a/translator/tocwconfig/sampleConfig/complete_linux_config.yaml b/translator/tocwconfig/sampleConfig/complete_linux_config.yaml
@@ -111,6 +111,9 @@ processors:
     awsentity/resource:
         entity_type: Resource
         platform: ec2
+    awsentity/service/emf:
+        entity_type: Service
+        platform: ec2
     awsentity/service/telegraf:
         entity_type: Service
         platform: ec2
@@ -380,6 +383,7 @@ service:
             exporters:
                 - awscloudwatchlogs/emf_logs
             processors:
+                - awsentity/service/emf
                 - batch/emf_logs
             receivers:
                 - udplog/emf_logs

diff --git a/translator/tocwconfig/sampleConfig/complete_windows_config.yaml b/translator/tocwconfig/sampleConfig/complete_windows_config.yaml
@@ -105,6 +105,9 @@ processors:
     awsentity/resource:
         entity_type: Resource
         platform: ec2
+    awsentity/service/emf:
+        entity_type: Service
+        platform: ec2
     awsentity/service/telegraf:
         entity_type: Service
         platform: ec2
@@ -260,6 +263,7 @@ service:
             exporters:
                 - awscloudwatchlogs/emf_logs
             processors:
+                - awsentity/service/emf
                 - batch/emf_logs
             receivers:
                 - udplog/emf_logs

diff --git a/translator/tocwconfig/sampleConfig/config_with_env.yaml b/translator/tocwconfig/sampleConfig/config_with_env.yaml
@@ -50,6 +50,9 @@ extensions:
         mode: ec2
         region: ${ENV_REGION}
 processors:
+    awsentity/service/emf:
+        entity_type: Service
+        platform: ec2
     batch/emf_logs:
         metadata_cardinality_limit: 1000
         send_batch_max_size: 0
@@ -92,6 +95,7 @@ service:
             exporters:
                 - awscloudwatchlogs/emf_logs
             processors:
+                - awsentity/service/emf
                 - batch/emf_logs
             receivers:
                 - tcplog/emf_logs

diff --git a/translator/tocwconfig/sampleConfig/emf_and_kubernetes_config.yaml b/translator/tocwconfig/sampleConfig/emf_and_kubernetes_config.yaml
@@ -411,6 +411,9 @@ extensions:
         region: us-east-1
         shared_credential_file: /root/.aws/credentials
 processors:
+    awsentity/service/emf:
+        entity_type: Service
+        platform: onPremise
     batch/containerinsights:
         metadata_cardinality_limit: 1000
         send_batch_max_size: 0
@@ -499,6 +502,7 @@ service:
             exporters:
                 - awscloudwatchlogs/emf_logs
             processors:
+                - awsentity/service/emf
                 - batch/emf_logs
             receivers:
                 - tcplog/emf_logs

diff --git a/translator/tocwconfig/sampleConfig/emf_and_kubernetes_with_gpu_config.yaml b/translator/tocwconfig/sampleConfig/emf_and_kubernetes_with_gpu_config.yaml
@@ -669,6 +669,9 @@ extensions:
         region: us-east-1
         shared_credential_file: /root/.aws/credentials
 processors:
+    awsentity/service/emf:
+        entity_type: Service
+        platform: onPremise
     batch/containerinsights:
         metadata_cardinality_limit: 1000
         send_batch_max_size: 0
@@ -1171,6 +1174,7 @@ service:
             exporters:
                 - awscloudwatchlogs/emf_logs
             processors:
+                - awsentity/service/emf
                 - batch/emf_logs
             receivers:
                 - tcplog/emf_logs

diff --git a/translator/tocwconfig/sampleConfig/emf_and_kubernetes_with_kueue_config.yaml b/translator/tocwconfig/sampleConfig/emf_and_kubernetes_with_kueue_config.yaml
@@ -491,6 +491,9 @@ extensions:
         region: us-east-1
         shared_credential_file: /root/.aws/credentials
 processors:
+    awsentity/service/emf:
+        entity_type: Service
+        platform: onPremise
     batch/containerinsights:
         metadata_cardinality_limit: 1000
         send_batch_max_size: 0
@@ -588,6 +591,7 @@ service:
             exporters:
                 - awscloudwatchlogs/emf_logs
             processors:
+                - awsentity/service/emf
                 - batch/emf_logs
             receivers:
                 - tcplog/emf_logs

diff --git a/translator/tocwconfig/sampleConfig/kueue_container_insights_config.yaml b/translator/tocwconfig/sampleConfig/kueue_container_insights_config.yaml
@@ -237,6 +237,9 @@ extensions:
         mode: ec2
         region: us-east-1
 processors:
+    awsentity/service/emf:
+        entity_type: Service
+        platform: ec2
     batch/containerinsights:
         metadata_cardinality_limit: 1000
         send_batch_max_size: 0
@@ -322,6 +325,7 @@ service:
             exporters:
                 - awscloudwatchlogs/emf_logs
             processors:
+                - awsentity/service/emf
                 - batch/emf_logs
             receivers:
                 - tcplog/emf_logs

diff --git a/translator/tocwconfig/sampleConfig/logs_and_kubernetes_config.yaml b/translator/tocwconfig/sampleConfig/logs_and_kubernetes_config.yaml
@@ -405,6 +405,9 @@ extensions:
         mode: ec2
         region: us-east-1
 processors:
+    awsentity/service/emf:
+        entity_type: Service
+        platform: ec2
     batch/containerinsights:
         metadata_cardinality_limit: 1000
         send_batch_max_size: 0
@@ -491,6 +494,7 @@ service:
             exporters:
                 - awscloudwatchlogs/emf_logs
             processors:
+                - awsentity/service/emf
                 - batch/emf_logs
             receivers:
                 - tcplog/emf_logs

diff --git a/translator/translate/otel/pipeline/emf_logs/translator.go b/translator/translate/otel/pipeline/emf_logs/translator.go
@@ -9,12 +9,15 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/confmap"
 
+	"github.com/aws/amazon-cloudwatch-agent/translator/context"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/common"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/exporter/awscloudwatchlogs"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/extension/agenthealth"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/awsentity"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/batchprocessor"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/receiver/tcplog"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/receiver/udplog"
+	"github.com/aws/amazon-cloudwatch-agent/translator/util/ecsutil"
 )
 
 var (
@@ -54,6 +57,9 @@ func (t *translator) Translate(conf *confmap.Conf) (*common.ComponentTranslators
 			agenthealth.NewTranslatorWithStatusCode(component.MustNewType("statuscode"), nil, true),
 		),
 	}
+	if !(context.CurrentContext().RunInContainer() && ecsutil.GetECSUtilSingleton().IsECS()) {
+		translators.Processors.Set(awsentity.NewTranslatorWithEntityType(awsentity.Service, "emf", false))
+	}
 	if serviceAddress, ok := common.GetString(conf, serviceAddressEMFKey); ok {
 		if strings.Contains(serviceAddress, common.Udp) {
 			translators.Receivers.Set(udplog.NewTranslatorWithName(common.PipelineNameEmfLogs))

diff --git a/translator/translate/otel/pipeline/emf_logs/translator_test.go b/translator/translate/otel/pipeline/emf_logs/translator_test.go
@@ -46,7 +46,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineType: "logs/emf_logs",
 				receivers:    []string{"tcplog/emf_logs", "udplog/emf_logs"},
-				processors:   []string{"batch/emf_logs"},
+				processors:   []string{"batch/emf_logs", "awsentity/service/emf"},
 				exporters:    []string{"awscloudwatchlogs/emf_logs"},
 				extensions:   []string{"agenthealth/logs", "agenthealth/statuscode"},
 			},
@@ -62,7 +62,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineType: "logs/emf_logs",
 				receivers:    []string{"tcplog/emf_logs", "udplog/emf_logs"},
-				processors:   []string{"batch/emf_logs"},
+				processors:   []string{"batch/emf_logs", "awsentity/service/emf"},
 				exporters:    []string{"awscloudwatchlogs/emf_logs"},
 				extensions:   []string{"agenthealth/logs", "agenthealth/statuscode"},
 			},
@@ -80,7 +80,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineType: "logs/emf_logs",
 				receivers:    []string{"udplog/emf_logs"},
-				processors:   []string{"batch/emf_logs"},
+				processors:   []string{"batch/emf_logs", "awsentity/service/emf"},
 				exporters:    []string{"awscloudwatchlogs/emf_logs"},
 				extensions:   []string{"agenthealth/logs", "agenthealth/statuscode"},
 			},
@@ -98,7 +98,7 @@ func TestTranslator(t *testing.T) {
 			want: &want{
 				pipelineType: "logs/emf_logs",
 				receivers:    []string{"tcplog/emf_logs"},
-				processors:   []string{"batch/emf_logs"},
+				processors:   []string{"batch/emf_logs", "awsentity/service/emf"},
 				exporters:    []string{"awscloudwatchlogs/emf_logs"},
 				extensions:   []string{"agenthealth/logs", "agenthealth/statuscode"},
 			},