feat(executors): capture metrics for all executors

Author: AndreyMarkinPPC

libs/core/garf/core/api_clients.py

@@ -28,22 +28,14 @@
 import pydantic
 import requests
 import smart_open
-from garf.core import exceptions, query_editor
+from garf.core import exceptions, query_editor, telemetry
 from garf.core.telemetry import tracer
 from opentelemetry import metrics, trace
 from typing_extensions import TypeAlias, override
 
 ApiRowElement: TypeAlias = Union[int, float, str, bool, list, dict, None]
 ApiResponseRow: TypeAlias = dict[str, ApiRowElement]
 
-meter = metrics.get_meter('garf.core')
-
-api_counter = meter.create_counter(
-  'garf_api_call_total',
-  unit='1',
-  description='Counts number of requests to API',
-)
-
 
 class GarfApiResponse(pydantic.BaseModel):
   """Base class for specifying response."""
@@ -71,12 +63,12 @@ class GarfApiError(exceptions.GarfError):
   ipaddress.ip_network(n)
   for n in (
     '169.254.0.0/16',  # link-local — AWS IMDS & GCE metadata endpoint
-    '127.0.0.0/8',     # IPv4 loopback
-    '10.0.0.0/8',      # RFC-1918 private
-    '172.16.0.0/12',   # RFC-1918 private
+    '127.0.0.0/8',  # IPv4 loopback
+    '10.0.0.0/8',  # RFC-1918 private
+    '172.16.0.0/12',  # RFC-1918 private
     '192.168.0.0/16',  # RFC-1918 private
-    '::1/128',         # IPv6 loopback
-    'fe80::/10',       # IPv6 link-local
+    '::1/128',  # IPv6 loopback
+    'fe80::/10',  # IPv6 link-local
   )
 ]
 
@@ -126,7 +118,7 @@ def call_api(
     span = trace.get_current_span()
     response = self.get_response(request, **kwargs)
     span.set_attribute('num_rows_api_response', len(response.results))
-    api_counter.add(1, {'api.client.class': self.__class__.__name__})
+    telemetry.api_counter.add(1, {'api.client.class': self.__class__.__name__})
     return response
 
   @abc.abstractmethod
@@ -283,4 +275,4 @@ def _field_converter(field: str):
     return int(field)
   with contextlib.suppress(ValueError):
     return float(field)
-  return field
+  return field

libs/core/garf/core/telemetry.py

@@ -13,8 +13,20 @@
 # limitations under the License.
 
 # pylint: disable=C0330, g-bad-import-order, g-multiple-import
-from opentelemetry import trace
+from opentelemetry import metrics, trace
 
 tracer = trace.get_tracer(
   instrumenting_module_name='garf.core',
 )
+meter = metrics.get_meter('garf.core')
+
+api_counter = meter.create_counter(
+  'garf_api_call_total',
+  unit='1',
+  description='Counts number of requests to API',
+)
+
+
+cache_size_meter = meter.create_gauge(
+  'garf_cache_size_bytes', unit='By', description='Size of garf cache in bytes'
+)

libs/executors/garf/executors/api_executor.py

@@ -33,16 +33,9 @@
 )
 from garf.executors.telemetry import tracer
 from garf.io.writers import abs_writer
-from opentelemetry import metrics, trace
+from opentelemetry import trace
 
 logger = logging.getLogger(__name__)
-meter = metrics.get_meter('garf.executors')
-
-api_counter = meter.create_counter(
-  'garf_api_execute_total',
-  unit='1',
-  description='Counts number of API executions',
-)
 
 
 class ApiQueryExecutor(executor.Executor):
@@ -121,16 +114,12 @@ def _execute(
     logger.debug('starting query %s', query)
     title = pathlib.Path(title).name.split('.')[0]
     try:
-      results = self.fetcher.fetch(
+      return self.fetcher.fetch(
         query_specification=query,
         args=context.query_parameters,
         title=title,
         **context.fetcher_parameters,
       )
-      api_counter.add(
-        1, {'api.client.class': self.fetcher.api_client.__class__.__name__}
-      )
-      return results
     except Exception as e:
       logger.error('%s generated an exception: %s', title, str(e))
       raise exceptions.GarfExecutorError(

libs/executors/garf/executors/executor.py

@@ -17,10 +17,11 @@
 import asyncio
 import inspect
 import logging
+import time
 from typing import Optional
 
 from garf.core import query_editor, report, report_fetcher
-from garf.executors import execution_context, query_processor
+from garf.executors import execution_context, query_processor, telemetry
 from garf.executors.telemetry import tracer
 from garf.io.writers import abs_writer
 from opentelemetry import trace
@@ -58,7 +59,9 @@ def execute(
     Returns:
       Report with data if query returns some data otherwise empty Report.
     """
+    start_time = time.perf_counter()
     span = trace.get_current_span()
+    executor_attributes = {'executor.class': self.__class__.__name__}
     query_spec = (
       query_editor.QuerySpecification(
         text=query, title=title, args=context.query_parameters
@@ -76,14 +79,25 @@ def execute(
     if self.preprocessors:
       _handle_processors(processors=self.preprocessors, context=context)
     results = self._execute(query=query_text, title=title, context=context)
+    if hasattr(self, 'fetcher'):
+      fetcher_attributes = {
+        'api.client.class': self.fetcher.api_client.__class__.__name__
+      }
+      executor_attributes.update(fetcher_attributes)
+    telemetry.executor_counter.add(1, executor_attributes)
     if (results or results.results_placeholder) and (
       self.writers or context.writer
     ):
       writer_clients = self.writers or context.writer_clients
-      return write_many(writer_clients, results, title)
+      write_outputs = write_many(writer_clients, results, title)
+      duration = time.perf_counter() - start_time
+      telemetry.executor_histogram.record(duration, executor_attributes)
+      return write_outputs
     span.set_attribute('execute.num_results', len(results))
     if self.postprocessors:
       _handle_processors(processors=self.postprocessors, context=context)
+    duration = time.perf_counter() - start_time
+    telemetry.executor_histogram.record(duration, executor_attributes)
     return results
 
   def _execute(
@@ -207,12 +221,17 @@ def write_many(
 ) -> Optional[str]:
   writing_results = []
   for writer_client in writer_clients:
+    start_time = time.perf_counter()
     logger.debug(
       'Start writing data for query %s via %s writer',
       title,
       type(writer_client),
     )
     writing_result = writer_client.write(results, title)
+    duration = time.perf_counter() - start_time
+    telemetry.write_histogram.record(
+      duration, {'writer_class': writer_client.__class__.__name__}
+    )
     logger.debug(
       'Finish writing data for query %s via %s writer',
       title,

libs/executors/garf/executors/telemetry.py

@@ -13,8 +13,26 @@
 # limitations under the License.
 
 # pylint: disable=C0330, g-bad-import-order, g-multiple-import
-from opentelemetry import trace
+from opentelemetry import metrics, trace
 
 tracer = trace.get_tracer(
   instrumenting_module_name='garf.executors',
 )
+meter = metrics.get_meter('garf.executors')
+
+executor_counter = meter.create_counter(
+  'garf_execute_total',
+  unit='1',
+  description='Counts number of executor invocations',
+)
+
+executor_histogram = meter.create_histogram(
+  'garf_execute_duration_seconds',
+  unit='s',
+  description='Measures execution duration in seconds',
+)
+write_histogram = meter.create_histogram(
+  'garf_write_duration_seconds',
+  unit='s',
+  description='Measures report writes duration in seconds',
+)