tracer-cloud--opensre/tests/utils/cloudwatch_logger.py

"""
CloudWatch logging utilities (stateless).

Infrastructure code for logging to AWS CloudWatch.
"""

import os
from datetime import UTC, datetime

import boto3
from botocore.config import Config
from botocore.exceptions import ClientError

from tests.utils.cloudwatch_helpers import (
    build_cloudwatch_console_url,
    verify_logs_in_cloudwatch,
)


def _log_group_prefix() -> str:
    """Return the CloudWatch log group prefix, allowing staged migration overrides."""
    value = os.getenv("CLOUDWATCH_LOG_GROUP_PREFIX", "/opensre/ai-investigations").strip()
    return value.rstrip("/") or "/opensre/ai-investigations"


def _build_logs_client(region: str):
    timeout_raw = os.getenv("CLOUDWATCH_TIMEOUT_SECONDS", "5")
    attempts_raw = os.getenv("CLOUDWATCH_MAX_ATTEMPTS", "2")
    try:
        timeout = float(timeout_raw)
    except ValueError:
        timeout = 5.0
    try:
        max_attempts = int(attempts_raw)
    except ValueError:
        max_attempts = 2

    config = Config(
        connect_timeout=timeout,
        read_timeout=timeout,
        retries={"max_attempts": max_attempts, "mode": "standard"},
    )
    return boto3.client("logs", region_name=region, config=config)


def _verify_logs_enabled() -> bool:
    value = os.getenv("CLOUDWATCH_VERIFY_LOGS", "1").strip().lower()
    return value not in {"0", "false", "no"}


def send_to_cloudwatch(
    log_group: str,
    log_stream: str,
    message: str,
    region: str = "us-east-1",
    client=None,
) -> bool:
    """Send log message to AWS CloudWatch Logs (stateless)."""
    from contextlib import suppress

    cw_client = client or _build_logs_client(region)

    try:
        with suppress(cw_client.exceptions.ResourceAlreadyExistsException):
            cw_client.create_log_group(logGroupName=log_group)

        with suppress(cw_client.exceptions.ResourceAlreadyExistsException):
            cw_client.create_log_stream(logGroupName=log_group, logStreamName=log_stream)

        timestamp_ms = int(datetime.now(UTC).timestamp() * 1000)
        cw_client.put_log_events(
            logGroupName=log_group,
            logStreamName=log_stream,
            logEvents=[{"timestamp": timestamp_ms, "message": message}],
        )
        return True
    except ClientError as err:
        error_code = err.response.get("Error", {}).get("Code", "")
        if error_code in {"AccessDenied", "AccessDeniedException", "UnauthorizedOperation"}:
            return False
        raise


def build_error_log_message(
    error_message: str,
    traceback_str: str,
    run_id: str,
    pipeline_name: str,
) -> str:
    """Build structured error log message (stateless)."""
    return f"""ERROR: {error_message}

Pipeline: {pipeline_name}
Run ID: {run_id}
Timestamp: {datetime.now(UTC).isoformat()}

Traceback:
{traceback_str}
"""


def log_error_to_cloudwatch(
    error: Exception,
    error_traceback: str,
    pipeline_name: str,
    run_id: str,
    test_name: str,
    region: str,
) -> dict:
    """
    Log error to CloudWatch (stateless).

    Args:
        error: The exception
        error_traceback: Full traceback
        pipeline_name: Pipeline name
        run_id: Run identifier
        test_name: Test/demo name (constructs log group: <prefix>/{test_name})
        region: AWS region

    Returns:
        Dict with log_group, log_stream, cloudwatch_url, error_message, logs_verified
    """
    error_message = str(error)
    log_stream = run_id
    log_group = f"{_log_group_prefix()}/{test_name}"

    log_message = build_error_log_message(
        error_message=error_message,
        traceback_str=error_traceback,
        run_id=run_id,
        pipeline_name=pipeline_name,
    )

    cw_client = _build_logs_client(region)
    logs_written = send_to_cloudwatch(log_group, log_stream, log_message, region, client=cw_client)

    logs_present = False
    if logs_written and _verify_logs_enabled():
        logs_present = verify_logs_in_cloudwatch(cw_client, log_group, log_stream)

    cloudwatch_url = build_cloudwatch_console_url(log_group, log_stream, region)

    return {
        "log_group": log_group,
        "log_stream": log_stream,
        "cloudwatch_url": cloudwatch_url,
        "error_message": error_message,
        "logs_written": logs_written,
        "logs_verified": logs_present,
    }