Add OTLP logs pipeline for application signals

Author: jj22ee

translator/cmdutil/translatorutil.go

@@ -251,7 +251,64 @@ func TranslateJsonMapToYamlConfig(jsonConfigValue interface{}) (interface{}, err
 	if err != nil {
 		return nil, err
 	}
-	return mapstructure.Marshal(cfg)
+	result, err := mapstructure.Marshal(cfg)
+	if err != nil {
+		return nil, err
+	}
+	// configopaque.String values are nil'd during marshal to prevent secret leakage.
+	// Restore non-secret headers for otlphttp/application_signals exporter (log group/stream).
+	injectAppSignalsLogsHeaders(result, jsonConfigValue)
+	return result, nil
+}
+
+// injectAppSignalsLogsHeaders restores x-aws-log-group and x-aws-log-stream headers
+// on the otlphttp/application_signals exporter. These are nil'd by NilHookFunc during
+// mapstructure marshal because configopaque.String is treated as a secret, but log
+// group/stream names are not sensitive and must survive the YAML round-trip.
+func injectAppSignalsLogsHeaders(yamlMap map[string]any, jsonConfig interface{}) {
+	jsonMap, ok := jsonConfig.(map[string]interface{})
+	if !ok {
+		return
+	}
+	exporters, ok := yamlMap["exporters"].(map[string]any)
+	if !ok {
+		return
+	}
+	exporterCfg, ok := exporters["otlphttp/application_signals"].(map[string]any)
+	if !ok {
+		return
+	}
+
+	// Read log_group_name and log_stream_name from JSON config
+	logGroup := "/aws/application-signals/data"
+	logStream := "default"
+	if logs, ok := jsonMap["logs"].(map[string]interface{}); ok {
+		if lc, ok := logs["logs_collected"].(map[string]interface{}); ok {
+			if as, ok := lc["application_signals"].(map[string]interface{}); ok {
+				if v, ok := as["log_group_name"].(string); ok && v != "" {
+					logGroup = v
+				}
+				if v, ok := as["log_stream_name"].(string); ok && v != "" {
+					logStream = v
+				}
+			} else if as, ok := lc["app_signals"].(map[string]interface{}); ok {
+				if v, ok := as["log_group_name"].(string); ok && v != "" {
+					logGroup = v
+				}
+				if v, ok := as["log_stream_name"].(string); ok && v != "" {
+					logStream = v
+				}
+			}
+		}
+	}
+
+	headers, ok := exporterCfg["headers"].(map[string]any)
+	if !ok {
+		headers = make(map[string]any)
+		exporterCfg["headers"] = headers
+	}
+	headers["x-aws-log-group"] = logGroup
+	headers["x-aws-log-stream"] = logStream
 }
 
 func ConfigToTomlFile(config interface{}, tomlConfigFilePath string) error {

translator/config/schema.json

@@ -633,6 +633,34 @@
             },
             "windows_events": {
               "$ref": "#/definitions/logsDefinition/definitions/logsWindowsEventsDefinition"
+            },
+            "application_signals": {
+              "type": "object",
+              "properties": {
+                "log_group_name": {
+                  "type": "string",
+                  "minLength": 1
+                },
+                "log_stream_name": {
+                  "type": "string",
+                  "minLength": 1
+                }
+              },
+              "additionalProperties": true
+            },
+            "app_signals": {
+              "type": "object",
+              "properties": {
+                "log_group_name": {
+                  "type": "string",
+                  "minLength": 1
+                },
+                "log_stream_name": {
+                  "type": "string",
+                  "minLength": 1
+                }
+              },
+              "additionalProperties": true
             }
           },
           "minProperties": 1,

translator/translate/otel/common/common.go

@@ -136,12 +136,15 @@ const (
 var (
 	AppSignalsTraces          = ConfigKey(TracesKey, TracesCollectedKey, AppSignals)
 	AppSignalsMetrics         = ConfigKey(LogsKey, MetricsCollectedKey, AppSignals)
+	AppSignalsLogs            = ConfigKey(LogsKey, LogsCollectedKey, AppSignals)
 	AppSignalsTracesFallback  = ConfigKey(TracesKey, TracesCollectedKey, AppSignalsFallback)
 	AppSignalsMetricsFallback = ConfigKey(LogsKey, MetricsCollectedKey, AppSignalsFallback)
+	AppSignalsLogsFallback    = ConfigKey(LogsKey, LogsCollectedKey, AppSignalsFallback)
 
 	AppSignalsConfigKeys = map[pipeline.Signal][]string{
 		pipeline.SignalTraces:  {AppSignalsTraces, AppSignalsTracesFallback},
 		pipeline.SignalMetrics: {AppSignalsMetrics, AppSignalsMetricsFallback},
+		pipeline.SignalLogs:    {AppSignalsLogs, AppSignalsLogsFallback},
 	}
 	SystemMetricsEnabledConfigKey = ConfigKey(AgentKey, SystemMetricsEnabledKey)
 	JmxConfigKey                  = ConfigKey(MetricsKey, MetricsCollectedKey, JmxKey)

translator/translate/otel/exporter/otlphttp/translator.go

@@ -0,0 +1,54 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package otlphttp
+
+import (
+	"fmt"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configauth"
+	"go.opentelemetry.io/collector/confmap"
+	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/otlphttpexporter"
+
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/agent"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/common"
+)
+
+type translator struct {
+	name    string
+	factory exporter.Factory
+}
+
+var _ common.ComponentTranslator = (*translator)(nil)
+
+func NewTranslatorWithName(name string) common.ComponentTranslator {
+	return &translator{name, otlphttpexporter.NewFactory()}
+}
+
+func (t *translator) ID() component.ID {
+	return component.NewIDWithName(t.factory.Type(), t.name)
+}
+
+// Translate creates an otlphttp exporter config that sends OTLP logs to the
+// CloudWatch OTLP endpoint with SigV4 authentication.
+func (t *translator) Translate(_ *confmap.Conf) (component.Config, error) {
+	cfg := t.factory.CreateDefaultConfig().(*otlphttpexporter.Config)
+
+	region := agent.Global_Config.Region
+	if region == "" {
+		return nil, fmt.Errorf("region is required for otlphttp exporter")
+	}
+
+	cfg.ClientConfig.Endpoint = fmt.Sprintf("https://logs.%s.amazonaws.com", region)
+	cfg.ClientConfig.Auth = &configauth.Authentication{
+		AuthenticatorID: component.NewID(component.MustNewType(common.SigV4Auth)),
+	}
+
+	// Note: x-aws-log-group and x-aws-log-stream headers are injected as raw strings
+	// in translatorutil.go:injectAppSignalsLogsHeaders() because configopaque.String
+	// values are nil'd during mapstructure marshal (NilHookFunc) to prevent secret leakage.
+
+	return cfg, nil
+}

translator/translate/otel/pipeline/applicationsignals/translator.go

@@ -15,9 +15,11 @@ import (
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/exporter/awsemf"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/exporter/awsxray"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/exporter/debug"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/exporter/otlphttp"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/extension/agenthealth"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/extension/awsproxy"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/extension/k8smetadata"
+	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/extension/sigv4auth"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/awsapplicationsignals"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/awsentity"
 	"github.com/aws/amazon-cloudwatch-agent/translator/translate/otel/processor/metricstransformprocessor"
@@ -58,6 +60,20 @@ func (t *translator) Translate(conf *confmap.Conf) (*common.ComponentTranslators
 		Extensions: common.NewTranslatorMap[component.Config, component.ID](),
 	}
 
+	if t.signal == pipeline.SignalLogs {
+		// OTLP Logs pipeline: receive OTLP logs from instrumentation, forward to
+		// CloudWatch OTLP endpoint via otlphttp exporter with SigV4 auth.
+		// No processors needed — logs are forwarded as-is to preserve full OTLP structure.
+		if enabled, _ := common.GetBool(conf, common.AgentDebugConfigKey); enabled {
+			translators.Exporters.Set(debug.NewTranslator(common.WithName(common.AppSignals)))
+		}
+		translators.Exporters.Set(otlphttp.NewTranslatorWithName(common.AppSignals))
+		translators.Extensions.Set(sigv4auth.NewTranslator())
+		translators.Extensions.Set(agenthealth.NewTranslator(agenthealth.LogsName, []string{agenthealth.OperationPutLogEvents}))
+		translators.Extensions.Set(agenthealth.NewTranslatorWithStatusCode(agenthealth.StatusCodeName, nil, true))
+		return translators, nil
+	}
+
 	if t.signal == pipeline.SignalMetrics {
 		translators.Processors.Set(metricstransformprocessor.NewTranslatorWithName(common.AppSignals))
 	}

translator/translate/otel/pipeline/applicationsignals/translator_test.go

@@ -339,3 +339,77 @@ func TestTranslatorMetricsForECS(t *testing.T) {
 		})
 	}
 }
+
+func TestTranslatorLogs(t *testing.T) {
+	type want struct {
+		receivers  []string
+		processors []string
+		exporters  []string
+		extensions []string
+	}
+	tt := NewTranslator(pipeline.SignalLogs)
+	assert.EqualValues(t, "logs/application_signals", tt.ID().String())
+	testCases := map[string]struct {
+		input      map[string]interface{}
+		want       *want
+		wantErr    error
+		isEKSCache func() eksdetector.IsEKSCache
+	}{
+		"WithoutLogsCollectedKey": {
+			input:   map[string]interface{}{},
+			wantErr: &common.MissingKeyError{ID: tt.ID(), JsonKey: fmt.Sprint(common.AppSignalsLogs)},
+		},
+		"WithAppSignalsEnabledLogs": {
+			input: map[string]interface{}{
+				"logs": map[string]interface{}{
+					"logs_collected": map[string]interface{}{
+						"application_signals": map[string]interface{}{},
+					},
+				},
+			},
+			want: &want{
+				receivers:  []string{"otlp/grpc_0_0_0_0_4315", "otlp/http_0_0_0_0_4316"},
+				processors: []string{},
+				exporters:  []string{"otlphttp/application_signals"},
+				extensions: []string{"sigv4auth", "agenthealth/logs", "agenthealth/statuscode"},
+			},
+			isEKSCache: eksdetector.TestIsEKSCacheEKS,
+		},
+		"WithAppSignalsLogsAndDebug": {
+			input: map[string]interface{}{
+				"agent": map[string]interface{}{
+					"debug": true,
+				},
+				"logs": map[string]interface{}{
+					"logs_collected": map[string]interface{}{
+						"application_signals": map[string]interface{}{},
+					},
+				},
+			},
+			want: &want{
+				receivers:  []string{"otlp/grpc_0_0_0_0_4315", "otlp/http_0_0_0_0_4316"},
+				processors: []string{},
+				exporters:  []string{"debug/application_signals", "otlphttp/application_signals"},
+				extensions: []string{"sigv4auth", "agenthealth/logs", "agenthealth/statuscode"},
+			},
+			isEKSCache: eksdetector.TestIsEKSCacheEKS,
+		},
+	}
+	for name, testCase := range testCases {
+		t.Run(name, func(t *testing.T) {
+			eksdetector.IsEKS = testCase.isEKSCache
+			conf := confmap.NewFromStringMap(testCase.input)
+			got, err := tt.Translate(conf)
+			assert.Equal(t, testCase.wantErr, err)
+			if testCase.want == nil {
+				assert.Nil(t, got)
+			} else {
+				require.NotNil(t, got)
+				assert.Equal(t, testCase.want.receivers, collections.MapSlice(got.Receivers.Keys(), component.ID.String))
+				assert.Equal(t, testCase.want.processors, collections.MapSlice(got.Processors.Keys(), component.ID.String))
+				assert.Equal(t, testCase.want.exporters, collections.MapSlice(got.Exporters.Keys(), component.ID.String))
+				assert.Equal(t, testCase.want.extensions, collections.MapSlice(got.Extensions.Keys(), component.ID.String))
+			}
+		})
+	}
+}

translator/translate/otel/translate_otel.go

@@ -73,6 +73,7 @@ func Translate(jsonConfig interface{}, os string) (*otelcol.Config, error) {
 	translators.Merge(containerInsightsTranslators)
 	translators.Set(applicationsignals.NewTranslator(pipeline.SignalTraces))
 	translators.Set(applicationsignals.NewTranslator(pipeline.SignalMetrics))
+	translators.Set(applicationsignals.NewTranslator(pipeline.SignalLogs))
 	translators.Merge(prometheus.NewTranslators(conf))
 	translators.Set(emf_logs.NewTranslator())
 	translators.Set(xray.NewTranslator())