Re: [PR] Add schema migration to supervisor-child comm [airflow]
uranusjr commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3290915722 ## task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py: ## Review Comment: Similar to the schema file, we can always change this as long as the version names stay the same. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
uranusjr merged PR #67235: URL: https://github.com/apache/airflow/pull/67235 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
uranusjr commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3290912721 ## task-sdk/src/airflow/sdk/execution_time/schema/schema.json: ## Review Comment: I’ll keep the file here for now, we can always remove it later. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
kaxil commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3282316972 ## task-sdk/src/airflow/sdk/execution_time/supervisor.py: ## @@ -548,6 +549,8 @@ class WatchedSubprocess: _process: psutil.Process = attrs.field(repr=False) """File descriptor for request handling.""" +_subprocess_schema_version: str | None = None Review Comment: Worth adding docstring about 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
uranusjr commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3280689105 ## task-sdk/src/airflow/sdk/execution_time/supervisor.py: ## @@ -548,6 +549,8 @@ class WatchedSubprocess: _process: psutil.Process = attrs.field(repr=False) """File descriptor for request handling.""" +_subprocess_schema_version: str | None = None Review Comment: This is intended, the override will happen in the Java Coordinator PR. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
github-actions[bot] commented on PR #67235:
URL: https://github.com/apache/airflow/pull/67235#issuecomment-4504435339
`uv.lock` on `main` just moved via
[#67121](https://github.com/apache/airflow/pull/67121) ("Add LlamaIndex
operators to common.ai provider"), commit
[`b64c302`](https://github.com/apache/airflow/commit/b64c302b2e9ab8d6810494551862b3b423b2b6d3)
and this PR currently conflicts.
Quickest fix:
```bash
git fetch upstream main && git rebase upstream/main
rm uv.lock && uv lock
git add uv.lock && git rebase --continue
git push --force-with-lease
```
_Automated nudge — ignore if you're not ready to rebase. This comment is
updated in place on future `uv.lock` bumps._
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
kaxil commented on code in PR #67235:
URL: https://github.com/apache/airflow/pull/67235#discussion_r3277765789
##
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##
@@ -730,35 +733,51 @@ def _create_log_forwarder(self, loggers, name,
log_level=logging.INFO) -> Callab
def _on_socket_closed(self, sock: socket):
# We want to keep servicing this process until we've read up to EOF
from all the sockets.
-
with suppress(KeyError):
self.selector.unregister(sock)
del self._open_sockets[sock]
+def _serialize_response(self, msg: BaseModel | ErrorResponse, **dump_opts)
-> dict[str, Any]:
+if self._subprocess_schema_version is not None:
+migrator = get_schema_version_migrator()
+msg = migrator.downgrade(msg, self._subprocess_schema_version,
dump_kwargs=dump_opts)
Review Comment:
This call crashes the first time the migration path actually runs.
`dump_kwargs=dump_opts` is absorbed by `downgrade`'s `**dump_opts`, then
forwarded to `msg.model_dump(**{**dump_opts, "mode": "json"})` at
`migrator.py:156` -- which becomes `model_dump(dump_kwargs={...},
mode="json")`. Pydantic rejects `dump_kwargs` as an unexpected keyword and
raises `TypeError`.
CI is green only because the integration-test fixture monkeypatches
`_serialize_response` with a corrected implementation (see
`test_integration.py` `mock_version_migrator` fixture), so this production code
path is never exercised under test.
Fix:
```python
msg = migrator.downgrade(msg, self._subprocess_schema_version, **dump_opts)
```
Then add a real test (no monkeypatch on `_serialize_response`) that drives
`send_msg` end-to-end against a non-`None` `_subprocess_schema_version`.
##
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##
@@ -548,6 +549,8 @@ class WatchedSubprocess:
_process: psutil.Process = attrs.field(repr=False)
"""File descriptor for request handling."""
+_subprocess_schema_version: str | None = None
Review Comment:
This attribute has no production writer, so the entire migration feature is
dead code on `main`. There is no constructor arg (`attrs.define` ignores plain
class attributes without `attrs.field`, so it's not even an attrs field), no
setter or property assignment anywhere in `supervisor.py`, and no handshake
message that negotiates the runtime's schema version.
Result: in production the `if self._subprocess_schema_version is not None:`
gates at lines 740 and 759 are always False, and the schema-migration seam
never runs. The only callers that flip it are the integration tests, which
assign `ws._subprocess_schema_version = pinned_version` directly on the
instance.
Either wire the version in through the child handshake before merging, or be
explicit in the PR description that this is the data-structures-only half and
the negotiation lands in a follow-up. Right now the PR description claims "The
supervisor use the migrations to upgrade ... and downgrade messages" which
overstates what's actually hooked up.
##
task-sdk/.pre-commit-config.yaml:
##
@@ -43,6 +43,8 @@ repos:
^src/airflow/sdk/definitions/deadline\.py$|
^src/airflow/sdk/definitions/dag\.py$|
^src/airflow/sdk/definitions/_internal/types\.py$|
+ ^src/airflow/sdk/execution_time/coordinator\.py$|
Review Comment:
`coordinator.py` does not exist in this PR -- searching the diff and the
task-sdk tree shows no `src/airflow/sdk/execution_time/coordinator.py`.
Pre-emptive exclusions for unmerged files attract drift (the file lands later
with a different path, the exclusion silently turns into a no-op). Drop this
line until the file actually lands; add it back in the PR that introduces
`coordinator.py`.
##
task-sdk/src/airflow/sdk/execution_time/schema/migrator.py:
##
@@ -0,0 +1,214 @@
+# 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.
+"""
+In-process bidirectional migration for supervisor schema bodies.
+
+:class:`SchemaVersionMigrator` walks a :class:`~cadwyn.VersionBundle`
+itself rather than going through Cadwyn's HTTP runner so the supervisor
+can d
Re: [PR] Add schema migration to supervisor-child comm [airflow]
jason810496 commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3274194034 ## task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py: ## Review Comment: > especially since there are additional supervisor-only models that may change as well but aren’t tracked in the Execution API spec Exactly, this is what the migrator and the versions bundle here trying to resolve. Yes, there're partial of data models are directly from Execution API spec. but most of them aren't define in Execution API, like all the data models defined in `task-sdk/src/airflow/sdk/execution_time/comms.py`. From my perspective, no matter the data models define in Execution API or not, all the data models consumed by `CommsDecoder` should be versioned somewhere. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
jason810496 commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3274194034 ## task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py: ## Review Comment: > especially since there are additional supervisor-only models that may change as well but aren’t tracked in the Execution API spec Exactly, this is what the migrator and the versions bundle here trying to resolve. Yes, there're partial of data models are directly from Execution API spec. but most of them aren't define in Execution API, like `task-sdk/src/airflow/sdk/execution_time/comms.py`. From my perspective, no matter the data models define in Execution API or not, all the data models consumed by `CommsDecoder` should be versioned somewhere. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
jason810496 commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3274153042 ## task-sdk/src/airflow/sdk/execution_time/schema/schema.json: ## Review Comment: There's JSON schema for Dag serialization format `airflow-core/src/airflow/serialization/schema.json`, no harm to introduce explicit JSON schema snapshot for the migration schema IMHO. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
uranusjr commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3273285770 ## task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py: ## Review Comment: Yes, but there’s not an established way to forward those migrations here, especially since there are additional supervisor-only models that may change as well but aren’t tracked in the Execution API spec. I’m open to ideas; the currently way doesn’t really provide a good way to keep shared migrations in sync. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
uranusjr commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3273285770 ## task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py: ## Review Comment: Yes, but there’s not an established way to forward those migrations here, especially since there are additional supervisor-only models that may change as well but aren’t tracked in the Execution API spec. I’m open to ideas; the currently way can be awkward keeping migrations in sync 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
uranusjr commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3273285770 ## task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py: ## Review Comment: Yes, but there’s not an established way to forward those migrations here, especially since there are additional supervisor-only models that may change as well but aren’t tracked in the Execution API spec. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
uranusjr commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3273273721 ## task-sdk/src/airflow/sdk/execution_time/schema/schema.json: ## Review Comment: Hm, I guess we don’t since we would publish this somewhere for the Java SDK to consume. On the other hand it’d be a while for us to get this merged and then have the infrastructure ready. Maybe have this for now before we set things up on airflow.apache.org so the Java SDK has somewhere to point to, and remove it after setup and Java SDK points to it instead? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
ashb commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3273035152 ## task-sdk/src/airflow/sdk/execution_time/schema/versions/__init__.py: ## Review Comment: I sort of see why we have this, but it feels weird to have a the version history "duplicated" on the client side. Don't a lot of the schemas/models implicitly have a dependency on the Exec API side versioning 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] Add schema migration to supervisor-child comm [airflow]
ashb commented on code in PR #67235: URL: https://github.com/apache/airflow/pull/67235#discussion_r3273025585 ## task-sdk/src/airflow/sdk/execution_time/schema/schema.json: ## Review Comment: Do we need this file committed? I'd been liking the fact that we don't have generated files committed in the execution API to date. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
[PR] Add schema migration to supervisor-child comm [airflow]
uranusjr opened a new pull request, #67235: URL: https://github.com/apache/airflow/pull/67235 With foreign language SDKs, it may be possible the two sides of supervisor comm have different versions. This adds a migration layer at the supervisor (server) side, so an SDK (client) using a lower version of the schema may be able to communicate to the server. This adds: (subpoints describe what each component is for, but those are not covered in this PR) 1. A JSON schema file to describe the models used by the supervisor. This is automatically kept up-to-date with pre-commit hooks. * This should be published with versions for SDKs to use. * An SDK should use a version of this schema to talk to the supervisor. 2. A structure to host Cadwyn migration files inside `execution_time`. * Whenever the supervisor schema changes, a migration should be added to the SDK. 3. The supervisor use the migrations to upgrade the SDK’s message to match the schema version used by the supervisor, and downgrade messages before they are sent to the SDK to match the schema version used by the SDK. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
