AIP-84 Adding logging actions (apache#47556)

closes: apache#44057

Author: vatsrahul1001

airflow/api_fastapi/core_api/routes/public/dags.py

@@ -57,6 +57,7 @@
     DAGResponse,
 )
 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 AirflowException, DagNotFound
 from airflow.models import DAG, DagModel
 from airflow.models.dagrun import DagRun
@@ -207,6 +208,7 @@ def get_dag_details(dag_id: str, session: SessionDep, request: Request) -> DAGDe
             status.HTTP_404_NOT_FOUND,
         ]
     ),
+    dependencies=[Depends(action_logging())],
 )
 def patch_dag(
     dag_id: str,
@@ -249,6 +251,7 @@ def patch_dag(
             status.HTTP_404_NOT_FOUND,
         ]
     ),
+    dependencies=[Depends(action_logging())],
 )
 def patch_dags(
     patch_body: DAGPatchBody,
@@ -310,6 +313,7 @@ def patch_dags(
             status.HTTP_422_UNPROCESSABLE_ENTITY,
         ]
     ),
+    dependencies=[Depends(action_logging())],
 )
 def delete_dag(
     dag_id: str,

airflow/api_fastapi/core_api/routes/public/pools.py

@@ -42,6 +42,7 @@
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
 from airflow.api_fastapi.core_api.security import requires_access_pool
 from airflow.api_fastapi.core_api.services.public.pools import BulkPoolService
+from airflow.api_fastapi.logging.decorators import action_logging
 from airflow.models.pool import Pool
 
 pools_router = AirflowRouter(tags=["Pool"], prefix="/pools")
@@ -56,7 +57,7 @@
             status.HTTP_404_NOT_FOUND,
         ]
     ),
-    dependencies=[Depends(requires_access_pool(method="DELETE"))],
+    dependencies=[Depends(requires_access_pool(method="DELETE")), Depends(action_logging())],
 )
 def delete_pool(
     pool_name: str,
@@ -130,7 +131,7 @@ def get_pools(
             status.HTTP_404_NOT_FOUND,
         ]
     ),
-    dependencies=[Depends(requires_access_pool(method="PUT"))],
+    dependencies=[Depends(requires_access_pool(method="PUT")), Depends(action_logging())],
 )
 def patch_pool(
     pool_name: str,
@@ -182,7 +183,7 @@ def patch_pool(
     responses=create_openapi_http_exception_doc(
         [status.HTTP_409_CONFLICT]
     ),  # handled by global exception handler
-    dependencies=[Depends(requires_access_pool(method="POST"))],
+    dependencies=[Depends(requires_access_pool(method="POST")), Depends(action_logging())],
 )
 def post_pool(
     body: PoolBody,
@@ -196,7 +197,7 @@ def post_pool(
 
 @pools_router.patch(
     "",
-    dependencies=[Depends(requires_access_pool(method="PUT"))],
+    dependencies=[Depends(requires_access_pool(method="PUT")), Depends(action_logging())],
 )
 def bulk_pools(
     request: BulkBody[PoolBody],

airflow/api_fastapi/core_api/routes/public/task_instances.py

@@ -60,6 +60,7 @@
     TaskInstancesBatchBody,
 )
 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 TaskNotFound
 from airflow.models import Base, DagRun
 from airflow.models.dag import DAG
@@ -463,6 +464,7 @@ def get_task_instances(
 @task_instances_router.post(
     task_instances_prefix + "/list",
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    dependencies=[Depends(action_logging())],
 )
 def get_task_instances_batch(
     dag_id: Literal["~"],
@@ -601,6 +603,7 @@ def get_mapped_task_instance_try_details(
 @task_instances_router.post(
     "/clearTaskInstances",
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    dependencies=[Depends(action_logging())],
 )
 def post_clear_task_instances(
     dag_id: str,
@@ -805,12 +808,14 @@ def patch_task_instance_dry_run(
     responses=create_openapi_http_exception_doc(
         [status.HTTP_404_NOT_FOUND, status.HTTP_400_BAD_REQUEST, status.HTTP_409_CONFLICT],
     ),
+    dependencies=[Depends(action_logging())],
 )
 @task_instances_router.patch(
     task_instances_prefix + "/{task_id}/{map_index}",
     responses=create_openapi_http_exception_doc(
         [status.HTTP_404_NOT_FOUND, status.HTTP_400_BAD_REQUEST, status.HTTP_409_CONFLICT],
     ),
+    dependencies=[Depends(action_logging())],
 )
 def patch_task_instance(
     dag_id: str,

tests/api_fastapi/core_api/routes/public/test_dags.py

@@ -30,6 +30,7 @@
 from airflow.utils.types import DagRunTriggeredByType, DagRunType
 
 from tests_common.test_utils.db import clear_db_dags, clear_db_runs, clear_db_serialized_dags
+from tests_common.test_utils.logs import check_last_log
 
 pytestmark = pytest.mark.db_test
 
@@ -257,14 +258,15 @@ class TestPatchDag(TestDagEndpoint):
         ],
     )
     def test_patch_dag(
-        self, test_client, query_params, dag_id, body, expected_status_code, expected_is_paused
+        self, test_client, query_params, dag_id, body, expected_status_code, expected_is_paused, session
     ):
         response = test_client.patch(f"/public/dags/{dag_id}", json=body, params=query_params)
 
         assert response.status_code == expected_status_code
         if expected_status_code == 200:
             body = response.json()
             assert body["is_paused"] == expected_is_paused
+            check_last_log(session, dag_id=dag_id, event="patch_dag", logical_date=None)
 
 
 class TestPatchDags(TestDagEndpoint):
@@ -312,7 +314,14 @@ class TestPatchDags(TestDagEndpoint):
         ],
     )
     def test_patch_dags(
-        self, test_client, query_params, body, expected_status_code, expected_ids, expected_paused_ids
+        self,
+        test_client,
+        query_params,
+        body,
+        expected_status_code,
+        expected_ids,
+        expected_paused_ids,
+        session,
     ):
         response = test_client.patch("/public/dags", json=body, params=query_params)
 
@@ -322,6 +331,7 @@ def test_patch_dags(
             assert [dag["dag_id"] for dag in body["dags"]] == expected_ids
             paused_dag_ids = [dag["dag_id"] for dag in body["dags"] if dag["is_paused"]]
             assert paused_dag_ids == expected_paused_ids
+            check_last_log(session, dag_id=DAG1_ID, event="patch_dag", logical_date=None)
 
 
 class TestDagDetails(TestDagEndpoint):
@@ -496,6 +506,7 @@ def test_delete_dag(
         status_code_details,
         has_running_dagruns,
         is_create_dag,
+        session,
     ):
         if is_create_dag:
             self._create_dag_for_deletion(
@@ -510,3 +521,5 @@ def test_delete_dag(
 
         details_response = test_client.get(f"{API_PREFIX}/{dag_id}/details")
         assert details_response.status_code == status_code_details
+        if details_response.status_code == 204:
+            check_last_log(session, dag_id=dag_id, event="delete_dag", logical_date=None)

tests/api_fastapi/core_api/routes/public/test_pools.py

@@ -22,6 +22,7 @@
 from airflow.utils.session import provide_session
 
 from tests_common.test_utils.db import clear_db_pools
+from tests_common.test_utils.logs import check_last_log
 
 pytestmark = pytest.mark.db_test
 
@@ -64,6 +65,7 @@ def test_delete_should_respond_204(self, test_client, session):
         assert response.status_code == 204
         pools = session.query(Pool).all()
         assert len(pools) == 2
+        check_last_log(session, dag_id=None, event="delete_pool", logical_date=None)
 
     def test_delete_should_respond_401(self, unauthenticated_test_client):
         response = unauthenticated_test_client.delete(f"/public/pools/{POOL1_NAME}")
@@ -300,6 +302,8 @@ def test_should_respond_200(
                 del error["url"]
 
         assert body == expected_response
+        if response.status_code == 200:
+            check_last_log(session, dag_id=None, event="patch_pool", logical_date=None)
 
     def test_should_respond_401(self, unauthenticated_test_client):
         response = unauthenticated_test_client.patch(f"/public/pools/{POOL1_NAME}", params={}, json={})
@@ -356,6 +360,7 @@ def test_should_respond_200(self, test_client, session, body, expected_status_co
 
         assert response.json() == expected_response
         assert session.query(Pool).count() == n_pools + 1
+        check_last_log(session, dag_id=None, event="post_pool", logical_date=None)
 
     def test_should_respond_401(self, unauthenticated_test_client):
         response = unauthenticated_test_client.post("/public/pools", json={})
@@ -751,6 +756,7 @@ def test_bulk_pools(self, test_client, actions, expected_results, session):
         response_data = response.json()
         for key, value in expected_results.items():
             assert response_data[key] == value
+        check_last_log(session, dag_id=None, event="bulk_pools", logical_date=None)
 
     def test_should_respond_401(self, unauthenticated_test_client):
         response = unauthenticated_test_client.patch("/public/pools", json={})

tests/api_fastapi/core_api/routes/public/test_task_instances.py

@@ -46,6 +46,7 @@
     clear_db_runs,
     clear_rendered_ti_fields,
 )
+from tests_common.test_utils.logs import check_last_log
 from tests_common.test_utils.mock_operators import MockOperator
 
 pytestmark = pytest.mark.db_test
@@ -1394,6 +1395,7 @@ def test_should_respond_200(
         assert response.status_code == 200, body
         assert expected_ti_count == body["total_entries"]
         assert expected_ti_count == len(body["task_instances"])
+        check_last_log(session, dag_id="~", event="get_task_instances_batch", logical_date=None)
 
     def test_should_respond_200_for_order_by(self, test_client, session):
         dag_id = "example_python_operator"
@@ -2061,6 +2063,7 @@ def test_should_respond_200(
         )
         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)
 
     @pytest.mark.parametrize("flag", ["include_future", "include_past"])
     def test_dag_run_with_future_or_past_flag_returns_400(self, test_client, session, flag):
@@ -2990,6 +2993,7 @@ def test_should_call_mocked_api(self, mock_set_ti_state, test_client, session):
             state=self.NEW_STATE,
             task_id=self.TASK_ID,
         )
+        check_last_log(session, dag_id=self.DAG_ID, event="patch_task_instance", logical_date=None)
 
     def test_should_update_task_instance_state(self, test_client, session):
         self.create_task_instances(session)