Последние года 4 я использовал Apache Airflow исключительно в облаке, преимущественно в Amazon —
Amazon Managed Apache Airflow. И как обычно бывает, в облаках всё так или иначе между собой связано. Логи хранятся в Cloud Watch, воркеры запускаются в изолированной среде (Amazon Fargate). С июля месяца я стал активно использовать self-hosted Airflow на своих серверах (для своих личных целей), и в целях экономии храню всё в файлах. Так уж получилось, что задачу с регулярной "чисткой" я постоянно откладывал и вот настал час X, когда всё легко из-за исчерпания inodes в файловой системе. Для этого случая я написал DAG, который каждый день в полночь чистит папки со старыми логами, делюсь с вами вдруг он пригодится:
import os
import shutil
from datetime import datetime, timedelta
import pendulum
import structlog
from airflow.sdk import DAG, task
logger = structlog.get_logger(__name__)
@task
def cleanup_airflow_logs(days_to_keep):
log_base_path = os.environ.get("AIRFLOW_HOME", "/opt/airflow") + "/logs"
cutoff_date = datetime.now() - timedelta(days=days_to_keep)
for root, dirs, files in os.walk(log_base_path):
for dir_name in dirs:
dir_path = os.path.join(root, dir_name)
try:
if os.path.getmtime(dir_path) < cutoff_date.timestamp():
logger.info(f"Deleting old log directory: {dir_path}")
shutil.rmtree(dir_path)
except Exception as e:
logger.error(f"Error deleting directory {dir_path}: {e}")
with DAG(
dag_id="airflow_log_cleanup_dag",
start_date=pendulum.datetime(2025, 10, 1, tz="Asia/Almaty"),
schedule="@daily", # Run daily at midnight
catchup=False,
default_args={
"owner": "airflow",
"retries": 2,
"retry_delay": timedelta(minutes=5),
},
max_active_runs=1,
) as dag:
cleanup_airflow_logs(days_to_keep=14)
Здесь учитывается стандартный шаблон именования логов и директорий, включающий дату и время. Я по привычке использую
structlog для ведения логов.
