TG-201: Integrate Apache Airflow orchestration for background CSV reload

dags/openfoodfacts_ingestion.py

@@ -0,0 +1,126 @@
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+from airflow.providers.docker.operators.docker import DockerOperator
+from airflow.exceptions import AirflowSkipException
+from datetime import datetime, timedelta
+import os
+import hashlib
+import requests
+import urllib.request
+
+DATA_DIR = "/opt/airflow/data"
+INGEST_FILE = "en.openfoodfacts.org.products.csv"
+URL = "https://static.openfoodfacts.org/data/en.openfoodfacts.org.products.csv"
+
+default_args = {
+    'owner': 'airflow',
+    'depends_on_past': False,
+    'start_date': datetime(2026, 1, 1),
+    'email_on_failure': False,
+    'email_on_retry': False,
+    'retries': 1,
+    'retry_delay': timedelta(minutes=5),
+}
+
+dag = DAG(
+    'openfoodfacts_ingestion',
+    default_args=default_args,
+    description='Automated Data Freshness Pipeline for OpenFoodFacts',
+    schedule_interval='0 4 * * *', # Daily at 04:00
+    catchup=False,
+)
+
+def download_and_validate(**kwargs):
+    os.makedirs(DATA_DIR, exist_ok=True)
+    file_path = os.path.join(DATA_DIR, INGEST_FILE)
+    
+    print("Downloading dataset...")
+    # Downloading stream to handle large files
+    try:
+        urllib.request.urlretrieve(URL, file_path)
+    except Exception as e:
+        print(f"Failed to download: {e}")
+        raise
+    
+    print("Calculating checksum...")
+    md5_hash = hashlib.md5()
+    with open(file_path, "rb") as f:
+        for byte_block in iter(lambda: f.read(4096), b""):
+            md5_hash.update(byte_block)
+    new_checksum = md5_hash.hexdigest()
+    
+    checksum_file = os.path.join(DATA_DIR, "checksum.md5")
+    old_checksum = ""
+    if os.path.exists(checksum_file):
+        with open(checksum_file, "r") as f:
+            old_checksum = f.read().strip()
+            
+    if new_checksum == old_checksum:
+        print("Checksum matches previously processed file. Skipping ingestion.")
+        raise AirflowSkipException("Dataset is already up to date.")
+        
+    print("Checksum mismatch: File is new or modified. Ingestion required.")
+    
+    # Push new checksum to XCom so the next task can save it upon success
+    kwargs['ti'].xcom_push(key='new_checksum', value=new_checksum)
+    return True
+
+def save_checksum(**kwargs):
+    new_checksum = kwargs['ti'].xcom_pull(key='new_checksum', task_ids='validate_freshness')
+    checksum_file = os.path.join(DATA_DIR, "checksum.md5")
+    with open(checksum_file, "w") as f:
+        f.write(new_checksum)
+    print("Checksum saved successfully.")
+
+t1_validate = PythonOperator(
+    task_id='validate_freshness',
+    python_callable=download_and_validate,
+    provide_context=True,
+    dag=dag,
+)
+
+# DockerOperator requires the docker socket to be mounted to the airflow container
+# It will spawn a container using the same image as our ingest service
+t2_ingest = DockerOperator(
+    task_id='trigger_ingestion_container',
+    image='food_project-ingest',
+    api_version='auto',
+    auto_remove=True,
+    command='./ingest_csv.py /data/en.openfoodfacts.org.products.csv',
+    docker_url='unix://var/run/docker.sock',
+    network_mode='food_project_default',
+    # We must mount the local data dir into the ingest container so it can see the CSV
+    # We use the relative host path since the docker socket resolves from the host's perspective!
+    # Airflow runs in Docker, but the socket is the Host's socket.
+    mounts=[
+        # Host path -> Container path
+        # Assuming the host project is in /home/francois/food_project
+        # Note: This hardcoding is necessary when triggering sibling containers via socket
+        # unless using complex volume bindings.
+    ],
+    environment={
+        'DB_HOST': 'mysql',
+        'DB_USER': 'food_loader',
+        'DB_PASS': 'BTSai123'
+    },
+    mount_tmp_dir=False,
+    dag=dag,
+)
+
+# Because host paths can vary, it's safer to use the named volume or rely on the fact 
+# that docker-compose already created the image. 
+# Wait, the ingest image COPY . /app. So the script is already inside. 
+# But the CSV is in the host's ./data directory. 
+# To fix the host path mount dynamically without hardcoding /home/francois/food_project:
+# The DockerOperator can mount volumes like this: "food_project_data:/data" but we don't have a named volume for data.
+# Let's map it via volumes argument.
+t2_ingest.volumes = ['/home/francois/food_project/data:/data']
+
+t3_save_checksum = PythonOperator(
+    task_id='save_checksum',
+    python_callable=save_checksum,
+    provide_context=True,
+    dag=dag,
+)
+
+t1_validate >> t2_ingest >> t3_save_checksum

docker-compose.yml

@@ -0,0 +1,110 @@
+services:
+  mysql:
+    build:
+      context: ./docker/mysql
+    ports:
+      - "3307:3306"
+    volumes:
+      - mysql_data:/var/lib/mysql
+      - ./my.cnf:/etc/mysql/conf.d/custom_ai_app.cnf
+      - ./init.sql:/docker-entrypoint-initdb.d/1-init.sql
+    environment:
+      - MYSQL_ROOT_PASSWORD=${MYSQL_ROOT_PASSWORD}
+    healthcheck:
+      test: ["CMD", "mysqladmin", "ping", "-h", "localhost"]
+      interval: 10s
+      timeout: 5s
+      retries: 20
+    restart: always
+
+  ingest:
+    build:
+      context: .
+      dockerfile: docker/ingest/Dockerfile
+    environment:
+      - DB_HOST=mysql
+      - DB_USER=food_loader
+      - DB_PASS=${DB_LOADER_PASS}
+    volumes:
+      - ./:/app
+    profiles:
+      - manual
+
+  ollama:
+    image: ollama/ollama:latest
+    volumes:
+      - ollama_data:/root/.ollama
+    restart: always
+
+  searxng:
+    image: searxng/searxng:latest
+    ports:
+      - "8080:8080"
+    volumes:
+      - ./searxng:/etc/searxng
+    environment:
+      - SEARXNG_BASE_URL=http://localhost:8080/
+    restart: always
+
+  app:
+    build:
+      context: .
+      dockerfile: docker/app/Dockerfile
+    ports:
+      - "8502:8501"
+    environment:
+      - DB_HOST=mysql
+      - DB_USER=food_reader
+      - DB_PASS=${DB_READER_PASS}
+      - APP_AUTH_USER=food_app_auth
+      - APP_AUTH_PASS=${DB_APP_AUTH_PASS}
+      - OLLAMA_HOST=http://ollama:11434
+      - SEARXNG_HOST=http://searxng:8080
+    restart: always
+
+  nginx:
+    image: nginx:latest
+    ports:
+      - "80:80"
+    volumes:
+      - ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
+    restart: always
+
+  zabbix-server:
+    image: zabbix/zabbix-server-mysql:ubuntu-7.0-latest
+    environment:
+      - DB_SERVER_HOST=mysql
+      - MYSQL_USER=zabbix
+      - MYSQL_PASSWORD=${MYSQL_ZABBIX_PASSWORD}
+      - ZBX_SNMPTRAPPER=1
+    restart: always
+    ports:
+      - "10051:10051"
+
+  zabbix-web:
+    image: zabbix/zabbix-web-nginx-mysql:ubuntu-7.0-latest
+    ports:
+      - "8081:8080"
+      - "8444:8443"
+    environment:
+      - DB_SERVER_HOST=mysql
+      - MYSQL_USER=zabbix
+      - MYSQL_PASSWORD=${MYSQL_ZABBIX_PASSWORD}
+      - ZBX_SERVER_HOST=zabbix-server
+      - PHP_TZ=Europe/Paris
+    restart: always
+
+  zabbix-agent:
+    image: zabbix/zabbix-agent:ubuntu-7.0-latest
+    environment:
+      - ZBX_HOSTNAME=DistributedNode
+      - ZBX_SERVER_HOST=zabbix-server
+    privileged: true
+    pid: "host"
+    volumes:
+      - /var/run:/var/run
+    restart: always
+
+volumes:
+  mysql_data:
+  ollama_data:

scripts/setup_deploy.py

@@ -175,6 +175,37 @@ monitoring_services = """
       - /var/run:/var/run
     restart: always
 """
+airflow_services = """
+  airflow-webserver:
+    image: apache/airflow:2.8.1
+    environment:
+      - AIRFLOW__CORE__EXECUTOR=SequentialExecutor
+      - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=sqlite:////opt/airflow/airflow.db
+      - AIRFLOW__CORE__LOAD_EXAMPLES=False
+    ports:
+      - "8082:8080"
+    volumes:
+      - ./dags:/opt/airflow/dags
+      - ./logs:/opt/airflow/logs
+      - ./data:/opt/airflow/data
+      - /var/run/docker.sock:/var/run/docker.sock
+    command: webserver
+    restart: always
+
+  airflow-scheduler:
+    image: apache/airflow:2.8.1
+    environment:
+      - AIRFLOW__CORE__EXECUTOR=SequentialExecutor
+      - AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=sqlite:////opt/airflow/airflow.db
+      - AIRFLOW__CORE__LOAD_EXAMPLES=False
+    volumes:
+      - ./dags:/opt/airflow/dags
+      - ./logs:/opt/airflow/logs
+      - ./data:/opt/airflow/data
+      - /var/run/docker.sock:/var/run/docker.sock
+    command: bash -c "airflow db migrate && airflow users create --role Admin --username admin --email admin --firstname admin --lastname admin --password admin && airflow scheduler"
+    restart: always
+"""
 
 header = "services:\n"
 footer = """
@@ -186,12 +217,12 @@ volumes:
 compose_content = header
 
 if choice == "1":
-    compose_content += mysql_service + ingest_service + ai_services + app_service + monitoring_services
+    compose_content += mysql_service + ingest_service + ai_services + app_service + monitoring_services + airflow_services
 elif choice == "2":
     compose_content += ai_services + app_service
     footer = "volumes:\n  ollama_data:\n"
 elif choice == "3":
-    compose_content += mysql_service + ingest_service
+    compose_content += mysql_service + ingest_service + airflow_services
     footer = "volumes:\n  mysql_data:\n"
 elif choice == "4":
     compose_content += monitoring_services