From 7b156e980cf1efaf5dcd46d6ed7a3b069ececea7 Mon Sep 17 00:00:00 2001
From: Andrei Markin <amarkin@google.com>
Date: Fri, 17 Jul 2026 17:54:35 +0500
Subject: [PATCH] fix(executors): update workflow specific metrics

* expose number of requested executions before each workflow step
* enrich worklow attributes in garf_workflow_run_requested metric
---
 .../garf/executors/entrypoints/server.py        |  6 ++----
 .../garf/executors/workflows/workflow.py        | 14 ++++++++++++++
 .../garf/executors/workflows/workflow_runner.py | 17 +++++------------
 3 files changed, 21 insertions(+), 16 deletions(-)

diff --git a/libs/executors/garf/executors/entrypoints/server.py b/libs/executors/garf/executors/entrypoints/server.py
index 78c1242f..128c4f9f 100644
--- a/libs/executors/garf/executors/entrypoints/server.py
+++ b/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,
diff --git a/libs/executors/garf/executors/workflows/workflow.py b/libs/executors/garf/executors/workflows/workflow.py
index 10075ab7..a848a499 100644
--- a/libs/executors/garf/executors/workflows/workflow.py
+++ b/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
diff --git a/libs/executors/garf/executors/workflows/workflow_runner.py b/libs/executors/garf/executors/workflows/workflow_runner.py
index 6b6734ef..4df32320 100644
--- a/libs/executors/garf/executors/workflows/workflow_runner.py
+++ b/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,