add delete mode to datastore which inits _attempt

metaflow/cli.py

@@ -432,7 +432,7 @@ def echo_unicode(line, **kwargs):
  help="Show both stdout and stderr of the task.",
 )
 @click.pass_obj
-def delete_logs(obj, input_path, stdout=None, stderr=None, both=None, timestamps=False):
+def delete_logs(obj, input_path, stdout=None, stderr=None, both=None):
  types = set()
  if stdout:
  types.add("stdout")
@@ -469,7 +469,7 @@ def delete_logs(obj, input_path, stdout=None, stderr=None, both=None, timestamps
  run_id=run_id,
  step_name=step_name,
  task_id=task_id,
- mode="w",
+ mode="d",
  allow_not_done=True,
  )
  ]

metaflow/datastore/task_datastore.py

@@ -173,6 +173,18 @@ def __init__(
  if data_obj is not None:
  self._objects = data_obj.get("objects", {})
  self._info = data_obj.get("info", {})
+ elif self._mode == "d":
+ self._objects = {}
+ self._info = {}
+
+ if self._attempt is None:
+ for i in range(metaflow_config.MAX_ATTEMPTS):
+ check_meta = self._metadata_name_for_attempt(
+ self.METADATA_ATTEMPT_SUFFIX, i
+ )
+ if self.has_metadata(check_meta, add_attempt=False):
+ self._attempt = i
+
  else:
  raise DataException("Unknown datastore mode: '%s'" % self._mode)
 
@@ -750,7 +762,7 @@ def save_logs(self, logsource, stream_data):
  to_store_dict[n] = data
  self._save_file(to_store_dict)
 
- @require_mode("w")
+ @require_mode("d")
  def delete_logs(self, logsources, stream, attempt_override=None):
  paths = [
  self._metadata_name_for_attempt(