Full support for top level JSON Schema validation for DagRun params?

[philippefutureboy]

I am super grateful for the introduction of Advanced Params using json-schema (#17100), which has greatly improved the DX and QA-X of our data engineering pipeline.

This being said, yesterday we had an incident that turned out that could have been avoided would the implementation be slightly different.
One of our non-airflow servers called our Airflow API with a request to trigger a DagRun with a previous version of the configuration supported by said DAG.
It resulted in unexpected behaviour, and we scrambled to prevent further fallout.

Beyond the desync between the APIs of the two systems, the underlying issue was the following: The current implementation allowed "additionalProperties" at the top level. Example:

Given the following DAG:

with airflow.DAG(
    # ...
    params={
        "a": Param(
            #...
        ),
        "b": Param(
            #...
        ),
    },
) as dag:
    # ...

There does not seem to be a way to validate that no additional properties (e.g. "c", "d", etc.) be provided to the conf.
This is important because in our case it would have prevented running a configuration that was supposed to filter the actions of the DAG, but did not because the props had a dot instead of an underscore.

So our patch was to do the following:

1. Install jsonschema@^3.2.0
2. Extract the params config as a JSON dict:

{
  "properties": {
    "a": {...},
    "a": {...}
  },
  "additionalProperties": false
}

3. Dynamically loop through the params as follows:

    params={
            key: Param(**prop)
            for key, prop in PARAMS_JSON_SCHEMA["properties"].items()
    }

4. Do a second validation in an initial PythonOperator task:

    dag_run = context["dag_run"]
    try:
        dag_conf = dag_run.conf
    except:
        dag_conf = {}

    jsonschema.validate(instance=dag_conf, schema=PARAMS_JSON_SCHEMA)

Hence the suggestion is:

Is there a way that the changes introduced in #17100 be modified to allow top-level jsonschema validation?

[msumit]

Hi @philippefutureboy the main reason to keep the params interface as it is to make sure that we don't break the backward compatibility. To solve your use case I believe some modifications can be done in ParamsDict class. However, wanted to understand why can't just make use of params keys directly, thus ignoring any extra keys passed into it?

[philippefutureboy]

Hey @msumit!

Thanks for #17100 and for your response.

However, wanted to understand why can't just make use of params keys directly, thus ignoring any extra keys passed into it?

Well the exact issue in our case was not that we used the extra keys, it's that none of the expected keys were passed.
This is what we got:

{"tenant_ids": ["<id>"], "rti_ids": ["<id>"], "force": true}

When the new API expects

{"filters": {"tenant.ids": ["<id>"],  "rti.ids": ["<id>"]}, "force": true}

So since tenant_ids and rti_ids were top level instead of within "filters", our DAG did not filter its actions accordingly.
The thing is, a conf without filter:

{"filters": {}, "force": true}

is still a valid configuration because we want to be able to mass generate analyses for our clients, whether for testing purpose or live purpose.
So what we did on top of the steps described in OP is that we added a two-man rule parameter to ensure "force" and empty filters only work when said two-man rule parameter is also true, e.g. {"filters": {}, "force": true, "dragonforce": true}

Let me know if you still have some questions :)

[msumit]

@philippefutureboy thanks for the explanation of your use case. I believe something can be done by extending ParamsDict class & using that instead of DAG's params, however that would still need some changes in core airflow code.