Support export OTel metrics to be EMF in starndard output (#437)

*Issue #, if available:*
Currently, OpenTelemetry (OTel) metrics users in AWS Lambda must export
metrics via synchronous calls. This often results in increased function
duration or lost metric data, due to how Lambda handles execution
lifecycle — particularly during the [freeze
phase](https://serverlessfolks.com/lambda-code-execution-freezethaw#heading-freeze).


*Description of changes:*
This PR introduces a new ConsoleEmfExporter, which exports OTel metrics
to standard output using the CloudWatch Embedded Metric Format (EMF).
In the Lambda environment, logs written to standard output are
automatically forwarded to CloudWatch by Lambda's built-in logging
agent. This makes ConsoleEmfExporter a simple and efficient way to
export OTel metrics in Lambda, avoiding the overhead and reliability
issues associated with synchronous metric export.

By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice.

---------

Co-authored-by: Jonathan Lee <[email protected]>

Author: wangzlei

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/aws_opentelemetry_configurator.py

@@ -122,7 +122,14 @@ class OtlpLogHeaderSetting(NamedTuple):
     log_group: Optional[str]
     log_stream: Optional[str]
     namespace: Optional[str]
-    is_valid: bool
+
+    def is_valid(self) -> bool:
+        """Check if the log header setting is valid by ensuring both log_group and log_stream are present."""
+        return self.log_group is not None and self.log_stream is not None
+
+
+# Singleton cache for OtlpLogHeaderSetting
+_otlp_log_header_setting_cache: Optional[OtlpLogHeaderSetting] = None
 
 
 class AwsOpenTelemetryConfigurator(_OTelSDKConfigurator):
@@ -440,7 +447,7 @@ def _customize_logs_exporter(log_exporter: LogExporter) -> LogExporter:
 
         if isinstance(log_exporter, OTLPLogExporter):
 
-            if _validate_and_fetch_logs_header().is_valid:
+            if _fetch_logs_header().is_valid():
                 endpoint, region = _extract_endpoint_and_region_from_otlp_endpoint(logs_endpoint)
                 # Setting default compression mode to Gzip as this is the behavior in upstream's
                 # collector otlp http exporter:
@@ -627,18 +634,23 @@ def _extract_endpoint_and_region_from_otlp_endpoint(endpoint: str):
     return endpoint, region
 
 
-def _validate_and_fetch_logs_header() -> OtlpLogHeaderSetting:
-    """Checks if x-aws-log-group and x-aws-log-stream are present in the headers in order to send logs to
-    AWS OTLP Logs endpoint."""
+def _fetch_logs_header() -> OtlpLogHeaderSetting:
+    """Returns the OTLP log header setting as a singleton instance."""
+    global _otlp_log_header_setting_cache  # pylint: disable=global-statement
+
+    if _otlp_log_header_setting_cache is not None:
+        return _otlp_log_header_setting_cache
 
     logs_headers = os.environ.get(OTEL_EXPORTER_OTLP_LOGS_HEADERS)
 
     if not logs_headers:
-        _logger.warning(
-            "Improper configuration: Please configure the environment variable OTEL_EXPORTER_OTLP_LOGS_HEADERS "
-            "to include x-aws-log-group and x-aws-log-stream"
-        )
-        return OtlpLogHeaderSetting(None, None, None, False)
+        if not _is_lambda_environment():
+            _logger.warning(
+                "Improper configuration: Please configure the environment variable OTEL_EXPORTER_OTLP_LOGS_HEADERS "
+                "to include x-aws-log-group and x-aws-log-stream"
+            )
+        _otlp_log_header_setting_cache = OtlpLogHeaderSetting(None, None, None)
+        return _otlp_log_header_setting_cache
 
     log_group = None
     log_stream = None
@@ -656,9 +668,14 @@ def _validate_and_fetch_logs_header() -> OtlpLogHeaderSetting:
             elif key == AWS_EMF_METRICS_NAMESPACE and value:
                 namespace = value
 
-    is_valid = log_group is not None and log_stream is not None
+    _otlp_log_header_setting_cache = OtlpLogHeaderSetting(log_group, log_stream, namespace)
+    return _otlp_log_header_setting_cache
+
 
-    return OtlpLogHeaderSetting(log_group, log_stream, namespace, is_valid)
+def _clear_logs_header_cache():
+    """Clear the singleton cache for OtlpLogHeaderSetting. Used primarily for testing."""
+    global _otlp_log_header_setting_cache  # pylint: disable=global-statement
+    _otlp_log_header_setting_cache = None
 
 
 def _get_metric_export_interval():
@@ -773,8 +790,25 @@ def _check_emf_exporter_enabled() -> bool:
 
 
 def _create_emf_exporter():
-    """Create and configure the CloudWatch EMF exporter."""
+    """
+    Create the appropriate EMF exporter based on the environment and configuration.
+
+    Returns:
+        ConsoleEmfExporter for Lambda without log headers log group and stream
+        AwsCloudWatchEmfExporter for other cases (when conditions are met)
+        None if CloudWatch exporter cannot be created
+    """
     try:
+        log_header_setting = _fetch_logs_header()
+
+        # Lambda without valid logs http headers - use Console EMF exporter
+        if _is_lambda_environment() and not log_header_setting.is_valid():
+            # pylint: disable=import-outside-toplevel
+            from amazon.opentelemetry.distro.exporter.aws.metrics.console_emf_exporter import ConsoleEmfExporter
+
+            return ConsoleEmfExporter(namespace=log_header_setting.namespace)
+
+        # For non-Lambda environment or Lambda with valid headers - use CloudWatch EMF exporter
         session = get_aws_session()
         # Check if botocore is available before importing the EMF exporter
         if not session:
@@ -786,9 +820,7 @@ def _create_emf_exporter():
             AwsCloudWatchEmfExporter,
         )
 
-        log_header_setting = _validate_and_fetch_logs_header()
-
-        if not log_header_setting.is_valid:
+        if not log_header_setting.is_valid():
             return None
 
         return AwsCloudWatchEmfExporter(