feat: Validate k8s resource requirements for CPU, Memory and Ephemera…

…l Volumes (#944)

**Pull Request Checklist**
- [x] Fixes #943
- [x] Tests added
- [ ] Documentation/examples added
- [ ] [Good commit messages](https://cbea.ms/git-commit/) and/or PR
title

**Description of PR**

Currently, resource requirements are only partially validated for binary
units with `hera.workflows.validators.validate_storage_units()` function
for string values. The current implementaion misses some edge cases
where the input is missing numerial values and only the unit is provided
(i.e. `"Ki"` or `"Mi"`), and the comparison between request and limit is
limited to int values for `cpu_request` and `cpu_limit`.

This PR 
- Refactors validation for binary units
  - Fixes float values not compared for CPU requests/limits
- Adds validation for decimal units representation of CPU
requests/limits in string format
- Supported units are based on
https://github.com/kubernetes/apimachinery/blob/dc7e034c86479d49be4b0eefad307621e10caa0e/pkg/api/resource/quantity.go#L48
   - Units are optional and "0.5" and "1" are still valid
- Adds converters from decimal and binary units to float values
- Adds comparison of decimal units (CPU) and binary units (Memory and
Ephemeral Volumes) representations of resource requirements
- Adds tests for `hera/workflows/resources.py`

---------

Signed-off-by: KengoA <[email protected]>

src/hera/workflows/converters.py

@@ -0,0 +1,76 @@
+"""Utility functions for converting binary and decimal units to float values."""
+import re
+
+_decimal_multipliers = {
+    "m": 1e-3,
+    "k": 1e3,
+    "M": 1e6,
+    "G": 1e9,
+    "T": 1e12,
+    "P": 1e15,
+    "E": 1e18,
+}
+
+_binary_multipliers = {
+    "Ki": 2**10,
+    "Mi": 2**20,
+    "Gi": 2**30,
+    "Ti": 2**40,
+    "Pi": 2**50,
+    "Ei": 2**60,
+}
+
+
+def _convert_decimal_units(value: str) -> float:
+    """Converts the given decimal units to a float value. If no unit is given, the value is multiplied by 1.
+
+    Args:
+        value (str): The value to convert the decimal units of. The supported units are ['m', 'k', 'M', 'G', 'T', 'P', 'E'].
+        Note that the units are case sensitive.
+
+    Raises:
+        ValueError: When the identified unit is not a supported one.
+
+    Returns:
+        float: Float value of the given decimal units.
+    """
+    pattern = r"^\s*([+-]?\d+(?:\.\d+)?)([mkMGTPE]?)\s*$"
+    matches = re.match(pattern, value)
+
+    if matches:
+        value, unit = matches.groups()
+        return float(value) * _decimal_multipliers.get(unit, 1)
+    else:
+        raise ValueError(
+            f"Invalid decimal units for input: {value}. Supported units are ['m', 'k', 'M', 'G', 'T', 'P', 'E']."
+        )
+
+
+def _convert_binary_units(value: str) -> float:
+    """Converts the given binary units to a float value. If no unit is given, the value is multiplied by 1.
+
+    Args:
+        value (str): The value to convert the binary unit of. The supported units are ['Ki', 'Mi', 'Gi', 'Ti', 'Pi', 'Ei'].
+        Note that the units are case sensitive.
+
+    Raises:
+        ValueError: When the identified unit is not a supported one.
+
+    Returns:
+        float: Float value of the given binary units.
+    """
+    pattern = r"^\s*([+-]?\d+(?:\.\d+)?)([KMGTPE]i)?\s*$"
+    matches = re.match(pattern, value)
+
+    if matches:
+        value, unit = matches.groups()
+        return float(value) * _binary_multipliers.get(unit, 1)
+    else:
+        raise ValueError(
+            f"Invalid binary units for input: {value}. Supported units are ['Ki', 'Mi', 'Gi', 'Ti', 'Pi', 'Ei']."
+        )
+
+
+convert_memory_units = _convert_binary_units
+convert_storage_units = _convert_binary_units
+convert_cpu_units = _convert_decimal_units

src/hera/workflows/resources.py

@@ -5,8 +5,9 @@
     BaseModel as _BaseModel,
     root_validator,
 )
+from hera.workflows.converters import convert_cpu_units, convert_memory_units, convert_storage_units
 from hera.workflows.models import ResourceRequirements as _ModelResourceRequirements
-from hera.workflows.validators import validate_storage_units
+from hera.workflows.validators import validate_cpu_units, validate_memory_units, validate_storage_units
 
 
 # TODO: Move function?
@@ -70,23 +71,39 @@ def _check_specs(cls, values):
         ephemeral_limit: Optional[str] = values.get("ephemeral_limit")
 
         if memory_request is not None:
-            validate_storage_units(memory_request)
+            validate_memory_units(memory_request)
         if memory_limit is not None:
-            validate_storage_units(memory_limit)
+            validate_memory_units(memory_limit)
+            if memory_request is not None:
+                assert convert_memory_units(memory_request) <= convert_memory_units(
+                    memory_limit
+                ), "Memory request must be smaller or equal to limit"
 
         if ephemeral_request is not None:
             validate_storage_units(ephemeral_request)
-        if ephemeral_limit:
+        if ephemeral_limit is not None:
             validate_storage_units(ephemeral_limit)
+            if ephemeral_request is not None:
+                assert convert_storage_units(ephemeral_request) <= convert_storage_units(
+                    ephemeral_limit
+                ), "Ephemeral request must be smaller or equal to limit"
 
-        # TODO: add validation for CPU units if str
-        if cpu_limit is not None and isinstance(cpu_limit, int):
-            assert cpu_limit >= 0, "CPU limit must be positive"
-        if cpu_request is not None and isinstance(cpu_request, int):
+        if cpu_request is not None and isinstance(cpu_request, (int, float)):
             assert cpu_request >= 0, "CPU request must be positive"
-            if cpu_limit is not None and isinstance(cpu_limit, int):
+        if cpu_limit is not None and isinstance(cpu_limit, (int, float)):
+            assert cpu_limit >= 0, "CPU limit must be positive"
+            if cpu_request is not None and isinstance(cpu_request, (int, float)):
                 assert cpu_request <= cpu_limit, "CPU request must be smaller or equal to limit"
 
+        if cpu_request is not None and isinstance(cpu_request, str):
+            validate_cpu_units(cpu_request)
+        if cpu_limit is not None and isinstance(cpu_limit, str):
+            validate_cpu_units(cpu_limit)
+            if cpu_request is not None and isinstance(cpu_request, str):
+                assert convert_cpu_units(cpu_request) <= convert_cpu_units(
+                    cpu_limit
+                ), "CPU request must be smaller or equal to limit"
+
         return values
 
     def build(self) -> _ModelResourceRequirements:

src/hera/workflows/validators.py

@@ -43,30 +43,52 @@ def validate_name(name: str, max_length: Optional[int] = None, generate_name: bo
     return name
 
 
-def validate_storage_units(value: str) -> None:
-    """Validates the units of the given value.
+def _validate_binary_units(value: str) -> None:
+    """Validates the binary units of the given value.
 
-    The given value is expected to satisfy a unit/value format that specifies a resource requirement such as 500Mi,
+    The given value is expected to satisfy a unit/value format that specifies a binary resource requirement such as 500Mi,
     1Gi, etc.
 
     Parameters
     ----------
     value: str
-        The value to validate the units of.
+        The value to validate the binary unit of. The supported units are ['Ki', 'Mi', 'Gi', 'Ti', 'Pi', 'Ei']. Note that the units are case sensitive.
 
     Raises:
     ------
     ValueError
-        When the units cannot be extracted from the given value.
-    AssertionError
-        When the identified unit is not a supported one. The supported units are ['Ki', 'Mi', 'Gi', 'Ti', 'Pi', 'Ei'].
+        When the identified unit is not a supported one.
     """
-    supported_units = ["Ki", "Mi", "Gi", "Ti", "Pi", "Ei"]
-
-    pattern = r"[A-Za-z]+"
-    unit_search = re.search(pattern, value)
-    if not unit_search:
-        raise ValueError("could not extract units out of the passed in value")
-    else:
-        unit = unit_search.group(0)
-        assert unit in supported_units, f"unsupported unit for parsed value {value}"
+    pattern = r"^\s*(\d+(?:\.\d+)?)([KMGTPE]i)\s*$"
+    if not re.match(pattern, value):
+        raise ValueError(
+            f"Invalid binary unit for input: {value}. Supported units are ['Ki', 'Mi', 'Gi', 'Ti', 'Pi', 'Ei']."
+        )
+
+
+def _validate_decimal_units(value: str) -> None:
+    """Validates the decimal units of the given value.
+
+    The given value is expected to satisfy a unit/value format that specifies a decimal resource requirement such as 500m,
+    2k, etc. Note that the units are optional and accepts values such as int and float values in string e.g. "0.5" and "1".
+
+    Parameters
+    ----------
+    value: str
+        The value to validate the decimal unit of. The supported units are ['m', 'k', 'M', 'G', 'T', 'P', 'E']. Note that the units are case sensitive.
+
+    Raises:
+    ------
+    ValueError
+        When the identified unit is not a supported one.
+    """
+    pattern = r"^\s*(\d+(?:\.\d+)?)([mkMGTPE]?)\s*$"
+    if not re.match(pattern, value):
+        raise ValueError(
+            f"Invalid decimal unit for input: {value}. Supported units are ['m', 'k', 'M', 'G', 'T', 'P', 'E']."
+        )
+
+
+validate_storage_units = _validate_binary_units
+validate_memory_units = _validate_binary_units
+validate_cpu_units = _validate_decimal_units

tests/test_unit/test_converters.py

@@ -0,0 +1,58 @@
+import pytest
+
+from hera.workflows.converters import _convert_binary_units, _convert_decimal_units
+
+
+@pytest.mark.parametrize(
+    "value, expected",
+    [
+        ("500m", 0.5),
+        ("2k", 2000.0),
+        ("1.5M", 1500000.0),
+        ("42", 42.0),
+    ],
+)
+def test_convert_decimal_units(value, expected):
+    assert expected == _convert_decimal_units(value)
+
+
+@pytest.mark.parametrize(
+    "value, expected",
+    [
+        ("500Ki", 512000.0),
+        ("1Mi", 1048576.0),
+        ("2Gi", 2147483648.0),
+        ("42", 42.0),
+        ("0.5", 0.5),
+    ],
+)
+def test_convert_binary_units(value, expected):
+    assert expected == _convert_binary_units(value)
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "1.5Z",
+        "abc",
+        "1.5Ki",
+        "1.5Mi",
+    ],
+)
+def test_convert_decimal_units_invalid(value):
+    with pytest.raises(ValueError, match="Invalid decimal units"):
+        _convert_decimal_units(value)
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "1.5Z",
+        "abc",
+        "500m",
+        "2k",
+    ],
+)
+def test_convert_binary_units_invalid(value):
+    with pytest.raises(ValueError, match="Invalid binary units"):
+        _convert_binary_units(value)

tests/test_unit/test_resources.py

@@ -0,0 +1,86 @@
+import pytest
+
+from hera.workflows.models import ResourceRequirements
+from hera.workflows.resources import Resources
+
+
+class TestResources:
+    @pytest.mark.parametrize(
+        "cpu_request, cpu_limit, memory_request, memory_limit, ephemeral_request, ephemeral_limit",
+        [
+            ("500m", "500m", None, None, None, None),
+            ("500m", "1000m", None, None, None, None),
+            ("0.5", "0.5", None, None, None, None),
+            ("0.5", "1", None, None, None, None),
+            (0.5, 0.5, None, None, None, None),
+            (0.5, 1.0, None, None, None, None),
+            (1, 1, None, None, None, None),
+            (1, 2, None, None, None, None),
+            (1, 2, "1Gi", "1Gi", None, None),
+            (1, 2, "1Gi", "2Gi", None, None),
+            (1, 2, "256Mi", "256Mi", None, None),
+            (1, 2, "256Mi", "512Mi", "50Gi", None),
+            (1, 2, "256Mi", "512Mi", "50Gi", "50Gi"),
+            (1, 2, "256Mi", "512Mi", "50Gi", "100Gi"),
+        ],
+    )
+    def test_build_valid(
+        self, cpu_request, cpu_limit, memory_request, memory_limit, ephemeral_request, ephemeral_limit
+    ) -> None:
+        resources = Resources(
+            cpu_request=cpu_request,
+            cpu_limit=cpu_limit,
+            memory_request=memory_request,
+            memory_limit=memory_limit,
+            ephemeral_request=ephemeral_request,
+            ephemeral_limit=ephemeral_limit,
+        )
+        requirements = resources.build()
+
+        assert isinstance(requirements, ResourceRequirements)
+
+        if cpu_request is not None:
+            assert requirements.requests["cpu"].__root__ == str(cpu_request)
+
+        if cpu_limit is not None:
+            assert requirements.limits["cpu"].__root__ == str(cpu_limit)
+
+        if memory_request is not None:
+            assert requirements.requests["memory"].__root__ == memory_request
+
+        if memory_limit is not None:
+            assert requirements.limits["memory"].__root__ == memory_limit
+
+        if ephemeral_request is not None:
+            assert requirements.requests["ephemeral-storage"].__root__ == ephemeral_request
+
+        if ephemeral_limit is not None:
+            assert requirements.limits["ephemeral-storage"].__root__ == ephemeral_limit
+
+    @pytest.mark.parametrize(
+        "cpu_request, cpu_limit, memory_request, memory_limit, ephemeral_request, ephemeral_limit, error_message",
+        [
+            ("500a", None, None, None, None, None, "Invalid decimal unit"),
+            (None, None, "500k", None, None, None, "Invalid binary unit"),
+            (-1, 1, None, None, None, None, "must be positive"),
+            (-0.5, -0.5, None, None, None, None, "must be positive"),
+            (0.5, 0.2, None, None, None, None, "request must be smaller or equal to limit"),
+            (3, 2, None, None, None, None, "request must be smaller or equal to limit"),
+            ("1", "0.5", None, None, None, None, "request must be smaller or equal to limit"),
+            ("1000m", "800m", None, None, None, None, "request must be smaller or equal to limit"),
+            ("1", "1", "1Gi", "512Mi", None, None, "request must be smaller or equal to limit"),
+            ("1", "1", "1Gi", "1Gi", "100Gi", "50Gi", "request must be smaller or equal to limit"),
+        ],
+    )
+    def test_build_invalid(
+        self, cpu_request, cpu_limit, memory_request, memory_limit, ephemeral_request, ephemeral_limit, error_message
+    ) -> None:
+        with pytest.raises(ValueError, match=error_message):
+            _ = Resources(
+                cpu_request=cpu_request,
+                cpu_limit=cpu_limit,
+                memory_request=memory_request,
+                memory_limit=memory_limit,
+                ephemeral_request=ephemeral_request,
+                ephemeral_limit=ephemeral_limit,
+            )

tests/test_unit/test_validators.py

@@ -0,0 +1,36 @@
+import pytest
+
+from hera.workflows.validators import _validate_binary_units, _validate_decimal_units
+
+
+@pytest.mark.parametrize("value", ["500Ki", "1Mi", "2Gi", "1Ti", "1.5Pi", "1.5Ei"])
+def test_validate_binary_units_valid(value):
+    _validate_binary_units(value)
+
+
+@pytest.mark.parametrize(
+    "value, error_message",
+    [("Mi", "Invalid binary unit"), ("5K", "Invalid binary unit"), ("Ti", "Invalid binary unit")],
+)
+def test_validate_binary_units_invalid(value, error_message):
+    with pytest.raises(ValueError, match=error_message):
+        _validate_binary_units(value)
+
+
+@pytest.mark.parametrize("value", ["0.5", "1", "500m", "2k", "1.5M"])
+def test_validate_decimal_units_valid(value):
+    _validate_decimal_units(value)
+
+
+@pytest.mark.parametrize(
+    "value, error_message",
+    [
+        ("abc", "Invalid decimal unit"),
+        ("K", "Invalid decimal unit"),
+        ("2e", "Invalid decimal unit"),
+        ("1.5Z", "Invalid decimal unit"),
+    ],
+)
+def test_validate_decimal_units_invalid(value, error_message):
+    with pytest.raises(ValueError, match=error_message):
+        _validate_decimal_units(value)