edenfunf opened a new pull request, #65739:
URL: https://github.com/apache/airflow/pull/65739
<!--
Fix for #65729: triggerer supervisor unable to ship logs to remote Azure
blob connection.
-->
When a trigger finishes, `TriggerRunnerSupervisor._handle_request` calls
`factory.upload_to_remote()` in the supervisor (main) process to ship the
per-trigger log file to the configured remote storage. The call path is:
```
WasbRemoteLogIO.write
-> WasbHook.load_string -> WasbHook.upload -> WasbHook._get_blob_client
-> WasbHook.blob_service_client -> WasbHook.get_conn
-> Hook.get_connection -> Connection.get -> _get_connection
```
`_get_connection` picks secrets backends via
`ensure_secrets_backend_loaded()`, which auto-detects context:
1. `SUPERVISOR_COMMS` set -> worker client chain
2. `_AIRFLOW_PROCESS_CONTEXT=server` -> server chain (includes
`MetastoreBackend`)
3. fallback -> **only** `EnvironmentVariablesBackend`
The triggerer supervisor process sets neither marker, so it falls into the
fallback chain and cannot see connections stored only in the metastore —
regardless of provider. The user in #65729 reports this with WASB, but the same
bug hits S3/GCS/etc. The result is `AirflowNotFoundException: The conn_id
'<id>' isn't defined` and the per-trigger log
(`attempt=1.log.trigger.<id>.log`) is never uploaded.
`SchedulerJobRunner._execute`, `DagFileProcessorManager.run` and
`airflow.api_fastapi.main` already set `_AIRFLOW_PROCESS_CONTEXT=server` for
the same reason. The triggerer was simply missing from that list.
## What this PR does
- In `TriggererJobRunner._execute()`, set `_AIRFLOW_PROCESS_CONTEXT=server`
before doing anything else. Identical pattern and comment as scheduler.
- Add a regression test `test_execute_marks_process_context_as_server` in
`airflow-core/tests/unit/jobs/test_triggerer_job.py` asserting the env var is
set after `_execute()` runs.
## Why the trigger subprocess is unaffected
The async trigger subprocess inherits this env var from the parent, but its
`TriggerRunner.arun()` first calls `await self.init_comms()`, which sets
`task_runner.SUPERVISOR_COMMS` before any trigger code runs.
`ensure_secrets_backend_loaded()` checks `SUPERVISOR_COMMS` before the env var,
so the subprocess still uses the correct client chain. No trigger-side
behaviour changes.
## Verification
Reproduced on both Linux (Ubuntu 22.04) and Windows by driving
`WasbRemoteLogIO.write` in the supervisor context with a metastore-backed fake
connection and `BlobServiceClient` stub:
```
Before fix: AirflowNotFoundException: The conn_id `logs_to_azure_blob` isn't
defined
(identical stack trace to issue #65729)
After fix: BlobServiceClient instantiated -> upload proceeds
```
Full `test_triggerer_job.py` (37 tests) and
`task-sdk/tests/.../test_secrets.py::TestContextDetection` (3 tests) pass after
the change.
---
##### Was generative AI tooling used to co-author this PR?
- [ ] Yes (please specify the tool below)
---
closes: #65729
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
