amoghrajesh commented on code in PR #57744:
URL: https://github.com/apache/airflow/pull/57744#discussion_r2544547380
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -845,38 +848,64 @@
Returns:
Connection object or None if not found.
"""
+ print(f"🔵 _fetch_remote_logging_conn() called with conn_id={conn_id}")
+
# Since we need to use the API Client directly, we can't use
Connection.get as that would try to use
# SUPERVISOR_COMMS
# TODO: Store in the SecretsCache if its enabled - see #48858
if conn_id in _REMOTE_LOGGING_CONN_CACHE:
+ print(f"🔵 Connection {conn_id} found in cache")
return _REMOTE_LOGGING_CONN_CACHE[conn_id]
+ print(f"🔵 Connection {conn_id} not in cache, fetching...")
backends = ensure_secrets_backend_loaded()
+ print(f"🔵 Loaded {len(backends)} secrets backends: {[type(b).__name__ for
b in backends]}")
+
for secrets_backend in backends:
+ print(f"🔵 Trying secrets backend: {type(secrets_backend).__name__}")
try:
conn = secrets_backend.get_connection(conn_id=conn_id)
if conn:
+ print(f"🔵 Connection {conn_id} found in secrets backend
{type(secrets_backend).__name__}")
+ print(f"🔵 Connection URI: {conn.get_uri() if hasattr(conn,
'get_uri') else 'N/A'}")
_REMOTE_LOGGING_CONN_CACHE[conn_id] = conn
return conn
- except Exception:
+ else:
+ print(f"🔵 Secrets backend {type(secrets_backend).__name__}
returned None")
+ except Exception as e:
+ print(f"❌ Error getting connection from
{type(secrets_backend).__name__}: {e}")
+ import traceback
+ traceback.print_exc()
log.exception(
"Unable to retrieve connection from secrets backend (%s). "
"Checking subsequent secrets backend.",
type(secrets_backend).__name__,
)
- conn = client.connections.get(conn_id)
- if isinstance(conn, ConnectionResponse):
- conn_result = ConnectionResult.from_conn_response(conn)
- from airflow.sdk.definitions.connection import Connection
-
- result: Connection | None =
Connection(**conn_result.model_dump(exclude={"type"}, by_alias=True))
- else:
+ print(f"🔵 Connection not found in secrets backends, trying API client...")
+ try:
+ conn = client.connections.get(conn_id)
+ print(f"🔵 API client returned: {type(conn)}")
+ if isinstance(conn, ConnectionResponse):
+ conn_result = ConnectionResult.from_conn_response(conn)
+ from airflow.sdk.definitions.connection import Connection
+
+ result: Connection | None =
Connection(**conn_result.model_dump(exclude={"type"}, by_alias=True))
+ print(f"🔵 Converted to Connection object: {result}")
+ print(f"🔵 Connection URI: {result.get_uri() if hasattr(result,
'get_uri') else 'N/A'}")
Review Comment:
This is only temporary, I will revert the comments once debugging is done.
In any case, these are developer / local localstack creds, all good.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -895,27 +924,44 @@
The connection details are fetched eagerly on every invocation to avoid
retaining
per-task API client instances in global caches.
"""
+ print(f"🔵 _remote_logging_conn() context manager entered")
from airflow.sdk.log import load_remote_conn_id, load_remote_log_handler
- if load_remote_log_handler() is None or not (conn_id :=
load_remote_conn_id()):
+ handler = load_remote_log_handler()
+ print(f"🔵 load_remote_log_handler() returned: {handler}")
+ conn_id = load_remote_conn_id()
+ print(f"🔵 load_remote_conn_id() returned: {conn_id}")
+
+ if handler is None or not conn_id:
+ print(f"⚠️ No handler or conn_id - skipping connection setup.
handler={handler}, conn_id={conn_id}")
# Nothing to do
yield
return
+ print(f"🔵 Fetching connection {conn_id}...")
# Fetch connection details on-demand without caching the entire API client
instance
conn = _fetch_remote_logging_conn(conn_id, client)
if conn:
key = f"AIRFLOW_CONN_{conn_id.upper()}"
old = os.getenv(key)
- os.environ[key] = conn.get_uri()
+ conn_uri = conn.get_uri()
+ print(f"🔵 Setting environment variable {key}={conn_uri[:100] if
len(conn_uri) > 100 else conn_uri}")
Review Comment:
This is only temporary, I will revert the comments once debugging is done.
In any case, these are developer / local localstack creds, all good.
--
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]
