add local root for @pypi

metaflow/plugins/kubernetes/kubernetes.py

@@ -698,7 +698,7 @@ def wait_for_launch(job):
  t = time.time()
  time.sleep(update_delay(time.time() - start_time))
 
- _make_prefix = lambda: b"[%s] " % util.to_bytes(self._job.id)
+ prefix = lambda: b"[%s] " % util.to_bytes(self._job.id)
 
  stdout_tail = get_log_tailer(stdout_location, self._datastore.TYPE)
  stderr_tail = get_log_tailer(stderr_location, self._datastore.TYPE)
@@ -708,7 +708,7 @@ def wait_for_launch(job):
 
  # 2) Tail logs until the job has finished
  tail_logs(
- prefix=_make_prefix(),
+ prefix=prefix(),
  stdout_tail=stdout_tail,
  stderr_tail=stderr_tail,
  echo=echo,

metaflow/plugins/kubernetes/kubernetes_jobsets.py

@@ -571,10 +571,8 @@ def dump(self):
  namespace=self._kwargs["namespace"],
  ),
  spec=client.V1PodSpec(
- ## --- jobset require podspec deets start----
  subdomain=self._kwargs["subdomain"],
  set_hostname_as_fqdn=True,
- ## --- jobset require podspec deets end ----
  # Timeout is set on the pod and not the job (important!)
  active_deadline_seconds=self._kwargs[
  "timeout_in_seconds"

metaflow/plugins/pypi/pypi_decorator.py

@@ -93,6 +93,12 @@ def step_init(self, flow, graph, step, decos, environment, flow_datastore, logge
  ),
  )
  )
+ # TODO: This code snippet can be done away with by altering the constructor of
+ # MetaflowEnvironment. A good first-task exercise.
+ # Avoid circular import
+ from metaflow.plugins.datastores.local_storage import LocalStorage
+
+ environment.set_local_root(LocalStorage.get_datastore_root_from_config(logger))
 
  def is_attribute_user_defined(self, name):
  return name in self._user_defined_attributes