Emit project related tags from within project decorator (#695)

* Emit project related tags from within project decorator

* update test

metaflow/cli.py

@@ -870,19 +870,6 @@ def start(ctx,
                            if e.TYPE == environment][0](ctx.obj.flow)
     ctx.obj.environment.validate_environment(echo)
 
-    ctx.obj.datastore = DATASTORES[datastore]
-    ctx.obj.datastore_root = datastore_root
-
-    # It is important to initialize flow decorators early as some of the
-    # things they provide may be used by some of the objects initialize after.
-    decorators._init_flow_decorators(ctx.obj.flow,
-                                     ctx.obj.graph,
-                                     ctx.obj.environment,
-                                     ctx.obj.datastore,
-                                     ctx.obj.logger,
-                                     echo,
-                                     deco_options)
-
     ctx.obj.monitor = Monitor(monitor, ctx.obj.environment, ctx.obj.flow.name)
     ctx.obj.monitor.start()
 
@@ -898,6 +885,17 @@ def start(ctx,
           ctx.obj.datastore.get_datastore_root_from_config(ctx.obj.echo)
     ctx.obj.datastore_root = ctx.obj.datastore.datastore_root = datastore_root
 
+    # It is important to initialize flow decorators early as some of the
+    # things they provide may be used by some of the objects initialize after.
+    decorators._init_flow_decorators(ctx.obj.flow,
+                                     ctx.obj.graph,
+                                     ctx.obj.environment,
+                                     ctx.obj.datastore,
+                                     ctx.obj.metadata,
+                                     ctx.obj.logger,
+                                     echo,
+                                     deco_options)
+
     if decospecs:
         decorators._attach_decorators(ctx.obj.flow, decospecs)
 

metaflow/decorators.py

@@ -135,7 +135,15 @@ def __init__(self, *args, **kwargs):
         self._flow_decorators.append(self)
         super(FlowDecorator, self).__init__(*args, **kwargs)
 
-    def flow_init(self, flow, graph,  environment, datastore, logger, echo, options):
+    def flow_init(self,
+                  flow,
+                  graph,
+                  environment,
+                  datastore,
+                  metadata,
+                  logger,
+                  echo,
+                  options):
         """
         Called when all decorators have been created for this flow.
         """
@@ -431,10 +439,18 @@ def _attach_decorators_to_step(step, decospecs):
             deco = decos[deconame]._parse_decorator_spec(decospec)
             step.decorators.append(deco)
 
-def _init_flow_decorators(flow, graph, environment, datastore, logger, echo, deco_options):
+def _init_flow_decorators(flow,
+                          graph,
+                          environment,
+                          datastore,
+                          metadata,
+                          logger,
+                          echo,
+                          deco_options):
     for deco in flow._flow_decorators.values():
         opts = {option: deco_options[option] for option in deco.options}
-        deco.flow_init(flow, graph, environment, datastore, logger, echo, opts)
+        deco.flow_init(flow, graph, environment, 
+                       datastore, metadata, logger, echo, opts)
 
 
 def _init_step_decorators(flow, graph, environment, datastore, logger):

metaflow/metadata/metadata.py

@@ -462,9 +462,6 @@ def _tags(self):
             tags.append('metaflow_r_version:' + env['metaflow_r_version'])
         if 'r_version_code' in env:
             tags.append('r_version:' + env['r_version_code'])
-        if 'project_name' in current:
-            tags.append('project:' + current.project_name)
-            tags.append('project_branch:' + current.branch_name)
         return tags
 
     def _register_code_package_metadata(self, run_id, step_name, task_id):

metaflow/plugins/aws/step_functions/schedule_decorator.py

@@ -8,7 +8,15 @@ class ScheduleDecorator(FlowDecorator):
                 'daily': True,
                 'hourly': False}
 
-    def flow_init(self, flow, graph,  environment, datastore, logger, echo, options):
+    def flow_init(self,
+                  flow,
+                  graph,
+                  environment,
+                  datastore,
+                  metadata,
+                  logger,
+                  echo,
+                  options):
         # Currently supports quartz cron expressions in UTC as defined in
         # https://docs.aws.amazon.com/eventbridge/latest/userguide/scheduled-events.html#cron-expressions
         if self.attributes['cron']:

metaflow/plugins/conda/conda_flow_decorator.py

@@ -37,7 +37,15 @@ class MyFlow(FlowSpec):
                 'python': None,
                 'disabled': None}
 
-    def flow_init(self, flow, graph,  environment, datastore, logger, echo, options):
+    def flow_init(self,
+                  flow,
+                  graph,
+                  environment,
+                  datastore,
+                  metadata,
+                  logger,
+                  echo,
+                  options):
         if environment.TYPE != 'conda':
             raise InvalidEnvironmentException('The *@conda* decorator requires '
                                               '--environment=conda')

metaflow/plugins/project_decorator.py

@@ -35,6 +35,7 @@ def flow_init(self,
                   graph,
                   environment,
                   datastore,
+                  metadata,
                   logger,
                   echo,
                   options):
@@ -54,6 +55,9 @@ def flow_init(self,
                              'is_user_branch': is_user_branch,
                              'is_production': options['production'],
                              'project_flow_name': project_flow_name})
+        metadata.add_sticky_tags(sys_tags=[
+            'project:%s' % project_name,
+            'project_branch:%s' % branch_name])
 
     def get_top_level_options(self):
         return list(self._option_values.items())

test/core/metaflow_extensions/plugins/flow_options.py

@@ -13,5 +13,5 @@ class FlowDecoratorWithOptions(FlowDecorator):
         )
     }
 
-    def flow_init(self, flow, graph, environment, flow_datastore, logger, echo, options):
+    def flow_init(self, flow, graph, environment, flow_datastore, metadata, logger, echo, options):
         current._update_env({'foobar_value': options['foobar']})