Search metric

components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py

@@ -154,6 +154,16 @@ def _observe_outstanding_tasks(_options: metrics.CallbackOptions):
     unit="s",
     description="Duration of query tasks",
 )
+uncompressed_bytes_scanned_histogram = meter.create_histogram(
+    "clp.query.uncompressed_bytes_scanned",
+    unit="By",
+    description="Uncompressed bytes of archive data selected for query jobs",
+)
+compressed_bytes_scanned_histogram = meter.create_histogram(
+    "clp.query.compressed_bytes_scanned",
+    unit="By",
+    description="Compressed (on-disk) bytes of archive data selected for query jobs",
+)
 
 
 class DispatchExecutor:
@@ -539,7 +549,10 @@ def _get_archives_for_search_without_datasets(
         where_clause = " WHERE " + " AND ".join(filter_clauses)
 
     table = get_archives_table_name(table_prefix, None)
-    query = f"SELECT id AS archive_id, end_timestamp FROM {table}{where_clause}"
+    query = (
+        f"SELECT id AS archive_id, end_timestamp, uncompressed_size, size AS compressed_size"
+        f" FROM {table}{where_clause}"
+    )
     query += " ORDER BY end_timestamp DESC"
 
     with contextlib.closing(db_conn.cursor(dictionary=True)) as cursor:
@@ -572,7 +585,8 @@ def get_archives_for_search(
     for ds in datasets:
         table = get_archives_table_name(table_prefix, ds)
         union_parts.append(
-            f"SELECT id AS archive_id, end_timestamp, '{ds}' AS dataset FROM {table}{where_clause}"
+            f"SELECT id AS archive_id, end_timestamp, uncompressed_size, size AS compressed_size,"
+            f" '{ds}' AS dataset FROM {table}{where_clause}"
         )
     query = " UNION ALL ".join(union_parts) + " ORDER BY end_timestamp DESC"
 
@@ -1016,6 +1030,8 @@ async def handle_finished_search_job(
         duration=duration,
     ):
         job_duration_histogram.record(duration)
+        uncompressed_bytes_scanned_histogram.record(job.uncompressed_bytes_scanned)
+        compressed_bytes_scanned_histogram.record(job.compressed_bytes_scanned)
         if new_job_status == QueryJobStatus.SUCCEEDED:
             logger.info(f"Completed job {job_id}.")
         elif reducer_failed:
@@ -1443,6 +1459,8 @@ def _handle_new_search_job(
         num_archives_to_search=len(archives_for_search),
         num_archives_searched=0,
         remaining_archives_for_search=archives_for_search,
+        uncompressed_bytes_scanned=sum(a["uncompressed_size"] for a in archives_for_search),
+        compressed_bytes_scanned=sum(a["compressed_size"] for a in archives_for_search),
     )
 
     if search_config.aggregation_config is not None:

components/job-orchestration/job_orchestration/scheduler/scheduler_data.py

@@ -87,6 +87,8 @@ class SearchJob(QueryJob):
     num_archives_to_search: int
     num_archives_searched: int
     remaining_archives_for_search: list[dict[str, Any]]
+    uncompressed_bytes_scanned: int = 0
+    compressed_bytes_scanned: int = 0
     reducer_acquisition_task: asyncio.Task | None = None
     reducer_handler_msg_queues: ReducerHandlerMessageQueues | None = None
 

docs/src/user-docs/reference-telemetry.md

@@ -55,6 +55,8 @@ Emitted by long-running CLP services to track duration and rate distributions:
 | compression-worker    | `clp.compression.output_rate`   | Histogram | Rate of compressed bytes output per task                   |
 | query-scheduler       | `clp.query.job.duration`        | Histogram | Duration of query jobs                                     |
 | query-scheduler       | `clp.query.task.duration`       | Histogram | Duration of query tasks                                    |
+| query-scheduler       | `clp.query.uncompressed_bytes_scanned` | Histogram | Uncompressed bytes of archive data selected for query jobs |
+| query-scheduler       | `clp.query.compressed_bytes_scanned` | Histogram | Compressed (on-disk) bytes of archive data selected for query jobs |
 
 #### Deployment topology gauges