allow tagging of metadata and add attempt status to metadata table (#371)

* allow tagging of metadata and add attempt status to metadata table

* update name, value, and type of metadata

Co-authored-by: Ferras Hamad <[email protected]>

Author: ferras

metaflow/datastore/local.py

@@ -229,7 +229,7 @@ def done(self):
                 raise
         self.metadata.register_metadata(
             self.run_id, self.step_name, self.task_id,
-            [MetaDatum(field='attempt-done', value=str(self.attempt), type='attempt-done')])
+            [MetaDatum(field='attempt-done', value=str(self.attempt), type='attempt-done', tags=[])])
 
         self._is_done_set = True
 

metaflow/datastore/s3.py

@@ -254,7 +254,7 @@ def done(self):
 
         self.metadata.register_metadata(
             self.run_id, self.step_name, self.task_id,
-            [MetaDatum(field='attempt-done', value=str(self.attempt), type='attempt-done')])
+            [MetaDatum(field='attempt-done', value=str(self.attempt), type='attempt-done', tags=[])])
 
         self._is_done_set = True
 

metaflow/metadata/metadata.py

@@ -12,7 +12,7 @@
                           'name ds_type url type sha')
 
 MetaDatum = namedtuple('MetaDatum',
-                       'field value type')
+                       'field value type tags')
 
 
 class MetadataProviderMeta(type):
@@ -430,6 +430,7 @@ def _metadata_to_json(self, run_id, step_name, task_id, metadata):
             'field_name': datum.field,
             'type': datum.type,
             'value': datum.value,
+            'tags': datum.tags,
             'user_name': user,
             'ts_epoch': int(round(time.time() * 1000))} for datum in metadata]
 
@@ -457,7 +458,8 @@ def _register_code_package_metadata(self, run_id, step_name, task_id):
             metadata.append(MetaDatum(
                 field='code-package',
                 value=json.dumps({'ds_type': code_ds, 'sha': code_sha, 'location': code_url}),
-                type='code-package'))
+                type='code-package',
+                tags=[]))
         if metadata:
             self.register_metadata(run_id, step_name, task_id, metadata)
 

metaflow/plugins/aws/batch/batch_decorator.py

@@ -182,7 +182,7 @@ def task_pre_step(self,
         meta['aws-batch-job-attempt'] = os.environ['AWS_BATCH_JOB_ATTEMPT']
         meta['aws-batch-ce-name'] = os.environ['AWS_BATCH_CE_NAME']
         meta['aws-batch-jq-name'] = os.environ['AWS_BATCH_JQ_NAME']    
-        entries = [MetaDatum(field=k, value=v, type=k) for k, v in meta.items()]
+        entries = [MetaDatum(field=k, value=v, type=k, tags=[]) for k, v in meta.items()]
         # Register book-keeping metadata for debugging.
         metadata.register_metadata(run_id, step_name, task_id, entries)
 

metaflow/plugins/aws/step_functions/step_functions_decorator.py

@@ -24,7 +24,7 @@ def task_pre_step(self,
         meta['aws-step-functions-execution'] = os.environ['METAFLOW_RUN_ID']
         meta['aws-step-functions-state-machine'] =\
                                         os.environ['SFN_STATE_MACHINE']
-        entries = [MetaDatum(field=k, value=v, type=k) for k, v in meta.items()]
+        entries = [MetaDatum(field=k, value=v, type=k, tags=[]) for k, v in meta.items()]
         # Register book-keeping metadata for debugging.
         metadata.register_metadata(run_id, step_name, task_id, entries)
 

metaflow/plugins/conda/conda_step_decorator.py

@@ -224,7 +224,8 @@ def task_pre_step(
         meta.register_metadata(run_id, step_name, task_id,
                                    [MetaDatum(field='conda_env_id',
                                               value=self._env_id(),
-                                              type='conda_env_id')])
+                                              type='conda_env_id',
+                                              tags=[])])
 
     def runtime_step_cli(self, cli_args, retry_count, max_user_code_retries):
         if self.is_enabled() and 'batch' not in cli_args.commands:

metaflow/runtime.py

@@ -642,7 +642,8 @@ def save_logs(self, logtype, logs):
                                'ds_type': self._ds.TYPE,
                                'location': location,
                                'attempt': self.retries}),
-                           type='log_path')]
+                           type='log_path',
+                           tags=[])]
         self.metadata.register_metadata(self.run_id,
                                         self.step,
                                         self.task_id,

metaflow/task.py

@@ -257,10 +257,12 @@ def run_step(self,
                                         task_id,
                                         [MetaDatum(field='attempt',
                                                    value=str(retry_count),
-                                                   type='attempt'),
+                                                   type='attempt',
+                                                   tags=[]),
                                          MetaDatum(field='origin-run-id',
                                                    value=str(origin_run_id),
-                                                   type='origin-run-id')])
+                                                   type='origin-run-id',
+                                                   tags=[])])
 
         step_func = getattr(self.flow, step_name)
         node = self.flow._graph[step_name]
@@ -447,6 +449,18 @@ def run_step(self,
             }
             logger.log(msg)
 
+            attempt_ok = str(bool(int(self.flow._task_ok)))
+            self.metadata.register_metadata(run_id,
+                                            step_name,
+                                            task_id,
+                                            [MetaDatum(field='attempt_ok',
+                                                       value=attempt_ok,
+                                                       type='internal_attempt_status',
+                                                       tags=["attempt_id:{0}".
+                                                       format(str(retry_count))
+                                                             ])
+                                             ])
+
             output.save_metadata('task_end', {})
             output.persist(self.flow)