Upload 7 files

Dockerfile

@@ -1,37 +1,68 @@
 FROM ubuntu:22.04
 
+LABEL maintainer="subhamgiri460"
+LABEL description="Apache Airflow on HuggingFace Docker Space"
+
 ENV DEBIAN_FRONTEND=noninteractive
 ENV AIRFLOW_HOME=/opt/airflow
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
 
+# ---- System packages ----
 RUN apt-get update && apt-get install -y --no-install-recommends \
-    python3 \
-    python3-pip \
-    python3-venv \
-    nginx \
-    git \
-    supervisor \
-    curl \
+        python3 \
+        python3-pip \
+        python3-venv \
+        nginx \
+        git \
+        supervisor \
+        curl \
+        libpq-dev \
     && rm -rf /var/lib/apt/lists/* \
     && useradd -m -u 1000 airflow
 
-RUN mkdir -p /var/log/supervisor /var/log/nginx /run/nginx /var/lib/nginx /opt/airflow/dags \
-    && chown -R airflow:airflow /opt/airflow /var/log/supervisor /var/log/nginx /run/nginx /var/lib/nginx
+# ---- Directories & permissions ----
+RUN mkdir -p \
+        /var/log/supervisor \
+        /var/log/nginx \
+        /run/nginx \
+        /var/lib/nginx \
+        /run/nginx/client_temp \
+        /run/nginx/proxy_temp \
+        /run/nginx/fastcgi_temp \
+        /run/nginx/uwsgi_temp \
+        /run/nginx/scgi_temp \
+        /opt/airflow/dags \
+        /opt/airflow/logs \
+    && chown -R airflow:airflow \
+        /opt/airflow \
+        /var/log/supervisor \
+        /var/log/nginx \
+        /run/nginx \
+        /var/lib/nginx
 
 USER airflow
-
 ENV PATH="/home/airflow/.local/bin:${PATH}"
 
+# ---- Python dependencies (cached layer) ----
 COPY --chown=airflow:airflow requirements.txt /tmp/requirements.txt
-RUN pip install --no-cache-dir -r /tmp/requirements.txt
+RUN pip install --no-cache-dir --upgrade pip \
+    && pip install --no-cache-dir -r /tmp/requirements.txt \
+    && rm /tmp/requirements.txt
 
-COPY --chown=airflow:airflow config.sh /config.sh
-COPY --chown=airflow:airflow supervisord.conf /etc/supervisord.conf
-COPY --chown=airflow:airflow nginx.conf /etc/nginx/nginx.conf
-COPY --chown=airflow:airflow entrypoint.sh /entrypoint.sh
-COPY --chown=airflow:airflow refresh_service.py /refresh_service.py
+# ---- Application files ----
+COPY --chown=airflow:airflow config.sh          /config.sh
+COPY --chown=airflow:airflow supervisord.conf    /etc/supervisord.conf
+COPY --chown=airflow:airflow nginx.conf          /etc/nginx/nginx.conf
+COPY --chown=airflow:airflow entrypoint.sh       /entrypoint.sh
+COPY --chown=airflow:airflow refresh_service.py  /refresh_service.py
 
 RUN chmod +x /entrypoint.sh /config.sh
 
+# ---- Health check ----
+HEALTHCHECK --interval=30s --timeout=10s --start-period=120s --retries=3 \
+    CMD curl -sf http://localhost:7860/health || exit 1
+
 EXPOSE 7860
 
 ENTRYPOINT ["/bin/bash", "/entrypoint.sh"]

config.sh

@@ -1,28 +1,36 @@
 #!/bin/bash
 # ===========================================
-# Airflow Configuration - SINGLE SOURCE OF TRUTH
+# Airflow Configuration — Single Source of Truth
 # ===========================================
-# All configurable values are defined here with defaults
-# Override by setting environment variables or HF Secrets
+# Override any value by setting the corresponding environment
+# variable (e.g. as an HF Secret).
 
 # ---- Directory Paths ----
 export AIRFLOW_HOME="${AIRFLOW_HOME:-/opt/airflow}"
 export DAGS_DIR="${AIRFLOW_HOME}/dags"
 export LOGS_DIR="${AIRFLOW_HOME}/logs"
 
-# ---- Database Connection (Set via HF Secret for production) ----
-export AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="${AIRFLOW__DATABASE__SQL_ALCHEMY_CONN:-postgresql://postgres.ffetkispaxytayrnmxvj:uKInpRkBXsCUfVg1@aws-1-ap-south-1.pooler.supabase.com:5432/postgres}"
+# ---- Database ----
+# Default: local SQLite so the container always boots.
+# For production, set AIRFLOW__DATABASE__SQL_ALCHEMY_CONN as an HF Secret
+# pointing to your PostgreSQL instance.
+export AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="${AIRFLOW__DATABASE__SQL_ALCHEMY_CONN:-sqlite:///${AIRFLOW_HOME}/airflow.db}"
 
-# Prevent DB Connection Exhaustion on Hugging Face / Supabase Free Tier
-export AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_SIZE="1"
-export AIRFLOW__DATABASE__SQL_ALCHEMY_MAX_OVERFLOW="1"
-export AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_RECYCLE="1800"
+# Pool tuning — keeps connections low for Supabase / free-tier DBs
+export AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_SIZE="${AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_SIZE:-2}"
+export AIRFLOW__DATABASE__SQL_ALCHEMY_MAX_OVERFLOW="${AIRFLOW__DATABASE__SQL_ALCHEMY_MAX_OVERFLOW:-2}"
+export AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_RECYCLE="${AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_RECYCLE:-1800}"
+export AIRFLOW__DATABASE__SQL_ALCHEMY_POOL_PRE_PING="True"
 
 # ---- DAG Repository ----
 export DAG_REPO_URL="${DAG_REPO_URL:-https://github.com/subhamgiri460/myworkflows.git}"
 export DAG_REPO_BRANCH="${DAG_REPO_BRANCH:-main}"
+# For private repos: set DAG_REPO_TOKEN to a GitHub PAT (fine-grained or classic)
+# The token is injected into the URL at runtime — never stored in config.
+export DAG_REPO_TOKEN="${DAG_REPO_TOKEN:-}"
 
-# ---- Admin Credentials (Set via HF Secret for production) ----
+# ---- Admin Credentials ----
+# Set these as HF Secrets for production.
 export AIRFLOW_ADMIN_USER="${AIRFLOW_ADMIN_USER:-admin}"
 export AIRFLOW_ADMIN_PASSWORD="${AIRFLOW_ADMIN_PASSWORD:-admin}"
 
@@ -31,18 +39,49 @@ export AIRFLOW_WEB_PORT="${AIRFLOW_WEB_PORT:-8080}"
 export REFRESH_SERVICE_PORT="${REFRESH_SERVICE_PORT:-5000}"
 export NGINX_PORT="${NGINX_PORT:-7860}"
 
-# ---- Airflow Core Settings ----
+# ---- Core ----
 export AIRFLOW__CORE__LOAD_EXAMPLES="${AIRFLOW__CORE__LOAD_EXAMPLES:-False}"
 export AIRFLOW__CORE__EXECUTOR="${AIRFLOW__CORE__EXECUTOR:-LocalExecutor}"
-export AIRFLOW__CORE__PARALLELISM="2"
-export AIRFLOW__CORE__MAX_ACTIVE_TASKS_PER_DAG="2"
+export AIRFLOW__CORE__PARALLELISM="${AIRFLOW__CORE__PARALLELISM:-2}"
+export AIRFLOW__CORE__MAX_ACTIVE_TASKS_PER_DAG="${AIRFLOW__CORE__MAX_ACTIVE_TASKS_PER_DAG:-2}"
+export AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION="${AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION:-True}"
 
-# ---- Airflow Webserver Settings ----
+# Auto-generate Fernet key if not set (required for connection encryption)
+if [ -z "${AIRFLOW__CORE__FERNET_KEY:-}" ]; then
+    export AIRFLOW__CORE__FERNET_KEY=$(python3 -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" 2>/dev/null || echo "")
+fi
+
+# ---- Webserver ----
 export AIRFLOW__WEBSERVER__WEB_SERVER_PORT="${AIRFLOW_WEB_PORT}"
 export AIRFLOW__WEBSERVER__WEB_SERVER_MASTER_TIMEOUT="${AIRFLOW__WEBSERVER__WEB_SERVER_MASTER_TIMEOUT:-300}"
 export AIRFLOW__WEBSERVER__WORKER_CLASS="${AIRFLOW__WEBSERVER__WORKER_CLASS:-gevent}"
-export AIRFLOW__WEBSERVER__SECRET_KEY="${AIRFLOW__WEBSERVER__SECRET_KEY:-hf-spaces-default-secret-key-12345}"
-export AIRFLOW__WEBSERVER__WORKERS="1"
+export AIRFLOW__WEBSERVER__SECRET_KEY="${AIRFLOW__WEBSERVER__SECRET_KEY:-$(python3 -c 'import secrets; print(secrets.token_hex(32))' 2>/dev/null)}"
+export AIRFLOW__WEBSERVER__WORKERS="${AIRFLOW__WEBSERVER__WORKERS:-1}"
+export AIRFLOW__WEBSERVER__EXPOSE_CONFIG="${AIRFLOW__WEBSERVER__EXPOSE_CONFIG:-False}"
+
+# ---- Scheduler ----
+export AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL="${AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL:-30}"
+
+# ---- API ----
+export AIRFLOW__API__AUTH_BACKENDS="${AIRFLOW__API__AUTH_BACKENDS:-airflow.api.auth.backend.session}"
+
+# ---- Logging ----
+export AIRFLOW__LOGGING__LOGGING_LEVEL="${AIRFLOW__LOGGING__LOGGING_LEVEL:-INFO}"
 
-# ---- Airflow Scheduler Settings ----
-export AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL="${AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL:-30}"
+# ---- Validation ----
+validate_config() {
+    local warn=0
+    if [[ "${AIRFLOW__DATABASE__SQL_ALCHEMY_CONN}" == sqlite* ]]; then
+        echo "[WARN] Using SQLite — not suitable for production. Set AIRFLOW__DATABASE__SQL_ALCHEMY_CONN as an HF Secret."
+        # SQLite requires SequentialExecutor
+        export AIRFLOW__CORE__EXECUTOR="SequentialExecutor"
+        warn=1
+    fi
+    if [ "${AIRFLOW_ADMIN_PASSWORD}" = "admin" ]; then
+        echo "[WARN] Using default admin password. Set AIRFLOW_ADMIN_PASSWORD as an HF Secret."
+        warn=1
+    fi
+    if [ "${warn}" -eq 0 ]; then
+        echo "[OK] Configuration validated."
+    fi
+}

entrypoint.sh

@@ -1,46 +1,138 @@
 #!/bin/bash
-set -e
+set -euo pipefail
 
-echo "=== Loading configuration ==="
+# ---- Helpers ----
+log() { echo "[$(date '+%Y-%m-%d %H:%M:%S')] $*"; }
+
+# Graceful shutdown: forward signals to child processes
+trap 'log "Caught signal, shutting down..."; kill -TERM $(jobs -p) 2>/dev/null; exit 0' SIGTERM SIGINT SIGQUIT
+
+# ---- Load Configuration ----
+log "Loading configuration..."
 source /config.sh
+validate_config
+
+log "Configuration:"
+log "  DB backend : $(echo "${AIRFLOW__DATABASE__SQL_ALCHEMY_CONN}" | sed -E 's|://[^@]*@|://***:***@|')"
+log "  DAG repo   : ${DAG_REPO_URL} (branch: ${DAG_REPO_BRANCH})"
+log "  Executor   : ${AIRFLOW__CORE__EXECUTOR}"
+log "  Ports      : nginx=${NGINX_PORT}  webserver=${AIRFLOW_WEB_PORT}  refresh=${REFRESH_SERVICE_PORT}"
+
+# ---- Ensure directories ----
+mkdir -p "${DAGS_DIR}" "${LOGS_DIR}"
 
-echo "Configuration:"
-echo "  DAG_REPO_URL: ${DAG_REPO_URL}"
-echo "  DAG_REPO_BRANCH: ${DAG_REPO_BRANCH}"
-echo "  DAGS_DIR: ${DAGS_DIR}"
-echo "  AIRFLOW_WEB_PORT: ${AIRFLOW_WEB_PORT}"
-echo "  REFRESH_SERVICE_PORT: ${REFRESH_SERVICE_PORT}"
+# ---- Wait for Database (if PostgreSQL) ----
+if [[ "${AIRFLOW__DATABASE__SQL_ALCHEMY_CONN}" == postgresql* ]]; then
+    DB_WAIT_TIMEOUT="${DB_WAIT_TIMEOUT:-60}"
+    log "Waiting for PostgreSQL (timeout: ${DB_WAIT_TIMEOUT}s)..."
+    elapsed=0
+    while [ "${elapsed}" -lt "${DB_WAIT_TIMEOUT}" ]; do
+        if python3 -c "
+import sqlalchemy, sys
+try:
+    engine = sqlalchemy.create_engine('${AIRFLOW__DATABASE__SQL_ALCHEMY_CONN}', connect_args={'connect_timeout': 5})
+    with engine.connect() as conn:
+        conn.execute(sqlalchemy.text('SELECT 1'))
+    sys.exit(0)
+except Exception:
+    sys.exit(1)
+" 2>/dev/null; then
+            log "PostgreSQL is ready."
+            break
+        fi
+        sleep 3
+        elapsed=$((elapsed + 3))
+    done
+    if [ "${elapsed}" -ge "${DB_WAIT_TIMEOUT}" ]; then
+        log "[ERROR] PostgreSQL not reachable after ${DB_WAIT_TIMEOUT}s."
+        log "Falling back to SQLite so the container still boots."
+        export AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="sqlite:///${AIRFLOW_HOME}/airflow.db"
+        export AIRFLOW__CORE__EXECUTOR="SequentialExecutor"
+    fi
+fi
+
+# ---- Build authenticated repo URL (if PAT is set) ----
+GIT_CLONE_URL="${DAG_REPO_URL}"
+if [ -n "${DAG_REPO_TOKEN}" ]; then
+    # Inject token: https://github.com/... → https://<token>@github.com/...
+    GIT_CLONE_URL=$(echo "${DAG_REPO_URL}" | sed "s|https://|https://${DAG_REPO_TOKEN}@|")
+    log "Using authenticated URL for DAG repo (PAT token set)."
+fi
 
-echo ""
-echo "=== Syncing DAG repository ==="
+# ---- Sync DAG Repository ----
+_create_sample_dag() {
+    log "Creating sample DAG for testing..."
+    cat > "${DAGS_DIR}/sample_dag.py" << 'SAMPLE_EOF'
+"""
+Sample DAG — created automatically because no DAG repository was configured.
+Replace this by setting DAG_REPO_URL or adding your own DAGs to /opt/airflow/dags.
+"""
+from datetime import datetime, timedelta
+from airflow import DAG
+from airflow.operators.bash import BashOperator
+
+with DAG(
+    dag_id="sample_hello_world",
+    description="A sample DAG to verify Airflow is working",
+    schedule=timedelta(hours=1),
+    start_date=datetime(2024, 1, 1),
+    catchup=False,
+    tags=["sample", "test"],
+) as dag:
+    hello = BashOperator(
+        task_id="say_hello",
+        bash_command='echo "Hello from Airflow! $(date)"',
+    )
+    check_env = BashOperator(
+        task_id="check_environment",
+        bash_command='echo "Python: $(python3 --version)" && echo "Airflow Home: $AIRFLOW_HOME"',
+    )
+    hello >> check_env
+SAMPLE_EOF
+    log "Sample DAG created at ${DAGS_DIR}/sample_dag.py"
+}
+
+log "Syncing DAG repository..."
 if [ ! -d "${DAGS_DIR}/.git" ]; then
+    # Fresh clone — clean any stale files (except sample_dag) first
     rm -rf "${DAGS_DIR:?}"/* 2>/dev/null || true
-    git clone --depth 1 --branch "${DAG_REPO_BRANCH}" "${DAG_REPO_URL}" "${DAGS_DIR}"
-    echo "Repository cloned successfully"
+    if git clone --depth 1 --branch "${DAG_REPO_BRANCH}" "${GIT_CLONE_URL}" "${DAGS_DIR}" 2>/dev/null; then
+        log "DAG repository cloned."
+    else
+        log "[WARN] Could not clone DAG repo (URL unreachable or private without token)."
+        _create_sample_dag
+    fi
 else
     cd "${DAGS_DIR}"
-    git fetch origin "${DAG_REPO_BRANCH}" && git reset --hard "origin/${DAG_REPO_BRANCH}"
-    echo "Repository updated successfully"
+    # If PAT changed, update the remote URL
+    git remote set-url origin "${GIT_CLONE_URL}" 2>/dev/null || true
+    if git fetch origin "${DAG_REPO_BRANCH}" 2>/dev/null && git reset --hard "origin/${DAG_REPO_BRANCH}" 2>/dev/null; then
+        log "DAG repository updated."
+    else
+        log "[WARN] Could not update DAG repo. Using existing DAGs."
+    fi
+fi
+
+# ---- Database Migrations ----
+log "Running database migrations..."
+airflow db migrate 2>&1 | tail -5
+log "Migrations complete."
+
+# ---- Create Admin User (idempotent) ----
+log "Ensuring admin user exists..."
+if ! airflow users list 2>/dev/null | grep -q "${AIRFLOW_ADMIN_USER}"; then
+    airflow users create \
+        --username "${AIRFLOW_ADMIN_USER}" \
+        --firstname Admin \
+        --lastname User \
+        --role Admin \
+        --email "${AIRFLOW_ADMIN_USER}@example.com" \
+        --password "${AIRFLOW_ADMIN_PASSWORD}"
+    log "Admin user '${AIRFLOW_ADMIN_USER}' created."
+else
+    log "Admin user '${AIRFLOW_ADMIN_USER}' already exists — skipping."
 fi
 
-echo ""
-echo "=== Running database migrations ==="
-airflow db init || true
-airflow db migrate || true
-
-echo ""
-echo "=== Creating admin user ==="
-# Delete old admin user if it exists to ensure fresh credentials
-airflow users delete -u "${AIRFLOW_ADMIN_USER}" || true
-
-airflow users create \
-    --username "${AIRFLOW_ADMIN_USER}" \
-    --firstname Admin \
-    --lastname User \
-    --role Admin \
-    --email "${AIRFLOW_ADMIN_USER}@example.com" \
-    --password "${AIRFLOW_ADMIN_PASSWORD}" || true
-
-echo ""
-echo "=== Starting services ==="
+# ---- Start Supervisor ----
+log "Starting services via supervisord..."
 exec /usr/bin/supervisord -c /etc/supervisord.conf

nginx.conf

@@ -9,57 +9,100 @@ events {
 http {
     include /etc/nginx/mime.types;
     default_type application/octet-stream;
-    
+
     log_format main '$remote_addr - $remote_user [$time_local] "$request" '
                     '$status $body_bytes_sent "$http_referer" "$http_user_agent"';
-    
+
     access_log /var/log/nginx/access.log main;
     sendfile on;
     keepalive_timeout 65;
-    
+
+    # Temp paths (writable by non-root)
     client_body_temp_path /run/nginx/client_temp;
     proxy_temp_path       /run/nginx/proxy_temp;
     fastcgi_temp_path     /run/nginx/fastcgi_temp;
     uwsgi_temp_path       /run/nginx/uwsgi_temp;
     scgi_temp_path        /run/nginx/scgi_temp;
-    
+
+    # Gzip compression
+    gzip on;
+    gzip_types text/plain text/css application/json application/javascript text/xml;
+    gzip_min_length 256;
+    gzip_vary on;
+
+    # WebSocket upgrade map (must be before server block)
+    map $http_upgrade $connection_upgrade {
+        default upgrade;
+        ''      '';
+    }
+
+    # Upstreams
     upstream airflow {
         server 127.0.0.1:8080;
+        keepalive 4;
     }
-    
+
     upstream refresh {
         server 127.0.0.1:5000;
+        keepalive 2;
     }
-    
+
     server {
         listen 7860;
         server_name _;
         client_max_body_size 100M;
-        
+
+        # Security headers
+        add_header X-Frame-Options SAMEORIGIN always;
+        add_header X-Content-Type-Options nosniff always;
+        add_header X-XSS-Protection "1; mode=block" always;
+        add_header Referrer-Policy strict-origin-when-cross-origin always;
+
+        # ---- Refresh service ----
         location /refresh {
             proxy_pass http://refresh;
             proxy_set_header Host $host;
+            proxy_set_header Connection "";
+            proxy_http_version 1.1;
             proxy_read_timeout 120s;
         }
-        
+
         location /config {
             proxy_pass http://refresh;
             proxy_set_header Host $host;
+            proxy_set_header Connection "";
+            proxy_http_version 1.1;
         }
-        
+
         location /health {
             proxy_pass http://refresh;
             proxy_set_header Host $host;
+            proxy_set_header Connection "";
+            proxy_http_version 1.1;
         }
-        
+
+        location /status {
+            proxy_pass http://refresh;
+            proxy_set_header Host $host;
+            proxy_set_header Connection "";
+            proxy_http_version 1.1;
+        }
+
+        # ---- Airflow webserver ----
         location / {
             proxy_pass http://airflow;
             proxy_set_header Host $host;
             proxy_set_header X-Real-IP $remote_addr;
             proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
             proxy_set_header X-Forwarded-Proto $scheme;
+            proxy_http_version 1.1;
+            proxy_set_header Connection "";
             proxy_read_timeout 300s;
             proxy_connect_timeout 75s;
+
+            # WebSocket support (for live log tailing)
+            proxy_set_header Upgrade $http_upgrade;
+            proxy_set_header Connection $connection_upgrade;
         }
     }
 }

refresh_service.py

@@ -1,76 +1,194 @@
+"""
+DAG Refresh Service — lightweight Flask sidecar for git-syncing DAGs.
+
+Endpoints:
+    GET/POST /refresh  — pull latest DAGs from the configured git repo
+    GET      /health   — liveness check (also pings Airflow webserver)
+    GET      /config   — show current (non-sensitive) configuration
+    GET      /status   — git SHA, last sync time, repo info
+"""
+
+import fcntl
+import logging
 import os
-from flask import Flask, jsonify
 import subprocess
+import time
+from datetime import datetime, timezone
+
+from flask import Flask, jsonify, request
+
+# ---- Logging ----
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+logger = logging.getLogger("refresh-service")
 
 app = Flask(__name__)
 
-DAGS_DIR = os.environ.get('AIRFLOW_HOME', '/opt/airflow') + '/dags'
-DAG_REPO_URL = os.environ.get('DAG_REPO_URL', 'https://github.com/subhamgiri460/myworkflows.git')
-DAG_REPO_BRANCH = os.environ.get('DAG_REPO_BRANCH', 'main')
+# ---- Configuration ----
+AIRFLOW_HOME = os.environ.get("AIRFLOW_HOME", "/opt/airflow")
+DAGS_DIR = os.path.join(AIRFLOW_HOME, "dags")
+DAG_REPO_URL = os.environ.get("DAG_REPO_URL", "https://github.com/subhamgiri460/myworkflows.git")
+DAG_REPO_BRANCH = os.environ.get("DAG_REPO_BRANCH", "main")
+DAG_REPO_TOKEN = os.environ.get("DAG_REPO_TOKEN", "")
+AIRFLOW_WEB_PORT = os.environ.get("AIRFLOW_WEB_PORT", "8080")
+REFRESH_SERVICE_PORT = os.environ.get("REFRESH_SERVICE_PORT", "5000")
+
+LOCK_FILE = "/tmp/dag_sync.lock"
+_last_sync: dict = {}
+
+
+def _get_authenticated_url() -> str:
+    """Inject PAT token into the repo URL for private repos."""
+    if DAG_REPO_TOKEN:
+        return DAG_REPO_URL.replace("https://", f"https://{DAG_REPO_TOKEN}@")
+    return DAG_REPO_URL
+
+
+def _run_git(args: list[str], cwd: str | None = None, timeout: int = 60) -> subprocess.CompletedProcess:
+    """Run a git command and return the result."""
+    return subprocess.run(
+        ["git"] + args,
+        cwd=cwd,
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+    )
+
+
+def _get_head_sha() -> str | None:
+    """Return the current HEAD SHA in the DAGs directory."""
+    try:
+        result = _run_git(["rev-parse", "--short", "HEAD"], cwd=DAGS_DIR, timeout=10)
+        return result.stdout.strip() if result.returncode == 0 else None
+    except Exception:
+        return None
+
+
+def _sync_repo() -> tuple[dict, int]:
+    """Clone or update the DAG repository. Returns (response_body, status_code)."""
+    global _last_sync
+
+    os.makedirs(DAGS_DIR, exist_ok=True)
+
+    auth_url = _get_authenticated_url()
+    is_update = os.path.isdir(os.path.join(DAGS_DIR, ".git"))
+
+    if is_update:
+        # Update remote URL in case PAT token changed
+        _run_git(["remote", "set-url", "origin", auth_url], cwd=DAGS_DIR)
 
+        fetch = _run_git(["fetch", "origin", DAG_REPO_BRANCH], cwd=DAGS_DIR)
+        if fetch.returncode != 0:
+            logger.error("git fetch failed: %s", fetch.stderr)
+            return {"status": "error", "message": f"git fetch failed: {fetch.stderr}"}, 500
 
-@app.route('/refresh', methods=['GET', 'POST'])
+        reset = _run_git(["reset", "--hard", f"origin/{DAG_REPO_BRANCH}"], cwd=DAGS_DIR)
+        if reset.returncode != 0:
+            logger.error("git reset failed: %s", reset.stderr)
+            return {"status": "error", "message": f"git reset failed: {reset.stderr}"}, 500
+        action = "updated"
+    else:
+        clone = _run_git(
+            ["clone", "--depth", "1", "--branch", DAG_REPO_BRANCH, auth_url, DAGS_DIR],
+            timeout=120,
+        )
+        if clone.returncode != 0:
+            logger.error("git clone failed: %s", clone.stderr)
+            return {"status": "error", "message": f"git clone failed: {clone.stderr}"}, 500
+        action = "cloned"
+
+    sha = _get_head_sha()
+    _last_sync = {
+        "action": action,
+        "sha": sha,
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    }
+
+    logger.info("DAGs %s — commit %s", action, sha)
+    return {
+        "status": "success",
+        "message": f"DAGs {action} successfully",
+        "repo": DAG_REPO_URL,
+        "branch": DAG_REPO_BRANCH,
+        "sha": sha,
+    }, 200
+
+
+# ---- Request logging ----
+@app.before_request
+def log_request():
+    logger.info("%s %s", request.method, request.path)
+
+
+# ---- Endpoints ----
+@app.route("/refresh", methods=["GET", "POST"])
 def refresh_dags():
+    """Pull latest DAGs from git. Uses a file lock to prevent concurrent syncs."""
     try:
-        if os.path.exists(os.path.join(DAGS_DIR, '.git')):
-            fetch_result = subprocess.run(
-                ['git', 'fetch', 'origin', DAG_REPO_BRANCH],
-                cwd=DAGS_DIR, capture_output=True, text=True, timeout=60
-            )
-            if fetch_result.returncode != 0:
-                return jsonify({
-                    'status': 'error',
-                    'message': f'Git fetch failed: {fetch_result.stderr}'
-                }), 500
-            
-            reset_result = subprocess.run(
-                ['git', 'reset', '--hard', f'origin/{DAG_REPO_BRANCH}'],
-                cwd=DAGS_DIR, capture_output=True, text=True, timeout=60
-            )
-            result = reset_result
-            action = 'updated'
-        else:
-            result = subprocess.run(
-                ['git', 'clone', '--depth', '1', '--branch', DAG_REPO_BRANCH, DAG_REPO_URL, DAGS_DIR],
-                capture_output=True, text=True, timeout=120
-            )
-            action = 'cloned'
-        
-        if result.returncode == 0:
-            return jsonify({
-                'status': 'success',
-                'message': f'DAGs {action} successfully',
-                'repo': DAG_REPO_URL,
-                'branch': DAG_REPO_BRANCH
-            }), 200
-        else:
-            return jsonify({
-                'status': 'error',
-                'message': result.stderr or 'Unknown error'
-            }), 500
-            
+        lock_fd = open(LOCK_FILE, "w")
+        acquired = False
+        try:
+            fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            acquired = True
+        except BlockingIOError:
+            return jsonify({"status": "busy", "message": "A sync is already in progress"}), 429
+
+        if acquired:
+            body, status = _sync_repo()
+            fcntl.flock(lock_fd, fcntl.LOCK_UN)
+            lock_fd.close()
+            return jsonify(body), status
+
     except subprocess.TimeoutExpired:
-        return jsonify({'status': 'error', 'message': 'Operation timed out'}), 500
+        return jsonify({"status": "error", "message": "Git operation timed out"}), 504
     except Exception as e:
-        return jsonify({'status': 'error', 'message': str(e)}), 500
+        logger.exception("Unexpected error during refresh")
+        return jsonify({"status": "error", "message": str(e)}), 500
 
 
-@app.route('/health')
+@app.route("/health")
 def health():
-    return jsonify({'status': 'healthy'}), 200
+    """Liveness probe — also checks if Airflow webserver is reachable."""
+    airflow_ok = False
+    try:
+        import urllib.request
+        resp = urllib.request.urlopen(f"http://127.0.0.1:{AIRFLOW_WEB_PORT}/health", timeout=5)
+        airflow_ok = resp.status == 200
+    except Exception:
+        pass
+
+    return jsonify({
+        "status": "healthy",
+        "airflow_webserver": "up" if airflow_ok else "starting",
+    }), 200
 
 
-@app.route('/config')
+@app.route("/config")
 def show_config():
+    """Return non-sensitive configuration values."""
+    return jsonify({
+        "dags_dir": DAGS_DIR,
+        "repo_url": DAG_REPO_URL,
+        "branch": DAG_REPO_BRANCH,
+        "airflow_web_port": AIRFLOW_WEB_PORT,
+        "refresh_service_port": REFRESH_SERVICE_PORT,
+    }), 200
+
+
+@app.route("/status")
+def status():
+    """Return the last sync status, current git SHA, and uptime."""
     return jsonify({
-        'dags_dir': DAGS_DIR,
-        'repo_url': DAG_REPO_URL,
-        'branch': DAG_REPO_BRANCH,
-        'airflow_web_port': os.environ.get('AIRFLOW_WEB_PORT', '8080'),
-        'refresh_service_port': os.environ.get('REFRESH_SERVICE_PORT', '5000')
+        "last_sync": _last_sync or "no sync yet",
+        "current_sha": _get_head_sha(),
+        "repo": DAG_REPO_URL,
+        "branch": DAG_REPO_BRANCH,
     }), 200
 
 
-if __name__ == '__main__':
-    port = int(os.environ.get('REFRESH_SERVICE_PORT', 5000))
-    app.run(host='0.0.0.0', port=port)
+if __name__ == "__main__":
+    port = int(REFRESH_SERVICE_PORT)
+    app.run(host="0.0.0.0", port=port)

requirements.txt

@@ -1,7 +1,7 @@
 apache-airflow[postgres]==2.10.3
-apache-airflow-providers-postgres
-apache-airflow-providers-ssh
-flask
-gunicorn
-gevent
-requests
+apache-airflow-providers-postgres==5.13.1
+apache-airflow-providers-ssh==3.14.0
+psycopg2-binary==2.9.10
+gunicorn==23.0.0
+gevent==24.11.1
+cryptography>=42.0.0

supervisord.conf

@@ -9,38 +9,58 @@ loglevel=info
 command=/usr/sbin/nginx -g "daemon off;"
 autostart=true
 autorestart=true
+startsecs=5
+startretries=3
+stopwaitsecs=10
+stopasgroup=true
+killasgroup=true
 stderr_logfile=/dev/stderr
 stderr_logfile_maxbytes=0
 stdout_logfile=/dev/stdout
 stdout_logfile_maxbytes=0
 priority=5
 
-[program:airflow-webserver]
-command=airflow webserver
+[program:refresh-service]
+command=gunicorn --bind 127.0.0.1:5000 --workers 1 --worker-class gevent --timeout 120 --chdir / refresh_service:app
 autostart=true
 autorestart=true
+startsecs=3
+startretries=5
+stopwaitsecs=10
+stopasgroup=true
+killasgroup=true
 stderr_logfile=/dev/stderr
 stderr_logfile_maxbytes=0
 stdout_logfile=/dev/stdout
 stdout_logfile_maxbytes=0
-priority=20
+priority=10
 
-[program:airflow-scheduler]
-command=airflow scheduler
+[program:airflow-webserver]
+command=airflow webserver
 autostart=true
 autorestart=true
+startsecs=10
+startretries=5
+stopwaitsecs=30
+stopasgroup=true
+killasgroup=true
 stderr_logfile=/dev/stderr
 stderr_logfile_maxbytes=0
 stdout_logfile=/dev/stdout
 stdout_logfile_maxbytes=0
-priority=30
+priority=20
 
-[program:refresh-service]
-command=gunicorn --bind 127.0.0.1:5000 --workers 1 --worker-class gevent --chdir / refresh_service:app
+[program:airflow-scheduler]
+command=airflow scheduler
 autostart=true
 autorestart=true
+startsecs=10
+startretries=5
+stopwaitsecs=30
+stopasgroup=true
+killasgroup=true
 stderr_logfile=/dev/stderr
 stderr_logfile_maxbytes=0
 stdout_logfile=/dev/stdout
 stdout_logfile_maxbytes=0
-priority=10
+priority=30