fix(executors): expose number of requested executions before workflow step

libs/executors/garf/executors/entrypoints/server.py

@@ -241,14 +241,13 @@ def execute_workflow(
   simulate: bool = False,
 ) -> dict[str, ApiExecutorResponse]:
   """Runs garf workflow till completion."""
-  telemetry.workflow_requested.add(1)
   try:
     execution_workflow = _init_workflow(
       workflow_file, workflow_path, config_file, config_path
     )
   except workflow.GarfWorkflowError as e:
-    telemetry.workflow_error_counter.add(1)
     raise fastapi.HTTPException(404, detail=str(e))
+  telemetry.workflow_requested.add(1, attributes=execution_workflow.attributes)
   result = tasks.execute_workflow(
     execution_workflow=execution_workflow.model_dump(),
     selected_aliases=selected_aliases,
@@ -276,15 +275,14 @@ async def execute_workflow_task(
   simulate: bool = False,
 ) -> dict[str, str]:
   """Creates a single operation for running garf workflow."""
-  telemetry.workflow_requested.add(1)
   span = trace.get_current_span()
   try:
     execution_workflow = _init_workflow(
       workflow_file, workflow_path, config_file, config_path
     )
   except workflow.GarfWorkflowError as e:
-    telemetry.workflow_error_counter.add(1)
     raise fastapi.HTTPException(404, detail=str(e))
+  telemetry.workflow_requested.add(1, attributes=execution_workflow.attributes)
   task = tasks.execute_workflow.delay(
     execution_workflow=execution_workflow.model_dump(),
     selected_aliases=selected_aliases,

libs/executors/garf/executors/workflows/workflow.py

@@ -295,3 +295,17 @@ def compile(self) -> None:
         else:
           new_queries.append(query.to_query(self.prefix))
       step.queries = new_queries
+
+  @property
+  def attributes(self) -> dict[str, str]:
+    workflow_attributes = {}
+    if name := self.name:
+      workflow_attributes.update({'workflow.name': name})
+    if version := self.metadata.version:
+      workflow_attributes.update({'workflow.version': version})
+    if config := self.execution_config:
+      if config_version := config.metadata.version:
+        workflow_attributes.update({'config.version': config_version})
+      if config_name := config.name:
+        workflow_attributes.update({'config.name': config_name})
+    return workflow_attributes

libs/executors/garf/executors/workflows/workflow_runner.py

@@ -74,26 +74,16 @@ def run(
   ) -> list[str]:
     span = trace.get_current_span()
     start_time = time.perf_counter()
-    workflow_attributes = {}
-    if name := self.workflow.name:
-      workflow_attributes.update({'workflow.name': name})
-    if version := self.workflow.metadata.version:
-      workflow_attributes.update({'workflow.version': version})
-    if config := self.workflow.execution_config:
-      if config_version := config.metadata.version:
-        workflow_attributes.update({'config.version': config_version})
-      if config_name := config.name:
-        workflow_attributes.update({'config.name': config_name})
+    if workflow_attributes := self.workflow.attributes:
+      span.set_attributes(workflow_attributes)
 
     steps = [step.fetcher for step in self.workflow.steps]
-    span.set_attributes(workflow_attributes)
     span.set_attributes(
       {
         'workflow.num_steps': len(steps),
         'workflow.fetchers': list(set(steps)),
       }
     )
-    span.set_attributes(workflow_attributes)
     self.workflow.compile()
     skipped_aliases = skipped_aliases or []
     selected_aliases = selected_aliases or []
@@ -170,6 +160,9 @@ def run(
         try:
           step_start_time = time.perf_counter()
           step_span.set_attribute('workflow.step.num_queries', len(batch))
+          telemetry.executor_requested_counter.add(
+            len(batch), attributes={'executor.source': step.fetcher}
+          )
           results = query_executor.execute_batch(
             batch,
             step.context,