add aws cloudwatch logs to airflow (#757)

* add aws cloudwatch logs to airflow

* PR commnet

* lint

Author: peterdudfield

terraform/modules/services/airflow/dags/india/forecast-site-dag.py

@@ -7,37 +7,43 @@
 from airflow.operators.latest_only import LatestOnlyOperator
 
 default_args = {
-    'owner': 'airflow',
-    'depends_on_past': False,
-    'start_date': datetime.now(tz=timezone.utc) - timedelta(hours=3),
-    'retries': 1,
-    'retry_delay': timedelta(minutes=1),
-    'max_active_runs':10,
-    'concurrency':10,
-    'max_active_tasks':10,
+    "owner": "airflow",
+    "depends_on_past": False,
+    "start_date": datetime.now(tz=timezone.utc) - timedelta(hours=3),
+    "retries": 1,
+    "retry_delay": timedelta(minutes=1),
+    "max_active_runs": 10,
+    "concurrency": 10,
+    "max_active_tasks": 10,
 }
 
 env = os.getenv("ENVIRONMENT", "development")
 subnet = os.getenv("ECS_SUBNET")
 security_group = os.getenv("ECS_SECURITY_GROUP")
 cluster = f"india-ecs-cluster-{env}"
 
-region = 'india'
+region = "india"
 
 # hour the forecast can run, not include 7,8,19,20
-hours = '0,1,2,3,4,5,6,9,10,11,12,13,14,15,16,17,18,21,22,23'
+hours = "0,1,2,3,4,5,6,9,10,11,12,13,14,15,16,17,18,21,22,23"
 
-with DAG(f'{region}-runvl-forecast', schedule_interval=f"0 {hours} * * *", default_args=default_args, concurrency=10, max_active_tasks=10) as dag:
+with DAG(
+    f"{region}-runvl-forecast",
+    schedule_interval=f"0 {hours} * * *",
+    default_args=default_args,
+    concurrency=10,
+    max_active_tasks=10,
+) as dag:
     dag.doc_md = "Run the forecast"
 
     latest_only = LatestOnlyOperator(task_id="latest_only")
 
     forecast = EcsRunTaskOperator(
-        task_id=f'{region}-forecast-ruvnl',
-        task_definition='forecast',
+        task_id=f"{region}-forecast-ruvnl",
+        task_definition="forecast",
         cluster=cluster,
         overrides={},
-        launch_type = "FARGATE",
+        launch_type="FARGATE",
         network_configuration={
             "awsvpcConfiguration": {
                 "subnets": [subnet],
@@ -46,22 +52,31 @@
             },
         },
         on_failure_callback=on_failure_callback,
-     task_concurrency = 10,
+        task_concurrency=10,
+        awslogs_group="/aws/ecs/forecast/forecast",
+        awslogs_stream_prefix="streaming/forecast-forecast",
+        awslogs_region="ap-south-1",
     )
 
     latest_only >> [forecast]
 
-with DAG(f'{region}-ad-forecast', schedule_interval=f"0,15,30,45 * * * *", default_args=default_args, concurrency=10, max_active_tasks=10) as dag:
+with DAG(
+    f"{region}-ad-forecast",
+    schedule_interval=f"0,15,30,45 * * * *",
+    default_args=default_args,
+    concurrency=10,
+    max_active_tasks=10,
+) as dag:
     dag.doc_md = "Run the forecast for client AD"
 
     latest_only = LatestOnlyOperator(task_id="latest_only")
 
     forecast = EcsRunTaskOperator(
-        task_id=f'{region}-forecast-ad',
-        task_definition='forecast-ad',
+        task_id=f"{region}-forecast-ad",
+        task_definition="forecast-ad",
         cluster=cluster,
         overrides={},
-        launch_type = "FARGATE",
+        launch_type="FARGATE",
         network_configuration={
             "awsvpcConfiguration": {
                 "subnets": [subnet],
@@ -70,7 +85,10 @@
             },
         },
         on_failure_callback=on_failure_callback,
-     task_concurrency = 10,
+        task_concurrency=10,
+        awslogs_group="/aws/ecs/forecast/forecast-ad",
+        awslogs_stream_prefix="streaming/forecast-ad-forecast",
+        awslogs_region="ap-south-1",
     )
 
     latest_only >> [forecast]

terraform/modules/services/airflow/dags/india/nwp-dag.py

@@ -8,25 +8,25 @@
 from utils.s3 import determine_latest_zarr
 
 default_args = {
-    'owner': 'airflow',
-    'depends_on_past': False,
-    'start_date': datetime.now(tz=timezone.utc) - timedelta(hours=1.5),
-    'retries': 1,
-    'retry_delay': timedelta(minutes=1),
-    'max_active_runs':10,
-    'concurrency':10,
-    'max_active_tasks':10,
+    "owner": "airflow",
+    "depends_on_past": False,
+    "start_date": datetime.now(tz=timezone.utc) - timedelta(hours=1.5),
+    "retries": 1,
+    "retry_delay": timedelta(minutes=1),
+    "max_active_runs": 10,
+    "concurrency": 10,
+    "max_active_tasks": 10,
 }
 
 env = os.getenv("ENVIRONMENT", "development")
 subnet = os.getenv("ECS_SUBNET")
 security_group = os.getenv("ECS_SECURITY_GROUP")
 cluster = f"india-ecs-cluster-{env}"
 
-region = 'india'
+region = "india"
 
 with DAG(
-    f'{region}-nwp-consumer',
+    f"{region}-nwp-consumer",
     schedule_interval="0 * * * *",
     default_args=default_args,
     concurrency=10,
@@ -37,40 +37,46 @@
     latest_only = LatestOnlyOperator(task_id="latest_only")
 
     nwp_consumer_ecmwf = EcsRunTaskOperator(
-         task_id=f'{region}-nwp-consumer-ecmwf-india',
-         task_definition='nwp-consumer-ecmwf-india',
-         cluster=cluster,
-         overrides={},
-         launch_type="FARGATE",
-         network_configuration={
-             "awsvpcConfiguration": {
-                 "subnets": [subnet],
-                 "securityGroups": [security_group],
-                 "assignPublicIp": "ENABLED",
-             },
-         },
-         task_concurrency=10,
+        task_id=f"{region}-nwp-consumer-ecmwf-india",
+        task_definition="nwp-consumer-ecmwf-india",
+        cluster=cluster,
+        overrides={},
+        launch_type="FARGATE",
+        network_configuration={
+            "awsvpcConfiguration": {
+                "subnets": [subnet],
+                "securityGroups": [security_group],
+                "assignPublicIp": "ENABLED",
+            },
+        },
+        task_concurrency=10,
+        awslogs_group="/aws/ecs/consumer/nwp-consumer-ecmwf-india",
+        awslogs_stream_prefix="streaming/nwp-consumer-ecmwf-india-consumer",
+        awslogs_region="ap-south-1",
     )
 
     nwp_consumer_gfs = EcsRunTaskOperator(
-         task_id=f'{region}-nwp-consumer-gfs-india',
-         task_definition='nwp-consumer-gfs-india',
-         cluster=cluster,
-         overrides={},
-         launch_type="FARGATE",
-         network_configuration={
-             "awsvpcConfiguration": {
-                 "subnets": [subnet],
-                 "securityGroups": [security_group],
-                 "assignPublicIp": "ENABLED",
-             },
-         },
-         task_concurrency=10,
+        task_id=f"{region}-nwp-consumer-gfs-india",
+        task_definition="nwp-consumer-gfs-india",
+        cluster=cluster,
+        overrides={},
+        launch_type="FARGATE",
+        network_configuration={
+            "awsvpcConfiguration": {
+                "subnets": [subnet],
+                "securityGroups": [security_group],
+                "assignPublicIp": "ENABLED",
+            },
+        },
+        task_concurrency=10,
+        awslogs_group="/aws/ecs/consumer/nwp-consumer-gfs-india",
+        awslogs_stream_prefix="streaming/nwp-consumer-gfs-india-consumer",
+        awslogs_region="ap-south-1",
     )
 
     nwp_consumer_metoffice = EcsRunTaskOperator(
-        task_id=f'{region}-nwp-consumer-metoffice-india',
-        task_definition='nwp-consumer-metoffice-india',
+        task_id=f"{region}-nwp-consumer-metoffice-india",
+        task_definition="nwp-consumer-metoffice-india",
         cluster=cluster,
         overrides={},
         launch_type="FARGATE",
@@ -82,20 +88,22 @@
             },
         },
         task_concurrency=10,
+        awslogs_group="/aws/ecs/consumer/nwp-consumer-metoffice-india",
+        awslogs_stream_prefix="streaming/nwp-consumer-metoffice-india-consumer",
+        awslogs_region="ap-south-1",
     )
     rename_zarr_metoffice = determine_latest_zarr.override(
-            task_id="determine_latest_zarr_metoffice",
+        task_id="determine_latest_zarr_metoffice",
     )(bucket=f"india-nwp-{env}", prefix="metoffice/data")
 
     rename_zarr_ecmwf = determine_latest_zarr.override(
         task_id="determine_latest_zarr_ecmwf",
-    )(bucket=f'india-nwp-{env}', prefix='ecmwf/data')
+    )(bucket=f"india-nwp-{env}", prefix="ecmwf/data")
 
     rename_zarr_gfs = determine_latest_zarr.override(
         task_id="determine_latest_zarr_gfs",
-    )(bucket=f'india-nwp-{env}', prefix='gfs/data')
+    )(bucket=f"india-nwp-{env}", prefix="gfs/data")
 
     latest_only >> nwp_consumer_ecmwf >> rename_zarr_ecmwf
     latest_only >> nwp_consumer_gfs >> rename_zarr_gfs
     latest_only >> nwp_consumer_metoffice >> rename_zarr_metoffice
-

terraform/modules/services/airflow/dags/india/runvnl-data-dag.py

@@ -22,10 +22,10 @@
 security_group = os.getenv("ECS_SECURITY_GROUP")
 cluster = f"india-ecs-cluster-{env}"
 
-region = 'india'
+region = "india"
 
 with DAG(
-    f'{region}-runvl-data-consumer',
+    f"{region}-runvl-data-consumer",
     schedule_interval="*/3 * * * *",
     default_args=default_args,
     concurrency=10,
@@ -36,8 +36,8 @@
     latest_only = LatestOnlyOperator(task_id="latest_only")
 
     runvl_data = EcsRunTaskOperator(
-        task_id=f'{region}-runvl-consumer',
-        task_definition='runvl-consumer',
+        task_id=f"{region}-runvl-consumer",
+        task_definition="runvl-consumer",
         cluster=cluster,
         overrides={},
         launch_type="FARGATE",
@@ -50,8 +50,9 @@
         },
         on_failure_callback=on_failure_callback,
         task_concurrency=10,
+        awslogs_group="/aws/ecs/consumer/runvl-consumer",
+        awslogs_stream_prefix="streaming/runvl-consumer-consumer",
+        awslogs_region="ap-south-1",
     )
 
-
     latest_only >> [runvl_data]
-

terraform/modules/services/airflow/dags/india/satellite-dag.py

@@ -14,7 +14,7 @@
     "max_active_runs": 10,
     "concurrency": 10,
     "max_active_tasks": 10,
-    "execution_timeout":timedelta(minutes=30),
+    "execution_timeout": timedelta(minutes=30),
 }
 
 env = os.getenv("ENVIRONMENT", "development")
@@ -24,10 +24,10 @@
 
 # Tasks can still be defined in terraform, or defined here
 
-region = 'india'
+region = "india"
 
 with DAG(
-    f'{region}-satellite-consumer',
+    f"{region}-satellite-consumer",
     schedule_interval="*/5 * * * *",
     default_args=default_args,
     concurrency=10,
@@ -39,8 +39,8 @@
     latest_only = LatestOnlyOperator(task_id="latest_only")
 
     sat_consumer = EcsRunTaskOperator(
-        task_id=f'{region}-satellite-consumer',
-        task_definition='sat-consumer',
+        task_id=f"{region}-satellite-consumer",
+        task_definition="sat-consumer",
         cluster=cluster,
         overrides={},
         launch_type="FARGATE",
@@ -52,7 +52,10 @@
             },
         },
         task_concurrency=10,
-        on_failure_callback=on_failure_callback
+        on_failure_callback=on_failure_callback,
+        awslogs_group="/aws/ecs/consumer/sat-consumer",
+        awslogs_stream_prefix="streaming/sat-consumer-consumer",
+        awslogs_region="ap-south-1",
     )
 
     latest_only >> sat_consumer