Introduce support for event-triggered workflows

metaflow/client/core.py

@@ -1,37 +1,31 @@
 from __future__ import print_function
-from datetime import datetime
+
+import json
 import os
 import tarfile
-import json
-from io import BytesIO
 from collections import namedtuple
+from datetime import datetime
+from io import BytesIO
 from itertools import chain
-from typing import (
-    Any,
-    Dict,
-    FrozenSet,
-    Iterable,
-    List,
-    Optional,
-    Tuple,
-)
+from typing import Any, Dict, FrozenSet, Iterable, List, Optional, Tuple
 
-from metaflow.metaflow_environment import MetaflowEnvironment
 from metaflow.current import current
+from metaflow.events import Trigger
 from metaflow.exception import (
-    MetaflowNotFound,
-    MetaflowNamespaceMismatch,
     MetaflowInternalError,
     MetaflowInvalidPathspec,
+    MetaflowNamespaceMismatch,
+    MetaflowNotFound,
 )
 from metaflow.includefile import IncludedFile
 from metaflow.metaflow_config import DEFAULT_METADATA, MAX_ATTEMPTS
+from metaflow.metaflow_environment import MetaflowEnvironment
 from metaflow.plugins import ENVIRONMENTS, METADATA_PROVIDERS
 from metaflow.unbounded_foreach import CONTROL_TASK_TAG
-from metaflow.util import cached_property, resolve_identity, to_unicode, is_stringish
+from metaflow.util import cached_property, is_stringish, resolve_identity, to_unicode
 
-from .filecache import FileCache
 from .. import INFO_FILE
+from .filecache import FileCache
 
 try:
     # python2
@@ -356,7 +350,8 @@ def __iter__(self) -> Iterable["MetaflowObject"]:
             Iterator over all children
         """
         query_filter = {}
-        # skip namespace filtering if _namespace_check is False
+
+        # skip namespace filtering if _namespace_check is unset.
         if self._namespace_check and current_namespace:
             query_filter = {"any_tags": current_namespace}
 
@@ -1901,6 +1896,24 @@ def replace_tags(self, tags_to_remove: Iterable[str], tags_to_add: Iterable[str]
         self._user_tags = frozenset(final_user_tags)
         self._tags = frozenset([*self._user_tags, *self._system_tags])
 
+    @property
+    def trigger(self) -> Optional[Trigger]:
+        """
+        Returns a container of events that triggered this run.
+
+        This returns None if the run was not triggered by any events.
+
+        Returns
+        -------
+        Trigger, optional
+            Container of triggering events
+        """
+        if "start" in self:
+            meta = self["start"].task.metadata_dict.get("execution-triggers")
+            if meta:
+                return Trigger(json.loads(meta))
+        return None
+
 
 class Flow(MetaflowObject):
     """

metaflow/events.py

@@ -0,0 +1,162 @@
+from collections import OrderedDict, namedtuple
+from datetime import datetime
+
+MetaflowEvent = namedtuple("MetaflowEvent", ["name", "id", "timestamp", "type"])
+MetaflowEvent.__doc__ = """
+    Container of metadata that identifies the event that triggered
+    the `Run` under consideration.
+
+    Attributes
+    ----------
+    name : str
+        name of the event.
+    id : str
+        unique identifier for the event.
+    timestamp : datetime
+        timestamp recording creation time for the event.
+    type : str
+        type for the event - one of `event` or `run`
+    """
+
+
+class Trigger(object):
+    """
+    Defines a container of event triggers' metadata.
+
+    """
+
+    def __init__(self, _meta=None):
+        if _meta is None:
+            _meta = []
+
+        _meta.sort(key=lambda x: x.get("timestamp") or float("-inf"), reverse=True)
+
+        self._runs = None
+        self._events = [
+            MetaflowEvent(
+                **{
+                    **obj,
+                    # Add timestamp as datetime. Guaranteed to exist for Metaflow
+                    # events - best effort for everything else.
+                    **(
+                        {"timestamp": datetime.fromtimestamp(obj["timestamp"])}
+                        if obj.get("timestamp")
+                        and isinstance(obj.get("timestamp"), int)
+                        else {}
+                    ),
+                }
+            )
+            for obj in _meta
+        ]
+
+    @classmethod
+    def from_runs(cls, run_objs):
+        run_objs.sort(key=lambda x: x.finished_at, reverse=True)
+        trigger = Trigger(
+            [
+                {
+                    "type": "run",
+                    "timestamp": run_obj.finished_at,
+                    "name": "metaflow.%s.%s" % (run_obj.parent.id, run_obj["end"].id),
+                    "id": run_obj.end_task.pathspec,
+                }
+                for run_obj in run_objs
+            ]
+        )
+        trigger._runs = run_objs
+        return trigger
+
+    @property
+    def event(self):
+        """
+        The `MetaflowEvent` object corresponding to the triggering event.
+
+        If multiple events triggered the run, this property is the latest event.
+
+        Returns
+        -------
+        MetaflowEvent, optional
+            The latest event that triggered the run, if applicable.
+        """
+        return next(iter(self._events), None)
+
+    @property
+    def events(self):
+        """
+        The list of `MetaflowEvent` objects correspondings to all the triggering events.
+
+        Returns
+        -------
+        List[MetaflowEvent], optional
+            List of all events that triggered the run
+        """
+        return list(self._events) or None
+
+    @property
+    def run(self):
+        """
+        The corresponding `Run` object if the triggering event is a Metaflow run.
+
+        In case multiple runs triggered the run, this property is the latest run.
+        Returns `None` if none of the triggering events are a `Run`.
+
+        Returns
+        -------
+        Run, optional
+            Latest Run that triggered this run, if applicable.
+        """
+        if self._runs is None:
+            self.runs
+        return next(iter(self._runs), None)
+
+    @property
+    def runs(self):
+        """
+        The list of `Run` objects in the triggering events.
+        Returns `None` if none of the triggering events are `Run` objects.
+
+        Returns
+        -------
+        List[Run], optional
+            List of runs that triggered this run, if applicable.
+        """
+        if self._runs is None:
+            # to avoid circular import
+            from metaflow import Run
+
+            self._runs = [
+                Run(
+                    # object id is the task pathspec for events that map to run
+                    obj.id[: obj.id.index("/", obj.id.index("/") + 1)],
+                    _namespace_check=False,
+                )
+                for obj in self._events
+                if obj.type == "run"
+            ]
+
+        return list(self._runs) or None
+
+    def __getitem__(self, key):
+        """
+        If triggering events are runs, `key` corresponds to the flow name of the triggering run. Returns a triggering `Run` object corresponding to the key. If triggering events are not runs, `key` corresponds to the event name and a `MetaflowEvent` object is returned.
+        """
+        if self.runs:
+            for run in self.runs:
+                if run.path_components[0] == key:
+                    return run
+        elif self.events:
+            for event in self.events:
+                if event.name == key:
+                    return event
+        raise KeyError(key)
+
+    def __iter__(self):
+        if self.events:
+            return iter(self.events)
+        return iter([])
+
+    def __contains__(self, id):
+        try:
+            return bool(self.__getitem__(id))
+        except KeyError:
+            return False

metaflow/integrations.py

@@ -8,7 +8,7 @@
 
 from metaflow.extension_support.integrations import process_integration_aliases
 
-# To enable an alias `metaflow.alias.get_s3_client` to
+# To enable an alias `metaflow.integrations.get_s3_client` to
 # `metaflow.plugins.aws.aws_client.get_aws_client`, use the following:
 #
 # ALIASES_DESC = [("get_s3_client", ".plugins.aws.aws_client.get_aws_client")]
@@ -17,6 +17,8 @@
 #  - name: name of the integration alias
 #  - obj: object it points to
 #
+ALIASES_DESC = [("ArgoEvent", ".plugins.argo.argo_events.ArgoEvent")]
+
 # Aliases can be enabled or disabled through configuration or extensions:
 #  - ENABLED_INTEGRATION_ALIAS: list of alias names to enable.
 #  - TOGGLE_INTEGRATION_ALIAS: if ENABLED_INTEGRATION_ALIAS is not set anywhere

metaflow/metaflow_config.py

@@ -285,6 +285,16 @@
 ARGO_WORKFLOWS_KUBERNETES_SECRETS = from_conf("ARGO_WORKFLOWS_KUBERNETES_SECRETS", "")
 ARGO_WORKFLOWS_ENV_VARS_TO_SKIP = from_conf("ARGO_WORKFLOWS_ENV_VARS_TO_SKIP", "")
 
+##
+# Argo Events Configuration
+##
+ARGO_EVENTS_SERVICE_ACCOUNT = from_conf("ARGO_EVENTS_SERVICE_ACCOUNT")
+ARGO_EVENTS_EVENT_BUS = from_conf("ARGO_EVENTS_EVENT_BUS", "default")
+ARGO_EVENTS_EVENT_SOURCE = from_conf("ARGO_EVENTS_EVENT_SOURCE")
+ARGO_EVENTS_EVENT = from_conf("ARGO_EVENTS_EVENT")
+ARGO_EVENTS_WEBHOOK_URL = from_conf("ARGO_EVENTS_WEBHOOK_URL")
+
+
 ##
 # Airflow Configuration
 ##

metaflow/plugins/__init__.py

@@ -1,6 +1,6 @@
 from metaflow.extension_support.plugins import (
-    process_plugins,
     merge_lists,
+    process_plugins,
     resolve_plugins,
 )
 
@@ -56,6 +56,8 @@
     ("conda_base", ".conda.conda_flow_decorator.CondaFlowDecorator"),
     ("schedule", ".aws.step_functions.schedule_decorator.ScheduleDecorator"),
     ("project", ".project_decorator.ProjectDecorator"),
+    ("trigger", ".events_decorator.TriggerDecorator"),
+    ("trigger_on_finish", ".events_decorator.TriggerOnFinishDecorator"),
 ]
 
 # Add environments here
@@ -137,28 +139,29 @@ def get_plugin_cli():
 AWS_CLIENT_PROVIDERS = resolve_plugins("aws_client_provider")
 SECRETS_PROVIDERS = resolve_plugins("secrets_provider")
 
+from .cards.card_modules import MF_EXTERNAL_CARDS
+
 # Cards; due to the way cards were designed, it is harder to make them fit
 # in the resolve_plugins mechanism. This should be OK because it is unlikely that
 # cards will need to be *removed*. No card should be too specific (for example, no
 # card should be something just for Airflow, or Argo or step-functions -- those should
 # be added externally).
 from .cards.card_modules.basic import (
-    DefaultCard,
-    TaskSpecCard,
-    ErrorCard,
     BlankCard,
+    DefaultCard,
     DefaultCardJSON,
+    ErrorCard,
+    TaskSpecCard,
 )
 from .cards.card_modules.test_cards import (
-    TestErrorCard,
-    TestTimeoutCard,
-    TestMockCard,
-    TestPathSpecCard,
     TestEditableCard,
     TestEditableCard2,
+    TestErrorCard,
+    TestMockCard,
     TestNonEditableCard,
+    TestPathSpecCard,
+    TestTimeoutCard,
 )
-from .cards.card_modules import MF_EXTERNAL_CARDS
 
 CARDS = [
     DefaultCard,