feat(logging)!: Add context variables to query logs.#2341
feat(logging)!: Add context variables to query logs.#2341rishikeshdevsot wants to merge 3 commits into
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (4)
WalkthroughQuery task execution and scheduling now use structured context variables for identifiers and query metadata. Extraction task dispatch passes the query job type, while task, completion, failure, cancellation, validation, and stream-state log messages use simplified text. ChangesQuery task context and execution
Estimated code review effort: 4 (Complex) | ~45 minutes Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
rishikeshdevsot
changed the title
rishikeshdevsot
changed the title
rishikeshdevsot
force-pushed
the
pr-query-log-correlation
branch
from
a61bedd to
9ba21a9
Compare
rishikeshdevsot
changed the title
rishikeshdevsot
force-pushed
the
pr-query-log-correlation
branch
from
9ba21a9 to
edd5c7e
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with 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.
Inline comments:
In
`@components/job-orchestration/job_orchestration/executor/query/fs_search_task.py`:
- Around line 58-60: Deserialize the job configuration only once in the search
wrapper, then pass the resulting SearchJobConfig to both the context setup logic
and search_entry_point. Update the relevant function signatures and callers so
search_entry_point no longer invokes msgpack.unpackb or
SearchJobConfig.model_validate, while preserving context["query"] and
context["query_hash"] population from the shared config.
- Around line 340-344: Move the potentially failing _get_search_task_log_context
call inside the task’s existing try/except structure so malformed job
configuration errors are handled by the SoftTimeLimitExceeded/Exception
handlers. Bind only a minimal job_id/task_id context before the try, then enrich
the logging context after successful context construction while preserving the
existing handlers and cross-log fields.
In
`@components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py`:
- Around line 1152-1153: In the exception handling around
try_getting_task_result(), replace logger.error(f"Job failed: {e}.") with
logger.exception using the same failure context so unexpected failures retain
their traceback; keep the TimeoutError branch separate and unchanged because it
already logs upstream.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 33764b30-ba7d-4066-98ef-badb8b89c8ee
📒 Files selected for processing (4)
components/job-orchestration/job_orchestration/executor/query/extract_stream_task.pycomponents/job-orchestration/job_orchestration/executor/query/fs_search_task.pycomponents/job-orchestration/job_orchestration/executor/query/utils.pycomponents/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py
| search_config = SearchJobConfig.model_validate(msgpack.unpackb(job_config_blob)) | ||
| context["query"] = search_config.query_string | ||
| context["query_hash"] = get_query_hash(search_config.query_string) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Job config blob is deserialized twice per task.
SearchJobConfig.model_validate(msgpack.unpackb(job_config_blob)) is called here and again in search_entry_point at line 283. Consider deserializing once in the search wrapper and passing the SearchJobConfig to both the context helper and the entry point, eliminating the duplicate parse.
🤖 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/executor/query/fs_search_task.py`
around lines 58 - 60, Deserialize the job configuration only once in the search
wrapper, then pass the resulting SearchJobConfig to both the context setup logic
and search_entry_point. Update the relevant function signatures and callers so
search_entry_point no longer invokes msgpack.unpackb or
SearchJobConfig.model_validate, while preserving context["query"] and
context["query_hash"] population from the shared config.
| with bound_contextvars( | ||
| **_get_search_task_log_context( | ||
| job_id, task_id, job_config_blob, archive_id, dataset | ||
| ) | ||
| except SoftTimeLimitExceeded: | ||
| logger.exception(f"Search task job_id={job_id} task_id={task_id} exceeded soft time limit.") | ||
| raise | ||
| ): |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Context setup exceptions bypass the task's error handlers.
_get_search_task_log_context deserializes and validates the job config blob (msgpack.unpackb + SearchJobConfig.model_validate), which can raise on a corrupted or malformed blob. Because this call is in the with expression — evaluated before the try block — any exception propagates uncaught by the SoftTimeLimitExceeded/Exception handlers below. The failure won't be logged with job_id/task_id context, undermining the PR's cross-log tracking goal.
Consider binding a minimal context first, then enriching inside the try:
🛡️ Proposed fix
def search(
self: Task,
job_id: str,
task_id: int,
job_config_blob: bytes,
archive_id: str,
clp_metadata_db_conn_params: dict,
results_cache_uri: str,
dataset: str | None = None,
) -> dict[str, Any]:
- with bound_contextvars(
- **_get_search_task_log_context(
- job_id, task_id, job_config_blob, archive_id, dataset
- )
- ):
- try:
+ with bound_contextvars(job_id=job_id, task_id=task_id):
+ try:
+ with bound_contextvars(
+ **_get_search_task_log_context(
+ job_id, task_id, job_config_blob, archive_id, dataset
+ )
+ ):
return search_entry_point(
job_id,
task_id,
job_config_blob,
archive_id,
clp_metadata_db_conn_params,
results_cache_uri,
dataset,
)
except SoftTimeLimitExceeded:
logger.exception("Search task exceeded soft time limit.")
raise
except Exception:
logger.exception("Search task failed with an unexpected exception.")
raise📝 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.
| with bound_contextvars( | |
| **_get_search_task_log_context( | |
| job_id, task_id, job_config_blob, archive_id, dataset | |
| ) | |
| except SoftTimeLimitExceeded: | |
| logger.exception(f"Search task job_id={job_id} task_id={task_id} exceeded soft time limit.") | |
| raise | |
| ): | |
| with bound_contextvars(job_id=job_id, task_id=task_id): | |
| try: | |
| with bound_contextvars( | |
| **_get_search_task_log_context( | |
| job_id, task_id, job_config_blob, archive_id, dataset | |
| ) | |
| ): | |
| return search_entry_point( | |
| job_id, | |
| task_id, | |
| job_config_blob, | |
| archive_id, | |
| clp_metadata_db_conn_params, | |
| results_cache_uri, | |
| dataset, | |
| ) | |
| except SoftTimeLimitExceeded: | |
| logger.exception("Search task exceeded soft time limit.") | |
| raise | |
| except Exception: | |
| logger.exception("Search task failed with an unexpected exception.") | |
| raise |
🤖 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/executor/query/fs_search_task.py`
around lines 340 - 344, Move the potentially failing
_get_search_task_log_context call inside the task’s existing try/except
structure so malformed job configuration errors are handled by the
SoftTimeLimitExceeded/Exception handlers. Bind only a minimal job_id/task_id
context before the try, then enrich the logging context after successful context
construction while preserving the existing handlers and cross-log fields.
| except Exception as e: | ||
| logger.error(f"Job failed: {e}.") |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
FILE="components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py"
echo "== around cited lines =="
sed -n '1120,1188p' "$FILE" | cat -n
echo
echo "== locate try_getting_task_result =="
rg -n "try_getting_task_result|TimeoutError|logger\.exception|logger\.error" "$FILE"Repository: y-scope/clp
Length of output: 6066
🏁 Script executed:
#!/bin/bash
set -euo pipefail
FILE="components/job-orchestration/job_orchestration/scheduler/query/query_scheduler.py"
sed -n '938,1015p' "$FILE" | cat -nRepository: y-scope/clp
Length of output: 3738
Preserve the traceback for unexpected job failures. logger.error(f"Job failed: {e}.") drops the stack trace for exceptions bubbling out of try_getting_task_result(). Switch this branch to logger.exception(...), but keep TimeoutError separate since that path already logs a traceback upstream.
🧰 Tools
🪛 Ruff (0.15.20)
[warning] 1152-1152: Do not catch blind exception: Exception
(BLE001)
[warning] 1153-1153: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
[warning] 1153-1153: Logging statement uses f-string
(G004)
🤖 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 1152 - 1153, In the exception handling around
try_getting_task_result(), replace logger.error(f"Job failed: {e}.") with
logger.exception using the same failure context so unexpected failures retain
their traceback; keep the TimeoutError branch separate and unchanged because it
already logs upstream.
1 participant
job_id,task_id,query_job_type,query,query_hash,archive_idanddatasetwhere available.Note
BREAKING CHANGE:
This PR changes the log text in query scheduler and worker.
Description
Checklist
breaking change.
Validation performed
'MapTask metrics system started'Summary by CodeRabbit