Add instructions for how to handle failures coming from cache policies (#3135)

* Add instructions for how to handle failures coming from cache policies

Signed-off-by: Eduardo Apolinario <eapolinario@users.noreply.github.com>

* Add Cache, CachePolicy, and VersionParameters to the API

Signed-off-by: Eduardo Apolinario <eapolinario@users.noreply.github.com>

* Give an example of how to set version in Cache

Signed-off-by: Eduardo Apolinario <eapolinario@users.noreply.github.com>

* Simplify error message matching

Signed-off-by: Eduardo Apolinario <eapolinario@users.noreply.github.com>

---------

Signed-off-by: Eduardo Apolinario <eapolinario@users.noreply.github.com>
Co-authored-by: Eduardo Apolinario <eapolinario@users.noreply.github.com>

Author: eapolinario

flytekit/__init__.py

@@ -178,6 +178,9 @@
    :toctree: generated/
 
    HashMethod
+   Cache
+   CachePolicy
+   VersionParameters
 
 Artifacts
 =========

flytekit/core/cache.py

@@ -85,8 +85,13 @@ def get_version(self, params: VersionParameters) -> str:
             return self.version
 
         task_hash = ""
-        for cache_instance in self._policies:
-            task_hash += cache_instance.get_version(self.salt, params)
+        for policy in self._policies:
+            try:
+                task_hash += policy.get_version(self.salt, params)
+            except Exception as e:
+                raise ValueError(
+                    f"Failed to generate version for cache policy {policy}. Please consider setting the version in the Cache definition, e.g. Cache(version='v1.2.3')"
+                ) from e
 
         hash_obj = hashlib.sha256(task_hash.encode())
         return hash_obj.hexdigest()

tests/flytekit/unit/core/test_cache.py

@@ -12,6 +12,11 @@ def get_version(self, salt: str, params: VersionParameters) -> str:
         return salt
 
 
+class ExceptionCachePolicy(CachePolicy):
+    def get_version(self, salt: str, params: VersionParameters) -> str:
+        raise Exception("This is an exception")
+
+
 @pytest.fixture
 def default_serialization_settings():
     default_image = Image(name="default", fqn="full/name", tag="some-tag")
@@ -167,3 +172,12 @@ def t_cached_explicit_version(a: int) -> int:
     serialized_t_cached_explicit_version = get_serializable_task(OrderedDict(), default_serialization_settings, t_cached_explicit_version)
     assert serialized_t_cached_explicit_version.template.metadata.discoverable == True
     assert serialized_t_cached_explicit_version.template.metadata.discovery_version == "a-version"
+
+
+def test_cache_policy_exception(default_serialization_settings):
+    # Set the address of the ExceptionCachePolicy in the error message so that the test is robust to changes in the
+    # address of the ExceptionCachePolicy class
+    with pytest.raises(ValueError, match="Failed to generate version for cache policy"):
+        @task(cache=Cache(policies=ExceptionCachePolicy()))
+        def t_cached(a: int) -> int:
+            return a