This is an automated email from the ASF dual-hosted git repository.
amoghrajesh pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 68ee32a61f4 Authenticate only once per task process to external
secrets backends (#71756)
68ee32a61f4 is described below
commit 68ee32a61f41ae65122d2804100f581ccd5d088e
Author: Amogh Desai <[email protected]>
AuthorDate: Wed Aug 26 10:32:31 2026 +0530
Authenticate only once per task process to external secrets backends
(#71756)
---
task-sdk/src/airflow/sdk/configuration.py | 30 +++++++++++------
task-sdk/tests/conftest.py | 10 ++++++
.../tests/task_sdk/execution_time/test_secrets.py | 38 ++++++++++++++++++++++
3 files changed, 68 insertions(+), 10 deletions(-)
diff --git a/task-sdk/src/airflow/sdk/configuration.py
b/task-sdk/src/airflow/sdk/configuration.py
index 395ecb5b491..e6bcf0ca429 100644
--- a/task-sdk/src/airflow/sdk/configuration.py
+++ b/task-sdk/src/airflow/sdk/configuration.py
@@ -304,22 +304,32 @@ def initialize_secrets_backends(
return backend_list
+_secrets_backend_cache: dict[tuple[str, ...], list] = {}
+
+
+def clear_secrets_backends_cache() -> None:
+ """Drop the memoised backends so the next load rebuilds them from the
current config."""
+ _secrets_backend_cache.clear()
+
+
def ensure_secrets_loaded(
default_backends: list[str] = _SERVER_DEFAULT_SECRETS_SEARCH_PATH,
) -> list:
"""
- Ensure that all secrets backends are loaded.
+ Return the secrets backends for the given search path, building them once
per process.
- If the secrets_backend_list contains only 2 default backends, reload it.
+ A backend holds an authenticated client, so rebuilding one per lookup
makes every secret
+ fetch authenticate again against the remote store. Nothing is memoised
until a custom
+ backend is configured, so one appearing after the first lookup is still
picked up.
"""
- # Check if the secrets_backend_list contains only 2 default backends.
-
- # Check if we are loading the backends for worker too by checking if the
default_backends is equal
- # to _SERVER_DEFAULT_SECRETS_SEARCH_PATH.
- secrets_backend_list = initialize_secrets_backends()
- if len(secrets_backend_list) == 2 or default_backends !=
_SERVER_DEFAULT_SECRETS_SEARCH_PATH:
- return initialize_secrets_backends(default_backends=default_backends)
- return secrets_backend_list
+ key = tuple(default_backends)
+ if key not in _secrets_backend_cache:
+ backends =
initialize_secrets_backends(default_backends=default_backends)
+ # Equal lengths mean nothing was prepended, so no custom backend is
configured yet.
+ if len(backends) == len(default_backends):
+ return backends
+ _secrets_backend_cache[key] = backends
+ return _secrets_backend_cache[key]
def initialize_config() -> AirflowSDKConfigParser:
diff --git a/task-sdk/tests/conftest.py b/task-sdk/tests/conftest.py
index eed48a1554c..a7d14ef1b54 100644
--- a/task-sdk/tests/conftest.py
+++ b/task-sdk/tests/conftest.py
@@ -174,6 +174,16 @@ def _disable_ol_plugin():
airflow.sdk.plugins_manager._get_plugins = old
[email protected](autouse=True)
+def _clear_secrets_backends_cache():
+ """Keep memoised backends from leaking config between tests."""
+ from airflow.sdk.configuration import clear_secrets_backends_cache
+
+ clear_secrets_backends_cache()
+ yield
+ clear_secrets_backends_cache()
+
+
@pytest.fixture(autouse=True)
def _cleanup_async_resources(request):
"""
diff --git a/task-sdk/tests/task_sdk/execution_time/test_secrets.py
b/task-sdk/tests/task_sdk/execution_time/test_secrets.py
index b97a85f3693..2fe41e328ce 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_secrets.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_secrets.py
@@ -20,8 +20,13 @@ from __future__ import annotations
import pytest
from airflow.sdk.api.datamodels._generated import ConnectionResponse
+from airflow.sdk.configuration import clear_secrets_backends_cache,
ensure_secrets_loaded
from airflow.sdk.exceptions import AirflowSecretsBackendAccessDenied, ErrorType
from airflow.sdk.execution_time.comms import ConnectionResult, ErrorResponse,
VariableResult
+from airflow.sdk.execution_time.secrets import (
+ _SERVER_DEFAULT_SECRETS_SEARCH_PATH,
+ DEFAULT_SECRETS_SEARCH_PATH_WORKERS,
+)
from airflow.sdk.execution_time.secrets.execution_api import
ExecutionAPISecretsBackend
@@ -302,3 +307,36 @@ class TestContextDetection:
assert "EnvironmentVariablesBackend" in backend_classes
assert "MetastoreBackend" not in backend_classes
assert "ExecutionAPISecretsBackend" not in backend_classes
+
+
+class TestSecretsBackendMemoisation:
+ BACKEND =
"airflow.secrets.environment_variables.EnvironmentVariablesBackend"
+
+ def test_nothing_is_memoised_until_a_custom_backend_is_configured(self):
+ first =
ensure_secrets_loaded(default_backends=_SERVER_DEFAULT_SECRETS_SEARCH_PATH)
+
+ assert
ensure_secrets_loaded(default_backends=_SERVER_DEFAULT_SECRETS_SEARCH_PATH) is
not first
+
+ def test_backends_are_built_once_with_a_custom_backend(self, monkeypatch):
+ monkeypatch.setenv("AIRFLOW__SECRETS__BACKEND", self.BACKEND)
+
+ first =
ensure_secrets_loaded(default_backends=_SERVER_DEFAULT_SECRETS_SEARCH_PATH)
+
+ assert
ensure_secrets_loaded(default_backends=_SERVER_DEFAULT_SECRETS_SEARCH_PATH) is
first
+
+ def test_worker_chain_is_memoised_separately_from_the_server_chain(self,
monkeypatch):
+ monkeypatch.setenv("AIRFLOW__SECRETS__BACKEND", self.BACKEND)
+
+ server =
ensure_secrets_loaded(default_backends=_SERVER_DEFAULT_SECRETS_SEARCH_PATH)
+ worker =
ensure_secrets_loaded(default_backends=DEFAULT_SECRETS_SEARCH_PATH_WORKERS)
+
+ assert server is not worker
+ assert
ensure_secrets_loaded(default_backends=DEFAULT_SECRETS_SEARCH_PATH_WORKERS) is
worker
+
+ def test_clearing_the_cache_rebuilds_backends(self, monkeypatch):
+ monkeypatch.setenv("AIRFLOW__SECRETS__BACKEND", self.BACKEND)
+
+ first =
ensure_secrets_loaded(default_backends=_SERVER_DEFAULT_SECRETS_SEARCH_PATH)
+ clear_secrets_backends_cache()
+
+ assert
ensure_secrets_loaded(default_backends=_SERVER_DEFAULT_SECRETS_SEARCH_PATH) is
not first
