Extend OTLP receiver to publish via EMF logs

internal/util/collections/collections.go

@@ -61,12 +61,12 @@ func NewPair[K any, V any](key K, value V) *Pair[K, V] {
 
 // Set is a map with a comparable K key and no
 // meaningful value.
-type Set[K comparable] map[K]any
+type Set[K comparable] map[K]struct{}
 
 // Add keys to the Set.
 func (s Set[K]) Add(keys ...K) {
 	for _, key := range keys {
-		s[key] = nil
+		s[key] = struct{}{}
 	}
 }
 

translator/config/sampleSchema/validOTLPMetrics.json

@@ -16,5 +16,23 @@
         }
       ]
     }
+  },
+  "logs": {
+    "metrics_collected": {
+      "otlp": [
+        {
+          "grpc_endpoint": "0.0.0.0:1234",
+          "http_endpoint": "0.0.0.0:2345"
+        },
+        {
+          "grpc_endpoint": "0.0.0.0:3456",
+          "http_endpoint": "0.0.0.0:4567",
+          "tls": {
+            "cert_file": "/path/to/cert.pem",
+            "key_file": "/path/to/key.pem"
+          }
+        }
+      ]
+    }
   }
 }

translator/config/schema.json

@@ -802,6 +802,9 @@
                 }
               },
               "additionalProperties": false
+            },
+            "otlp": {
+              "$ref": "#/definitions/otlpDefinitions"
             }
           },
           "additionalProperties": true

translator/tocwconfig/sampleConfig/amp_config_linux.yaml

@@ -122,7 +122,7 @@ service:
   - agenthealth/metrics
   - sigv4auth
   pipelines:
-    metrics/host:
+    metrics/host/cloudwatch:
       exporters:
       - awscloudwatch
       processors:

translator/tocwconfig/sampleConfig/complete_linux_config.yaml

@@ -107,7 +107,7 @@ processors:
         send_batch_max_size: 0
         send_batch_size: 8192
         timeout: 200ms
-    cumulativetodelta/hostDeltaMetrics:
+    cumulativetodelta/hostDeltaMetrics/cloudwatch:
         exclude:
             match_type: strict
             metrics:
@@ -362,7 +362,7 @@ service:
                 - batch/emf_logs
             receivers:
                 - udplog/emf_logs
-        metrics/host:
+        metrics/host/cloudwatch:
             exporters:
                 - awscloudwatch
             processors:
@@ -378,11 +378,11 @@ service:
                 - telegraf_netstat
                 - telegraf_processes
                 - telegraf_mem
-        metrics/hostDeltaMetrics:
+        metrics/hostDeltaMetrics/cloudwatch:
             exporters:
                 - awscloudwatch
             processors:
-                - cumulativetodelta/hostDeltaMetrics
+                - cumulativetodelta/hostDeltaMetrics/cloudwatch
                 - ec2tagger
                 - transform
             receivers:
@@ -394,9 +394,9 @@ service:
             processors:
                 - filter/jmx/0
                 - resource/jmx
-                - cumulativetodelta/jmx
                 - transform/jmx/0
                 - ec2tagger
+                - cumulativetodelta/jmx
             receivers:
                 - jmx/0
         metrics/jmx/cloudwatch/1:

translator/tocwconfig/sampleConfig/jmx_config_linux.yaml

@@ -161,7 +161,7 @@ service:
         - agenthealth/metrics
         - sigv4auth
     pipelines:
-        metrics/host:
+        metrics/host/cloudwatch:
             exporters:
                 - awscloudwatch
             processors:
@@ -184,8 +184,8 @@ service:
             processors:
                 - filter/jmx
                 - resource/jmx
-                - batch/jmx/amp
                 - transform/jmx
+                - batch/jmx/amp
             receivers:
                 - jmx
         metrics/jmx/cloudwatch:

translator/tocwconfig/sampleConfig/otlp_metrics_emf_config.conf

@@ -0,0 +1,24 @@
+[agent]
+  collection_jitter = "0s"
+  debug = false
+  flush_interval = "1s"
+  flush_jitter = "0s"
+  hostname = ""
+  interval = "60s"
+  logfile = "/opt/aws/amazon-cloudwatch-agent/logs/amazon-cloudwatch-agent.log"
+  logtarget = "lumberjack"
+  metric_batch_size = 1000
+  metric_buffer_limit = 10000
+  omit_hostname = false
+  precision = ""
+  quiet = false
+  round_interval = false
+
+[inputs]
+
+[outputs]
+
+  [[outputs.cloudwatchlogs]]
+    force_flush_interval = "30s"
+    log_stream_name = "i-UNKNOWN"
+    region = "us-west-2"

translator/tocwconfig/sampleConfig/otlp_metrics_emf_config.json

@@ -0,0 +1,15 @@
+{
+  "logs": {
+    "force_flush_interval": 30,
+    "metrics_collected": {
+      "otlp": {
+        "grpc_endpoint": "0.0.0.0:1234",
+        "http_endpoint": "0.0.0.0:2345",
+        "tls": {
+          "cert_file": "/path/to/cert.pem",
+          "key_file": "/path/to/key.pem"
+        }
+      }
+    }
+  }
+}

translator/tocwconfig/sampleConfig/otlp_metrics_emf_config.yaml

@@ -0,0 +1,122 @@
+exporters:
+  awsemf:
+    certificate_file_path: ""
+    detailed_metrics: false
+    dimension_rollup_option: NoDimensionRollup
+    disable_metric_extraction: false
+    eks_fargate_container_insights_enabled: false
+    endpoint: ""
+    enhanced_container_insights: false
+    imds_retries: 1
+    local_mode: false
+    log_group_name: /aws/cwagent
+    log_retention: 0
+    log_stream_name: ""
+    max_retries: 2
+    middleware: agenthealth/logs
+    namespace: CWAgent
+    no_verify_ssl: false
+    num_workers: 8
+    output_destination: cloudwatch
+    profile: ""
+    proxy_address: ""
+    region: us-west-2
+    request_timeout_seconds: 30
+    resource_arn: ""
+    resource_to_telemetry_conversion:
+      enabled: true
+    retain_initial_value_of_delta_metric: false
+    role_arn: ""
+    version: "0"
+extensions:
+  agenthealth/logs:
+    is_usage_data_enabled: true
+    stats:
+      operations:
+        - PutLogEvents
+      usage_flags:
+        mode: EC2
+        region_type: ACJ
+processors:
+  batch/hostDeltaMetrics/emf:
+    metadata_cardinality_limit: 1000
+    send_batch_max_size: 0
+    send_batch_size: 8192
+    timeout: 30s
+  cumulativetodelta/hostDeltaMetrics/emf:
+    exclude:
+      match_type: ""
+    include:
+      match_type: ""
+    initial_value: 2
+    max_staleness: 0s
+receivers:
+  otlp/metrics:
+    protocols:
+      grpc:
+        dialer:
+          timeout: 0s
+        endpoint: 0.0.0.0:1234
+        include_metadata: false
+        max_concurrent_streams: 0
+        max_recv_msg_size_mib: 0
+        read_buffer_size: 524288
+        transport: tcp
+        write_buffer_size: 0
+        tls:
+          ca_file: ""
+          cert_file: /path/to/cert.pem
+          client_ca_file: ""
+          client_ca_file_reload: false
+          include_system_ca_certs_pool: false
+          key_file: /path/to/key.pem
+          max_version: ""
+          min_version: ""
+          reload_interval: 0s
+      http:
+        endpoint: 0.0.0.0:2345
+        include_metadata: false
+        logs_url_path: /v1/logs
+        max_request_body_size: 0
+        metrics_url_path: /v1/metrics
+        traces_url_path: /v1/traces
+        tls:
+          ca_file: ""
+          cert_file: /path/to/cert.pem
+          client_ca_file: ""
+          client_ca_file_reload: false
+          include_system_ca_certs_pool: false
+          key_file: /path/to/key.pem
+          max_version: ""
+          min_version: ""
+          reload_interval: 0s
+service:
+  extensions:
+    - agenthealth/logs
+  pipelines:
+    metrics/hostDeltaMetrics/emf:
+      exporters:
+        - awsemf
+      processors:
+        - batch/hostDeltaMetrics/emf
+        - cumulativetodelta/hostDeltaMetrics/emf
+      receivers:
+        - otlp/metrics
+  telemetry:
+    logs:
+      development: false
+      disable_caller: false
+      disable_stacktrace: false
+      encoding: console
+      level: info
+      output_paths:
+        - /opt/aws/amazon-cloudwatch-agent/logs/amazon-cloudwatch-agent.log
+      sampling:
+        enabled: true
+        initial: 2
+        thereafter: 500
+        tick: 10s
+    metrics:
+      address: ""
+      level: None
+    traces: { }

translator/tocwconfig/tocwconfig_test.go

@@ -226,6 +226,14 @@ func TestOtlpMetricsConfig(t *testing.T) {
 	checkTranslation(t, "otlp_metrics_config", "windows", nil, "")
 }
 
+func TestOtlpMetricsEmfConfig(t *testing.T) {
+	resetContext(t)
+	context.CurrentContext().SetMode(config.ModeEC2)
+	checkTranslation(t, "otlp_metrics_emf_config", "linux", nil, "")
+	checkTranslation(t, "otlp_metrics_emf_config", "darwin", nil, "")
+	checkTranslation(t, "otlp_metrics_emf_config", "windows", nil, "")
+}
+
 func TestProcstatMemorySwapConfig(t *testing.T) {
 	resetContext(t)
 	context.CurrentContext().SetRunInContainer(false)

translator/translate/otel/common/destination.go

@@ -0,0 +1,28 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package common
+
+import "go.opentelemetry.io/collector/confmap"
+
+const (
+	DefaultDestination = ""
+)
+
+var (
+	metricsDestinationsKey = ConfigKey(MetricsKey, MetricsDestinationsKey)
+)
+
+func GetMetricsDestinations(conf *confmap.Conf) []string {
+	var destinations []string
+	if conf.IsSet(ConfigKey(metricsDestinationsKey, CloudWatchKey)) {
+		destinations = append(destinations, CloudWatchKey)
+	}
+	if conf.IsSet(ConfigKey(metricsDestinationsKey, AMPKey)) {
+		destinations = append(destinations, AMPKey)
+	}
+	if conf.IsSet(MetricsKey) && len(destinations) == 0 {
+		destinations = append(destinations, DefaultDestination)
+	}
+	return destinations
+}

translator/translate/otel/common/destination_test.go

@@ -0,0 +1,68 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: MIT
+
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+func TestGetMetricsDestinations(t *testing.T) {
+	testCases := map[string]struct {
+		input map[string]any
+		want  []string
+	}{
+		"WithNoMetrics": {
+			input: map[string]any{
+				"logs": map[string]any{},
+			},
+		},
+		"WithMetrics/Default": {
+			input: map[string]any{
+				"metrics": map[string]any{},
+			},
+			want: []string{DefaultDestination},
+		},
+		"WithMetrics/AMP": {
+			input: map[string]any{
+				"metrics": map[string]any{
+					"metrics_destinations": map[string]any{
+						"amp": map[string]any{},
+					},
+				},
+			},
+			want: []string{AMPKey},
+		},
+		"WithMetrics/CloudWatch": {
+			input: map[string]any{
+				"metrics": map[string]any{
+					"metrics_destinations": map[string]any{
+						"cloudwatch": map[string]any{},
+					},
+				},
+			},
+			want: []string{CloudWatchKey},
+		},
+		"WithMetrics/CloudWatch&AMP": {
+			input: map[string]any{
+				"metrics": map[string]any{
+					"metrics_destinations": map[string]any{
+						"cloudwatch": map[string]any{},
+						"amp":        map[string]any{},
+					},
+				},
+			},
+			want: []string{CloudWatchKey, AMPKey},
+		},
+	}
+	for name, testCase := range testCases {
+		t.Run(name, func(t *testing.T) {
+			conf := confmap.NewFromStringMap(testCase.input)
+			got := GetMetricsDestinations(conf)
+			assert.Equal(t, testCase.want, got)
+		})
+	}
+}