Merge remote-tracking branch 'upstream/master' into feature/configure…

…-sensor-namespace

metaflow/plugins/argo/argo_client.py

@@ -169,15 +169,15 @@ def delete_workflow_template(self, name):
                     json.loads(e.body)["message"] if e.body is not None else e.reason
                 )
 
-    def terminate_workflow(self, run_id):
+    def terminate_workflow(self, name):
         client = self._client.get()
         try:
             workflow = client.CustomObjectsApi().get_namespaced_custom_object(
                 group=self._group,
                 version=self._version,
                 namespace=self._namespace,
                 plural="workflows",
-                name=run_id,
+                name=name,
             )
         except client.rest.ApiException as e:
             raise ArgoClientException(
@@ -199,7 +199,7 @@ def terminate_workflow(self, run_id):
                 version=self._version,
                 namespace=self._namespace,
                 plural="workflows",
-                name=run_id,
+                name=name,
                 body=body,
             )
         except client.rest.ApiException as e:

metaflow/plugins/argo/argo_events.py

@@ -8,8 +8,8 @@
 
 from metaflow.exception import MetaflowException
 from metaflow.metaflow_config import (
-    ARGO_EVENTS_WEBHOOK_URL,
     ARGO_EVENTS_WEBHOOK_AUTH,
+    ARGO_EVENTS_WEBHOOK_URL,
     SERVICE_HEADERS,
 )
 

metaflow/plugins/argo/argo_workflows.py

@@ -7,20 +7,21 @@
 from collections import defaultdict
 from hashlib import sha1
 
-from metaflow import current, JSONType
-from metaflow.includefile import FilePathClass
+from metaflow import JSONType, current
 from metaflow.decorators import flow_decorators
 from metaflow.exception import MetaflowException
+from metaflow.includefile import FilePathClass
 from metaflow.metaflow_config import (
     ARGO_EVENTS_EVENT,
     ARGO_EVENTS_EVENT_BUS,
     ARGO_EVENTS_EVENT_SOURCE,
-    ARGO_EVENTS_SERVICE_ACCOUNT,
-    ARGO_EVENTS_SENSOR_NAMESPACE,
     ARGO_EVENTS_INTERNAL_WEBHOOK_URL,
+    ARGO_EVENTS_SENSOR_NAMESPACE,
+    ARGO_EVENTS_SERVICE_ACCOUNT,
+    ARGO_EVENTS_WEBHOOK_AUTH,
     ARGO_WORKFLOWS_ENV_VARS_TO_SKIP,
     ARGO_WORKFLOWS_KUBERNETES_SECRETS,
-    ARGO_EVENTS_WEBHOOK_AUTH,
+    ARGO_WORKFLOWS_UI_URL,
     AWS_SECRETS_MANAGER_DEFAULT_REGION,
     AZURE_STORAGE_BLOB_SERVICE_ENDPOINT,
     CARD_AZUREROOT,
@@ -39,15 +40,12 @@
     KUBERNETES_SANDBOX_INIT_SCRIPT,
     KUBERNETES_SECRETS,
     S3_ENDPOINT_URL,
+    S3_SERVER_SIDE_ENCRYPTION,
     SERVICE_HEADERS,
     SERVICE_INTERNAL_URL,
-    S3_SERVER_SIDE_ENCRYPTION,
     UI_URL,
-    ARGO_WORKFLOWS_UI_URL,
 )
-
 from metaflow.metaflow_config_funcs import config_values
-
 from metaflow.mflog import BASH_SAVE_LOGS, bash_capture_logs, export_mflog_env_vars
 from metaflow.parameters import deploy_time_eval
 from metaflow.plugins.kubernetes.kubernetes import (
@@ -292,6 +290,22 @@ def terminate(flow_name, name):
                 )
             )
 
+    @staticmethod
+    def get_workflow_status(flow_name, name):
+        client = ArgoClient(namespace=KUBERNETES_NAMESPACE)
+        # TODO: Only look for workflows for the specified flow
+        workflow = client.get_workflow(name)
+        if workflow:
+            # return workflow phase for now
+            status = workflow.get("status", {}).get("phase")
+            return status
+        else:
+            raise ArgoWorkflowsException(
+                "No execution found for {flow_name}/{run_id} in Argo Workflows.".format(
+                    flow_name=flow_name, run_id=name
+                )
+            )
+
     @staticmethod
     def suspend(name):
         client = ArgoClient(namespace=KUBERNETES_NAMESPACE)

metaflow/plugins/argo/argo_workflows_cli.py

@@ -10,10 +10,10 @@
 from metaflow._vendor import click
 from metaflow.exception import MetaflowException, MetaflowInternalError
 from metaflow.metaflow_config import (
-    SERVICE_VERSION_CHECK,
-    UI_URL,
     ARGO_WORKFLOWS_UI_URL,
     KUBERNETES_NAMESPACE,
+    SERVICE_VERSION_CHECK,
+    UI_URL,
 )
 from metaflow.package import MetaflowPackage
 
@@ -788,6 +788,27 @@ def validate_token(name, token_prefix, authorize, instructions_fn=None):
     return True
 
 
+@argo_workflows.command(help="Fetch flow execution status on Argo Workflows.")
+@click.argument("run-id", required=True, type=str)
+@click.pass_obj
+def status(obj, run_id):
+    if not run_id.startswith("argo-"):
+        raise RunIdMismatch(
+            "Run IDs for flows executed through Argo Workflows begin with 'argo-'"
+        )
+    obj.echo(
+        "Fetching status for run *{run_id}* for {flow_name} ...".format(
+            run_id=run_id, flow_name=obj.flow.name
+        ),
+        bold=True,
+    )
+    # Trim prefix from run_id
+    name = run_id[5:]
+    status = ArgoWorkflows.get_workflow_status(obj.flow.name, name)
+    if status is not None:
+        obj.echo_always(status)
+
+
 @argo_workflows.command(help="Terminate flow execution on Argo Workflows.")
 @click.option(
     "--authorize",

metaflow/plugins/cards/ui/package.json

@@ -28,7 +28,7 @@
     "cypress-svelte-unit-test": "^3.3.4",
     "eslint": "^8.44.0",
     "eslint-plugin-svelte3": "^3.4.0",
-    "postcss": "^8.4.24",
+    "postcss": "^8.4.31",
     "rollup": "^2.63.0",
     "rollup-plugin-css-only": "^3.1.0",
     "rollup-plugin-livereload": "^2.0.5",

metaflow/plugins/cards/ui/yarn.lock

@@ -2541,10 +2541,10 @@ postcss-value-parser@^4.1.0, postcss-value-parser@^4.2.0:
   resolved "https://registry.yarnpkg.com/postcss-value-parser/-/postcss-value-parser-4.2.0.tgz#723c09920836ba6d3e5af019f92bc0971c02e514"
   integrity sha512-1NNCs6uurfkVbeXG4S8JFT9t19m45ICnif8zWLd5oPSZ50QnwMfK+H3jv408d4jw/7Bttv5axS5IiHoLaVNHeQ==
 
-postcss@^8.4.24:
-  version "8.4.24"
-  resolved "https://registry.yarnpkg.com/postcss/-/postcss-8.4.24.tgz#f714dba9b2284be3cc07dbd2fc57ee4dc972d2df"
-  integrity sha512-M0RzbcI0sO/XJNucsGjvWU9ERWxb/ytp1w6dKtxTKgixdtQDq4rmx/g8W1hnaheq9jgwL/oyEdH5Bc4WwJKMqg==
+postcss@^8.4.31:
+  version "8.4.31"
+  resolved "https://registry.yarnpkg.com/postcss/-/postcss-8.4.31.tgz#92b451050a9f914da6755af352bdc0192508656d"
+  integrity sha512-PS08Iboia9mts/2ygV3eLpY5ghnUcfLV/EXTOW1E2qYxJKGGBUtNjN76FYHnMs36RmARn41bC0AZmn+rR0OVpQ==
   dependencies:
     nanoid "^3.3.6"
     picocolors "^1.0.0"

metaflow/plugins/pypi/bootstrap.py

@@ -90,7 +90,7 @@
         f"""set -e;
         if ! command -v micromamba >/dev/null 2>&1; then
             mkdir micromamba;
-            wget -qO- https://micro.mamba.pm/api/micromamba/{architecture}/latest | python -c "import sys, bz2; sys.stdout.buffer.write(bz2.decompress(sys.stdin.buffer.read()))" | tar -xv -C $(pwd)/micromamba bin/micromamba --strip-components 1;
+            python -c "import requests, bz2, sys; data = requests.get('https://micro.mamba.pm/api/micromamba/{architecture}/latest').content; sys.stdout.buffer.write(bz2.decompress(data))" | tar -xv -C $(pwd)/micromamba bin/micromamba --strip-components 1;
             export PATH=$PATH:$(pwd)/micromamba;
             if ! command -v micromamba >/dev/null 2>&1; then
                 echo "Failed to install Micromamba!";

metaflow/plugins/pypi/conda_decorator.py

@@ -122,7 +122,7 @@ def step_init(self, flow, graph, step, decos, environment, flow_datastore, logge
 
         environment.set_local_root(LocalStorage.get_datastore_root_from_config(logger))
 
-        self.disabled = not self.environment.get_environment(
+        self.disabled = self.environment.is_disabled(
             next(step for step in self.flow if step.name == self.step)
         )
 

metaflow/plugins/pypi/conda_environment.py

@@ -181,6 +181,15 @@ def interpreter(self, step_name):
         # User workloads are executed through the conda environment's interpreter.
         return self.solvers["conda"].interpreter(id_)
 
+    def is_disabled(self, step):
+        for decorator in step.decorators:
+            # @conda decorator is guaranteed to exist thanks to self.decospecs
+            if decorator.name in ["conda", "pypi"]:
+                # handle @conda/@pypi(disabled=True)
+                disabled = decorator.attributes["disabled"]
+                return disabled or str(disabled).lower() != "false"
+        return False
+
     @functools.lru_cache(maxsize=None)
     def get_environment(self, step):
         environment = {}

setup.py

@@ -1,6 +1,6 @@
 from setuptools import setup, find_packages
 
-version = "2.10.2"
+version = "2.10.3"
 
 setup(
     include_package_data=True,