Repository: apache/airflow
Pull request: Add default weight rule configuration option
closes: https://github.com/apache/airflow/issues/12807
<!--
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 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/
-->

---
**^ Add meaningful description above**

Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+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 [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).


Problem statement:
add default weight_rule to airflow.cfg
**Description**

It would be nice if the weight_rule default value could be managed by a global config

suggested config:
```
# Weighting method used for the effective total priority weight of the task. 
# Options are: { downstream | upstream | absolute } default is
default_weight_rule = downstream
```
**Use case / motivation**

In some pipeline, you really need to have absolute weight, and then you have to add a line in each task definition which is annoying

Hints from issue discussion:
>In some pipeline, you really need to have absolute weight, and then you have to add a line in each task definition which is annoying

Why not define it once in `default_args`?
Then it needs to be done in each dag, right?
> Then it needs to be done in each dag, right?

Yes

Guess that is the point of issue!

On Thu, Feb 4, 2021 at 4:37 PM eladkal <notifications@github.com> wrote:

> Then it needs to be done in each dag, right?
>
> Yes
>
> —
> You are receiving this because you authored the thread.
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/airflow/issues/12807#issuecomment-773399489>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AAG447LNSTCWEVHQ57KA7SDS5K5LDANCNFSM4UNG56PA>
> .
>

> Guess that is the point of issue!

You described that you want to avoid specifying the same value of `weight_rule` per task.
For that you can define it once in `default_args`.

Maybe [task policy](https://airflow.apache.org/docs/apache-airflow/stable/concepts.html#task-level-cluster-policy) can solve some of it? (Though not exactly the same as this overwrites values provided by the user)
I really think having that one configurable for entire cluster could be
helpful!

On Thu, Feb 4, 2021, 4:56 PM eladkal <notifications@github.com> wrote:

> Guess that is the point of issue!
>
> You described that you want to avoid specifying the same value of
> weight_rule per task.
> For that you can define it once in default_args.
>
> Though it's not exactly what you are describing maybe task policy
> <https://airflow.apache.org/docs/apache-airflow/stable/concepts.html#task-level-cluster-policy>
> can solve some of it.
>
> —
> You are receiving this because you authored the thread.
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/airflow/issues/12807#issuecomment-773412937>,
> or unsubscribe
> <https://github.com/notifications/unsubscribe-auth/AAG447JD64NOSWACDNYS2ADS5K7TJANCNFSM4UNG56PA>
> .
>

Commit message: Add configuration validation

Patch to review:
diff --git a/airflow/config_templates/config.yml b/airflow/config_templates/config.yml
index 42d9bca822d13..4c7cfa89dbee0 100644
--- a/airflow/config_templates/config.yml
+++ b/airflow/config_templates/config.yml
@@ -354,6 +354,13 @@
       type: string
       example: ~
       default: "0"
+    - name: default_task_weight_rule
+      description: |
+        The weighting method used for the effective total priority weight of the task
+      version_added: 2.2.0
+      type: string
+      example: ~
+      default: "downstream"
     - name: min_serialized_dag_update_interval
       description: |
         Updating serialized DAG can not be faster than a minimum interval to reduce database write rate.
diff --git a/airflow/config_templates/default_airflow.cfg b/airflow/config_templates/default_airflow.cfg
index c7ede491d9b50..464d5f1ed62b7 100644
--- a/airflow/config_templates/default_airflow.cfg
+++ b/airflow/config_templates/default_airflow.cfg
@@ -207,6 +207,9 @@ dag_discovery_safe_mode = True
 # The number of retries each task is going to have by default. Can be overridden at dag or task level.
 default_task_retries = 0
 
+# The weighting method used for the effective total priority weight of the task
+default_task_weight_rule = downstream
+
 # Updating serialized DAG can not be faster than a minimum interval to reduce database write rate.
 min_serialized_dag_update_interval = 30
 
diff --git a/airflow/configuration.py b/airflow/configuration.py
index fb2078a01ad0c..40534b2e62ec8 100644
--- a/airflow/configuration.py
+++ b/airflow/configuration.py
@@ -38,6 +38,7 @@
 from airflow.secrets import DEFAULT_SECRETS_SEARCH_PATH, BaseSecretsBackend
 from airflow.utils import yaml
 from airflow.utils.module_loading import import_string
+from airflow.utils.weight_rule import WeightRule
 
 log = logging.getLogger(__name__)
 
@@ -193,6 +194,8 @@ class AirflowConfigParser(ConfigParser):
         },
     }
 
+    enums = {"default_task_weight_rule": WeightRule}
+
     # This method transforms option names on every read, get, or set operation.
     # This changes from the default behaviour of ConfigParser from lowercasing
     # to instead be case-preserving
@@ -212,6 +215,8 @@ def validate(self):
 
         self._validate_config_dependencies()
 
+        self._validate_enums()
+
         for section, replacement in self.deprecated_values.items():
             for name, info in replacement.items():
                 old, new, version = info
@@ -229,6 +234,14 @@ def validate(self):
 
         self.is_validated = True
 
+    def _validate_enums(self):
+        """Validate that enum type config has an accepted value"""
+        for setting, enum_class in self.enums.items():
+            if self.has_option('core', setting):
+                value = self.get("core", setting)
+                if not enum_class.is_valid(value):
+                    raise AirflowConfigException(f"{value} is not an accepted config for {setting}")
+
     def _validate_config_dependencies(self):
         """
         Validate that config values aren't invalid given other config values
diff --git a/airflow/models/baseoperator.py b/airflow/models/baseoperator.py
index 800881f1b68b7..58b9e468354dc 100644
--- a/airflow/models/baseoperator.py
+++ b/airflow/models/baseoperator.py
@@ -497,7 +497,7 @@ def __init__(
         params: Optional[Dict] = None,
         default_args: Optional[Dict] = None,
         priority_weight: int = 1,
-        weight_rule: str = WeightRule.DOWNSTREAM,
+        weight_rule: str = conf.get('core', 'default_task_weight_rule', fallback=WeightRule.DOWNSTREAM),
         queue: str = conf.get('operators', 'default_queue'),
         pool: Optional[str] = None,
         pool_slots: int = 1,
diff --git a/tests/core/test_configuration.py b/tests/core/test_configuration.py
index d651de13cd26e..f3a7305e5665a 100644
--- a/tests/core/test_configuration.py
+++ b/tests/core/test_configuration.py
@@ -696,3 +696,12 @@ def test_run_command(self):
 
     def test_confirm_unittest_mod(self):
         assert conf.get('core', 'unit_test_mode')
+
+    def test_enum_default_task_weight_rule_from_conf(self):
+        test_conf = AirflowConfigParser(default_config='')
+        test_conf.read_dict({'core': {'default_task_weight_rule': 'sidestream'}})
+        with pytest.raises(AirflowConfigException) as ctx:
+            test_conf.validate()
+        exception = str(ctx.value)
+        message = "sidestream is not an accepted config for default_task_weight_rule"
+        assert message == exception
diff --git a/tests/models/test_baseoperator.py b/tests/models/test_baseoperator.py
index a6c1ee5a47a38..f8fb99007fcbc 100644
--- a/tests/models/test_baseoperator.py
+++ b/tests/models/test_baseoperator.py
@@ -30,6 +30,7 @@
 from airflow.utils.edgemodifier import Label
 from airflow.utils.task_group import TaskGroup
 from airflow.utils.trigger_rule import TriggerRule
+from airflow.utils.weight_rule import WeightRule
 from tests.models import DEFAULT_DATE
 from tests.test_utils.mock_operators import DeprecatedOperator, MockNamedTuple, MockOperator
 
@@ -565,6 +566,14 @@ def test_replace_dummy_trigger_rule(self, rule):
 
             assert op1.trigger_rule == TriggerRule.ALWAYS
 
+    def test_weight_rule_default(self):
+        op = BaseOperator(task_id="test_task")
+        assert WeightRule.DOWNSTREAM == op.weight_rule
+
+    def test_weight_rule_override(self):
+        op = BaseOperator(task_id="test_task", weight_rule="upstream")
+        assert WeightRule.UPSTREAM == op.weight_rule
+
 
 def test_init_subclass_args():
     class InitSubclassOp(BaseOperator):


Code review comments:
diff_hunk: @@ -229,6 +234,14 @@ def validate(self):
 
         self.is_validated = True
 
+    def _validate_enums(self):
+        """Validate that enum type config has an accepted value"""
+        for setting, enum_class in self.enums.items():
+            if self.has_option('core', setting):
+                value = self.get("core", setting)
+                if not enum_class.is_valid(value):
+                    raise AirflowConfigException(f"{value} is not an accepted config for {setting}")
path: airflow/configuration.py
text: @user1:
```suggestion
                    raise AirflowConfigException(f"{value} is not an accepted config for  [{section}] {setting}")
```

diff_hunk: @@ -229,6 +234,14 @@ def validate(self):
 
         self.is_validated = True
 
+    def _validate_enums(self):
+        """Validate that enum type config has an accepted value"""
+        for setting, enum_class in self.enums.items():
+            if self.has_option('core', setting):
+                value = self.get("core", setting)
path: airflow/configuration.py
text: @user1:
```suggestion
        for (section, setting), enum_class in self.enums.items():
            if self.has_option(section, setting):
                value = self.get(section, setting)
```

diff_hunk: @@ -193,6 +194,8 @@ class AirflowConfigParser(ConfigParser):
         },
     }
 
+    enums = {"default_task_weight_rule": WeightRule}
path: airflow/configuration.py
text: @user1:
```suggestion
    enums = {("core", "default_task_weight_rule"): WeightRule}
```