Merge branch 'master' of https://github.com/Netflix/metaflow into fea…

…ture/enable-subclassing-flows

.github/workflows/test.yml

@@ -61,7 +61,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-20.04]
-        ver: ['4.4']
+        ver: ['4.4.1']
 
     steps:
     - uses: actions/checkout@ee0669bd1cc54295c223e0bb666b733df41de1c5 # v2.7.0

metaflow/__init__.py

@@ -101,9 +101,7 @@ class and related decorators.
 # Flow spec
 from .flowspec import FlowSpec
 
-from .parameters import Parameter, JSONTypeClass
-
-JSONType = JSONTypeClass()
+from .parameters import Parameter, JSONTypeClass, JSONType
 
 # data layer
 # For historical reasons, we make metaflow.plugins.datatools accessible as
@@ -149,6 +147,7 @@ class and related decorators.
     from .runner.metaflow_runner import Runner
     from .runner.nbrun import NBRunner
     from .runner.deployer import Deployer
+    from .runner.deployer import DeployedFlow
     from .runner.nbdeploy import NBDeployer
 
 __ext_tl_modules__ = []

metaflow/cli.py

@@ -717,6 +717,20 @@ def resume(
     if runtime.should_skip_clone_only_execution():
         return
 
+    current._update_env(
+        {
+            "run_id": runtime.run_id,
+        }
+    )
+    _system_logger.log_event(
+        level="info",
+        module="metaflow.resume",
+        name="start",
+        payload={
+            "msg": "Resuming run",
+        },
+    )
+
     with runtime.run_heartbeat():
         if clone_only:
             runtime.clone_original_run()
@@ -776,20 +790,34 @@ def run(
     write_file(run_id_file, runtime.run_id)
 
     obj.flow._set_constants(obj.graph, kwargs)
-    runtime.print_workflow_info()
-    runtime.persist_constants()
-
-    if runner_attribute_file:
-        with open(runner_attribute_file, "w", encoding="utf-8") as f:
-            json.dump(
-                {
-                    "run_id": runtime.run_id,
-                    "flow_name": obj.flow.name,
-                    "metadata": obj.metadata.metadata_str(),
-                },
-                f,
-            )
-    runtime.execute()
+    current._update_env(
+        {
+            "run_id": runtime.run_id,
+        }
+    )
+    _system_logger.log_event(
+        level="info",
+        module="metaflow.run",
+        name="start",
+        payload={
+            "msg": "Starting run",
+        },
+    )
+    with runtime.run_heartbeat():
+        runtime.print_workflow_info()
+        runtime.persist_constants()
+
+        if runner_attribute_file:
+            with open(runner_attribute_file, "w", encoding="utf-8") as f:
+                json.dump(
+                    {
+                        "run_id": runtime.run_id,
+                        "flow_name": obj.flow.name,
+                        "metadata": obj.metadata.metadata_str(),
+                    },
+                    f,
+                )
+        runtime.execute()
 
 
 def write_file(file_path, content):

metaflow/client/core.py

@@ -39,7 +39,7 @@
 from .filecache import FileCache
 
 if TYPE_CHECKING:
-    from metaflow.metadata import MetadataProvider
+    from metaflow.metadata_provider import MetadataProvider
 
 try:
     # python2
@@ -143,7 +143,7 @@ def default_metadata() -> str:
     if default:
         current_metadata = default[0]
     else:
-        from metaflow.plugins.metadata import LocalMetadataProvider
+        from metaflow.plugins.metadata_providers import LocalMetadataProvider
 
         current_metadata = LocalMetadataProvider
     return get_metadata()

metaflow/clone_util.py

@@ -1,5 +1,5 @@
 import time
-from .metadata import MetaDatum
+from .metadata_provider import MetaDatum
 
 
 def clone_task_helper(