Write action log to DB when DAG run is trigged via API

[kolfild26]

Based on #28435

The purpose is to save to Log table an airflow user who created a dagrun via REST API.
Currently, such event as trigger (from UI) and cli_dag_trigger (from CLI) are being written to metadb, But the api endpoint dagrun request is missing there.

I suggest to add a log entry in dag_run_endpoint.py like this:

ID	dttm	dag_id	task_id	event	execution_date	owner
143130	2023-01-17 17:30:05.498895	test_rest		rest_dag_trigger	2023-01-17 17:30:05.482352	my_user

'rest_dag_trigger' is discussable, it might be someone could advise a better text. I just named it in the likeness of 'cli_dag_trigger'.
The other columns are obvious.

[ephraimbuddy]

Instead of this, there's action_logging decorator you should use, see

airflow/airflow/api_connexion/endpoints/connection_endpoint.py

Line 51 in 5d84b59

@action_logging(

[kolfild26]

@ephraimbuddy thanks, my dev env was a bit outdated so I didn't have this decorator.
Updated and changed the way for db insertion. Please review.

[ephraimbuddy]

With the action_logging, do we still need this?

[kolfild26]

The point was to write the log table. To do so we first need to create Log object. That's what we actually do here.
We don't write to dag_run table, thus cannot use DagRun object as itself.
And also, owner is only accessible from flask.g.
So we need to "construct" Log object

[ephraimbuddy]

Can you look at the code for the action_logging?

airflow/airflow/www/decorators.py

Lines 105 to 112 in be0e353

	log = Log(
	event=event or f.__name__,
	task_instance=None,
	owner=user,
	extra=str(extra_fields),
	task_id=params.get("task_id"),
	dag_id=params.get("dag_id"),
	)

[ephraimbuddy]

We can have a global constant for this prefix. Check the usage in other places.

[kolfild26]

The same, as above. In other places we are logging the "main" object to its correspondent table. For instance, in connection_endpoint.py we are logging the Connection object to connection table. In variable_endpoint.py we are logging the Variable object to variable table.

Here is the difference, in dag_run_endpoint.py we are logging to log table rather than dag_run table.

Based on that I decided not to have prefix in a global constatnt. I mean if we want to add something more to dag_run table we will need RESOURCE_EVENT_PREFIX = "dag_run" rather than RESOURCE_EVENT_PREFIX = "log".

@ephraimbuddy so, the question, is it ok to use RESOURCE_EVENT_PREFIX or it's better to use smth like RESOURCE_EVENT_LOG_PREFIX ?

#example
RESOURCE_EVENT_LOG_PREFIX = "log" 
...
@action_logging(
    event=action_event_from_permission(
        prefix=RESOURCE_EVENT_LOG_PREFIX,
        permission=permissions.ACTION_CAN_CREATE,
    ),
)

[ephraimbuddy]

The same, as above. In other places we are logging the "main" object to its correspondent table. For instance, in connection_endpoint.py we are logging the Connection object to connection table. In variable_endpoint.py we are logging the Variable object to variable table.

This is wrong. The logging is all done in the log table through the action_logging decorator.

[kolfild26]

@ephraimbuddy
Thanks for these inputs. I was actually looking at cli_action_loggers.py. It differs in there. That's why.
For both questions, you're right. We only need a decorator to be used. Fixed.

[ephraimbuddy]

@ephraimbuddy Thanks for these inputs. I was actually looking at cli_action_loggers.py. It differs in there. That's why. For both questions, you're right. We only need a decorator to be used. Fixed.

We also need a test

[kolfild26]

We also need a test

Ok, will update once it's done.

[kolfild26]

Hi, @ephraimbuddy
I added test _check_last_log similarly it's done in test_connection_endpoint.py.
Tests were passed successfully.

[kolfild26]

Above there was a conflict with the very fresh commit b94f36b in main. It also changed the same test in the same row of code by coincidence.
I resolved and re-run tests locally. Everything's fine.