feat(logging)!: add context variables to compression logs

[rishikeshdevsot]

Description

Adds scoped log context to the compression scheduler and worker so compression logs include correlation fields such as job_id and task_id.
This makes it easier to follow a compression job across scheduler logs, Celery worker logs, and subprocess log dumps.

Note

BREAKING CHANGE:
This PR changes the log text in compression scheduler and worker.

Checklist

• The PR satisfies the contribution guidelines.
• This is a breaking change and that has been indicated in the PR title, OR this isn't a
  breaking change.
• Necessary docs have been updated, OR no docs need to be updated.

Validation performed

• Compression scheduler logs:

{"event": "Dispatched job with 1 tasks (0 remaining).", "job_id": 1, "logger": "compression_scheduler", "level": "info", "timestamp": "2026-07-09T20:29:02.453743Z", "filename": "compression_scheduler.py", "func_name": "_batch_and_submit_tasks", "lineno": 761}
{"event": "Compression task completed in 0.253056 second(s).", "task_id": 1, "job_id": 1, "logger": "compression_scheduler", "level": "info", "timestamp": "2026-07-09T20:29:02.957818Z", "filename": "compression_scheduler.py", "func_name": "poll_running_jobs", "lineno": 555}
{"event": "Job succeeded (1 tasks completed).", "job_id": 1, "logger": "compression_scheduler", "level": "info", "timestamp": "2026-07-09T20:29:02.958530Z", "filename": "compression_scheduler.py", "func_name": "_complete_compression_job", "lineno": 779}
{"event": "Dispatched job with 1 tasks (0 remaining).", "job_id": 2, "logger": "compression_scheduler", "level": "info", "timestamp": "2026-07-09T20:29:49.098165Z", "filename": "compression_scheduler.py", "func_name": "_batch_and_submit_tasks", "lineno": 761}
{"event": "Compression task failed with error: See logs /var/log/compression_worker/compression-job-2-task-2-stderr.log.", "task_id": 2, "job_id": 2, "logger": "compression_scheduler", "level": "error", "timestamp": "2026-07-09T20:29:49.601406Z", "filename": "compression_scheduler.py", "func_name": "poll_running_jobs", "lineno": 565}
{"event": "Job failed. See worker logs or status_msg for details.", "job_id": 2, "logger": "compression_scheduler", "level": "error", "timestamp": "2026-07-09T20:29:49.601686Z", "filename": "compression_scheduler.py", "func_name": "_handle_failed_compression_job", "lineno": 867}
{"event": "Dispatched job with 1 tasks (0 remaining).", "job_id": 3, "logger": "compression_scheduler", "level": "info", "timestamp": "2026-07-09T20:31:07.895661Z", "filename": "compression_scheduler.py", "func_name": "_batch_and_submit_tasks", "lineno": 761}
{"event": "Compression task failed with error: See logs /var/log/compression_worker/compression-job-3-task-3-stderr.log.", "task_id": 3, "job_id": 3, "logger": "compression_scheduler", "level": "error", "timestamp": "2026-07-09T20:31:08.399750Z", "filename": "compression_scheduler.py", "func_name": "poll_running_jobs", "lineno": 565}
{"event": "Job failed. See worker logs or status_msg for details.", "job_id": 3, "logger": "compression_scheduler", "level": "error", "timestamp": "2026-07-09T20:31:08.400038Z", "filename": "compression_scheduler.py", "func_name": "_handle_failed_compression_job", "lineno": 867}

• Compression worker logs:

{"event": "COMPRESSION STARTED.", "job_id": 1, "task_id": 1, "logger": "job_orchestration.executor.compress.celery_compress", "level": "info", "timestamp": "2026-07-09T20:29:02.452609Z", "lineno": 664, "filename": "compression_task.py", "func_name": "compression_entry_point"}
{"event": "COMPRESSION COMPLETED.", "job_id": 1, "task_id": 1, "logger": "job_orchestration.executor.compress.celery_compress", "level": "info", "timestamp": "2026-07-09T20:29:02.705532Z", "lineno": 678, "filename": "compression_task.py", "func_name": "compression_entry_point"}
{"event": "COMPRESSION STARTED.", "job_id": 2, "task_id": 2, "logger": "job_orchestration.executor.compress.celery_compress", "level": "info", "timestamp": "2026-07-09T20:29:49.044147Z", "lineno": 664, "filename": "compression_task.py", "func_name": "compression_entry_point"}
{"event": "Failed to compress, return_code=1", "job_id": 2, "task_id": 2, "logger": "job_orchestration.executor.compress.celery_compress", "level": "error", "timestamp": "2026-07-09T20:29:49.177246Z", "lineno": 602, "filename": "compression_task.py", "func_name": "run_clp"}
{"event": "Contents of compression-job-2-task-2-stderr.log:\n2026-07-09T20:29:49.076+00:00 [error] Direct ingestion of unstructured log-text is not supported from input /mnt/logs/tmp/clp-package-log-verify/input.log\n2026-07-09T20:29:49.147+00:00 [error] Encountered error while parsing input.\n", "job_id": 2, "task_id": 2, "logger": "job_orchestration.executor.compress.celery_compress", "level": "error", "timestamp": "2026-07-09T20:29:49.177724Z", "lineno": 18, "filename": "utils.py", "func_name": "log_file_contents"}
{"event": "COMPRESSION COMPLETED.", "job_id": 2, "task_id": 2, "logger": "job_orchestration.executor.compress.celery_compress", "level": "info", "timestamp": "2026-07-09T20:29:49.177968Z", "lineno": 678, "filename": "compression_task.py", "func_name": "compression_entry_point"}
{"event": "COMPRESSION STARTED.", "job_id": 3, "task_id": 3, "logger": "job_orchestration.executor.compress.celery_compress", "level": "info", "timestamp": "2026-07-09T20:31:07.898735Z", "lineno": 664, "filename": "compression_task.py", "func_name": "compression_entry_point"}
{"event": "Failed to compress, return_code=1", "job_id": 3, "task_id": 3, "logger": "job_orchestration.executor.compress.celery_compress", "level": "error", "timestamp": "2026-07-09T20:31:08.047156Z", "lineno": 602, "filename": "compression_task.py", "func_name": "run_clp"}
{"event": "Contents of compression-job-3-task-3-stderr.log:\n2026-07-09T20:31:07.915+00:00 [error] Direct ingestion of unstructured log-text is not supported from input /mnt/logs/tmp/clp-package-log-verify/input.log\n2026-07-09T20:31:08.017+00:00 [error] Encountered error while parsing input.\n", "job_id": 3, "task_id": 3, "logger": "job_orchestration.executor.compress.celery_compress", "level": "error", "timestamp": "2026-07-09T20:31:08.047560Z", "lineno": 18, "filename": "utils.py", "func_name": "log_file_contents"}
{"event": "COMPRESSION COMPLETED.", "job_id": 3, "task_id": 3, "logger": "job_orchestration.executor.compress.celery_compress", "level": "info", "timestamp": "2026-07-09T20:31:08.047773Z", "lineno": 678, "filename": "compression_task.py", "func_name": "compression_entry_point"}

Summary by CodeRabbit

Summary by CodeRabbit

• Improvements
  • Upgraded compression and scheduling logs to consistently include job/task context for easier tracing.
  • Standardized lifecycle messaging to clearly mark compression start and completion.
  • Improved exception logging around compression and scheduling by capturing errors with consistent contextual information, including soft time limit and unexpected failures, without changing outcomes.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro

Run ID: 643dde42-0ac6-4dcb-91e9-df776d806ff3

📥 Commits

Reviewing files that changed from the base of the PR and between 8e7a960 and a3d59b9.

📒 Files selected for processing (2)

• components/job-orchestration/job_orchestration/executor/compress/celery_compress.py
• components/job-orchestration/job_orchestration/executor/compress/spider_compress.py

---

Walkthrough

Compression scheduler and executor paths now bind job and task identifiers with structlog context variables. Related log messages no longer interpolate identifiers directly, while compression, scheduling, status updates, metrics, and dispatch behaviour remain unchanged.

Changes

Compression logging context

Layer / File(s)	Summary
Scheduler context binding components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py	_schedule_job and poll_running_jobs bind job/task identifiers through bound_contextvars; scheduler lifecycle messages use parameterised logging without explicit job identifiers.
Executor context binding components/job-orchestration/job_orchestration/executor/compress/celery_compress.py, components/job-orchestration/job_orchestration/executor/compress/spider_compress.py, components/job-orchestration/job_orchestration/executor/compress/compression_task.py	Celery and spider execution bind job/task identifiers, add exception logging, and use static compression start/completion messages.

Estimated code review effort: 3 (Moderate) | ~20 minutes

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 77.78% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately summarizes the main change: adding context variables to compression logging.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[jackluo923]

bound_contextvars only attaches job_id and task_id while execution remains inside this
with block. The soft-timeout path is fine because it logs inside the block. For any other
exception from compression_entry_point—for example, configuration validation, run_clp, or a
database update—Python first exits the with block and removes these values. Celery then catches
the exception and emits its task-failure log without the CLP job and task IDs.

The sequence is:

1. Bind job_id and task_id.
2. compression_entry_point raises an unexpected exception.
3. Execution leaves the with block, clearing the context.
4. Celery logs the failure without the correlation fields.

The Spider wrapper has equivalent behavior. _schedule_job has the same issue for exceptions not
caught locally: its bound scope ends before main logs Error in scheduling., so that log lacks
job_id.

Please either log a general exception inside each bound scope before re-raising it:

except Exception:
    logger.exception("Compression task failed unexpectedly.")
    raise

Or bind and clear the context through the worker/scheduler lifecycle so framework failure logging
still sees it.

Normal and explicitly handled paths are correct; this only affects unexpected failures, where
correlation is most valuable.