You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "anishgirianish (via GitHub)" <gi...@apache.org> on 2026/01/04 20:57:45 UTC
[PR] Add token refresh mechanism for Execution API (#59553) [airflow]
anishgirianish opened a new pull request, #60108:
URL: https://github.com/apache/airflow/pull/60108
Tasks waiting in Celery queue may have their JWT tokens expire before execution starts. This adds a token refresh endpoint that allows the supervisor to refresh expired tokens before task execution.
Changes:
- Add /token/refresh endpoint to Execution API
- Add client-side token refresh logic in supervisor.py
- Add tests for the new endpoint
Fixes: #59553
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
<!-- Please keep an empty line above the dashes. -->
---
---
Summary
This PR implements a token refresh mechanism for the Execution API to fix issue #59553 where tasks waiting in Celery queue fail because their JWT tokens expire before execution starts.
Problem
When using Celery executor with limited worker concurrency (e.g., GPU workers), tasks can queue for extended periods. The JWT tokens issued at task submission time may expire before the worker picks up the task, causing authentication failures with "Token has expired" errors.
Solution
Added a token refresh endpoint that allows the supervisor to detect and refresh expired tokens before task execution begins.
Changes
New Files
airflow-core/src/airflow/api_fastapi/execution_api/routes/tokens.py
- New /execution/token/refresh endpoint
- Accepts expired tokens and validates signature (not expiry)
- Verifies task instance exists and is in valid state (QUEUED or RUNNING)
- Issues new token with fresh expiration time
- Security: Only refreshes tokens for tasks that are legitimately waiting
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_tokens.py
- Comprehensive tests for the token refresh endpoint
- Tests for valid refresh, invalid tokens, non-existent tasks, and invalid task states
Modified Files
airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py
- Registers the new token router on the execution API
task-sdk/src/airflow/sdk/execution_time/supervisor.py
- Added _decode_token_claims() - Decodes JWT payload without verification
- Added _is_token_expired_or_expiring() - Checks if token is expired or expiring within buffer (60s)
- Added _refresh_token() - Calls the refresh endpoint to get a new token
- Token check happens at supervisor startup before task execution begins
task-sdk/src/airflow/sdk/api/datamodels/_generated.py
- Auto-generated datamodel updates for the new endpoint
task-sdk/tests/task_sdk/execution_time/test_supervisor.py
- Tests for token decode, expiry detection, and refresh helper functions
Testing
Environment Variables for Long-Queue Testing
When testing with tasks that queue for extended periods, use these settings:
# Increase task queued timeout (default: 600s / 10 min)
export AIRFLOW__SCHEDULER__TASK_QUEUED_TIMEOUT=86400 # 24 hours
# Increase stuck-in-queued retries (default: 2)
export AIRFLOW__SCHEDULER__NUM_STUCK_IN_QUEUED_RETRIES=50
# Celery visibility timeout - already defaults to 86400 (24h)
export AIRFLOW__CELERY__VISIBILITY_TIMEOUT=86400
# Worker prefetch multiplier - already defaults to 1
export AIRFLOW__CELERY__WORKER_PREFETCH_MULTIPLIER=1
Test Scenario
1. Create a DAG with tasks targeting a limited-concurrency queue (e.g., queue="gpu_queue")
2. Trigger multiple DAG runs (e.g., 20 runs)
3. Start only 1 worker for that queue: airflow celery worker -q gpu_queue --concurrency 1
4. Tasks will queue and their tokens will expire
5. With this fix, tokens are refreshed before execution starts
Expected Logs
When token refresh occurs:
Token expired or expiring, refreshing before execution
Token refreshed successfully
Security Considerations
- The refresh endpoint validates the JWT signature (ensures token was issued by this server)
- Only refreshes tokens for task instances in QUEUED or RUNNING state
- Invalid or tampered tokens are rejected
- The endpoint is registered on the unauthenticated router since expired tokens cannot pass normal auth
closes: #59553
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragment file, named `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2991889424
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -3234,6 +3264,44 @@ def test_invalid_scope_value_rejected(self, client, session, create_task_instanc
assert resp.status_code == 403
assert "Invalid token scope" in resp.json()["detail"]
+ def test_workload_scope_accepted_on_run_endpoint(
+ self, client, session, create_task_instance, time_machine
+ ):
+ """workload scoped tokens should be accepted on the /run endpoint."""
+ instant = timezone.parse("2024-10-31T12:00:00Z")
+ time_machine.move_to(instant, tick=False)
+
+ ti = create_task_instance(
+ task_id="test_workload_run",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=instant,
+ dag_id=str(uuid4()),
+ )
+ session.commit()
+
+ validator = mock.AsyncMock(spec=JWTValidator)
+ validator.avalidated_claims.side_effect = lambda cred, validators: {
+ "sub": str(ti.id),
+ "scope": "workload",
+ "exp": 9999999999,
+ "iat": 1000000000,
+ }
+ lifespan.registry.register_value(JWTValidator, validator)
Review Comment:
This `JWTValidator` registration is dead code -- the `client` fixture's `mock_jwt_bearer` overrides `_jwt_bearer` via FastAPI dependency overrides, so FastAPI never calls the real `_jwt_bearer` (which would use `JWTValidator` from the registry). Every request through `client` gets `scope: "execution"` regardless of what's registered here.
The test passes because execution-scoped tokens are allowed on `/run`, not because workload-scoped tokens are. To actually test workload token acceptance, the test needs to either:
1. Remove the `_jwt_bearer` dependency override for this test and let the real auth flow use this `JWTValidator`, or
2. Override `mock_jwt_bearer` to return `TIToken(..., claims={..., "scope": "workload"})` instead of the conftest's hardcoded `"scope": "execution"`.
(Re-review follow-up: this is a new finding from the latest push.)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3131749813
##########
airflow-core/docs/security/jwt_token_authentication.rst:
##########
@@ -201,16 +201,26 @@ Token structure (Execution API)
Token scopes (Execution API)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-The Execution API defines two token scopes:
+The Execution API defines two token scopes with different lifetimes:
**workload**
- A restricted scope accepted only on endpoints that explicitly opt in via
- ``Security(require_auth, scopes=["token:workload"])``. Used for endpoints that
- manage task state transitions.
+ A token embedded in the workload JSON payload when the Scheduler
+ dispatches a task. This scope is accepted only on endpoints that explicitly opt in
+ via ``Security(require_auth, scopes=["token:workload"])``. The longer lifetime
Review Comment:
Removed. Thank you.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3102624900
##########
airflow-core/docs/security/jwt_token_authentication.rst:
##########
@@ -201,16 +201,26 @@ Token structure (Execution API)
Token scopes (Execution API)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-The Execution API defines two token scopes:
+The Execution API defines two token scopes with different lifetimes:
**workload**
- A restricted scope accepted only on endpoints that explicitly opt in via
- ``Security(require_auth, scopes=["token:workload"])``. Used for endpoints that
- manage task state transitions.
+ A longer-lived token embedded in the workload JSON payload when the Scheduler
Review Comment:
Nit: The term "longer" striked me a bit... because default is then after rework the same duration like during execution time. So maybe a bit neutralizing? (Otherwise calling it "a potential longer living" might sound also very odd)
```suggestion
A token embedded in the workload JSON payload when the Scheduler
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4048433934
@anishgirianish Tons of conflict, could you rebase please
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3031435531
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -3234,6 +3264,44 @@ def test_invalid_scope_value_rejected(self, client, session, create_task_instanc
assert resp.status_code == 403
assert "Invalid token scope" in resp.json()["detail"]
+ def test_workload_scope_accepted_on_run_endpoint(
+ self, client, session, create_task_instance, time_machine
+ ):
+ """workload scoped tokens should be accepted on the /run endpoint."""
+ instant = timezone.parse("2024-10-31T12:00:00Z")
+ time_machine.move_to(instant, tick=False)
+
+ ti = create_task_instance(
+ task_id="test_workload_run",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=instant,
+ dag_id=str(uuid4()),
+ )
+ session.commit()
+
+ validator = mock.AsyncMock(spec=JWTValidator)
+ validator.avalidated_claims.side_effect = lambda cred, validators: {
+ "sub": str(ti.id),
+ "scope": "workload",
+ "exp": 9999999999,
+ "iat": 1000000000,
+ }
+ lifespan.registry.register_value(JWTValidator, validator)
Review Comment:
Removed the standalone JWTValidator registration. The TestTokenTypeValidation class now only registers it
explicitly when using _use_real_jwt_bearer fixture, so it's no longer dead code.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3132796957
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -142,6 +145,12 @@ async def dispatch(self, request: Request, call_next):
validator: JWTValidator = await services.aget(JWTValidator)
claims = await validator.avalidated_claims(token, {})
+ # Workload tokens are long-lived and meant to survive queue
+ # wait times so avoid refreshing them. If avalidated_claims
+ # raises for a workload token, the outer except handles it.
+ if claims.get("scope") == "workload":
+ return response
+
now = int(time.time())
validity = conf.getint("execution_api", "jwt_expiration_time")
refresh_when_less_than = max(int(validity * 0.20), 30)
Review Comment:
Updated/Added thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3132791192
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -230,6 +238,17 @@ def ti_run(
last_heartbeat_at=timezone.utcnow(),
)
+ execution_token: str | None = None
+ if token.claims.scope == "workload":
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except (jwt.PyJWTError, ValueError, KeyError):
Review Comment:
Went with later one, dropped the try/expcet entierly.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961453328
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -447,15 +447,33 @@ def signing_arg(self) -> AllowedPrivateKeys | str:
assert self._secret_key
return self._secret_key
- def generate(self, extras: dict[str, Any] | None = None, headers: dict[str, Any] | None = None) -> str:
+ def generate_workload_token(self, sub: str) -> str:
+ """Generate a long-lived workload token for executor queues."""
+ from airflow.configuration import conf
+
+ workload_valid_for = conf.getint(
+ "execution_api", "jwt_workload_token_expiration_time", fallback=86400
+ )
+ return self.generate(
+ extras={"sub": sub, "scope": "workload"},
+ valid_for=workload_valid_for,
+ )
Review Comment:
This went through serveral of iteration, and the implementation has evolved along the way, so description was outdated, but now I have updated it
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3131687862
##########
airflow-core/docs/security/jwt_token_authentication.rst:
##########
@@ -201,16 +201,26 @@ Token structure (Execution API)
Token scopes (Execution API)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-The Execution API defines two token scopes:
+The Execution API defines two token scopes with different lifetimes:
**workload**
- A restricted scope accepted only on endpoints that explicitly opt in via
- ``Security(require_auth, scopes=["token:workload"])``. Used for endpoints that
- manage task state transitions.
+ A token embedded in the workload JSON payload when the Scheduler
+ dispatches a task. This scope is accepted only on endpoints that explicitly opt in
+ via ``Security(require_auth, scopes=["token:workload"])``. The longer lifetime
Review Comment:
I know I had this before, but on seeing this now I think this is an implementation detail we should remove from the public docs -- this is for people to know how Airflow auth works/operates, not how to develope Airflow. The rest is 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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2957077744
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -246,6 +246,36 @@ def test_ti_run_state_to_running(
)
assert response.status_code == 409
+ def test_ti_run_returns_execution_token(self, client, session, create_task_instance, time_machine):
+ """PATCH /run should return an X-Execution-Token header on success."""
+ instant = timezone.parse("2024-10-31T12:00:00Z")
+ time_machine.move_to(instant, tick=False)
+
+ ti = create_task_instance(
+ task_id="test_exec_token",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=instant,
+ dag_id=str(uuid4()),
+ )
+ session.commit()
+
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "test-host",
+ "unixname": "test-user",
+ "pid": 100,
+ "start_date": "2024-10-31T12:00:00Z",
+ },
+ )
+
+ assert response.status_code == 200
+ assert "X-Execution-Token" in response.headers
+ assert response.headers["X-Execution-Token"] == "mock-execution-token"
+
Review Comment:
New behavior allows workload-scoped tokens to call `PATCH /run`, but there is no regression test asserting that a token with `scope="workload"` is accepted on `/run` (only that workload tokens are rejected elsewhere). Adding a test that uses the real JWT bearer/validator to return `scope="workload"` and expects 200 from `/run` would prevent accidental removal of the `token:workload` allowance.
##########
devel-common/src/tests_common/test_utils/mock_executor.py:
##########
@@ -59,6 +59,7 @@ def __init__(self, do_update=True, *args, **kwargs):
# Mock JWT generator for token generation
mock_jwt_generator = MagicMock()
mock_jwt_generator.generate.return_value = "mock-token"
+ mock_jwt_generator.generate_workload_token.return_value = "mock-workload-token"
Review Comment:
`MagicMock()` is created without `spec`/`autospec` for the JWT generator. With the new `generate_workload_token` call path, an unspec'd mock can hide attribute/method mismatches and make tests less reliable. Prefer `MagicMock(spec=JWTGenerator)` (or `create_autospec`) so incorrect API usage fails fast.
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -447,15 +447,33 @@ def signing_arg(self) -> AllowedPrivateKeys | str:
assert self._secret_key
return self._secret_key
- def generate(self, extras: dict[str, Any] | None = None, headers: dict[str, Any] | None = None) -> str:
+ def generate_workload_token(self, sub: str) -> str:
+ """Generate a long-lived workload token for executor queues."""
+ from airflow.configuration import conf
+
+ workload_valid_for = conf.getint(
+ "execution_api", "jwt_workload_token_expiration_time", fallback=86400
+ )
+ return self.generate(
+ extras={"sub": sub, "scope": "workload"},
+ valid_for=workload_valid_for,
+ )
Review Comment:
The PR description mentions adding `JWTBearerWorkloadDep`, `TOKEN_SCOPE_WORKLOAD`, and `SCOPE_MAPPING`, but these identifiers don't exist in the code changes. The implementation here uses a `scope="workload"` claim plus `token:*` FastAPI Security scopes/`ExecutionAPIRoute` logic instead. Please update the PR description to reflect the actual approach so reviewers/users aren’t looking for non-existent components.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -92,18 +94,21 @@
@ti_id_router.patch(
"/{task_instance_id}/run",
status_code=status.HTTP_200_OK,
+ dependencies=[Security(require_auth, scopes=["token:execution", "token:workload"])],
responses={
Review Comment:
This endpoint adds an extra `Security(require_auth, scopes=[...])` dependency solely to widen allowed token types, but `require_auth` is already applied at the router level. This likely causes `require_auth` to run twice per request for `/run`, which is redundant and can be confusing. Consider using a non-executing marker dependency (or another mechanism) to add the `token:*` scopes for `ExecutionAPIRoute` without re-running the full auth dependency.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -92,18 +94,21 @@
@ti_id_router.patch(
"/{task_instance_id}/run",
status_code=status.HTTP_200_OK,
+ dependencies=[Security(require_auth, scopes=["token:execution", "token:workload"])],
responses={
status.HTTP_404_NOT_FOUND: {"description": "Task Instance not found"},
status.HTTP_409_CONFLICT: {"description": "The TI is already in the requested state"},
HTTP_422_UNPROCESSABLE_CONTENT: {"description": "Invalid payload for the state transition"},
},
response_model_exclude_unset=True,
)
-def ti_run(
+async def ti_run(
task_instance_id: UUID,
ti_run_payload: Annotated[TIEnterRunningPayload, Body()],
+ response: Response,
session: SessionDep,
dag_bag: DagBagDep,
+ services=DepContainer,
Review Comment:
`ti_run` was changed to `async def` but still uses `SessionDep` (sync SQLAlchemy session) and performs many blocking DB operations directly in the event loop. This can reduce throughput for the Execution API under load. Consider keeping this endpoint sync (and avoid awaiting the svcs container directly), or migrate the DB access to `AsyncSessionDep`/async SQLAlchemy so the endpoint can safely be async.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961427807
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -92,18 +94,21 @@
@ti_id_router.patch(
"/{task_instance_id}/run",
status_code=status.HTTP_200_OK,
+ dependencies=[Security(require_auth, scopes=["token:execution", "token:workload"])],
responses={
status.HTTP_404_NOT_FOUND: {"description": "Task Instance not found"},
status.HTTP_409_CONFLICT: {"description": "The TI is already in the requested state"},
HTTP_422_UNPROCESSABLE_CONTENT: {"description": "Invalid payload for the state transition"},
},
response_model_exclude_unset=True,
)
-def ti_run(
+async def ti_run(
task_instance_id: UUID,
ti_run_payload: Annotated[TIEnterRunningPayload, Body()],
+ response: Response,
session: SessionDep,
dag_bag: DagBagDep,
+ services=DepContainer,
Review Comment:
Done
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3003907577
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -332,10 +343,29 @@ async def always_allow(request: Request):
)
return TIToken(id=ti_id, claims={"scope": "execution"})
+ # Override _container (the svcs service locator behind DepContainer).
+ # The default _container reads request.app.state.svcs_registry, but
+ # Cadwyn's versioned sub-apps don't inherit the main app's state,
+ # so lookups raise ServiceNotFoundError. This registry provides
+ # services needed by routes called during dag.test().
+ #
Review Comment:
added comment thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2980465748
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -2069,6 +2069,15 @@ execution_api:
type: integer
example: ~
default: "600"
+ jwt_workload_token_expiration_time:
+ description: |
+ Seconds until workload JWT tokens expire. These long-lived tokens are sent
+ with task workloads to executors and can only call the /run endpoint.
+ Set long enough to cover maximum expected queue wait time.
+ version_added: 3.2.0
Review Comment:
+1
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -952,7 +952,10 @@ def __init__(self, *, base_url: str | None, dry_run: bool = False, token: str, *
)
def _update_auth(self, response: httpx.Response):
- if new_token := response.headers.get("Refreshed-API-Token"):
+ if new_token := response.headers.get("X-Execution-Token"):
+ log.debug("Received execution token, swapping auth")
+ self.auth = BearerAuth(new_token)
+ elif new_token := response.headers.get("Refreshed-API-Token"):
Review Comment:
I agree too.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -281,14 +286,18 @@ def ti_run(
if ti.next_method:
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
-
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id)})
Review Comment:
Do we need to set the scope as `execution` here to be explicit?
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -281,14 +286,18 @@ def ti_run(
if ti.next_method:
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
-
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id)})
+ response.headers["X-Execution-Token"] = execution_token
Review Comment:
If `services.get(JWTGenerator)` raises or fails for whatever reason / if `generate` raises, the API would return 500 here, but we committed the TI as running earlier. To avoid this, I suggest to move this up in the try/except itself.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2723676113
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -78,6 +79,12 @@
if TYPE_CHECKING:
from sqlalchemy.sql.dml import Update
+log = structlog.get_logger(__name__)
+
+JWTBearerWorkloadDep = Depends(JWTBearerWorkloadScope(path_param_name="task_instance_id"))
+
+ti_run_router = VersionedAPIRouter(dependencies=[JWTBearerWorkloadDep])
Review Comment:
removed extra router in the recent commit
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2694572709
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -296,25 +296,74 @@ class InProcessExecutionAPI:
@cached_property
def app(self):
if not self._app:
+ import os
+ from base64 import urlsafe_b64encode
+
from airflow.api_fastapi.common.dagbag import create_dag_bag
from airflow.api_fastapi.execution_api.app import create_task_execution_api_app
from airflow.api_fastapi.execution_api.deps import (
JWTBearerDep,
JWTBearerTIPathDep,
)
from airflow.api_fastapi.execution_api.routes.connections import has_connection_access
+ from airflow.api_fastapi.execution_api.routes.task_instances import JWTBearerWorkloadDep
from airflow.api_fastapi.execution_api.routes.variables import has_variable_access
from airflow.api_fastapi.execution_api.routes.xcoms import has_xcom_access
+ from airflow.configuration import conf
+
+ # Ensure JWT secret is available for in-process execution.
+ # The /run endpoint needs JWTGenerator to issue execution tokens.
+ # If the config option is empty, generate a random one for the duration of this process.
+ if not conf.get("api_auth", "jwt_secret", fallback=None):
+ logger.debug(
+ "`api_auth/jwt_secret` is not set, generating a temporary one for in-process execution"
+ )
+ conf.set("api_auth", "jwt_secret", urlsafe_b64encode(os.urandom(16)).decode())
Review Comment:
`os.urandom` shouldn't be used for anything security related -- please use https://docs.python.org/3/library/secrets.html, likely `secrets.token_urlsafe(16)`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2694618755
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -78,6 +79,12 @@
if TYPE_CHECKING:
from sqlalchemy.sql.dml import Update
+log = structlog.get_logger(__name__)
+
+JWTBearerWorkloadDep = Depends(JWTBearerWorkloadScope(path_param_name="task_instance_id"))
+
+ti_run_router = VersionedAPIRouter(dependencies=[JWTBearerWorkloadDep])
Review Comment:
We shouldn't need this extra router, when we already have `router`, and we can then declare the JWTBearerWorkloadDep on the path
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2694574837
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -296,25 +296,74 @@ class InProcessExecutionAPI:
@cached_property
def app(self):
if not self._app:
+ import os
+ from base64 import urlsafe_b64encode
+
from airflow.api_fastapi.common.dagbag import create_dag_bag
from airflow.api_fastapi.execution_api.app import create_task_execution_api_app
from airflow.api_fastapi.execution_api.deps import (
JWTBearerDep,
JWTBearerTIPathDep,
)
from airflow.api_fastapi.execution_api.routes.connections import has_connection_access
+ from airflow.api_fastapi.execution_api.routes.task_instances import JWTBearerWorkloadDep
from airflow.api_fastapi.execution_api.routes.variables import has_variable_access
from airflow.api_fastapi.execution_api.routes.xcoms import has_xcom_access
+ from airflow.configuration import conf
+
+ # Ensure JWT secret is available for in-process execution.
+ # The /run endpoint needs JWTGenerator to issue execution tokens.
+ # If the config option is empty, generate a random one for the duration of this process.
+ if not conf.get("api_auth", "jwt_secret", fallback=None):
+ logger.debug(
+ "`api_auth/jwt_secret` is not set, generating a temporary one for in-process execution"
+ )
+ conf.set("api_auth", "jwt_secret", urlsafe_b64encode(os.urandom(16)).decode())
Review Comment:
That said, I'm a bit surprised we need to to this at all here -- isn't this done elsewhere already?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-3811693774
I haven't forgotten about my suggestion - I'm still working on it in parallel with other things
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2980434615
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -3226,6 +3256,44 @@ def test_invalid_scope_value_rejected(self, client, session, create_task_instanc
assert resp.status_code == 403
assert "Invalid token scope" in resp.json()["detail"]
+ def test_workload_scope_accepted_on_run_endpoint(
Review Comment:
Maybe we already have a test for this, but I think the negative test is important too -- that the workload scoped token is _not_ accepted on other endpoints. You don't have to exhaustively check all endpoints, but testing at least one other to ensure it gives a 4xx is worthwhile I think
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3088750164
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -2093,6 +2093,15 @@ execution_api:
type: integer
example: ~
default: "600"
+ jwt_workload_token_expiration_time:
Review Comment:
I thought a moment about this and actually am thinking... do we need a config parameter for this? There is a parameter `task_queued_timeout` already (see https://airflow.apache.org/docs/apache-airflow/stable/configurations-ref.html#task-queued-timeout) - after this tasks are made failed if they starve in the queue.
Would it be also an idea to make expiration of JWT equal to the queue timeout? (Maybe a few seconds more as buffer...) - I see no reason making this JWT defautl to 24h if the task is evicted from queued state after 10min per default (which usually is a good idea to set higher in a loaded environment)
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -287,14 +292,24 @@ def ti_run(
if ti.next_method:
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
-
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except Exception:
Review Comment:
Do we need a broad exception here or can we point to specific one(s)?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3093765663
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -2093,6 +2093,15 @@ execution_api:
type: integer
example: ~
default: "600"
+ jwt_workload_token_expiration_time:
Review Comment:
I think you are right Jens, good catch.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3103987159
##########
airflow-core/docs/security/jwt_token_authentication.rst:
##########
@@ -201,16 +201,26 @@ Token structure (Execution API)
Token scopes (Execution API)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
-The Execution API defines two token scopes:
+The Execution API defines two token scopes with different lifetimes:
**workload**
- A restricted scope accepted only on endpoints that explicitly opt in via
- ``Security(require_auth, scopes=["token:workload"])``. Used for endpoints that
- manage task state transitions.
+ A longer-lived token embedded in the workload JSON payload when the Scheduler
Review Comment:
updated thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4297930420
Hi @potiuk, thank you so much for the detailed review! I have addressed the feedback in the latest push. Would like to request you for your re-review whenever you get a chance.
Thank you.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961436273
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -92,18 +94,21 @@
@ti_id_router.patch(
"/{task_instance_id}/run",
status_code=status.HTTP_200_OK,
+ dependencies=[Security(require_auth, scopes=["token:execution", "token:workload"])],
responses={
Review Comment:
JWTBearer caches on request scope so validation only runs once. The endpoint-level Security is needed for ExecutionAPIRoute to pick up the token:workload scope at registration time.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961406770
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -92,18 +94,21 @@
@ti_id_router.patch(
"/{task_instance_id}/run",
status_code=status.HTTP_200_OK,
+ dependencies=[Security(require_auth, scopes=["token:execution", "token:workload"])],
responses={
status.HTTP_404_NOT_FOUND: {"description": "Task Instance not found"},
status.HTTP_409_CONFLICT: {"description": "The TI is already in the requested state"},
HTTP_422_UNPROCESSABLE_CONTENT: {"description": "Invalid payload for the state transition"},
},
response_model_exclude_unset=True,
)
-def ti_run(
+async def ti_run(
Review Comment:
thank you. updated
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3003906744
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -281,14 +286,18 @@ def ti_run(
if ti.next_method:
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
-
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id)})
Review Comment:
yep not its explicitly set, thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3003909398
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -286,14 +291,18 @@ def ti_run(
if ti.next_method:
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
-
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id)})
Review Comment:
wrapped this with its own try/except, thank you.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3132773016
##########
airflow-core/src/airflow/executors/workloads/base.py:
##########
@@ -76,7 +76,12 @@ class BaseWorkloadSchema(BaseModel):
@staticmethod
def generate_token(sub_id: str, generator: JWTGenerator | None = None) -> str:
- return generator.generate({"sub": sub_id}) if generator else ""
+ if not generator:
+ return ""
+ return generator.generate(
Review Comment:
Thank you so much. Adopted this thanks
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4314025504
> Thanks for sticking with this @anishgirianish!
@ashb Thank you so much
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4308465854
> #protm
@jscheffl thank you so much
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3125232646
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -289,13 +295,24 @@ def ti_run(
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
context.start_date = ti.start_date
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except jwt.PyJWTError:
+ log.exception("Failed to generate execution token for task instance %s", task_instance_id)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Token generation failed"
+ )
+ response.headers["Refreshed-API-Token"] = execution_token
Review Comment:
Done, `ti_run` now accepts `token: TIToken = CurrentTIToken` and the `generator.generate(...)` call is gated on `if token.claims.scope == "workload":`.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -289,13 +295,24 @@ def ti_run(
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
context.start_date = ti.start_date
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except jwt.PyJWTError:
+ log.exception("Failed to generate execution token for task instance %s", task_instance_id)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Token generation failed"
+ )
+ response.headers["Refreshed-API-Token"] = execution_token
Review Comment:
Added a comment on the `response.headers["Refreshed-API-Token"]` assignment.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -289,13 +295,24 @@ def ti_run(
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
context.start_date = ti.start_date
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except jwt.PyJWTError:
Review Comment:
Broadened to `except (jwt.PyJWTError, ValueError, KeyError)` as suggested.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -289,13 +295,24 @@ def ti_run(
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
context.start_date = ti.start_date
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except jwt.PyJWTError:
+ log.exception("Failed to generate execution token for task instance %s", task_instance_id)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Token generation failed"
+ )
+ response.headers["Refreshed-API-Token"] = execution_token
+
+ return context
Review Comment:
Moved token generation before `session.execute(query)` so a `generate()` failure leaves the TI in QUEUED rather than stuck in RUNNING with no recovery path.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -289,13 +295,24 @@ def ti_run(
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
context.start_date = ti.start_date
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except jwt.PyJWTError:
+ log.exception("Failed to generate execution token for task instance %s", task_instance_id)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Token generation failed"
+ )
+ response.headers["Refreshed-API-Token"] = execution_token
+
+ return context
Review Comment:
Moved token generation before `session.execute(query)` ,
##########
airflow-core/tests/unit/api_fastapi/execution_api/conftest.py:
##########
@@ -55,6 +59,10 @@ async def mock_jwt_bearer(request: Request):
exec_app.dependency_overrides[_jwt_bearer] = mock_jwt_bearer
with TestClient(app, headers={"Authorization": "Bearer fake"}) as client:
+ mock_generator = MagicMock(spec=JWTGenerator)
+ mock_generator.generate.return_value = "mock-execution-token"
+ lifespan.registry.register_value(JWTGenerator, mock_generator)
Review Comment:
Removed the `register_value` call from the shared fixture entirely; the mock generator is now registered inside `test_ti_run_returns_execution_token` only, where it's needed.
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -281,6 +281,36 @@ def test_ti_run_state_to_running(
)
assert response.status_code == 409
+ def test_ti_run_returns_execution_token(self, client, session, create_task_instance, time_machine):
+ """PATCH /run should return a Refreshed-API-Token header on success."""
+ instant = timezone.parse("2024-10-31T12:00:00Z")
+ time_machine.move_to(instant, tick=False)
+
+ ti = create_task_instance(
+ task_id="test_exec_token",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=instant,
+ dag_id=str(uuid4()),
+ )
+ session.commit()
+
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "test-host",
+ "unixname": "test-user",
+ "pid": 100,
+ "start_date": "2024-10-31T12:00:00Z",
+ },
+ )
+
+ assert response.status_code == 200
+ assert "Refreshed-API-Token" in response.headers
+ assert response.headers["Refreshed-API-Token"] == "mock-execution-token"
Review Comment:
Added `mock_gen.generate.assert_called_once()` plus assertions on `extras["scope"] == "execution"` and `extras["sub"] == str(ti.id)`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3131873700
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -418,6 +418,10 @@ class JWTGenerator:
kid: str = attrs.field(default=attrs.Factory(_generate_kid, takes_self=True))
valid_for: float
+ workload_valid_for: float = attrs.field(
Review Comment:
`JWTGenerator` is used by auth managers (FAB, Simple) via `BaseAuthManager._get_token_signer`, not just the execution API. Adding a class-level field that unconditionally reads `[scheduler] task_queued_timeout` at construction couples a general auth primitive to scheduler config. In an api-server-only or auth-manager-only deployment, this pulls a key that's irrelevant to the caller and potentially absent.
Can you move `workload_valid_for` out of `JWTGenerator` and compute it at the workload-generation site (e.g., in `BaseWorkloadSchema.generate_token`)? `generate()` already accepts a per-call `valid_for` override, so the generator doesn't need to know about this concept.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -230,6 +238,17 @@ def ti_run(
last_heartbeat_at=timezone.utcnow(),
)
+ execution_token: str | None = None
Review Comment:
Token minting runs before the state-transition executes. If the state update fails (SQLAlchemyError at the `session.execute(query)` below, or the client hits the 409 path further down), you've minted a token that's thrown away.
More importantly, in the duplicate-request branch earlier in this handler (where we return early for a TI already in RUNNING), a workload-token call still mints a new execution token even though the worker already has one from its first request.
Can this block move down to just before `return context` so it only fires on the success path?
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -281,6 +284,54 @@ def test_ti_run_state_to_running(
)
assert response.status_code == 409
+ def test_ti_run_returns_execution_token(
+ self, client, exec_app, session, create_task_instance, time_machine
+ ):
+ """PATCH /run with a workload token should swap to an execution-scoped token."""
+ instant = timezone.parse("2024-10-31T12:00:00Z")
+ time_machine.move_to(instant, tick=False)
+
+ ti = create_task_instance(
+ task_id="test_exec_token",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=instant,
+ dag_id=str(uuid4()),
+ )
+ session.commit()
+
+ mock_gen = mock.MagicMock(spec=JWTGenerator)
+ mock_gen.generate.return_value = "mock-execution-token"
+ lifespan.registry.register_value(JWTGenerator, mock_gen)
Review Comment:
`lifespan.registry.register_value(JWTGenerator, mock_gen)` registers into module-level state that persists past this test. I don't see a teardown that removes it, so later tests in the session will see this mock generator instead of the real one -- that's a hidden test-pollution source.
Same pattern exists in `TestTokenTypeValidation` with `JWTValidator`.
Can you add an autouse fixture (or `finally` block) that restores/clears the registry entry after each test? The PR description mentions "registry teardown to prevent test pollution" but I can't find the actual teardown in this file.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -230,6 +238,17 @@ def ti_run(
last_heartbeat_at=timezone.utcnow(),
)
+ execution_token: str | None = None
+ if token.claims.scope == "workload":
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except (jwt.PyJWTError, ValueError, KeyError):
Review Comment:
This catch is too broad. `generate()` is deterministic; `ValueError`/`KeyError` here would indicate a programming bug (bad extras dict, missing signing key) that should surface with a real stack trace, not a generic 500.
Also worth noting: `services.get(JWTGenerator)` raises `svcs.exceptions.ServiceNotFoundError`, which isn't in this tuple -- so if the registry is misconfigured you'll get an unhelpful 500 from the global handler anyway.
Suggest narrowing to `jwt.PyJWTError` only, or dropping the try/except entirely and letting the global handler log the real error.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -333,10 +342,30 @@ async def always_allow(request: Request):
claims = TIClaims(scope="execution")
return TIToken(id=ti_id, claims=claims)
+ # Override _container (the svcs service locator behind DepContainer).
+ # The default _container reads request.app.state.svcs_registry, but
+ # Cadwyn's versioned sub-apps don't inherit the main app's state,
+ # so lookups raise ServiceNotFoundError. This registry provides
+ # services needed by routes called during dag.test().
+ # Note: tokens generated by this stub are never validated since
+ # _jwt_bearer is overridden with always_allow in dag.test() mode.
+ stub_generator = JWTGenerator(
Review Comment:
The stub generator uses a per-process random secret (`secrets.token_urlsafe(32)`), so any token it signs is unverifiable against the real validator. The comment above says tokens aren't validated because `_jwt_bearer` is overridden, but `/run` now actively mints an execution token via `services.get(JWTGenerator).generate(...)` and returns it in `Refreshed-API-Token`.
Under `dag.test()` the current workload branch is skipped because `always_allow` returns `scope="execution"`, but this is a fragile invariant. Any future path that triggers the workload-token swap (or any test simulating a workload token through the in-process path) would hand the client a token no validator can verify.
Consider either reusing the real signing key from `get_signing_args()`, or not registering a `JWTGenerator` under in-process mode at all so callers fail loudly if they ever exercise that path.
##########
airflow-core/src/airflow/executors/workloads/base.py:
##########
@@ -76,7 +76,12 @@ class BaseWorkloadSchema(BaseModel):
@staticmethod
def generate_token(sub_id: str, generator: JWTGenerator | None = None) -> str:
- return generator.generate({"sub": sub_id}) if generator else ""
+ if not generator:
+ return ""
+ return generator.generate(
Review Comment:
Since `JWTGenerator` ideally shouldn't know about `task_queued_timeout`, this is the right place to read it:
```python
valid_for = conf.getfloat("scheduler", "task_queued_timeout")
return generator.generate(
extras={"sub": sub_id, "scope": "workload"},
valid_for=valid_for,
)
```
That also lets you drop `workload_valid_for` from `JWTGenerator` entirely.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -142,6 +145,12 @@ async def dispatch(self, request: Request, call_next):
validator: JWTValidator = await services.aget(JWTValidator)
claims = await validator.avalidated_claims(token, {})
+ # Workload tokens are long-lived and meant to survive queue
+ # wait times so avoid refreshing them. If avalidated_claims
+ # raises for a workload token, the outer except handles it.
+ if claims.get("scope") == "workload":
+ return response
+
now = int(time.time())
validity = conf.getint("execution_api", "jwt_expiration_time")
refresh_when_less_than = max(int(validity * 0.20), 30)
Review Comment:
The refresh-threshold denominator is `jwt_expiration_time`, which only makes sense for execution tokens. The new early-return at line 151 (`if claims.get("scope") == "workload": return response`) guards workload tokens today, but if someone removes that guard or adds a new scope with a different lifetime, this threshold math silently becomes wrong (e.g. comparing a 5-minute workload-token's remaining life against 20% of the 10-minute execution-token lifetime).
Worth computing the threshold from the actual token lifetime instead:
```python
token_lifetime = int(claims.get("exp", 0)) - int(claims.get("iat", 0))
refresh_when_less_than = max(int(token_lifetime * 0.20), 30)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961410559
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -282,6 +287,10 @@ def ti_run(
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
+ generator: JWTGenerator = await services.aget(JWTGenerator)
Review Comment:
thank you for pointing out, Moved and updated
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961426434
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -939,7 +939,10 @@ def __init__(self, *, base_url: str | None, dry_run: bool = False, token: str, *
)
def _update_auth(self, response: httpx.Response):
Review Comment:
Added a check to skip the workload token. Thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2957015411
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -92,18 +94,21 @@
@ti_id_router.patch(
"/{task_instance_id}/run",
status_code=status.HTTP_200_OK,
+ dependencies=[Security(require_auth, scopes=["token:execution", "token:workload"])],
responses={
status.HTTP_404_NOT_FOUND: {"description": "Task Instance not found"},
status.HTTP_409_CONFLICT: {"description": "The TI is already in the requested state"},
HTTP_422_UNPROCESSABLE_CONTENT: {"description": "Invalid payload for the state transition"},
},
response_model_exclude_unset=True,
)
-def ti_run(
+async def ti_run(
Review Comment:
Changing `ti_run` from `def` to `async def` will block the event loop. The function body does extensive synchronous SQLAlchemy calls (`session.execute()`, `session.scalars()`, etc.). FastAPI runs sync `def` handlers in a threadpool automatically, but `async def` handlers run directly on the event loop — so every synchronous DB call here becomes a blocking operation on the loop.
Since `_jwt_generator` is registered as a sync factory, you can use the sync `services.get(JWTGenerator)` instead of `await services.aget(JWTGenerator)`, and keep this function as `def`.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -282,6 +287,10 @@ def ti_run(
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
+ generator: JWTGenerator = await services.aget(JWTGenerator)
Review Comment:
This token generation block sits inside the `try: ... except SQLAlchemyError:` block (the `except` is at line 295). If `services.aget()` or `generator.generate()` raises a non-SQLAlchemy exception, it propagates as an unhandled 500.
Consider moving these 3 lines after the try/except, right before `return context`, so only the actual DB operations are covered by the SQLAlchemy catch.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4236857255
Hi @ashb @kaxil @amoghrajesh. Thank you very much for the last review. I have addressed all the feedback and updated the pr. I would like to request you for your review whenever you get a chance. Thank you so much.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4263165104
> LGTM, but as Jens pointed out we should remove the extra new validity we used and use the existing queued timeout.
>
> Oh, though I guess that could be not set, in which case those tokens would be valid for ever which probably isn't the correct behaviour. Hmmmm.
The queued timeout is defaulted to float(600.0) - so should not be end-less. (I believe the default of only 10min is way too short in my view - and today if you make your first setup and have the first real load really takes a long time to understand where the problem lies...)
<img width="740" height="479" alt="image" src="https://github.com/user-attachments/assets/7ee51809-db28-48d3-8ea4-3b962a2b7432" />
So we should point this out in the docs with this PR that JWT token lifetime is also influenced by this parameter.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2991889424
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -3234,6 +3264,44 @@ def test_invalid_scope_value_rejected(self, client, session, create_task_instanc
assert resp.status_code == 403
assert "Invalid token scope" in resp.json()["detail"]
+ def test_workload_scope_accepted_on_run_endpoint(
+ self, client, session, create_task_instance, time_machine
+ ):
+ """workload scoped tokens should be accepted on the /run endpoint."""
+ instant = timezone.parse("2024-10-31T12:00:00Z")
+ time_machine.move_to(instant, tick=False)
+
+ ti = create_task_instance(
+ task_id="test_workload_run",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=instant,
+ dag_id=str(uuid4()),
+ )
+ session.commit()
+
+ validator = mock.AsyncMock(spec=JWTValidator)
+ validator.avalidated_claims.side_effect = lambda cred, validators: {
+ "sub": str(ti.id),
+ "scope": "workload",
+ "exp": 9999999999,
+ "iat": 1000000000,
+ }
+ lifespan.registry.register_value(JWTValidator, validator)
Review Comment:
This `JWTValidator` registration is dead code -- the `client` fixture's `mock_jwt_bearer` overrides `_jwt_bearer` via FastAPI dependency overrides, so FastAPI never calls the real `_jwt_bearer` (which would use `JWTValidator` from the registry). Every request through `client` gets `scope: "execution"` regardless of what's registered here.
The test passes because execution-scoped tokens are allowed on `/run`, not because workload-scoped tokens are. To actually test workload token acceptance, the test needs to either:
1. Remove the `_jwt_bearer` dependency override for this test and let the real auth flow use this `JWTValidator`, or
2. Override `mock_jwt_bearer` to return `TIToken(..., claims={..., "scope": "workload"})` instead of the conftest's hardcoded `"scope": "execution"`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3003905420
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -952,7 +952,10 @@ def __init__(self, *, base_url: str | None, dry_run: bool = False, token: str, *
)
def _update_auth(self, response: httpx.Response):
- if new_token := response.headers.get("Refreshed-API-Token"):
+ if new_token := response.headers.get("X-Execution-Token"):
+ log.debug("Received execution token, swapping auth")
+ self.auth = BearerAuth(new_token)
+ elif new_token := response.headers.get("Refreshed-API-Token"):
Review Comment:
Yep, I agree as well, removed and now it only uses `Refreshed-API-Token`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add token refresh mechanism for Execution API (#59553) [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-3719154109
> Hi @tirkarthi, Thanks for pointing out the middleware approach from #55506 - that's helpful context.
>
> I took a stab at extending that pattern in #60197, handling expired tokens transparently in JWTBearer + middleware so no client-side changes are needed. Would love your thoughts on it.
>
> Totally happy to go with whichever approach the team feels is better!
>
> cc: @vincbeck @pierrejeambrun
Would love to hear @ashb or @amoghrajesh 's opinion on this one
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2694581443
##########
airflow-core/src/airflow/api_fastapi/execution_api/deps.py:
##########
@@ -87,15 +84,56 @@ async def __call__( # type: ignore[override]
else:
validators = self.required_claims
claims = await validator.avalidated_claims(creds.credentials, validators)
+
+ # Let subclasses validate scope
+ self._check_scope(claims)
+
return TIToken(id=claims["sub"], claims=claims)
+ except HTTPException:
+ raise
except Exception as err:
- log.warning(
- "Failed to validate JWT",
- exc_info=True,
- token=creds.credentials,
- )
+ log.warning("Failed to validate JWT", exc_info=True)
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Invalid auth token: {err}")
+ def _check_scope(self, claims: dict[str, Any]) -> None:
+ """Override in subclasses to validate scope. Raise HTTPException if invalid."""
+ pass
+
+
+class JWTBearer(_BaseJWTBearer):
+ """
+ JWT validation that rejects workload-scoped tokens.
+
+ Used for most Execution API endpoints. Workload-scoped tokens can only be used
+ on the /run endpoint, which exchanges them for short-lived execution tokens.
+ """
+
+ def _check_scope(self, claims: dict[str, Any]) -> None:
+ if claims.get("scope") == TOKEN_SCOPE_WORKLOAD:
Review Comment:
I wonder if this should be
```suggestion
if claims.get("scope"):
```
so that it works when other scopes might be added in future?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961446036
##########
devel-common/src/tests_common/test_utils/mock_executor.py:
##########
@@ -59,6 +59,7 @@ def __init__(self, do_update=True, *args, **kwargs):
# Mock JWT generator for token generation
mock_jwt_generator = MagicMock()
mock_jwt_generator.generate.return_value = "mock-token"
+ mock_jwt_generator.generate_workload_token.return_value = "mock-workload-token"
Review Comment:
Fixed thanks
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-3762934122
> I'm not sold yet on the way we are handling the dependencies/mutliple routers etc. Let me have a look at what options we have in FastAPI as i feel this should be possible in a nicer manner.
>
> I'm seeing if we can make use of https://fastapi.tiangolo.com/advanced/security/oauth2-scopes/ or something like it.
That sounds like a great approach! I'd be really grateful if you could share your findings on FastAPI OAuth2 scopes - happy to refactor based on your recommendations. Thank you so much for taking the time to look into this!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4102313582
Hi @kaxil @ashb, Thank you very much for the detailed review. I have addresed the feeback in the latest push. Would like to request you for your re-reveiw when ever you get a chance. Thank you so much
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3003901990
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -2069,6 +2069,15 @@ execution_api:
type: integer
example: ~
default: "600"
+ jwt_workload_token_expiration_time:
+ description: |
+ Seconds until workload JWT tokens expire. These long-lived tokens are sent
+ with task workloads to executors and can only call the /run endpoint.
+ Set long enough to cover maximum expected queue wait time.
+ version_added: 3.2.0
Review Comment:
Done thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-3787995719
Hi @ashb, thank you for the feedback on the earlier iteration.
As you suggested, I implemented scope validation using FastAPI's SecurityScopes mechanism. The approach uses a single router with JWTBearerBaseDep at the router level for basic auth, and scope checks are done at the endpoint/module level via JWTBearerDep (execution scope) and JWTBearerWorkloadDep (workload scope for /run only).
Would really appreciate another look when you get a chance. Happy to make any further adjustments based on your feedback. Thanks!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3066491837
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -311,9 +319,13 @@ class InProcessExecutionAPI:
@cached_property
def app(self):
if not self._app:
+ import svcs
+
+ from airflow.api_fastapi.auth.tokens import JWTGenerator
Review Comment:
`InProcessExecutionAPI.app` adds new in-method imports (`import svcs` / `from airflow.api_fastapi.auth.tokens import JWTGenerator`). Both modules are already imported at module scope in this file, so these imports are redundant and also violate the project guideline to keep imports at the top of the file. Please remove these local imports (and, if `_container` must stay local for a circular-import reason, add a brief comment explaining why).
```suggestion
```
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -76,6 +77,7 @@ def _jwt_generator():
generator = JWTGenerator(
valid_for=conf.getint("execution_api", "jwt_expiration_time"),
+ # workload_valid_for uses the attrs default factory which reads the same config key
Review Comment:
The inline comment says `workload_valid_for ... reads the same config key`, but `JWTGenerator.workload_valid_for` is sourced from `execution_api.jwt_workload_token_expiration_time` (a different key than `jwt_expiration_time`). This comment is misleading and should be updated to reference the correct config key (or removed).
```suggestion
# workload_valid_for uses the attrs default factory, which reads
# execution_api.jwt_workload_token_expiration_time
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3098066143
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -2093,6 +2093,15 @@ execution_api:
type: integer
example: ~
default: "600"
+ jwt_workload_token_expiration_time:
Review Comment:
Thank you, I have removed it and used task_queued_timeout and updated the docs accordingly
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3066989580
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -311,9 +319,13 @@ class InProcessExecutionAPI:
@cached_property
def app(self):
if not self._app:
+ import svcs
+
+ from airflow.api_fastapi.auth.tokens import JWTGenerator
Review Comment:
Nice catch thanks
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3120520807
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -289,13 +295,24 @@ def ti_run(
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
context.start_date = ti.start_date
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except jwt.PyJWTError:
+ log.exception("Failed to generate execution token for task instance %s", task_instance_id)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Token generation failed"
+ )
+ response.headers["Refreshed-API-Token"] = execution_token
Review Comment:
**§1 — `/run` unconditionally mints a new execution token, even when the incoming token was already `execution`-scoped.**
This path runs for every successful `/run`, regardless of incoming scope. Consequences:
- Wasted JWT signing on every `/run` from an already-execution-scoped caller (e.g. retries after transient errors when the client still holds a healthy execution token).
- The client will swap to the new token, **replacing a token with more remaining validity with a fresh one.** The existing `JWTReissueMiddleware` is careful about this and only refreshes when <20% remains — `/run` bypasses that rule entirely.
Suggestion: gate on incoming scope. You already have the parsed token available; accept it as a parameter (`token: TIToken = CurrentTIToken`) and do:
```python
if token.claims.scope == "workload":
execution_token = generator.generate(
extras={"sub": str(task_instance_id), "scope": "execution"}
)
response.headers["Refreshed-API-Token"] = execution_token
```
This also naturally eliminates the header-collision concern in §2.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -289,13 +295,24 @@ def ti_run(
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
context.start_date = ti.start_date
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except jwt.PyJWTError:
+ log.exception("Failed to generate execution token for task instance %s", task_instance_id)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Token generation failed"
+ )
+ response.headers["Refreshed-API-Token"] = execution_token
+
+ return context
Review Comment:
**§4 — Token generation runs after the DB commit, outside the `except SQLAlchemyError` block.**
The refactor moved `return context` out of the transaction `try` so the token can be generated post-commit. That's correct, but note the failure mode: if `generator.generate()` raises here, the DB transaction has already committed and the TI is in `running` state, yet the worker gets a 500 back. The worker retries `/run`, gets 409 (already running), and is stuck with its workload token — unable to make further API calls.
It's a narrow scenario (requires `generate()` to raise, which is rare), but the failure leaves a half-committed run with no recovery path for the SDK. Two options:
- Generate the token **before** the `running` state transition, so a token-gen failure doesn't leave the TI partially running.
- Return a more specific error code + log so the SDK can distinguish "I got 500 but the state changed" from "I got 500 and nothing happened".
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -289,13 +295,24 @@ def ti_run(
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
context.start_date = ti.start_date
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except jwt.PyJWTError:
+ log.exception("Failed to generate execution token for task instance %s", task_instance_id)
+ raise HTTPException(
+ status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Token generation failed"
+ )
+ response.headers["Refreshed-API-Token"] = execution_token
Review Comment:
**§2 — Header-name coordination with `JWTReissueMiddleware` is implicit.**
Both the middleware and this handler write `Refreshed-API-Token`. They don't collide today because the middleware skips workload tokens (in `app.py`), but the coupling is invisible — if someone later changes the middleware scope filter, both could write the header on the same response and the last one wins.
A one-line comment pointing at `JWTReissueMiddleware` ("deliberately same header; middleware skips workload tokens so there is no collision") would save future debugging. Fixing §1 removes the collision path but the comment is still worth having.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -289,13 +295,24 @@ def ti_run(
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
context.start_date = ti.start_date
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except jwt.PyJWTError:
Review Comment:
**§3 — `except jwt.PyJWTError:` is too narrow.**
`JWTGenerator.generate()` can also raise `ValueError` (misconfigured signing key), `KeyError` (missing config), or `AttributeError` from attrs — none inherit from `jwt.PyJWTError`. Those would escape as unhandled 500s with no structured context on this path.
Either broaden to `Exception` (the existing `log.exception` keeps the traceback) or enumerate explicitly, e.g. `(jwt.PyJWTError, ValueError, KeyError)`.
Not a correctness bug — just a less informative failure mode on misconfiguration.
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -281,6 +281,36 @@ def test_ti_run_state_to_running(
)
assert response.status_code == 409
+ def test_ti_run_returns_execution_token(self, client, session, create_task_instance, time_machine):
+ """PATCH /run should return a Refreshed-API-Token header on success."""
+ instant = timezone.parse("2024-10-31T12:00:00Z")
+ time_machine.move_to(instant, tick=False)
+
+ ti = create_task_instance(
+ task_id="test_exec_token",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=instant,
+ dag_id=str(uuid4()),
+ )
+ session.commit()
+
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "test-host",
+ "unixname": "test-user",
+ "pid": 100,
+ "start_date": "2024-10-31T12:00:00Z",
+ },
+ )
+
+ assert response.status_code == 200
+ assert "Refreshed-API-Token" in response.headers
+ assert response.headers["Refreshed-API-Token"] == "mock-execution-token"
Review Comment:
**§6 — Missing assertion on what scope the server asked for.**
This test asserts the header is present and equal to `"mock-execution-token"`, but doesn't verify the server actually asked the generator for `scope="execution"` (vs. accidentally `"workload"`, or a missing scope). Quick strengthen:
```python
mock_generator.generate.assert_called_once()
extras = mock_generator.generate.call_args.kwargs["extras"]
assert extras["scope"] == "execution"
assert extras["sub"] == str(ti.id)
```
The end-to-end JWT-decode path is covered by `test_generate_token_produces_workload_scope` on the workload side, so this is the symmetric check for the `/run` handler.
##########
airflow-core/tests/unit/api_fastapi/execution_api/conftest.py:
##########
@@ -55,6 +59,10 @@ async def mock_jwt_bearer(request: Request):
exec_app.dependency_overrides[_jwt_bearer] = mock_jwt_bearer
with TestClient(app, headers={"Authorization": "Bearer fake"}) as client:
+ mock_generator = MagicMock(spec=JWTGenerator)
+ mock_generator.generate.return_value = "mock-execution-token"
+ lifespan.registry.register_value(JWTGenerator, mock_generator)
Review Comment:
**§5 — Shared `lifespan.registry` mutated without teardown.**
This registers the mock on the shared `lifespan.registry` for the rest of the pytest session. `register_value` replaces rather than appends, so subsequent tests that override the same key are fine — but any test expecting the real `JWTGenerator` that doesn't override would now see this mock and get `"mock-execution-token"` back.
The PR body mentions "registry teardown to prevent test pollution" but I don't see an explicit teardown in the diff. Either add one after `yield client`, or move the registration into the individual tests that need it rather than into the shared fixture.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3132767827
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -418,6 +418,10 @@ class JWTGenerator:
kid: str = attrs.field(default=attrs.Factory(_generate_kid, takes_self=True))
valid_for: float
+ workload_valid_for: float = attrs.field(
Review Comment:
Moved, thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues (#59553) [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-3745560807
Please ping me directly when you want me to re-review this
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2694576839
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -296,25 +296,74 @@ class InProcessExecutionAPI:
@cached_property
def app(self):
if not self._app:
+ import os
+ from base64 import urlsafe_b64encode
+
from airflow.api_fastapi.common.dagbag import create_dag_bag
from airflow.api_fastapi.execution_api.app import create_task_execution_api_app
from airflow.api_fastapi.execution_api.deps import (
JWTBearerDep,
JWTBearerTIPathDep,
)
from airflow.api_fastapi.execution_api.routes.connections import has_connection_access
+ from airflow.api_fastapi.execution_api.routes.task_instances import JWTBearerWorkloadDep
from airflow.api_fastapi.execution_api.routes.variables import has_variable_access
from airflow.api_fastapi.execution_api.routes.xcoms import has_xcom_access
+ from airflow.configuration import conf
+
+ # Ensure JWT secret is available for in-process execution.
+ # The /run endpoint needs JWTGenerator to issue execution tokens.
+ # If the config option is empty, generate a random one for the duration of this process.
+ if not conf.get("api_auth", "jwt_secret", fallback=None):
+ logger.debug(
+ "`api_auth/jwt_secret` is not set, generating a temporary one for in-process execution"
+ )
+ conf.set("api_auth", "jwt_secret", urlsafe_b64encode(os.urandom(16)).decode())
self._app = create_task_execution_api_app()
# Set up dag_bag in app state for dependency injection
self._app.state.dag_bag = create_dag_bag()
+ self._app.state.jwt_generator = _jwt_generator()
+ self._app.state.jwt_validator = _jwt_validator()
+
+ # Why InProcessContainer instead of lifespan.registry or svcs.Container?
+ #
+ # The normal app uses @svcs.fastapi.lifespan which manages the registry lifecycle.
+ # In tests (conftest.py), lifespan.registry.register_value() works because the
+ # TestClient initializes the lifespan before requests. However, in InProcessExecutionAPI,
+ # the lifespan runs later (when transport is accessed), but services may be needed
+ # before that. Using lifespan.registry fails in CI with ServiceNotFoundError.
+ #
+ # This minimal container bypasses the svcs lifecycle and directly returns pre-created
+ # service instances from app.state. If you add new services, update this class.
+ from airflow.api_fastapi.execution_api.deps import _container
+
+ class InProcessContainer:
+ """Minimal container for in-process execution, bypassing svcs lifecycle."""
Review Comment:
Why can't we use svcs? Why do we need to implement our version of it?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3003910545
##########
airflow-core/tests/unit/api_fastapi/execution_api/conftest.py:
##########
@@ -53,6 +57,12 @@ async def mock_jwt_bearer(request: Request):
exec_app.dependency_overrides[_jwt_bearer] = mock_jwt_bearer
with TestClient(app, headers={"Authorization": "Bearer fake"}) as client:
+ mock_generator = MagicMock(spec=JWTGenerator)
+ mock_generator.generate.return_value = "mock-execution-token"
+ lifespan.registry.register_value(JWTGenerator, mock_generator)
+
yield client
+ lifespan.registry.close()
+
exec_app.dependency_overrides.pop(_jwt_bearer, None)
Review Comment:
remove the close() call. thanks
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3066987978
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -76,6 +77,7 @@ def _jwt_generator():
generator = JWTGenerator(
valid_for=conf.getint("execution_api", "jwt_expiration_time"),
+ # workload_valid_for uses the attrs default factory which reads the same config key
Review Comment:
updaated thanks
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -76,6 +77,7 @@ def _jwt_generator():
generator = JWTGenerator(
valid_for=conf.getint("execution_api", "jwt_expiration_time"),
+ # workload_valid_for uses the attrs default factory which reads the same config key
Review Comment:
updated thanks
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4048495854
@kaxil sure will do thanks
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2980446069
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -952,7 +952,10 @@ def __init__(self, *, base_url: str | None, dry_run: bool = False, token: str, *
)
def _update_auth(self, response: httpx.Response):
- if new_token := response.headers.get("Refreshed-API-Token"):
+ if new_token := response.headers.get("X-Execution-Token"):
+ log.debug("Received execution token, swapping auth")
+ self.auth = BearerAuth(new_token)
+ elif new_token := response.headers.get("Refreshed-API-Token"):
Review Comment:
Nit: do we even need a new header? Couldn't we use `Refreshed-API-Token` in both cases? Also if you do think a new header is worth it then remove the `X-` prefix -- that is not recommended by HTTP standards anymore.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2980988929
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -281,14 +286,18 @@ def ti_run(
if ti.next_method:
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
-
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id)})
+ response.headers["X-Execution-Token"] = execution_token
Review Comment:
I don't think it can ever fali really.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4300066932
Looks good - but I would love @ashb to take a final final look and merge :)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3132791192
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -230,6 +238,17 @@ def ti_run(
last_heartbeat_at=timezone.utcnow(),
)
+ execution_token: str | None = None
+ if token.claims.scope == "workload":
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except (jwt.PyJWTError, ValueError, KeyError):
Review Comment:
Went with later one, dropped the try/except entirely.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3132787471
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -230,6 +238,17 @@ def ti_run(
last_heartbeat_at=timezone.utcnow(),
)
+ execution_token: str | None = None
Review Comment:
Moved down just before `return context`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4308455518
#protm
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961442337
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -246,6 +246,36 @@ def test_ti_run_state_to_running(
)
assert response.status_code == 409
+ def test_ti_run_returns_execution_token(self, client, session, create_task_instance, time_machine):
+ """PATCH /run should return an X-Execution-Token header on success."""
+ instant = timezone.parse("2024-10-31T12:00:00Z")
+ time_machine.move_to(instant, tick=False)
+
+ ti = create_task_instance(
+ task_id="test_exec_token",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=instant,
+ dag_id=str(uuid4()),
+ )
+ session.commit()
+
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "test-host",
+ "unixname": "test-user",
+ "pid": 100,
+ "start_date": "2024-10-31T12:00:00Z",
+ },
+ )
+
+ assert response.status_code == 200
+ assert "X-Execution-Token" in response.headers
+ assert response.headers["X-Execution-Token"] == "mock-execution-token"
+
Review Comment:
Added thanks
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add token refresh mechanism for Execution API (#59553) [airflow]
Posted by "tirkarthi (via GitHub)" <gi...@apache.org>.
tirkarthi commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-3717031529
As per my understanding this was removed in https://github.com/apache/airflow/pull/55506 to use a middleware that refreshes token. Are you running an instance with execution api only separately with api-server? Could this middleware approach be extended for task-sdk calls too?
cc: @vincbeck @pierrejeambrun
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2694560832
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -458,6 +474,21 @@ def generate(self, extras: dict[str, Any] | None = None, headers: dict[str, Any]
headers["kid"] = self.kid
return jwt.encode(claims, self.signing_arg, algorithm=self.algorithm, headers=headers)
+ def generate_workload_token(self, sub: str) -> str:
+ """
+ Generate a long-lived workload token for task execution.
+
+ Workload tokens have a special 'scope' claim that restricts them to the /run endpoint only.
+ They are valid for longer (default 24h) to survive executor queue wait times.
+ """
+ from airflow.configuration import conf
+
+ workload_expiry = conf.getint("execution_api", "jwt_workload_token_expiration_time", fallback=86400)
Review Comment:
This default should come from a config option, probablu in the `execution_api` section to go alongside the `jwt_expiration_time` there.
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -458,6 +474,21 @@ def generate(self, extras: dict[str, Any] | None = None, headers: dict[str, Any]
headers["kid"] = self.kid
return jwt.encode(claims, self.signing_arg, algorithm=self.algorithm, headers=headers)
+ def generate_workload_token(self, sub: str) -> str:
+ """
+ Generate a long-lived workload token for task execution.
+
+ Workload tokens have a special 'scope' claim that restricts them to the /run endpoint only.
+ They are valid for longer (default 24h) to survive executor queue wait times.
+ """
+ from airflow.configuration import conf
+
+ workload_expiry = conf.getint("execution_api", "jwt_workload_token_expiration_time", fallback=86400)
Review Comment:
This default should come from a config option, probablu in the `execution_api` section to go alongside the `jwt_expiration_time` there.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2694565105
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -434,15 +437,28 @@ def signing_arg(self) -> AllowedPrivateKeys | str:
assert self._secret_key
return self._secret_key
- def generate(self, extras: dict[str, Any] | None = None, headers: dict[str, Any] | None = None) -> str:
- """Generate a signed JWT for the subject."""
+ def generate(
+ self,
+ extras: dict[str, Any] | None = None,
+ headers: dict[str, Any] | None = None,
+ expiry: int | None = None,
Review Comment:
Nit: we should probably call this argument `valid_for` since that is what it is being used for (i.e. it is not the absolute expiry time)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3003907836
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -142,6 +144,11 @@ async def dispatch(self, request: Request, call_next):
validator: JWTValidator = await services.aget(JWTValidator)
claims = await validator.avalidated_claims(token, {})
+ # Workload tokens are long-lived and meant to survive queue
+ # wait times so avoid refreshing them.
+ if claims.get("scope") == "workload":
+ return response
+
Review Comment:
added comment thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3066987978
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -76,6 +77,7 @@ def _jwt_generator():
generator = JWTGenerator(
valid_for=conf.getint("execution_api", "jwt_expiration_time"),
+ # workload_valid_for uses the attrs default factory which reads the same config key
Review Comment:
udpated thanks
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3132810023
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -281,6 +284,54 @@ def test_ti_run_state_to_running(
)
assert response.status_code == 409
+ def test_ti_run_returns_execution_token(
+ self, client, exec_app, session, create_task_instance, time_machine
+ ):
+ """PATCH /run with a workload token should swap to an execution-scoped token."""
+ instant = timezone.parse("2024-10-31T12:00:00Z")
+ time_machine.move_to(instant, tick=False)
+
+ ti = create_task_instance(
+ task_id="test_exec_token",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=instant,
+ dag_id=str(uuid4()),
+ )
+ session.commit()
+
+ mock_gen = mock.MagicMock(spec=JWTGenerator)
+ mock_gen.generate.return_value = "mock-execution-token"
+ lifespan.registry.register_value(JWTGenerator, mock_gen)
Review Comment:
Added an autouse fixture in execution_api/conftest.py that snapshots lifespan.registry._services before each test and restores it after. Type-agnostic, so it covers both JWTGenerator (here) and JWTValidator (TestTokenTypeValidation). Thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4306695997
Hi @kaxil, thank you so much for the review. I have addressed the feedback in the latest push. Would like to request you for your re-review whenever you get a chance.
Thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4312568642
Thanks for sticking with this @anishgirianish!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-3956229048
Hi @ashb, just checking in on this when you get a chance. Really appreciate the guidance so far!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961412357
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -447,15 +447,33 @@ def signing_arg(self) -> AllowedPrivateKeys | str:
assert self._secret_key
return self._secret_key
Review Comment:
Thank you. Done
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2957016171
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -447,15 +447,33 @@ def signing_arg(self) -> AllowedPrivateKeys | str:
assert self._secret_key
return self._secret_key
Review Comment:
`generate_workload_token` reads `conf.getint("execution_api", "jwt_workload_token_expiration_time")` on every call. Since this is called per-task during scheduling, consider reading the value once at generator construction time (like `valid_for` is).
Also, ashb's earlier comment on the prior PR (#62582) suggested this method may be unnecessary — callers could use `generate(extras={"sub": sub, "scope": "workload"}, valid_for=workload_expiry)` directly, keeping the generator generic.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -332,10 +336,23 @@ async def always_allow(request: Request):
)
return TIToken(id=ti_id, claims={"scope": "execution"})
+ # Override _container (the svcs service locator behind DepContainer).
+ # The default _container reads request.app.state.svcs_registry, but
+ # Cadwyn's versioned sub-apps don't inherit the main app's state,
+ # so lookups raise ServiceNotFoundError. This registry provides
Review Comment:
`_jwt_generator()` calls `get_signing_args(make_secret_key_if_needed=False)`, which raises `ValueError` when no signing key is configured. In the `InProcessExecutionAPI` context (used by `dag.test()`), users typically won't have JWT signing keys set up.
Since `always_allow` already bypasses auth checks, consider registering a stub/mock `JWTGenerator` here that returns a dummy token, instead of wiring up the real factory that requires signing key configuration.
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -2069,6 +2069,15 @@ execution_api:
type: integer
example: ~
default: "600"
+ jwt_workload_token_expiration_time:
+ description: |
+ Seconds until workload JWT tokens expire. These long-lived tokens are sent
+ with task workloads to executors and can only call the /run endpoint.
Review Comment:
`version_added` is `~` (null). Should be set to `3.2.0` (or whichever version this ships in).
##########
airflow-core/tests/unit/api_fastapi/execution_api/conftest.py:
##########
@@ -53,6 +57,11 @@ async def mock_jwt_bearer(request: Request):
exec_app.dependency_overrides[_jwt_bearer] = mock_jwt_bearer
with TestClient(app, headers={"Authorization": "Bearer fake"}) as client:
+ mock_generator = MagicMock(spec=JWTGenerator)
+ mock_generator.generate.return_value = "mock-execution-token"
+ mock_generator.generate_workload_token.return_value = "mock-workload-token"
+ lifespan.registry.register_value(JWTGenerator, mock_generator)
Review Comment:
`lifespan.registry.register_value(JWTGenerator, mock_generator)` mutates the module-level svcs registry, which persists across tests. The fixture cleans up `dependency_overrides` (line 67) but not this registry mutation. Other tests that depend on `JWTGenerator` from the registry could be affected by test ordering.
Consider adding a teardown step (after `yield`) to deregister/reset the `JWTGenerator` entry in the registry.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2957016853
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -939,7 +939,10 @@ def __init__(self, *, base_url: str | None, dry_run: bool = False, token: str, *
)
def _update_auth(self, response: httpx.Response):
Review Comment:
Priority order is correct: `X-Execution-Token` wins over `Refreshed-API-Token`.
One edge case to consider: the `JWTReissueMiddleware` will also try to reissue the inbound workload token if it's near expiry (within 20% of `jwt_expiration_time`). The reissued token inherits the `scope: "workload"` claim but gets `jwt_expiration_time` (600s default) validity instead of the original 24h. In practice this won't happen — a 24h workload token won't be within 120s of expiry on a single `/run` call — but the middleware should probably skip reissuing workload-scoped tokens to be safe against future shorter-lived workload tokens.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961423999
##########
airflow-core/tests/unit/api_fastapi/execution_api/conftest.py:
##########
@@ -53,6 +57,11 @@ async def mock_jwt_bearer(request: Request):
exec_app.dependency_overrides[_jwt_bearer] = mock_jwt_bearer
with TestClient(app, headers={"Authorization": "Bearer fake"}) as client:
+ mock_generator = MagicMock(spec=JWTGenerator)
+ mock_generator.generate.return_value = "mock-execution-token"
+ mock_generator.generate_workload_token.return_value = "mock-workload-token"
+ lifespan.registry.register_value(JWTGenerator, mock_generator)
Review Comment:
Done. Thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961422679
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -2069,6 +2069,15 @@ execution_api:
type: integer
example: ~
default: "600"
+ jwt_workload_token_expiration_time:
+ description: |
+ Seconds until workload JWT tokens expire. These long-lived tokens are sent
+ with task workloads to executors and can only call the /run endpoint.
Review Comment:
set to 3.2.0, but I think we are really close. Please let me know if this has to be deferred.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2980415983
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -2069,6 +2069,15 @@ execution_api:
type: integer
example: ~
default: "600"
+ jwt_workload_token_expiration_time:
+ description: |
+ Seconds until workload JWT tokens expire. These long-lived tokens are sent
+ with task workloads to executors and can only call the /run endpoint.
+ Set long enough to cover maximum expected queue wait time.
+ version_added: 3.2.0
Review Comment:
I don't think we'll merge this for 3.2.0, so
```suggestion
version_added: 3.2.1
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2991910313
##########
airflow-core/tests/unit/api_fastapi/execution_api/conftest.py:
##########
@@ -53,6 +57,12 @@ async def mock_jwt_bearer(request: Request):
exec_app.dependency_overrides[_jwt_bearer] = mock_jwt_bearer
with TestClient(app, headers={"Authorization": "Bearer fake"}) as client:
+ mock_generator = MagicMock(spec=JWTGenerator)
+ mock_generator.generate.return_value = "mock-execution-token"
+ lifespan.registry.register_value(JWTGenerator, mock_generator)
+
yield client
+ lifespan.registry.close()
+
exec_app.dependency_overrides.pop(_jwt_bearer, None)
Review Comment:
`lifespan.registry.close()` is new here (no other test file does this), and the registry is shared across all tests via `cached_app`. Closing it could break subsequent tests that try to look up services from the same registry. The existing pattern in other test files (e.g., `test_task_instances.py`, `test_router.py`) registers values on `lifespan.registry` without closing it afterward. I'd drop this `close()` call to match what the rest of the test suite does.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -142,6 +144,11 @@ async def dispatch(self, request: Request, call_next):
validator: JWTValidator = await services.aget(JWTValidator)
claims = await validator.avalidated_claims(token, {})
+ # Workload tokens are long-lived and meant to survive queue
+ # wait times so avoid refreshing them.
+ if claims.get("scope") == "workload":
+ return response
+
Review Comment:
The early return for workload tokens skips the refresh logic (correct), but it also skips the `except` block below. If `avalidated_claims` raises for a workload token, execution falls into the outer `except` and the response still gets returned (with a warning log). Might be worth a comment clarifying that workload token validation errors are handled by the outer catch.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -332,10 +343,29 @@ async def always_allow(request: Request):
)
return TIToken(id=ti_id, claims={"scope": "execution"})
+ # Override _container (the svcs service locator behind DepContainer).
+ # The default _container reads request.app.state.svcs_registry, but
+ # Cadwyn's versioned sub-apps don't inherit the main app's state,
+ # so lookups raise ServiceNotFoundError. This registry provides
+ # services needed by routes called during dag.test().
+ #
Review Comment:
The stub `JWTGenerator` uses `secrets.token_urlsafe(32)` as the secret key, so a new key is generated every time `InProcessExecutionAPI.app` is accessed. Since `app` is a `cached_property`, the key is stable for the lifetime of the object. But the token generated here by `ti_run` won't be validated by anything (since `_jwt_bearer` is also overridden with `always_allow`), so this stub only exists to satisfy the `services.get(JWTGenerator)` call. A brief comment noting that these tokens are never validated in `dag.test()` mode would help future readers.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -286,14 +291,18 @@ def ti_run(
if ti.next_method:
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
-
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id)})
Review Comment:
Token generation happens outside the `try...except SQLAlchemyError` block. If `services.get(JWTGenerator)` or `generator.generate()` raises (missing service, crypto error, etc.), the client gets a raw 500 with no useful detail. Worth wrapping this in its own try/except, or at minimum a log line, so operators can tell the difference between "database error" and "token generation failed".
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -418,6 +418,10 @@ class JWTGenerator:
kid: str = attrs.field(default=attrs.Factory(_generate_kid, takes_self=True))
valid_for: float
+ workload_valid_for: float = attrs.field(
Review Comment:
The `workload_valid_for` default reads from config via `_conf_factory`, and `_jwt_generator()` in `app.py` also reads the same config key and passes it explicitly. The explicit kwarg takes precedence, so the default factory never runs in production. Having two code paths that reference the same config key is easy to get out of sync -- consider dropping the attrs default (make it required like `valid_for`) and always passing it explicitly, or drop the explicit kwarg in `_jwt_generator()` and let the default handle it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4266562739
@ashb @jscheffl Thank you so much for the review! Addressed both comments:
1. Narrowed except Exception.
2. Dropped the new jwt_workload_token_expiration_time config; workload-token lifetime now derives from [scheduler] task_queued_timeout. Docs updated accordingly.
Would like to request you for re-review whenever you get a chance.
Thank you.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2694582987
##########
airflow-core/src/airflow/api_fastapi/execution_api/deps.py:
##########
@@ -87,15 +84,56 @@ async def __call__( # type: ignore[override]
else:
validators = self.required_claims
claims = await validator.avalidated_claims(creds.credentials, validators)
+
+ # Let subclasses validate scope
+ self._check_scope(claims)
+
return TIToken(id=claims["sub"], claims=claims)
+ except HTTPException:
+ raise
except Exception as err:
- log.warning(
- "Failed to validate JWT",
- exc_info=True,
- token=creds.credentials,
- )
+ log.warning("Failed to validate JWT", exc_info=True)
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=f"Invalid auth token: {err}")
+ def _check_scope(self, claims: dict[str, Any]) -> None:
+ """Override in subclasses to validate scope. Raise HTTPException if invalid."""
+ pass
+
+
+class JWTBearer(_BaseJWTBearer):
+ """
+ JWT validation that rejects workload-scoped tokens.
+
+ Used for most Execution API endpoints. Workload-scoped tokens can only be used
+ on the /run endpoint, which exchanges them for short-lived execution tokens.
+ """
+
+ def _check_scope(self, claims: dict[str, Any]) -> None:
+ if claims.get("scope") == TOKEN_SCOPE_WORKLOAD:
+ raise HTTPException(
+ status_code=status.HTTP_403_FORBIDDEN,
+ detail="Workload tokens cannot access this endpoint. Use the token from /run response.",
+ )
+
+
+class JWTBearerWorkloadScope(_BaseJWTBearer):
+ """
+ JWT validation that ONLY accepts workload-scoped tokens.
+
+ Used exclusively by the /run endpoint. Workload tokens have scope="ExecuteTaskWorkload"
+ and are long-lived to survive executor queue wait times. The /run endpoint validates
+ the workload token and issues a short-lived execution token for subsequent API calls.
+ """
+
+ def _check_scope(self, claims: dict[str, Any]) -> None:
+ scope = claims.get("scope")
+ # Reject if scope is explicitly set to something other than workload scope
+ if scope is not None and scope != TOKEN_SCOPE_WORKLOAD:
Review Comment:
```suggestion
if scope != TOKEN_SCOPE_WORKLOAD:
```
Crucially we want to fail if scope is None too!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2700795602
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -296,25 +296,74 @@ class InProcessExecutionAPI:
@cached_property
def app(self):
if not self._app:
+ import os
+ from base64 import urlsafe_b64encode
+
from airflow.api_fastapi.common.dagbag import create_dag_bag
from airflow.api_fastapi.execution_api.app import create_task_execution_api_app
from airflow.api_fastapi.execution_api.deps import (
JWTBearerDep,
JWTBearerTIPathDep,
)
from airflow.api_fastapi.execution_api.routes.connections import has_connection_access
+ from airflow.api_fastapi.execution_api.routes.task_instances import JWTBearerWorkloadDep
from airflow.api_fastapi.execution_api.routes.variables import has_variable_access
from airflow.api_fastapi.execution_api.routes.xcoms import has_xcom_access
+ from airflow.configuration import conf
+
+ # Ensure JWT secret is available for in-process execution.
+ # The /run endpoint needs JWTGenerator to issue execution tokens.
+ # If the config option is empty, generate a random one for the duration of this process.
+ if not conf.get("api_auth", "jwt_secret", fallback=None):
+ logger.debug(
+ "`api_auth/jwt_secret` is not set, generating a temporary one for in-process execution"
+ )
+ conf.set("api_auth", "jwt_secret", urlsafe_b64encode(os.urandom(16)).decode())
self._app = create_task_execution_api_app()
# Set up dag_bag in app state for dependency injection
self._app.state.dag_bag = create_dag_bag()
+ self._app.state.jwt_generator = _jwt_generator()
+ self._app.state.jwt_validator = _jwt_validator()
+
+ # Why InProcessContainer instead of lifespan.registry or svcs.Container?
+ #
+ # The normal app uses @svcs.fastapi.lifespan which manages the registry lifecycle.
+ # In tests (conftest.py), lifespan.registry.register_value() works because the
+ # TestClient initializes the lifespan before requests. However, in InProcessExecutionAPI,
+ # the lifespan runs later (when transport is accessed), but services may be needed
+ # before that. Using lifespan.registry fails in CI with ServiceNotFoundError.
+ #
+ # This minimal container bypasses the svcs lifecycle and directly returns pre-created
+ # service instances from app.state. If you add new services, update this class.
+ from airflow.api_fastapi.execution_api.deps import _container
+
+ class InProcessContainer:
+ """Minimal container for in-process execution, bypassing svcs lifecycle."""
Review Comment:
I'm still getting familiar with the codebase, but from what I understand, I added this to fix ServiceNotFoundError failures in CI. With InProcessExecutionAPI, the svcs lifespan runs later (when transport is accessed), but services like JWTGenerator are needed before that. This container bypasses the lifecycle and returns pre-created instances from app.state. I may well be missing something - if there's a cleaner pattern you'd recommend, I'd really appreciate the guidance!
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -296,25 +296,74 @@ class InProcessExecutionAPI:
@cached_property
def app(self):
if not self._app:
+ import os
+ from base64 import urlsafe_b64encode
+
from airflow.api_fastapi.common.dagbag import create_dag_bag
from airflow.api_fastapi.execution_api.app import create_task_execution_api_app
from airflow.api_fastapi.execution_api.deps import (
JWTBearerDep,
JWTBearerTIPathDep,
)
from airflow.api_fastapi.execution_api.routes.connections import has_connection_access
+ from airflow.api_fastapi.execution_api.routes.task_instances import JWTBearerWorkloadDep
from airflow.api_fastapi.execution_api.routes.variables import has_variable_access
from airflow.api_fastapi.execution_api.routes.xcoms import has_xcom_access
+ from airflow.configuration import conf
+
+ # Ensure JWT secret is available for in-process execution.
+ # The /run endpoint needs JWTGenerator to issue execution tokens.
+ # If the config option is empty, generate a random one for the duration of this process.
+ if not conf.get("api_auth", "jwt_secret", fallback=None):
+ logger.debug(
+ "`api_auth/jwt_secret` is not set, generating a temporary one for in-process execution"
+ )
+ conf.set("api_auth", "jwt_secret", urlsafe_b64encode(os.urandom(16)).decode())
self._app = create_task_execution_api_app()
# Set up dag_bag in app state for dependency injection
self._app.state.dag_bag = create_dag_bag()
+ self._app.state.jwt_generator = _jwt_generator()
+ self._app.state.jwt_validator = _jwt_validator()
+
+ # Why InProcessContainer instead of lifespan.registry or svcs.Container?
+ #
+ # The normal app uses @svcs.fastapi.lifespan which manages the registry lifecycle.
+ # In tests (conftest.py), lifespan.registry.register_value() works because the
+ # TestClient initializes the lifespan before requests. However, in InProcessExecutionAPI,
+ # the lifespan runs later (when transport is accessed), but services may be needed
+ # before that. Using lifespan.registry fails in CI with ServiceNotFoundError.
+ #
+ # This minimal container bypasses the svcs lifecycle and directly returns pre-created
+ # service instances from app.state. If you add new services, update this class.
+ from airflow.api_fastapi.execution_api.deps import _container
+
+ class InProcessContainer:
+ """Minimal container for in-process execution, bypassing svcs lifecycle."""
Review Comment:
I'm still getting familiar with the codebase, but from what I understand, I added this to fix ServiceNotFoundError failures in CI. With InProcessExecutionAPI, the svcs lifespan runs later (when transport is accessed), but services like JWTGenerator are needed before that. This container bypasses the lifecycle and returns pre-created instances from app.state. I may well be missing something - if there's a cleaner pattern you'd recommend, I'd really appreciate the guidance.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2700796690
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -296,25 +296,74 @@ class InProcessExecutionAPI:
@cached_property
def app(self):
if not self._app:
+ import os
+ from base64 import urlsafe_b64encode
+
from airflow.api_fastapi.common.dagbag import create_dag_bag
from airflow.api_fastapi.execution_api.app import create_task_execution_api_app
from airflow.api_fastapi.execution_api.deps import (
JWTBearerDep,
JWTBearerTIPathDep,
)
from airflow.api_fastapi.execution_api.routes.connections import has_connection_access
+ from airflow.api_fastapi.execution_api.routes.task_instances import JWTBearerWorkloadDep
from airflow.api_fastapi.execution_api.routes.variables import has_variable_access
from airflow.api_fastapi.execution_api.routes.xcoms import has_xcom_access
+ from airflow.configuration import conf
+
+ # Ensure JWT secret is available for in-process execution.
+ # The /run endpoint needs JWTGenerator to issue execution tokens.
+ # If the config option is empty, generate a random one for the duration of this process.
+ if not conf.get("api_auth", "jwt_secret", fallback=None):
+ logger.debug(
+ "`api_auth/jwt_secret` is not set, generating a temporary one for in-process execution"
+ )
+ conf.set("api_auth", "jwt_secret", urlsafe_b64encode(os.urandom(16)).decode())
Review Comment:
From my understanding, this is only for InProcessExecutionAPI which is used when running tasks without a separate API server - it generates a temporary secret if none is configured. I'm still exploring this part of the codebase, so please do correct me if I've misunderstood something.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4086496747
Hi @ashb, rebased on top of the SecurityScopes foundation, it made the integration really straightforward to build on. Thank you so much for putting that together. Ready for another look whenever you have a moment. Really appreciate all your guidance. Thank you.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2961416427
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -332,10 +336,23 @@ async def always_allow(request: Request):
)
return TIToken(id=ti_id, claims={"scope": "execution"})
+ # Override _container (the svcs service locator behind DepContainer).
+ # The default _container reads request.app.state.svcs_registry, but
+ # Cadwyn's versioned sub-apps don't inherit the main app's state,
+ # so lookups raise ServiceNotFoundError. This registry provides
Review Comment:
Registered a stub. thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2980439190
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -3226,6 +3256,44 @@ def test_invalid_scope_value_rejected(self, client, session, create_task_instanc
assert resp.status_code == 403
assert "Invalid token scope" in resp.json()["detail"]
+ def test_workload_scope_accepted_on_run_endpoint(
Review Comment:
I also think we should check this once or twice on other routers too -- the workload scoped tokens should not work _anywhere_ but the `/run` TI end point.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3003906180
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -281,14 +286,18 @@ def ti_run(
if ti.next_method:
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
-
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id)})
+ response.headers["X-Execution-Token"] = execution_token
Review Comment:
wrapped this with try/expect
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3003902690
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py:
##########
@@ -3226,6 +3256,44 @@ def test_invalid_scope_value_rejected(self, client, session, create_task_instanc
assert resp.status_code == 403
assert "Invalid token scope" in resp.json()["detail"]
+ def test_workload_scope_accepted_on_run_endpoint(
Review Comment:
added additional tests to cover these thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3098073235
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -287,14 +292,24 @@ def ti_run(
if ti.next_method:
context.next_method = ti.next_method
context.next_kwargs = ti.next_kwargs
-
- return context
except SQLAlchemyError:
log.exception("Error marking Task Instance state as running")
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Database error occurred"
)
+ try:
+ generator: JWTGenerator = services.get(JWTGenerator)
+ execution_token = generator.generate(extras={"sub": str(task_instance_id), "scope": "execution"})
+ except Exception:
Review Comment:
I have narrowed down the exception. Please let me know your thoughts on the narrow version. thanks
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4284098529
> Looks good to me. As security relevant would be good to have a couple more eyes on the PR prior merge.
>
> If we (hopefully not but for whatever reason) need an 3.2.1rc3, then I'd suggest putting this on the candidate list.
cc: @vatsrahul1001
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4305521637
> One doc nit/change then LGTM!
Thank you so much for the review. I have addresed this in latest push. thanks
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3132783010
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -333,10 +342,30 @@ async def always_allow(request: Request):
claims = TIClaims(scope="execution")
return TIToken(id=ti_id, claims=claims)
+ # Override _container (the svcs service locator behind DepContainer).
+ # The default _container reads request.app.state.svcs_registry, but
+ # Cadwyn's versioned sub-apps don't inherit the main app's state,
+ # so lookups raise ServiceNotFoundError. This registry provides
+ # services needed by routes called during dag.test().
+ # Note: tokens generated by this stub are never validated since
+ # _jwt_bearer is overridden with always_allow in dag.test() mode.
+ stub_generator = JWTGenerator(
Review Comment:
Thank you so much for this. Dissecting this actually led me to figure out that the entire override is not needed now as the token generation is now scope-gated.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -333,10 +342,30 @@ async def always_allow(request: Request):
claims = TIClaims(scope="execution")
return TIToken(id=ti_id, claims=claims)
+ # Override _container (the svcs service locator behind DepContainer).
+ # The default _container reads request.app.state.svcs_registry, but
+ # Cadwyn's versioned sub-apps don't inherit the main app's state,
+ # so lookups raise ServiceNotFoundError. This registry provides
+ # services needed by routes called during dag.test().
+ # Note: tokens generated by this stub are never validated since
+ # _jwt_bearer is overridden with always_allow in dag.test() mode.
+ stub_generator = JWTGenerator(
Review Comment:
Thank you so much for this. Disecting this actually led me to figure out that the entire override is not needed now as the token generation is now scope-gated.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil merged PR #60108:
URL: https://github.com/apache/airflow/pull/60108
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add token refresh mechanism for Execution API (#59553) [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-3717374643
Hi @tirkarthi,
Thanks for pointing out the middleware approach from #55506 - that's helpful context.
I took a stab at extending that pattern in #60197, handling expired tokens transparently in JWTBearer + middleware so no client-side changes are needed. Would love your thoughts on it.
Totally happy to go with whichever approach the team feels is better!
cc: @vincbeck @pierrejeambrun
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add token refresh mechanism for Execution API (#59553) [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r2675871772
##########
airflow-core/src/airflow/api_fastapi/execution_api/deps.py:
##########
@@ -87,13 +87,70 @@ async def __call__( # type: ignore[override]
else:
validators = self.required_claims
claims = await validator.avalidated_claims(creds.credentials, validators)
+
+ # Reject queue-scoped tokens - they can only be used on /run endpoint
+ # Only check if scope claim is present (allows backwards compatibility with tests)
Review Comment:
If this back compat is just for tests then we don't need it.
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -458,6 +461,39 @@ def generate(self, extras: dict[str, Any] | None = None, headers: dict[str, Any]
headers["kid"] = self.kid
return jwt.encode(claims, self.signing_arg, algorithm=self.algorithm, headers=headers)
+ def generate_queue_token(self, sub: str) -> str:
+ """
+ Generate a long-lived queue token for task workloads.
+
+ Queue tokens have a special 'scope' claim that restricts them to the /run endpoint only.
+ They are valid for longer (default 24h) to survive queue wait times.
+ """
+ from airflow.configuration import conf
+
+ queue_expiry = conf.getint("execution_api", "jwt_queue_token_expiration_time", fallback=86400)
+ now = int(datetime.now(tz=timezone.utc).timestamp())
+
+ claims = {
+ "jti": uuid.uuid4().hex,
+ "iss": self.issuer,
+ "aud": self.audience,
+ "nbf": now,
+ "exp": now + queue_expiry,
+ "iat": now,
+ "sub": sub,
+ "scope": TOKEN_SCOPE_QUEUE,
+ }
+
+ if claims["iss"] is None:
+ del claims["iss"]
+ if claims["aud"] is None:
+ del claims["aud"]
+
+ headers = {"alg": self.algorithm}
+ if self._private_key:
+ headers["kid"] = self.kid
+ return jwt.encode(claims, self.signing_arg, algorithm=self.algorithm, headers=headers)
Review Comment:
I don't think we need a new whole function for this -- the existing `generate()` could work already by doing:
```python
generator.generate({"sub": sub, "exp": now + queue_expiry})
```
If you think it's worth "packaging" that up, then make it call self.generate -- don't essentially duplicate it.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -296,15 +296,24 @@ class InProcessExecutionAPI:
@cached_property
def app(self):
if not self._app:
+ from unittest.mock import AsyncMock, MagicMock
+
+ from airflow.api_fastapi.auth.tokens import JWTValidator
from airflow.api_fastapi.common.dagbag import create_dag_bag
- from airflow.api_fastapi.execution_api.app import create_task_execution_api_app
from airflow.api_fastapi.execution_api.deps import (
+ DepContainer,
JWTBearerDep,
+ JWTBearerQueueDep,
JWTBearerTIPathDep,
)
from airflow.api_fastapi.execution_api.routes.connections import has_connection_access
from airflow.api_fastapi.execution_api.routes.variables import has_variable_access
from airflow.api_fastapi.execution_api.routes.xcoms import has_xcom_access
+ from airflow.configuration import conf
+
+ # Set a dummy JWT secret so the lifespan can create JWT services without failing.
+ if not conf.get("api_auth", "jwt_secret", fallback=None):
+ conf.set("api_auth", "jwt_secret", "in-process-test-secret-key")
Review Comment:
Never ever ever do this in production/runtime code. The risk of it being picked up and every install in the world having a signing key of "in-process-test-secret-key" is too large.
I knot this is just the InProcess class, but I'm still worried about doing this, doubly so as config is process global.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -315,28 +324,57 @@ async def always_allow(): ...
self._app.dependency_overrides[JWTBearerDep.dependency] = always_allow
self._app.dependency_overrides[JWTBearerTIPathDep.dependency] = always_allow
+ self._app.dependency_overrides[JWTBearerQueueDep.dependency] = always_allow
self._app.dependency_overrides[has_connection_access] = always_allow
self._app.dependency_overrides[has_variable_access] = always_allow
self._app.dependency_overrides[has_xcom_access] = always_allow
+ # Create a mock container that provides mock JWT services
+ mock_jwt_generator = MagicMock(spec=JWTGenerator)
+ mock_jwt_generator.generate.return_value = "mock-execution-token"
+
+ mock_jwt_validator = AsyncMock(spec=JWTValidator)
+ mock_jwt_validator.avalidated_claims.return_value = {"sub": "test", "exp": 9999999999}
+
+ class MockContainer:
+ """A mock svcs container that returns mock services."""
+
+ async def aget(self, svc_type):
+ if svc_type is JWTGenerator:
+ return mock_jwt_generator
+ if svc_type is JWTValidator:
+ return mock_jwt_validator
+ raise ValueError(f"Unknown service type: {svc_type}")
+
+ async def mock_container_dep():
+ return MockContainer()
+
+ self._app.dependency_overrides[DepContainer.dependency] = mock_container_dep
Review Comment:
Why do we need a _mock_ svcs container? Why not just use a real svcs container?
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -1822,6 +1822,17 @@ execution_api:
type: integer
example: ~
default: "600"
+ jwt_queue_token_expiration_time:
+ description: |
+ Number in seconds until the queue JWT token expires. Queue tokens are long-lived tokens
+ sent with task workloads to executors (e.g., Celery). They can only be used to call
+ the /run endpoint, which then issues a short-lived execution token.
+
+ This should be set long enough to cover the maximum expected queue wait time.
+ version_added: 3.1.0
Review Comment:
```suggestion
version_added: 3.1.7
```
##########
airflow-core/src/airflow/api_fastapi/execution_api/deps.py:
##########
Review Comment:
I'd rather we wrote the JWTBearer and the JWTBearerQueueScope queue in a layered approach (deps can depend on each other, or subclassing) - that way we only have to writ much of the validation once and only tweak the behaviour.
I think the layer approach would be best, so we have a base JWTBearer dep that does the basic validation, but nothing of the presence/absence of the queue scope, and then two deps that consume that returned TIToken to do the next layer.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -100,12 +100,15 @@
HTTP_422_UNPROCESSABLE_CONTENT: {"description": "Invalid payload for the state transition"},
},
response_model_exclude_unset=True,
+ dependencies=[JWTBearerQueueDep],
Review Comment:
Since this is the one and only place we use this dep (and it's also the only place we ever want to use this dep) I think it would be better if we moved JWTBearerQueueDep in to this file.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -315,28 +324,57 @@ async def always_allow(): ...
self._app.dependency_overrides[JWTBearerDep.dependency] = always_allow
self._app.dependency_overrides[JWTBearerTIPathDep.dependency] = always_allow
+ self._app.dependency_overrides[JWTBearerQueueDep.dependency] = always_allow
self._app.dependency_overrides[has_connection_access] = always_allow
self._app.dependency_overrides[has_variable_access] = always_allow
self._app.dependency_overrides[has_xcom_access] = always_allow
+ # Create a mock container that provides mock JWT services
+ mock_jwt_generator = MagicMock(spec=JWTGenerator)
+ mock_jwt_generator.generate.return_value = "mock-execution-token"
+
+ mock_jwt_validator = AsyncMock(spec=JWTValidator)
+ mock_jwt_validator.avalidated_claims.return_value = {"sub": "test", "exp": 9999999999}
+
+ class MockContainer:
+ """A mock svcs container that returns mock services."""
+
+ async def aget(self, svc_type):
+ if svc_type is JWTGenerator:
+ return mock_jwt_generator
+ if svc_type is JWTValidator:
+ return mock_jwt_validator
+ raise ValueError(f"Unknown service type: {svc_type}")
+
+ async def mock_container_dep():
+ return MockContainer()
+
+ self._app.dependency_overrides[DepContainer.dependency] = mock_container_dep
Review Comment:
What is this doing here? This looks like test-only code.
##########
airflow-core/src/airflow/api_fastapi/execution_api/app.py:
##########
@@ -315,28 +324,57 @@ async def always_allow(): ...
self._app.dependency_overrides[JWTBearerDep.dependency] = always_allow
self._app.dependency_overrides[JWTBearerTIPathDep.dependency] = always_allow
+ self._app.dependency_overrides[JWTBearerQueueDep.dependency] = always_allow
self._app.dependency_overrides[has_connection_access] = always_allow
self._app.dependency_overrides[has_variable_access] = always_allow
self._app.dependency_overrides[has_xcom_access] = always_allow
+ # Create a mock container that provides mock JWT services
+ mock_jwt_generator = MagicMock(spec=JWTGenerator)
+ mock_jwt_generator.generate.return_value = "mock-execution-token"
+
+ mock_jwt_validator = AsyncMock(spec=JWTValidator)
+ mock_jwt_validator.avalidated_claims.return_value = {"sub": "test", "exp": 9999999999}
+
+ class MockContainer:
+ """A mock svcs container that returns mock services."""
+
+ async def aget(self, svc_type):
+ if svc_type is JWTGenerator:
+ return mock_jwt_generator
+ if svc_type is JWTValidator:
+ return mock_jwt_validator
+ raise ValueError(f"Unknown service type: {svc_type}")
+
+ async def mock_container_dep():
+ return MockContainer()
+
+ self._app.dependency_overrides[DepContainer.dependency] = mock_container_dep
+
return self._app
@cached_property
def transport(self) -> httpx.WSGITransport:
import asyncio
+ import threading
import httpx
from a2wsgi import ASGIMiddleware
middleware = ASGIMiddleware(self.app)
+ lifespan_started = threading.Event()
# https://github.com/abersheeran/a2wsgi/discussions/64
async def start_lifespan(cm: AsyncExitStack, app: FastAPI):
await cm.enter_async_context(app.router.lifespan_context(app))
+ lifespan_started.set()
self._cm = AsyncExitStack()
asyncio.run_coroutine_threadsafe(start_lifespan(self._cm, self.app), middleware.loop)
+ # Wait for lifespan to complete before returning the transport
+ lifespan_started.wait(timeout=5.0)
Review Comment:
Why do we need this?
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/__init__.py:
##########
Review Comment:
Any reason we needed to change this order?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues (#59553) [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-3746299326
Hi @ashb
Thank you so much for taking the time to review this PR and for your valuable guidance. I truly appreciate the feedback.
I've made the updates based on your suggestions, and the PR is ready for another look at your convenience, whenever you have time works perfectly.
Thanks again for your patience and support!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "anishgirianish (via GitHub)" <gi...@apache.org>.
anishgirianish commented on code in PR #60108:
URL: https://github.com/apache/airflow/pull/60108#discussion_r3031436662
##########
airflow-core/src/airflow/api_fastapi/auth/tokens.py:
##########
@@ -418,6 +418,10 @@ class JWTGenerator:
kid: str = attrs.field(default=attrs.Factory(_generate_kid, takes_self=True))
valid_for: float
+ workload_valid_for: float = attrs.field(
Review Comment:
Fixed, _jwt_generator() in app.py no longer passes workload_valid_for explicitly. Only the attrs factory default in tokens.py reads the config now. thank you
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Two-token mechanism for task execution to prevent token expiration while tasks wait in executor queues [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #60108:
URL: https://github.com/apache/airflow/pull/60108#issuecomment-4237214097
Will look when I have a spare cycle, right now I'm digging in to issue 65010 though.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org