Skip to content

feat(telemetry): Add bytes-scanned metrics for query jobs. #2380

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rishikeshdevsot wants to merge 4 commits into
base: main
Choose a base branch
from
+38 −2
Open

feat(telemetry): Add bytes-scanned metrics for query jobs. #2380

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a2a2e97
draft changes
rishikeshdevsot Jul 15, 2026
5ff99e8
Merge remote-tracking branch 'origin/main' into search-metric
rishikeshdevsot Jul 15, 2026
690eda9
added counters
rishikeshdevsot Jul 17, 2026
4c066fe
Merge branch 'main' into search-metric
jackluo923 Jul 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 32 additions & 2 deletions components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,26 @@ 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 archive data selected for query jobs",
)
compressed_bytes_scanned_histogram = meter.create_histogram(
"clp.query.compressed_bytes_scanned",
unit="By",
description="Compressed archive data selected for query jobs",
)
uncompressed_bytes_scanned_counter = meter.create_counter(
"clp.query.uncompressed_bytes_scanned_total",
unit="By",
description="Cumulative uncompressed bytes of archive data selected for query jobs",
)
compressed_bytes_scanned_counter = meter.create_counter(
"clp.query.compressed_bytes_scanned_total",
unit="By",
description="Cumulative compressed (on-disk) bytes of archive data selected for query jobs",
)


class DispatchExecutor:
Expand Down Expand Up @@ -539,7 +559,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:
Expand Down Expand Up @@ -572,7 +595,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"

Expand Down Expand Up @@ -1016,6 +1040,10 @@ 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)
uncompressed_bytes_scanned_counter.add(job.uncompressed_bytes_scanned)
compressed_bytes_scanned_counter.add(job.compressed_bytes_scanned)
if new_job_status == QueryJobStatus.SUCCEEDED:
logger.info(f"Completed job {job_id}.")
elif reducer_failed:
Expand Down Expand Up @@ -1443,6 +1471,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),
)
Comment on lines +1510 to 1512

@coderabbitai coderabbitai Bot Jul 17, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win

Safeguard against None values when summing database results.

If uncompressed_size or size can be NULL in the database schema, a["uncompressed_size"] will evaluate to None, causing sum() to raise a TypeError. Consider defaulting to 0 to safely handle potential NULL values.

🛡️ Proposed fix 
-        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),
+        uncompressed_bytes_scanned=sum((a["uncompressed_size"] or 0) for a in archives_for_search),
+        compressed_bytes_scanned=sum((a["compressed_size"] or 0) for a in archives_for_search),
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
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),
)
uncompressed_bytes_scanned=sum((a["uncompressed_size"] or 0) for a in archives_for_search),
compressed_bytes_scanned=sum((a["compressed_size"] or 0) for a in archives_for_search),
)
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In
`@components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py`
around lines 1474 - 1476, Update the archive byte totals in the query scheduler
result construction to treat NULL database values as zero before summing.
Specifically, adjust the uncompressed_size and compressed_size lookups in the
sum expressions for uncompressed_bytes_scanned and compressed_bytes_scanned,
preserving the existing aggregation for non-NULL values.


if search_config.aggregation_config is not None:
Expand Down
2 changes: 2 additions & 0 deletions components/job-orchestration/job_orchestration/scheduler/scheduler_data.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
4 changes: 4 additions & 0 deletions docs/src/user-docs/reference-telemetry.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@ Emitted by long-running CLP services to track throughput:
| log-ingestor | `clp.ingest.total_num_objects` | Counter | Total objects (log events) ingested |
| query-scheduler | `clp.query.tasks.completed` | Counter | Number of completed query tasks |
| query-scheduler | `clp.query.tasks.failed` | Counter | Number of failed query tasks |
| query-scheduler | `clp.query.uncompressed_bytes_scanned_total` | Counter | Cumulative uncompressed bytes of archive data selected for query jobs |
| query-scheduler | `clp.query.compressed_bytes_scanned_total` | Counter | Cumulative compressed (on-disk) bytes of archive data selected for query jobs |

#### Operational up-down counters

Expand All @@ -55,6 +57,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

Expand Down
Loading