This is an automated email from the ASF dual-hosted git repository. tirkarthi pushed a commit to branch backport-bdc6147-v3-0-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 3314df439f1194b1be7b779aa3cac64282e0721d Author: Karthikeyan Singaravelan <[email protected]> AuthorDate: Tue Sep 2 20:40:42 2025 +0530 [v3-0-test] Don't create audit log for API calls with dry_run=True. (#55116) * Don't create audit log for API calls with dry_run=True. * Skip audit_log only for clear task instance and dag run dry runs. (cherry picked from commit bdc61474938554945892fd9ce1954a87aa3aad95) Co-authored-by: Karthikeyan Singaravelan <[email protected]> --- .../src/airflow/api_fastapi/logging/decorators.py | 4 +++ .../core_api/routes/public/test_dag_run.py | 9 +++++- .../core_api/routes/public/test_task_instances.py | 35 ++++++++++++++++++++-- 3 files changed, 45 insertions(+), 3 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/logging/decorators.py b/airflow-core/src/airflow/api_fastapi/logging/decorators.py index ad65c8caac8..01e57ad6e5b 100644 --- a/airflow-core/src/airflow/api_fastapi/logging/decorators.py +++ b/airflow-core/src/airflow/api_fastapi/logging/decorators.py @@ -83,6 +83,7 @@ def action_logging(event: str | None = None): ): """Log user actions.""" event_name = event or request.scope["endpoint"].__name__ + skip_dry_run_events = {"clear_dag_run", "post_clear_task_instances"} if not user: user_name = "anonymous" @@ -100,6 +101,9 @@ def action_logging(event: str | None = None): request_body = {} masked_body_json = {} + if event_name in skip_dry_run_events and request_body.get("dry_run", True): + return + fields_skip_logging = { "csrf_token", "_csrf_token", diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py index 4ab197c2f59..dc9fc2a4596 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py @@ -27,7 +27,7 @@ from sqlalchemy import select from airflow.api_fastapi.core_api.datamodels.dag_versions import DagVersionResponse from airflow.listeners.listener import get_listener_manager -from airflow.models import DagModel, DagRun +from airflow.models import DagModel, DagRun, Log from airflow.models.asset import AssetEvent, AssetModel from airflow.providers.standard.operators.empty import EmptyOperator from airflow.sdk.definitions.asset import Asset @@ -1208,6 +1208,13 @@ class TestClearDagRun: dag_run = session.scalar(select(DagRun).filter_by(dag_id=DAG1_ID, run_id=DAG1_RUN1_ID)) assert dag_run.state == DAG1_RUN1_STATE + logs = ( + session.query(Log) + .filter(Log.dag_id == DAG1_ID, Log.run_id == dag_run_id, Log.event == "clear_dag_run") + .count() + ) + assert logs == 0 + def test_clear_dag_run_not_found(self, test_client): response = test_client.post(f"/dags/{DAG1_ID}/dagRuns/invalid/clear", json={"dry_run": False}) assert response.status_code == 404 diff --git a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py index c0cdffd56b3..52f83780449 100644 --- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py +++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py @@ -31,7 +31,7 @@ from airflow.dag_processing.bundles.manager import DagBundlesManager from airflow.jobs.job import Job from airflow.jobs.triggerer_job_runner import TriggererJobRunner from airflow.listeners.listener import get_listener_manager -from airflow.models import DagRun, TaskInstance +from airflow.models import DagRun, Log, TaskInstance from airflow.models.baseoperator import BaseOperator from airflow.models.dagbag import DagBag from airflow.models.renderedtifields import RenderedTaskInstanceFields as RTIF @@ -2170,7 +2170,9 @@ class TestPostClearTaskInstances(TestTaskInstanceEndpoint): ) assert response.status_code == 200 assert response.json()["total_entries"] == expected_ti - check_last_log(session, dag_id=request_dag, event="post_clear_task_instances", logical_date=None) + + if not payload.get("dry_run", True): + check_last_log(session, dag_id=request_dag, event="post_clear_task_instances", logical_date=None) @pytest.mark.parametrize("flag", ["include_future", "include_past"]) def test_dag_run_with_future_or_past_flag_returns_400(self, test_client, session, flag): @@ -2764,6 +2766,35 @@ class TestPostClearTaskInstances(TestTaskInstanceEndpoint): assert response.status_code == 404 assert "DAG non-existent-dag not found" in response.text + @pytest.mark.parametrize( + "dry_run, audit_log_count", + [ + (True, 0), + (False, 1), + ], + ) + def test_dry_run_audit_log(self, test_client, session, dry_run, audit_log_count): + dag_id = "example_python_operator" + dag_run_id = "TEST_DAG_RUN_ID" + event = "post_clear_task_instances" + + payload = {"dry_run": dry_run, "dag_run_id": dag_run_id} + self.create_task_instances(session, dag_id) + + response = test_client.post( + f"/dags/{dag_id}/clearTaskInstances", + json=payload, + ) + + logs = ( + session.query(Log) + .filter(Log.dag_id == dag_id, Log.run_id == dag_run_id, Log.event == event) + .count() + ) + + assert response.status_code == 200 + assert logs == audit_log_count + class TestGetTaskInstanceTries(TestTaskInstanceEndpoint): def test_should_respond_200(self, test_client, session):
