Repository: apache/airflow
Pull request: AIP-84 Adding logging actions
closes: https://github.com/apache/airflow/issues/44057
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).


Problem statement:
AIP-84 Log User Actions / Server Accesses
### Body

Similarly to the legacy API, the new FastAPI API should also Log user actions in the `Log` table.


Like the `action_logging` decorator or a Middleware to implement similar behavior.

### Committer

- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.

Hints from issue discussion:
@pierrejeambrun I would like to work on this. Assigned myself cc: @pierrejeambrun

Commit message: implement review comments

Patch to review:
diff --git a/airflow/api_fastapi/core_api/routes/public/assets.py b/airflow/api_fastapi/core_api/routes/public/assets.py
index 335110282992c..50232d0dfde7c 100644
--- a/airflow/api_fastapi/core_api/routes/public/assets.py
+++ b/airflow/api_fastapi/core_api/routes/public/assets.py
@@ -232,6 +232,7 @@ def get_asset_events(
 @assets_router.post(
     "/assets/events",
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    dependencies=[Depends(action_logging())],
 )
 def create_asset_event(
     body: CreateAssetEventsBody,
@@ -413,6 +414,7 @@ def get_dag_asset_queued_event(
     "/assets/{asset_id}/queuedEvents",
     status_code=status.HTTP_204_NO_CONTENT,
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    dependencies=[Depends(action_logging())],
 )
 def delete_asset_queued_events(
     asset_id: int,
@@ -439,6 +441,7 @@ def delete_asset_queued_events(
             status.HTTP_404_NOT_FOUND,
         ]
     ),
+    dependencies=[Depends(action_logging())],
 )
 def delete_dag_asset_queued_events(
     dag_id: str,
@@ -463,6 +466,7 @@ def delete_dag_asset_queued_events(
             status.HTTP_404_NOT_FOUND,
         ]
     ),
+    dependencies=[Depends(action_logging())],
 )
 def delete_dag_asset_queued_event(
     dag_id: str,
diff --git a/airflow/api_fastapi/core_api/routes/public/backfills.py b/airflow/api_fastapi/core_api/routes/public/backfills.py
index 7c4dc9d32a2d0..8ef6c61a218eb 100644
--- a/airflow/api_fastapi/core_api/routes/public/backfills.py
+++ b/airflow/api_fastapi/core_api/routes/public/backfills.py
@@ -39,6 +39,7 @@
 from airflow.api_fastapi.core_api.openapi.exceptions import (
     create_openapi_http_exception_doc,
 )
+from airflow.api_fastapi.logging.decorators import action_logging
 from airflow.exceptions import DagNotFound
 from airflow.models import DagRun
 from airflow.models.backfill import (
@@ -148,6 +149,7 @@ def unpause_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
             status.HTTP_409_CONFLICT,
         ]
     ),
+    dependencies=[Depends(action_logging())],
 )
 def cancel_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
     b: Backfill = session.get(Backfill, backfill_id)
@@ -188,6 +190,7 @@ def cancel_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
 @backfills_router.post(
     path="",
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND, status.HTTP_409_CONFLICT]),
+    dependencies=[Depends(action_logging())],
 )
 def create_backfill(
     backfill_request: BackfillPostBody,
diff --git a/airflow/api_fastapi/core_api/routes/public/connections.py b/airflow/api_fastapi/core_api/routes/public/connections.py
index ceb522bd5ec0f..1ecf55665cff8 100644
--- a/airflow/api_fastapi/core_api/routes/public/connections.py
+++ b/airflow/api_fastapi/core_api/routes/public/connections.py
@@ -39,6 +39,7 @@
 )
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
 from airflow.api_fastapi.core_api.services.public.connections import BulkConnectionService
+from airflow.api_fastapi.logging.decorators import action_logging
 from airflow.configuration import conf
 from airflow.models import Connection
 from airflow.secrets.environment_variables import CONN_ENV_PREFIX
@@ -52,6 +53,7 @@
     "/{connection_id}",
     status_code=status.HTTP_204_NO_CONTENT,
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    dependencies=[Depends(action_logging())],
 )
 def delete_connection(
     connection_id: str,
@@ -129,6 +131,7 @@ def get_connections(
     responses=create_openapi_http_exception_doc(
         [status.HTTP_409_CONFLICT]
     ),  # handled by global exception handler
+    dependencies=[Depends(action_logging())],
 )
 def post_connection(
     post_body: ConnectionBody,
@@ -140,7 +143,7 @@ def post_connection(
     return connection
 
 
-@connections_router.patch("")
+@connections_router.patch("", dependencies=[Depends(action_logging())])
 def bulk_connections(
     request: BulkBody[ConnectionBody],
     session: SessionDep,
@@ -157,6 +160,7 @@ def bulk_connections(
             status.HTTP_404_NOT_FOUND,
         ]
     ),
+    dependencies=[Depends(action_logging())],
 )
 def patch_connection(
     connection_id: str,
@@ -233,6 +237,7 @@ def test_connection(
 @connections_router.post(
     "/defaults",
     status_code=status.HTTP_204_NO_CONTENT,
+    dependencies=[Depends(action_logging())],
 )
 def create_default_connections(
     session: SessionDep,
diff --git a/airflow/api_fastapi/core_api/routes/public/dag_parsing.py b/airflow/api_fastapi/core_api/routes/public/dag_parsing.py
index cfb82deadd696..da85534cc08c9 100644
--- a/airflow/api_fastapi/core_api/routes/public/dag_parsing.py
+++ b/airflow/api_fastapi/core_api/routes/public/dag_parsing.py
@@ -19,13 +19,14 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING
 
-from fastapi import HTTPException, Request, status
+from fastapi import Depends, HTTPException, Request, status
 from itsdangerous import BadSignature, URLSafeSerializer
 from sqlalchemy import select
 
 from airflow.api_fastapi.common.db.common import SessionDep
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.logging.decorators import action_logging
 from airflow.auth.managers.models.resource_details import DagDetails
 from airflow.models.dag import DagModel
 from airflow.models.dagbag import DagPriorityParsingRequest
@@ -40,6 +41,7 @@
     "",
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
     status_code=status.HTTP_201_CREATED,
+    dependencies=[Depends(action_logging())],
 )
 def reparse_dag_file(
     file_token: str,
diff --git a/tests/api_fastapi/core_api/routes/public/test_assets.py b/tests/api_fastapi/core_api/routes/public/test_assets.py
index 9b7c3352a00c1..044f94f96c321 100644
--- a/tests/api_fastapi/core_api/routes/public/test_assets.py
+++ b/tests/api_fastapi/core_api/routes/public/test_assets.py
@@ -42,6 +42,7 @@
 from tests_common.test_utils.asserts import assert_queries_count
 from tests_common.test_utils.db import clear_db_assets, clear_db_runs
 from tests_common.test_utils.format_datetime import from_datetime_to_zulu_without_ms
+from tests_common.test_utils.www import _check_last_log
 
 DEFAULT_DATE = datetime(2020, 6, 11, 18, 0, 0, tzinfo=timezone.utc)
 
@@ -894,6 +895,7 @@ def test_should_respond_204(self, test_client, session, create_dummy_dag):
         assert response.status_code == 204
         adrqs = session.query(AssetDagRunQueue).all()
         assert len(adrqs) == 0
+        _check_last_log(session, dag_id=dag_id, event="delete_dag_asset_queued_events", logical_date=None)
 
     def test_should_respond_404_invalid_dag(self, test_client):
         dag_id = "not_exists"
@@ -941,6 +943,7 @@ def test_should_respond_200(self, test_client, session):
             "created_dagruns": [],
             "timestamp": from_datetime_to_zulu_without_ms(DEFAULT_DATE),
         }
+        _check_last_log(session, dag_id=None, event="create_asset_event", logical_date=None)
 
     def test_invalid_attr_not_allowed(self, test_client, session):
         self.create_assets(session)
@@ -1066,6 +1069,7 @@ def test_should_respond_204(self, test_client, session, create_dummy_dag):
         response = test_client.delete(f"/public/assets/{asset_id}/queuedEvents")
         assert response.status_code == 204
         assert session.get(AssetDagRunQueue, (asset_id, dag_id)) is None
+        _check_last_log(session, dag_id=None, event="delete_asset_queued_events", logical_date=None)
 
     def test_should_respond_404(self, test_client):
         response = test_client.delete("/public/assets/1/queuedEvents")
@@ -1091,6 +1095,7 @@ def test_delete_should_respond_204(self, test_client, session, create_dummy_dag)
         assert response.status_code == 204
         adrq = session.query(AssetDagRunQueue).all()
         assert len(adrq) == 0
+        _check_last_log(session, dag_id=dag_id, event="delete_dag_asset_queued_events", logical_date=None)
 
     def test_should_respond_404(self, test_client):
         dag_id = "not_exists"
diff --git a/tests/api_fastapi/core_api/routes/public/test_backfills.py b/tests/api_fastapi/core_api/routes/public/test_backfills.py
index 04f61d736b152..cba6e678016a7 100644
--- a/tests/api_fastapi/core_api/routes/public/test_backfills.py
+++ b/tests/api_fastapi/core_api/routes/public/test_backfills.py
@@ -39,6 +39,7 @@
     clear_db_runs,
     clear_db_serialized_dags,
 )
+from tests_common.test_utils.www import _check_last_log
 
 pytestmark = [pytest.mark.db_test, pytest.mark.need_serialized_dag]
 
@@ -213,6 +214,7 @@ def test_create_backfill(self, repro_act, repro_exp, session, dag_maker, test_cl
             "to_date": to_date_iso,
             "updated_at": mock.ANY,
         }
+        _check_last_log(session, dag_id="TEST_DAG_1", event="create_backfill", logical_date=None)
 
     def test_dag_not_exist(self, session, test_client):
         session.query(DagModel).all()
@@ -544,6 +546,7 @@ def test_cancel_backfill(self, session, test_client):
         # get conflict when canceling already-canceled backfill
         response = test_client.put(f"/public/backfills/{backfill.id}/cancel")
         assert response.status_code == 409
+        _check_last_log(session, dag_id=None, event="cancel_backfill", logical_date=None)
 
     def test_cancel_backfill_end_states(self, dag_maker, session, test_client):
         """
diff --git a/tests/api_fastapi/core_api/routes/public/test_connections.py b/tests/api_fastapi/core_api/routes/public/test_connections.py
index 174c8f0346ead..f9cf0b8fd6e48 100644
--- a/tests/api_fastapi/core_api/routes/public/test_connections.py
+++ b/tests/api_fastapi/core_api/routes/public/test_connections.py
@@ -26,6 +26,7 @@
 from airflow.utils.session import provide_session
 
 from tests_common.test_utils.db import clear_db_connections
+from tests_common.test_utils.www import _check_last_log
 
 pytestmark = pytest.mark.db_test
 
@@ -100,6 +101,7 @@ def test_delete_should_respond_204(self, test_client, session):
         assert response.status_code == 204
         connection = session.query(Connection).all()
         assert len(connection) == 0
+        _check_last_log(session, dag_id=None, event="delete_connection", logical_date=None)
 
     def test_delete_should_respond_404(self, test_client):
         response = test_client.delete(f"/public/connections/{TEST_CONN_ID}")
@@ -208,6 +210,7 @@ def test_post_should_respond_201(self, test_client, session, body):
         assert response.status_code == 201
         connection = session.query(Connection).all()
         assert len(connection) == 1
+        _check_last_log(session, dag_id=None, event="post_connection", logical_date=None)
 
     @pytest.mark.parametrize(
         "body",
@@ -303,10 +306,11 @@ def test_post_should_respond_already_exist(self, test_client, body):
             ),
         ],
     )
-    def test_post_should_response_201_redacted_password(self, test_client, body, expected_response):
+    def test_post_should_response_201_redacted_password(self, test_client, body, expected_response, session):
         response = test_client.post("/public/connections", json=body)
         assert response.status_code == 201
         assert response.json() == expected_response
+        _check_last_log(session, dag_id=None, event="post_connection", logical_date=None, check_masked=True)
 
 
 class TestPatchConnection(TestConnectionEndpoint):
@@ -337,6 +341,7 @@ def test_patch_should_respond_200(self, test_client, body, session):
 
         response = test_client.patch(f"/public/connections/{TEST_CONN_ID}", json=body)
         assert response.status_code == 200
+        _check_last_log(session, dag_id=None, event="patch_connection", logical_date=None)
 
     @pytest.mark.parametrize(
         "body, updated_connection, update_mask",
@@ -577,6 +582,7 @@ def test_patch_should_response_200_redacted_password(self, test_client, session,
         response = test_client.patch(f"/public/connections/{TEST_CONN_ID}", json=body)
         assert response.status_code == 200
         assert response.json() == expected_response
+        _check_last_log(session, dag_id=None, event="post_connection", logical_date=None, check_masked=True)
 
 
 class TestConnection(TestConnectionEndpoint):
@@ -625,10 +631,11 @@ def test_should_respond_403_by_default(self, test_client, body):
 
 
 class TestCreateDefaultConnections(TestConnectionEndpoint):
-    def test_should_respond_204(self, test_client):
+    def test_should_respond_204(self, test_client, session):
         response = test_client.post("/public/connections/defaults")
         assert response.status_code == 204
         assert response.content == b""
+        _check_last_log(session, dag_id=None, event="create_default_connections", logical_date=None)
 
     @mock.patch("airflow.api_fastapi.core_api.routes.public.connections.db_create_default_connections")
     def test_should_call_db_create_default_connections(self, mock_db_create_default_connections, test_client):
@@ -929,9 +936,10 @@ class TestBulkConnections(TestConnectionEndpoint):
             ),
         ],
     )
-    def test_bulk_connections(self, test_client, actions, expected_results):
+    def test_bulk_connections(self, test_client, actions, expected_results, session):
         self.create_connections()
         response = test_client.patch("/public/connections", json=actions)
         response_data = response.json()
         for connection_id, value in expected_results.items():
             assert response_data[connection_id] == value
+        _check_last_log(session, dag_id=None, event="bulk_connections", logical_date=None)
diff --git a/tests/api_fastapi/core_api/routes/public/test_dag_parsing.py b/tests/api_fastapi/core_api/routes/public/test_dag_parsing.py
index fe78c193d389b..f5c79427f2ebe 100644
--- a/tests/api_fastapi/core_api/routes/public/test_dag_parsing.py
+++ b/tests/api_fastapi/core_api/routes/public/test_dag_parsing.py
@@ -26,6 +26,7 @@
 from airflow.utils.session import provide_session
 
 from tests_common.test_utils.db import clear_db_dag_parsing_requests, parse_and_sync_to_db
+from tests_common.test_utils.www import _check_last_log
 
 pytestmark = pytest.mark.db_test
 
@@ -65,6 +66,7 @@ def test_201_and_400_requests(self, url_safe_serializer, session, test_client):
         assert response.status_code == 409
         parsing_requests = session.scalars(select(DagPriorityParsingRequest)).all()
         assert parsing_requests[0].fileloc == test_dag.fileloc
+        _check_last_log(session, dag_id=None, event="reparse_dag_file", logical_date=None)
 
     def test_bad_file_request(self, url_safe_serializer, session, test_client):
         url = f"/public/parseDagFile/{url_safe_serializer.dumps('/some/random/file.py')}"
diff --git a/tests_common/test_utils/www.py b/tests_common/test_utils/www.py
index ccc888e3b57e6..989c445f8e8d3 100644
--- a/tests_common/test_utils/www.py
+++ b/tests_common/test_utils/www.py
@@ -27,6 +27,7 @@
 import pytest
 
 from airflow.models import Log
+from airflow.sdk.execution_time.secrets_masker import DEFAULT_SENSITIVE_FIELDS
 
 if TYPE_CHECKING:
     import jinja2
@@ -76,7 +77,8 @@ def check_content_not_in_response(text, resp, resp_code=200):
         assert text not in resp_html
 
 
-def _check_last_log(session, dag_id, event, logical_date, expected_extra=None):
+def _check_last_log(session, dag_id, event, logical_date, expected_extra=None, check_masked=False):
+    sensitive_fields = DEFAULT_SENSITIVE_FIELDS.copy()
     logs = (
         session.query(
             Log.dag_id,
@@ -98,7 +100,15 @@ def _check_last_log(session, dag_id, event, logical_date, expected_extra=None):
     assert len(logs) >= 1
     assert logs[0].extra
     if expected_extra:
+        print(f"json.loads(logs[0].extra) is {json.loads(logs[0].extra)}")
+        print(f"expected_extra is {expected_extra}")
         assert json.loads(logs[0].extra) == expected_extra
+    if check_masked:
+        extra_json = json.loads(logs[0].extra)
+        for k, v in extra_json.items():
+            if k in sensitive_fields:
+                assert v == "***", f"Expected masked value for {k}, but got {v}"
+
     session.query(Log).delete()
 
 
@@ -135,7 +145,12 @@ def _check_last_log_masked_connection(session, dag_id, event, logical_date):
     }
 
 
-def _check_last_log_masked_variable(session, dag_id, event, logical_date):
+def _check_last_log_masked_variable(
+    session,
+    dag_id,
+    event,
+    logical_date,
+):
     logs = (
         session.query(
             Log.dag_id,


Code review comments:
diff_hunk: @@ -98,7 +100,15 @@ def _check_last_log(session, dag_id, event, logical_date, expected_extra=None):
     assert len(logs) >= 1
     assert logs[0].extra
     if expected_extra:
+        print(f"json.loads(logs[0].extra) is {json.loads(logs[0].extra)}")
+        print(f"expected_extra is {expected_extra}")
path: tests_common/test_utils/www.py
text: @user1:
I don't think we need those prints.

@author:
Resolved