jason810496 commented on code in PR #74041:
URL: https://github.com/apache/airflow/pull/74041#discussion_r4163179075
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -2131,6 +2134,70 @@ def from_dict(cls, serialized_obj: dict) ->
SerializedDAG:
# Pass client_defaults directly to deserialize_dag
return cls.deserialize_dag(serialized_obj["dag"], client_defaults)
+ @classmethod
+ def fill_config_defaults(cls, serialized_obj: dict[str, Any]) -> None:
+ """
+ Fill in the Dag settings a serialized Dag leaves unset from the
Airflow config, as a Python Dag does.
+
+ A Lang-SDK runtime cannot read the Airflow config, so it leaves
``max_active_tasks``,
+ ``max_active_runs``, ``max_consecutive_failed_dag_runs``, ``catchup``
and
+ ``disable_bundle_versioning`` out unless the Dag sets them. A value
the Dag sets is kept.
+ *serialized_obj* is changed in place.
+ """
+ dag = serialized_obj.get("dag")
+ if not isinstance(dag, dict):
+ # validate_serialized_dag rejects it.
+ return
+ for key, get, section, option in (
+ ("max_active_tasks", conf.getint, "core",
"max_active_tasks_per_dag"),
Review Comment:
Done in 5c5ec77031. `receive()` now calls `fill_config_defaults` and
`validate_serialized_dag`, and the duplicate list is gone.
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -2131,6 +2134,70 @@ def from_dict(cls, serialized_obj: dict) ->
SerializedDAG:
# Pass client_defaults directly to deserialize_dag
return cls.deserialize_dag(serialized_obj["dag"], client_defaults)
+ @classmethod
+ def fill_config_defaults(cls, serialized_obj: dict[str, Any]) -> None:
+ """
+ Fill in the Dag settings a serialized Dag leaves unset from the
Airflow config, as a Python Dag does.
+
+ A Lang-SDK runtime cannot read the Airflow config, so it leaves
``max_active_tasks``,
+ ``max_active_runs``, ``max_consecutive_failed_dag_runs``, ``catchup``
and
+ ``disable_bundle_versioning`` out unless the Dag sets them. A value
the Dag sets is kept.
+ *serialized_obj* is changed in place.
+ """
+ dag = serialized_obj.get("dag")
+ if not isinstance(dag, dict):
+ # validate_serialized_dag rejects it.
+ return
+ for key, get, section, option in (
+ ("max_active_tasks", conf.getint, "core",
"max_active_tasks_per_dag"),
+ ("max_active_runs", conf.getint, "core",
"max_active_runs_per_dag"),
+ (
+ "max_consecutive_failed_dag_runs",
+ conf.getint,
+ "core",
+ "max_consecutive_failed_dag_runs_per_dag",
+ ),
+ ("catchup", conf.getboolean, "scheduler", "catchup_by_default"),
+ ("disable_bundle_versioning", conf.getboolean, "dag_processor",
"disable_bundle_versioning"),
+ ):
+ if key not in dag:
+ dag[key] = get(section, option)
+
+ @classmethod
+ def validate_serialized_dag(cls, serialized_obj: dict[str, Any]) -> None:
+ """
+ Check that a serialized Dag, such as one a Lang-SDK runtime produced,
can be stored and loaded.
+
+ It must match the JSON schema, deserialize, and have no cycle in its
task graph.
+ *serialized_obj* is not changed.
+
+ :raises DeserializationError: if it does not.
+ """
+ from jsonschema import ValidationError
Review Comment:
Done in e154d35e33.
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -2131,6 +2134,70 @@ def from_dict(cls, serialized_obj: dict) ->
SerializedDAG:
# Pass client_defaults directly to deserialize_dag
return cls.deserialize_dag(serialized_obj["dag"], client_defaults)
+ @classmethod
+ def fill_config_defaults(cls, serialized_obj: dict[str, Any]) -> None:
+ """
+ Fill in the Dag settings a serialized Dag leaves unset from the
Airflow config, as a Python Dag does.
+
+ A Lang-SDK runtime cannot read the Airflow config, so it leaves
``max_active_tasks``,
+ ``max_active_runs``, ``max_consecutive_failed_dag_runs``, ``catchup``
and
+ ``disable_bundle_versioning`` out unless the Dag sets them. A value
the Dag sets is kept.
+ *serialized_obj* is changed in place.
+ """
+ dag = serialized_obj.get("dag")
+ if not isinstance(dag, dict):
+ # validate_serialized_dag rejects it.
+ return
+ for key, get, section, option in (
+ ("max_active_tasks", conf.getint, "core",
"max_active_tasks_per_dag"),
+ ("max_active_runs", conf.getint, "core",
"max_active_runs_per_dag"),
+ (
+ "max_consecutive_failed_dag_runs",
+ conf.getint,
+ "core",
+ "max_consecutive_failed_dag_runs_per_dag",
+ ),
+ ("catchup", conf.getboolean, "scheduler", "catchup_by_default"),
+ ("disable_bundle_versioning", conf.getboolean, "dag_processor",
"disable_bundle_versioning"),
+ ):
+ if key not in dag:
+ dag[key] = get(section, option)
+
+ @classmethod
+ def validate_serialized_dag(cls, serialized_obj: dict[str, Any]) -> None:
+ """
+ Check that a serialized Dag, such as one a Lang-SDK runtime produced,
can be stored and loaded.
+
+ It must match the JSON schema, deserialize, and have no cycle in its
task graph.
+ *serialized_obj* is not changed.
+
+ :raises DeserializationError: if it does not.
+ """
+ from jsonschema import ValidationError
+
+ dag = serialized_obj.get("dag")
+ dag_id = dag.get("dag_id") if isinstance(dag, dict) else None
+ try:
+ cls.validate_schema(serialized_obj)
+ except ValidationError as e:
+ raise DeserializationError(
+ dag_id, f"Dag {dag_id!r} does not match the schema:
{e.message}"
Review Comment:
Done in e154d35e33. The message now includes `e.json_path`, for example
`$.dag.max_active_runs`.
##########
airflow-core/tests/unit/serialization/test_dag_serialization.py:
##########
@@ -5155,3 +5155,110 @@ def get_weight(self, ti):
op = BaseOperator(task_id="empty_task",
weight_rule=NotRegisteredPriorityWeightStrategy())
with pytest.raises(ValueError, match="Unknown priority strategy"):
OperatorSerialization.serialize(op)
+
+
+class TestValidateSerializedDag:
+ @staticmethod
+ def _serialize() -> dict:
+ with DAG(dag_id="checked_dag", schedule=None) as dag:
+ BaseOperator(task_id="extract") >> BaseOperator(task_id="load")
+ return DagSerialization.to_dict(dag)
+
+ def test_accepts_a_dag_that_loads(self):
+ data = self._serialize()
+ data["__version"] = 2
+ before = copy.deepcopy(data)
+
+ DagSerialization.validate_serialized_dag(data)
+
+ assert data == before
+
+ @pytest.mark.parametrize(
+ ("change", "error"),
+ [
+ pytest.param(
+ {"max_active_runs": "many"},
+ "Dag 'checked_dag' does not match the schema: 'many' is not of
type 'number'",
+ id="schema",
+ ),
+ pytest.param(
+ {"timetable": {"__type": "no.such.Timetable", "__var": {}}},
+ "Dag 'checked_dag' cannot be deserialized:
TimetableNotRegistered: ",
+ id="deserialize",
+ ),
+ ],
+ )
+ def test_rejects_a_dag_that_does_not_load(self, change, error):
+ data = self._serialize()
+ data["dag"].update(change)
+
+ with pytest.raises(DeserializationError, match=f"^{re.escape(error)}"):
+ DagSerialization.validate_serialized_dag(data)
+
+ def test_rejects_a_dag_with_a_cycle(self):
+ data = self._serialize()
+ load = next(task for task in data["dag"]["tasks"] if
task["__var"]["task_id"] == "load")
+ load["__var"]["downstream_task_ids"] = ["extract"]
+
+ with pytest.raises(DeserializationError, match="^Dag 'checked_dag' has
a cycle through task 'load'$"):
+ DagSerialization.validate_serialized_dag(data)
+
+
+class TestFillConfigDefaults:
+ CONFIG = {
+ ("core", "max_active_tasks_per_dag"): "7",
+ ("core", "max_active_runs_per_dag"): "3",
+ ("core", "max_consecutive_failed_dag_runs_per_dag"): "5",
+ ("scheduler", "catchup_by_default"): "True",
Review Comment:
Done in e154d35e33. The two booleans now have different config values, and a
`downstream_task_ids=["ghost"]` case covers the `__cause__` unwrap.
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -2131,6 +2134,70 @@ def from_dict(cls, serialized_obj: dict) ->
SerializedDAG:
# Pass client_defaults directly to deserialize_dag
return cls.deserialize_dag(serialized_obj["dag"], client_defaults)
+ @classmethod
+ def fill_config_defaults(cls, serialized_obj: dict[str, Any]) -> None:
+ """
+ Fill in the Dag settings a serialized Dag leaves unset from the
Airflow config, as a Python Dag does.
+
+ A Lang-SDK runtime cannot read the Airflow config, so it leaves
``max_active_tasks``,
+ ``max_active_runs``, ``max_consecutive_failed_dag_runs``, ``catchup``
and
+ ``disable_bundle_versioning`` out unless the Dag sets them. A value
the Dag sets is kept.
+ *serialized_obj* is changed in place.
+ """
+ dag = serialized_obj.get("dag")
+ if not isinstance(dag, dict):
+ # validate_serialized_dag rejects it.
+ return
+ for key, get, section, option in (
+ ("max_active_tasks", conf.getint, "core",
"max_active_tasks_per_dag"),
+ ("max_active_runs", conf.getint, "core",
"max_active_runs_per_dag"),
+ (
+ "max_consecutive_failed_dag_runs",
+ conf.getint,
+ "core",
+ "max_consecutive_failed_dag_runs_per_dag",
+ ),
+ ("catchup", conf.getboolean, "scheduler", "catchup_by_default"),
+ ("disable_bundle_versioning", conf.getboolean, "dag_processor",
"disable_bundle_versioning"),
+ ):
+ if key not in dag:
+ dag[key] = get(section, option)
+
+ @classmethod
+ def validate_serialized_dag(cls, serialized_obj: dict[str, Any]) -> None:
+ """
+ Check that a serialized Dag, such as one a Lang-SDK runtime produced,
can be stored and loaded.
+
+ It must match the JSON schema, deserialize, and have no cycle in its
task graph.
+ *serialized_obj* is not changed.
+
+ :raises DeserializationError: if it does not.
+ """
+ from jsonschema import ValidationError
+
+ dag = serialized_obj.get("dag")
+ dag_id = dag.get("dag_id") if isinstance(dag, dict) else None
+ try:
+ cls.validate_schema(serialized_obj)
Review Comment:
Done in e154d35e33. Every `tasks` entry must now be an operator with a
string `task_id`, and a repeated `task_id` is rejected. I didn't enforce the
full `#/definitions/operator`: `to_dict` validates every Python Dag, and that
definition has never been checked. 4f3cecbd67 syncs the Java SDK copy of the
schema, and f1791c6a22 fixes two tests that stored a serialized Dag proxy,
which this check now catches.
##########
airflow-core/adr/lang-sdk/0004-dag-parsing.md:
##########
@@ -216,9 +216,11 @@ The language runtime must produce a `DagFileParsingResult`
that matches Python A
| `start_date` | float (epoch) | if set | Unwrapped from `__type`/`__var` |
| `end_date` | float (epoch) | if set | Unwrapped from `__type`/`__var` |
| `tags` | list | if non-empty | Unwrapped from `__type`/`__var` |
-| `catchup` | bool | if `true` | |
-| `max_active_tasks` | int | if non-default | |
-| `max_active_runs` | int | if non-default | |
+| `catchup` | bool | if set | Airflow fills an unset field from its config |
Review Comment:
Done in ea4f7637ad. Both lines now say a runtime may leave out the marked
fields, and Airflow fills them in before it loads the Dag.
##########
airflow-core/src/airflow/serialization/serialized_objects.py:
##########
@@ -2131,6 +2134,70 @@ def from_dict(cls, serialized_obj: dict) ->
SerializedDAG:
# Pass client_defaults directly to deserialize_dag
return cls.deserialize_dag(serialized_obj["dag"], client_defaults)
+ @classmethod
+ def fill_config_defaults(cls, serialized_obj: dict[str, Any]) -> None:
Review Comment:
Not here, it's a follow-up. To fill `client_defaults` on the server, each
runtime first has to always write the task fields the author set, even when
they equal the default. That changes the runtime contract, so it needs its own
ADR change.
##########
airflow-core/src/airflow/serialization/schema.json:
##########
@@ -257,7 +257,18 @@
},
"tasks": {
"type": "array",
- "additionalProperties": { "$ref": "#/definitions/operator" }
+ "items": {
+ "type": "object",
+ "required": [ "__type", "__var" ],
+ "properties": {
+ "__type": { "const": "operator" },
+ "__var": {
+ "type": "object",
+ "required": [ "task_id" ],
+ "properties": { "task_id": { "type": "string" } }
+ }
Review Comment:
Actually, the Python Dag (task-sdk) can't even pass the `"$ref":
"#/definitions/operator"` schema. All the example Dags we have will fail. In
another word the `"#/definitions/operator"` is wrong for a long time before.
This will be the thing we need to fix in the further follow-up.
This PR is the first PR that started doing the actual validation.
Once we figure out the up-to-date schema of `"$ref":
"#/definitions/operator"`, then can wire it up like the shape as you mentioned.
So I will keep it as-is for now.
--
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]
