Update SM2A workers python version to 3.11

dags/example_dag.py

@@ -5,13 +5,14 @@
 from airflow import DAG
 from airflow.operators.dummy_operator import DummyOperator as EmptyOperator
 from airflow.operators.python import PythonOperator
-
+import sys
 
 def log_task(text: str):
     logging.info(text)
 
 
 def discover_from_cmr_task(text):
+    print(f"Python version used {sys.version}")
     log_task(text)
 
 

dags/slack_notifications.py

@@ -1,10 +1,13 @@
 from airflow.contrib.operators.slack_webhook_operator import SlackWebhookOperator
+from airflow.models.variable import Variable
 
 
 def slack_alert(context, circle, status):
     slack_conn_id = "slack_connection_id"
     ti = context.get("task_instance")
     pocs = ti.dag_run.conf.get("pocs", [])
+    vars = Variable.get("aws_dags_variables", deserialize_json=True)
+    sm2a_base_url = f'https://{vars.get("SM2A_BASE_URL", "localhost:8080")}'
     slack_msg = """
             :{circle}: Task {status}. 
             *Task*: {task}
@@ -18,7 +21,7 @@ def slack_alert(context, circle, status):
         task=ti.task_id,
         dag=ti.dag_id,
         exec_date=context.get("data_interval_start"),
-        log_url=ti.log_url,
+        log_url=ti.log_url.replace("http://localhost:8080", sm2a_base_url),
     )
     if pocs:
         pocs_mentions = " ".join([f"<@{poc}>" for poc in pocs])

infrastructure/aws_data.tf

@@ -8,7 +8,7 @@ terraform {
 
 
 provider "aws" {
-  alias = "aws_current"
+  alias   = "aws_current"
   profile = var.aws_profile
   region  = var.aws_region
 }

infrastructure/main.tf

@@ -180,7 +180,7 @@ resource "aws_iam_policy" "s3_bucket_access" {
         Action = [
           "s3:*",
         ]
-        Effect   = "Allow"
+        Effect = "Allow"
         Resource = [
           "arn:aws:s3:::*",
           "arn:aws:s3:::*/*"
@@ -201,7 +201,7 @@ resource "aws_iam_role_policy_attachment" "lambda_basic_execution" {
 }
 
 resource "aws_iam_role_policy_attachment" "lambda_s3_access" {
-  role = aws_iam_role.lambda_execution_role.name
+  role       = aws_iam_role.lambda_execution_role.name
   policy_arn = aws_iam_policy.s3_bucket_access.arn
 }
 

sm2a/airflow_worker/Dockerfile

@@ -18,15 +18,24 @@ WORKDIR /opt/airflow
 RUN chown $UNAME:$GID /opt/airflow
 
 RUN apt-get -y update \
-        && apt install -y python3-pip \
         && apt-get install -y --no-install-recommends gcc libc6-dev libcurl4-openssl-dev libssl-dev \
         && apt-get autoremove -yqq --purge \
         && apt-get clean \
         && rm -rf /var/lib/apt/lists/*
 
 USER airflow
 
-ENV PATH $PATH:/home/airflow/.local/bin
+
+RUN mkdir -p /home/airflow/miniconda3 \
+    && curl -o /home/airflow/miniconda3/miniconda.sh https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh \
+    && bash /home/airflow/miniconda3/miniconda.sh -b -u -p /home/airflow/miniconda3 \
+    && rm /home/airflow/miniconda3/miniconda.sh
+
+# Install python 3.11
+RUN /home/airflow/miniconda3/bin/conda create -n py11 --yes python=3.11
+
+ENV PATH $PATH:/home/airflow/miniconda3/envs/py11/bin
+
 
 COPY --chown=airflow:airflow airflow_worker/requirements.txt "${AIRFLOW_HOME}/requirements.txt"
 

sm2a/infrastructure/main.tf

@@ -81,7 +81,7 @@ module "sma-base" {
   domain_name = var.domain_name
   stage       = var.stage
   subdomain   = var.subdomain
-  worker_cmd  = ["/home/airflow/.local/bin/airflow", "celery", "worker"]
+  worker_cmd  = ["airflow", "celery", "worker"]
 
   airflow_custom_variables = {
     EVENT_BUCKET          = var.state_bucketname
@@ -91,6 +91,7 @@ module "sma-base" {
     VECTOR_SECRET_NAME    = var.vector_secret_name
     ASSUME_ROLE_READ_ARN = var.assume_role_read_arn
     ASSUME_ROLE_WRITE_ARN = var.assume_role_write_arn
+    SM2A_BASE_URL = module.sma-base.airflow_url
   }
 }