Initial setup to trigger log upload lambda for broadcast alerts

app/celery/broadcast_message_tasks.py

@@ -5,6 +5,7 @@
 from flask import current_app
 
 from app import cbc_proxy_client, notify_celery
+from app.celery.log_ingest_tasks import request_log_ingest_task
 from app.clients.cbc_proxy import CBCProxyRetryableException
 from app.dao.broadcast_message_dao import (
     create_broadcast_provider_message,
@@ -131,6 +132,11 @@ def send_broadcast_event(broadcast_event_id):
             send_broadcast_provider_message.apply_async(
                 kwargs={"broadcast_event_id": broadcast_event_id, "provider": provider}, queue=QueueNames.HIGH_PRIORITY
             )
+
+        request_log_ingest_task.apply_async(
+            kwargs={"broadcast_event_id": broadcast_event_id}, queue=QueueNames.BROADCASTS
+        )
+
     except Exception as e:
         current_app.logger.exception(
             f"Failed to send broadcast (event id {broadcast_event_id})",

app/celery/log_ingest_tasks.py

@@ -0,0 +1,142 @@
+import json
+
+import boto3
+import botocore.exceptions
+from emergency_alerts_utils.celery import TaskNames
+from flask import current_app
+
+from app import notify_celery
+from app.models import BroadcastEvent
+
+
+@notify_celery.task(name=TaskNames.REQUEST_LOG_INGEST)
+def request_log_ingest_task(broadcast_event_id):
+    """
+    Invokes the operator portal log upload Lambda to send invite emails to MNOs.
+    """
+    try:
+        current_app.logger.info("Starting request_log_ingest_task", extra={"broadcast_event_id": broadcast_event_id})
+
+        broadcast_event = BroadcastEvent.query.get(broadcast_event_id)
+        if not broadcast_event:
+            current_app.logger.error(
+                f"Broadcast event {broadcast_event_id} not found", extra={"broadcast_event_id": broadcast_event_id}
+            )
+            return False
+
+        broadcast = broadcast_event.broadcast_message
+
+        payload = {
+            "alert_reference": broadcast.reference,
+            "environment": current_app.config.get("ENVIRONMENT"),
+            "broadcast_start": (
+                broadcast_event.transmitted_starts_at.isoformat() if broadcast_event.transmitted_starts_at else None
+            ),
+            "broadcast_end": (
+                broadcast_event.transmitted_finishes_at.isoformat() if broadcast_event.transmitted_finishes_at else None
+            ),
+            "mnos": [
+                {"mno_id": provider.upper()} for provider in broadcast_event.service.get_available_broadcast_providers()
+            ],
+        }
+
+        current_app.logger.info(
+            "Built Lambda payload", extra={"alert_reference": broadcast.reference, "payload": json.dumps(payload)}
+        )
+
+        lambda_arn = current_app.config.get("LOG_UPLOAD_LAMBDA_ARN")
+
+        current_app.logger.info(
+            "About to invoke Lambda", extra={"lambda_arn": lambda_arn, "has_lambda_arn": lambda_arn is not None}
+        )
+
+        if not lambda_arn:
+            current_app.logger.error("LOG_UPLOAD_LAMBDA_ARN not configured!")
+            return False
+
+        success = _invoke_log_upload_lambda(lambda_arn, payload)
+
+        if success:
+            current_app.logger.info(
+                "Successfully invoked log upload Lambda",
+                extra={"broadcast_reference": broadcast.reference, "broadcast_event_id": broadcast_event_id},
+            )
+        else:
+            current_app.logger.error(
+                "Failed to invoke log upload Lambda",
+                extra={"broadcast_reference": broadcast.reference, "broadcast_event_id": broadcast_event_id},
+            )
+
+        return success
+
+    except Exception as e:
+        current_app.logger.exception(
+            f"Error in request_log_ingest_task for broadcast_event {broadcast_event_id}",
+            extra={
+                "exception_type": type(e).__name__,
+                "python_module": __name__,
+                "exception": str(e),
+                "broadcast_event_id": broadcast_event_id,
+            },
+        )
+        raise
+
+
+def _invoke_log_upload_lambda(lambda_name, payload):
+    """
+    Invoke the log upload Lambda function
+    """
+    lambda_client = boto3.client("lambda", region_name="eu-west-2")
+    payload_bytes = bytes(json.dumps(payload), encoding="utf8")
+
+    try:
+        current_app.logger.info(
+            f"Calling log upload lambda {lambda_name}",
+            extra={
+                "lambda_payload": str(payload),
+                "lambda_invocation_type": "Event",
+            },
+        )
+
+        response = lambda_client.invoke(
+            FunctionName=lambda_name,
+            InvocationType="Event",
+            Payload=payload_bytes,
+        )
+
+        if response["StatusCode"] == 202:
+            current_app.logger.info(
+                f"Successfully invoked log upload lambda {lambda_name}",
+                extra={
+                    "status_code": response["StatusCode"],
+                },
+            )
+            return True
+        else:
+            current_app.logger.error(
+                f"Error invoking log upload lambda {lambda_name}",
+                extra={
+                    "status_code": response["StatusCode"],
+                },
+            )
+            return False
+
+    except botocore.exceptions.ClientError as e:
+        current_app.logger.error(
+            f"Boto3 ClientError calling log upload lambda {lambda_name}",
+            extra={
+                "python_module": __name__,
+                "error": str(e),
+            },
+        )
+        return False
+
+    except Exception as e:
+        current_app.logger.error(
+            f"Unexpected error calling log upload lambda {lambda_name}",
+            extra={
+                "python_module": __name__,
+                "error": str(e),
+            },
+        )
+        return False

app/config.py

@@ -66,6 +66,16 @@ class Config(object):
     CBC_PROXY_ENABLED = True
     ENABLED_CBCS = {BroadcastProvider.EE, BroadcastProvider.THREE, BroadcastProvider.O2, BroadcastProvider.VODAFONE}
 
+    LOG_UPLOAD_LAMBDA_ARN = os.getenv(
+        "LOG_UPLOAD_LAMBDA_ARN",
+        "arn:aws:lambda:eu-west-2:435684131547:function:mno-portal-development-log-upload-handler",
+    )
+
+    MNO_EMAILS_EE = os.getenv("MNO_EMAILS_EE", "")
+    MNO_EMAILS_O2 = os.getenv("MNO_EMAILS_O2", "")
+    MNO_EMAILS_THREE = os.getenv("MNO_EMAILS_THREE", "")
+    MNO_EMAILS_VODAFONE = os.getenv("MNO_EMAILS_VODAFONE", "")
+
     SQLALCHEMY_ENGINE_OPTIONS = {
         "pool_size": int(os.environ.get("SQLALCHEMY_POOL_SIZE", 5)),
         "pool_timeout": 30,
@@ -144,7 +154,7 @@ class Config(object):
 
     SERVICE: Literal["api", "celery"] = os.environ.get("SERVICE")
     QUEUE_NAME = QueueNames.BROADCASTS if SERVICE == "api" else QueueNames.PERIODIC
-    TASK_IMPORTS = "broadcast_message_tasks" if SERVICE == "api" else "scheduled_tasks"
+    TASK_IMPORTS = ["broadcast_message_tasks", "log_ingest_tasks"] if SERVICE == "api" else ["scheduled_tasks"]
 
     CELERY = {
         "broker_url": "filesystem://",
@@ -234,6 +244,11 @@ class Hosted(Config):
     CBC_PROXY_ENABLED = True
     DEBUG = False
 
+    LOG_UPLOAD_LAMBDA_ARN = os.getenv(
+        "LOG_UPLOAD_LAMBDA_ARN",
+        "arn:aws:lambda:eu-west-2:435684131547:function:mno-portal-development-log-upload-handler",
+    )
+
     TENANT_PREFIX = f"{os.environ.get('TENANT')}-" if os.environ.get("TENANT") is not None else ""
     ENVIRONMENT = os.getenv("ENVIRONMENT")
     ENVIRONMENT_PREFIX = ENVIRONMENT if ENVIRONMENT != "development" else "dev"
@@ -242,7 +257,7 @@ class Hosted(Config):
     SQS_QUEUE_BASE_URL = os.getenv("SQS_QUEUE_BASE_URL")
     SERVICE = os.environ.get("SERVICE")
     QUEUE_NAME = QueueNames.BROADCASTS if SERVICE == "api" else QueueNames.PERIODIC
-    TASK_IMPORTS = "broadcast_message_tasks" if SERVICE == "api" else "scheduled_tasks"
+    TASK_IMPORTS = ["broadcast_message_tasks", "log_ingest_tasks"] if SERVICE == "api" else ["scheduled_tasks"]
 
     BEAT_SCHEDULE = {
         TaskNames.RUN_HEALTH_CHECK: {
@@ -312,7 +327,7 @@ class Hosted(Config):
             "task_acks_late": True,
         },
         "timezone": "UTC",
-        "imports": [f"app.celery.{TASK_IMPORTS}"],
+        "imports": [f"app.celery.{task}" for task in TASK_IMPORTS],
         "task_queues": [Queue(QUEUE_NAME, Exchange("default"), routing_key=QUEUE_NAME)],
         "worker_max_tasks_per_child": 10,
         "beat_schedule": BEAT_SCHEDULE,
@@ -352,6 +367,8 @@ class Test(Config):
 
     GOVUK_ALERTS_S3_BUCKET_NAME = "test-govuk-alerts-bucket"
 
+    LOG_UPLOAD_LAMBDA_ARN = os.getenv("LOG_UPLOAD_LAMBDA_ARN", "")
+
 
 configs = {
     "local": Config,

tests/app/celery/test_broadcast_message_tasks.py

@@ -67,6 +67,8 @@ def test_send_broadcast_event_calls_publish_govuk_alerts_task(
         "app.celery.broadcast_message_tasks.send_broadcast_provider_message",
     )
 
+    mocker.patch("app.celery.broadcast_message_tasks.request_log_ingest_task")
+
     mock = mocker.patch("app.celery.broadcast_message_tasks.notify_celery.send_task")
 
     with set_config(notify_api, "ENABLED_CBCS", {"ee", "vodafone"}):