Add tracking of triggering user to Dag runs

[jscheffl]

This PR adds a feature to Airflow that trigering user is tracked on the Dag run level. So far if you have usages where users trigger manual you needed to find-out which user it was by looking into audit log.

The user is tracked with unix user name when using CLI or airflowctl, in web UI or REST cases the authenticated user is used.
In case a backfill is started via UI the user who started the backfill propagates the runs of Dags.

FYI @clellmann@wolfdn@AutomationDev85

[jscheffl]

Note: Marking as DRAFT as I assume some pytests fail and need to be adjusted.

[potiuk]

nice! Highly requested feature.

[dheerajturaga]

Just for reference, in Airflow 3, retrieving the triggering user from event logs requires the following approach. I've encapsulated the access_token handling within the get_airflow_client_configuration method for clarity.

That said, there may be scenarios where the logical date does not align as expected, which introduces additional complexity—particularly in our unit tests where we need to mock API responses. Given that over 50% of our DAGs rely on this functionality, its absence could significantly delay our adoption of Airflow 3.

I hope this concern is understandable and that accommodating this request is feasible. I truly appreciate your consideration.

def_find_owner_v3(dag_run=None) ->str|None:
""" This is only for Airflow3, use the Airflow Client API to fetch the event logs """# Only run for manual runsifdag_run.run_type.name.upper() !='MANUAL':
logger.error(f"Not manually triggered. run_type: {dag_run.run_type}")
return# Cant co-relate if logical date is missing for dag runifnotdag_run.logical_date:
logger.error(f"No logical date available for this run, cant find owner")
returnlogical_date=dag_run.logical_date.strftime('%Y-%m-%dT%H:%M:%S.%f')[:-3] +'Z'importairflow_client.clientfromairflow_client.client.restimportApiExceptionfromcustom.api.utilsimportget_airflow_client_configurationconfiguration=get_airflow_client_configuration()
# Enter a context with an instance of the API clientwithairflow_client.client.ApiClient(configuration) asapi_client:
# Create an instance of the API classapi_instance=airflow_client.client.EventLogApi(api_client)
event='trigger_dag_run'try:
logger.info("#################################################")
logger.info(f"DAG ID: {dag_run.dag_id}")
logger.info(f"RUN ID: {dag_run.run_id}")
logger.info(f"Logical Date: {logical_date}")
logger.info("#################################################")
# Get Event Logsapi_response=api_instance.get_event_logs(dag_id=dag_run.dag_id, event=event)
ifnotapi_response:
logger.error("No trigger events found!")
returnforeventinreversed(api_response.event_logs):
logger.info(event)
ifevent.extra:
event_info= {}
event_info=json.loads(event.extra)
if"logical_date"inevent_info:
iflogical_date==event_info["logical_date"]:
logger.info(f"Matching Event: {event}")
logger.info(f"Dag triggered by: {event.owner}")
returnevent.ownerexceptExceptionase:
raiseAirflowException("Exception when calling EventLogApi->get_event_logs: %s\n"%e)

[pierrejeambrun]

Just a few nits, questions, but looks good to me.

[ashb]

actually just because adding one field to the schema which consumes a few bytes... many more fields on task instance which also would be candidated to optimize DB. We never had such a long discussion in other PRs adding e.g. "triggered_by" which then is falling into the same category.

That's because it's not about adding a column/it's not about what the code itself but how we do the feature, and it's about what is the most sustainable way to develop Airflow for the long term; is this short term approach right (which we will likely have in place for years), or should we spend more time to build a longer term and more generic approach.

[potiuk]

actually just because adding one field to the schema which consumes a few bytes... many more fields on task instance which also would be candidated to optimize DB. We never had such a long discussion in other PRs adding e.g. "triggered_by" which then is falling into the same category.

That's because it's not about adding a column/it's not about what the code itself but how we do the feature, and it's about what is the most sustainable way to develop Airflow for the long term; is this short term approach right (which we will likely have in place for years), or should we spend more time to build a longer term and more generic approach.

I think we are all well aware of that. And we have different views what would be more generic way and apparently slightly different design assumptions. Nobody who wants to add user here does it because they think it's "short term gain" - the arguments here are that this is a good design decision.

There are clearly voices that audit log is a differen thing than data model of the app and they should not be mixed. That's one of the design views here. I think we should jointly make a dcieison based on those different design assumptions.

IMHO design where 'triggering user" is part of the task instance model is a good design decision.

[ashb]

I think we are all well aware of that.

@potiuk Are you though? I don't really think you are, else you wouldn't have insulted my intentions by saying I was bike-shedding and having a pointless discussion. My point was not about what color to paint a bike shed, but if we should build a bike shed or a train station.

[potiuk]

Indeed too many discussions - 4 people approving. several having doubts, and complaints about bikesheddint. Let's just merge and move on.

[jscheffl]

Thanks for merging.

Please calm down the argumentation. I was waiting before merge if there are real objections. I think the current PR was not a workaround and yes there could be cleaner and nicer solutions. I am open for somebody to change and clean-up but the additional field actually is not rocket science. So welcoming other PRs to make a real cool solution but we should focus our energy on the real critical items, I think there are sufficient other places to improve as well.