Priority Weights: Examples are not reproducible

[ninsbl]

What do you see as an issue?

I am trying to implement a custom weight rule, following the documentation (https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/priority-weight.html#priority-weights).

When I try using the examples as a basis for customization, there seems to be a missing link between the example plugin code and the example DAG code. If I use the example DAG code it throws an import error, that the module custom_weight_rule_module is not found. And when I store the plugin code in a plugin prioritization_plugins and adjust the import, I get the following warning/error message in the Airflow UI:
'prioritization_plugins.DecreasingPriorityStrategy' is not one of ['downstream', 'upstream', 'absolute'] Failed validating 'enum' in schema['properties']['weight_rule']: {'description': 'Weight rule.', 'type': 'string', 'enum': ['downstream', 'upstream', 'absolute']} On instance['weight_rule']:
and all tasks have priority_weight=1

As a result I cannot figure out how to use cusom weight rules...

Solving the problem

It would be very helpful for new users of the feature to have a working, fully reproducible example that covers all necessary steps for using cusom weight rules.

Anything else

I would be willing to provide a PR, but I can not figure out how to get the feature working. Thus, I can not provide an improved documentation...

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[ninsbl]

See also: #38222 where it is discussed that is not obvious for first time users how to implement the feature in own plugin + DAG code...

[potiuk]

I marked it as good first issue, hopefully someone will improve it.

[azharizz]

i will take this issue, please assign to me thank you

[ninsbl]

i will take this issue, please assign to me thank you

@azharizz : Thanks! Very much appreciated!

After some more attempts, I am no longer sure this is just a documentation issue, The WeightRule Enum used for validation of weight_rule input does not seem to contain the added, custom weight rules, although I could verify that the plugin is registered and validate_and_load_priority_weight_strategy returns an object of the PriorityWeightStrategy class:

from airflow.task.priority_strategy import validate_and_load_priority_weight_strategy

validate_and_load_priority_weight_strategy('decreasing_priority_weight_strategy.DecreasingPriorityStrategy')
<decreasing_priority_weight_strategy.DecreasingPriorityStrategy object at 0x7f46c65bb390>

Since WeightRule is an Enum, it cannot be easily extended,
Maybe, the weight_rule configuration (

airflow/airflow/configuration.py

Line 389 in 8fbdf75

("core", "default_task_weight_rule"): sorted(WeightRule.all_weight_rules()),

) needs to be extended from plugins too??? It is the marshmallow schema validation that seems to fail...

Maybe something like this is needed in the configuration:

from enum import Enum
from airflow import plugins_manager
plugins_manager.initialize_priority_weight_strategy_plugins()
plugins_manager.priority_weight_strategy_classes

(...)

    enums_options = {
    ("core", "default_task_weight_rule"): Enum(
        "WeightRule", {str(k.upper()): str(k) for k in sorted(set(plugins_manager.priority_weight_strategy_classes.keys()))}
        ).__members__.values(),
    (...)

Or the weight_rule validation can be skipped completely at the configuration stage, as this is done later with the validate_and_load_priority_weight_strategy function?

Anyway, if anyone has an example showing how this feature is implemented in a DAG properly, I would be very greatful! Likewise for any hints on fixing the issue if it is not documentation only...

[mohanraj-web-devops]

Can i work on this issue @ninsbl

[eladkal]

cc @hussein-awala

[ninsbl]

OK, I did some more digging and I am now relatively sure this is actually a bug, and not just short-coming in the documentation.
The priority weight is computed according to my priority weight strategy, when and if I remove the schema validation for WeightRuleField in airflow/api_connexion/schemas/task_schema.py.

That said, priority weight does not seem to work as I expect (it does not seem to have effect when I clear the state of a bunch of DagRuns), but that is a different issue...