UN-2123 [FEAT] Propagate request_id across services and workers - #2229
UN-2123 [FEAT] Propagate request_id across services and workers#2229Deepak-Kesavan wants to merge 1 commit into
Conversation
Complete end-to-end request_id (X-Request-ID) correlation so a single ID can be used to filter logs across the backend, Celery workers and services in gcloud. - backend: before_task_publish signal injects the request-scoped request_id (from StateStore) into every published Celery task's message headers, with no per-call-site changes. - workers: task_prerun now prefers an explicit request_id from the message headers over payload-derived ids; a before_task_publish handler re-propagates it onto downstream worker->worker task chains. - workers: internal-API HTTP client now forwards X-Request-ID from the worker log context so backend callbacks share the originating request's id. - x2text-service: reads/mints X-Request-ID and logs it (self-contained, no new dependency). - logging: unified all services onto one canonical log format so request_id/ trace_id/span_id parse identically in gcloud. LLMW adoption and full OpenTelemetry trace activation are intentionally out of scope (separate tickets).
Summary by CodeRabbit
WalkthroughThe changes add request ID middleware and canonical logging fields, propagate request IDs through Celery task headers, bind them in workers, and forward them in outbound worker HTTP requests. ChangesRequest ID observability
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant FlaskApp
participant StateStore
participant Celery
participant WorkerLogger
participant BaseClient
FlaskApp->>FlaskApp: capture or generate request_id
FlaskApp->>StateStore: store request_id
Celery->>StateStore: read request_id before publish
StateStore-->>Celery: return request_id
Celery->>Celery: add request_id to task headers
Celery->>WorkerLogger: deliver task message
WorkerLogger->>WorkerLogger: bind request_id from headers
WorkerLogger->>BaseClient: provide current request_id
BaseClient->>BaseClient: send X-Request-ID
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 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 |
|
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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 `@x2text-service/app/logging_util.py`:
- Around line 31-33: Update RequestIDFilter.filter to call has_request_context()
before accessing Flask’s context-local g; assign the request ID from g only when
a request context exists, otherwise retain "-".
🪄 Autofix
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: CHILL
Plan: Pro Plus
Run ID: 6da06d9c-e764-45ab-a92a-7479f521926e
📒 Files selected for processing (7)
backend/backend/celery_service.pybackend/backend/celery_signals.pyunstract/core/src/unstract/core/flask/logging.pyworkers/shared/clients/base_client.pyworkers/shared/infrastructure/logging/logger.pyx2text-service/app/config.pyx2text-service/app/logging_util.py
| def filter(self, record: logging.LogRecord) -> bool: | ||
| record.request_id = getattr(g, "request_id", "-") if g else "-" | ||
| return True |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== files =="
git ls-files | rg '(^|/)logging_util\.py$|requirements|pyproject|setup|Pipfile|poetry|gunicorn|main|wsgi|app'
echo "== logging_util.py =="
if [ -f x2text-service/app/logging_util.py ]; then
cat -n x2text-service/app/logging_util.py
fi
echo "== flask imports/usages of RequestIDFilter/Logging =="
rg -n "RequestIDFilter|LoggingMiddleware|logging_util|log_config|handlers|gunicorn|request_id|from flask import|import flask" -S .Repository: Zipstack/unstract
Length of output: 50373
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== inspect Flask proxy g semantics from installed package if available =="
python3 - <<'PY'
try:
from flask import g, has_request_context
print("flask_available=True")
print(has_request_context())
try:
_ = bool(g)
print("bool_g_outside_request_context=True")
except Exception as e:
print(f"bool_g_outside_request_context_error={type(e).__name__}: {e}")
try:
_ = getattr(g, "request_id", "-") if g else "-"
print("getattr_if_bool_g_outside_request_context=True")
except Exception as e:
print(f"getattr_if_bool_g_outside_request_context_error={type(e).__name__}: {e}")
except Exception as e:
print(f"flask_available_or_probe_error={type(e).__name__}: {e}")
PYRepository: Zipstack/unstract
Length of output: 305
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== similar core logging filter =="
cat -n unstract/core/src/unstract/core/flask/logging.py
echo "== exact matching RequestIDFilter definitions using Flask g =="
python3 - <<'PY'
from pathlib import Path
import re
patterns = [
re.compile(r"class RequestIDFilter\(logging\.Filter\):"),
re.compile(r"getattr\(\bg\b, .request_id"),
re.compile(r"\n\s+if \bg\b else \"\-\""),
]
for path in Path(".").rglob("*.py"):
text = path.read_text(errors="ignore")
if "RequestIDFilter" in text and "g, \"request_id\"" in text:
print("\n---", path, "---")
for i, m in enumerate(re.finditer(r"class RequestIDFilter\(logging\.Filter\):", text)):
print(f"definition instance {i+1} at line 1..")
line = text[:m.start()].count("\n") + 1
print(line)
start = m.start()
end = text.find("\nclass ", start + 1)
snippet = text[start:end if end != -1 else None]
print(snippet[:2000])
PYRepository: Zipstack/unstract
Length of output: 4964
🌐 Web query:
Flask g request_context proxy RuntimeError outside application context has_request_context documentation
💡 Result:
The RuntimeError: Working outside of application context or Working outside of request context in Flask occurs when you attempt to access proxy objects (like g, current_app, request, or session) without an active corresponding context [1][2][3]. The g object is a proxy to a namespace used to store data during a single request or application context [4]. It is bound to the application context, meaning it is only available when an application context is active [1][4]. To resolve these errors, you must ensure your code runs within the appropriate context: 1. Application Context: If you need to access current_app or g outside of a view function (e.g., in a CLI command, initialization script, or background task), wrap your code in a with app.app_context: block [1][2][5]. 2. Request Context: If you need to access request or session objects, you generally need an active HTTP request [3]. When testing, you can use with app.test_request_context: to simulate an environment [3][6]. 3. has_request_context: Flask provides a utility function, flask.has_request_context, which returns a boolean indicating whether a request context is currently active [4]. Similarly, flask.has_app_context can be used to check for an active application context [7]. These errors often arise when code is placed outside of a view function or is being executed in a context where Flask has not been initialized [1][2][3]. If you are attempting to interact with a Flask application from an external script, do not use Flask's request object; instead, use standard HTTP libraries like requests to send requests to the running application [8].
Citations:
- 1: https://flask.palletsprojects.com/en/latest/appcontext/
- 2: https://flask.palletsprojects.com/en/stable/appcontext/
- 3: https://flask.palletsprojects.com/en/stable/reqcontext/
- 4: https://github.com/pallets/flask/blob/main/docs/api.rst
- 5: https://sentry.io/answers/working-outside-of-application-context/
- 6: https://flask.palletsprojects.com/en/stable/api/
- 7: https://stackoverflow.com/questions/34122949/working-outside-of-application-context-flask
- 8: https://stackoverflow.com/questions/65283661/runtimeerror-working-outside-of-request-context-when-trying-to-send-post-reques
Guard Flask context access in RequestIDFilter.
Line 32 dereferences Flask’s context-local g, which can raise RuntimeError when logging runs outside an application/request context; use has_request_context() before accessing it.
Proposed fix
-from flask import Flask, g, request
+from flask import Flask, g, has_request_context, request
class RequestIDFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
- record.request_id = getattr(g, "request_id", "-") if g else "-"
+ record.request_id = (
+ getattr(g, "request_id", "-") if has_request_context() else "-"
+ )
return True📝 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.
| def filter(self, record: logging.LogRecord) -> bool: | |
| record.request_id = getattr(g, "request_id", "-") if g else "-" | |
| return True | |
| def filter(self, record: logging.LogRecord) -> bool: | |
| record.request_id = ( | |
| getattr(g, "request_id", "-") if has_request_context() else "-" | |
| ) | |
| return True |
🤖 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 `@x2text-service/app/logging_util.py` around lines 31 - 33, Update
RequestIDFilter.filter to call has_request_context() before accessing Flask’s
context-local g; assign the request ID from g only when a request context
exists, otherwise retain "-".
|
| Filename | Overview |
|---|---|
| backend/backend/celery_service.py | Imports the new signal module during Celery bootstrap so producer-side propagation is registered. |
| backend/backend/celery_signals.py | Safely injects the request-scoped ID into outgoing Celery message headers without overriding an explicit header. |
| workers/shared/infrastructure/logging/logger.py | Prefers propagated IDs during task context binding, forwards them to downstream tasks, and retains existing fallbacks and cleanup. |
| workers/shared/clients/base_client.py | Adds the current worker correlation ID to internal backend HTTP requests. |
| x2text-service/app/logging_util.py | Introduces standardized request-aware logging and middleware for x2text. |
| x2text-service/app/config.py | Initializes standardized logging and request-ID middleware during application creation. |
| unstract/core/src/unstract/core/flask/logging.py | Aligns shared Flask log output with the backend and worker correlation-field format. |
Sequence Diagram
sequenceDiagram
participant Client
participant Backend
participant Broker
participant WorkerA as Worker
participant WorkerB as Downstream Worker
participant X2Text
Client->>Backend: HTTP request with X-Request-ID
Backend->>Broker: Publish task with request_id header
Broker->>WorkerA: Deliver task
WorkerA->>WorkerA: Bind request_id to log context
WorkerA->>Broker: Publish downstream task with request_id
Broker->>WorkerB: Deliver downstream task
WorkerA->>Backend: Internal callback with X-Request-ID
Client->>X2Text: Request with optional X-Request-ID
X2Text->>X2Text: Reuse or mint request_id for logs
Reviews (1): Last reviewed commit: "UN-2123 [FEAT] Propagate request_id acro..." | Re-trigger Greptile
Unstract test resultsPer-group results
Critical paths
|
What
Completes end-to-end
request_id(X-Request-ID) correlation so a single IDcan filter logs across the backend, Celery workers and services in gcloud.
before_task_publishsignal(
backend/backend/celery_signals.py) injects the request-scopedrequest_id(from
StateStore) into every published Celery task's message headers — noper-
send_taskchanges required.task_prerunnow prefers an explicitrequest_idfrom themessage headers over payload-derived ids (
file_execution_id, etc.), and anew
before_task_publishhandler re-propagates it onto downstreamworker→worker task chains.
X-Request-IDfrom the worker log context, so backend callbacks share theoriginating request's id.
X-Request-IDand logs it via aself-contained logging module (no new dependency).
request_id/trace_id/span_idparse identically in gcloud.Why
Debugging across services is painful when logs can't be correlated. The backend
already assigned a
request_idper request (and the frontend forwards one), butit died at the backend boundary — it was never propagated to the Celery workers
or echoed on worker→backend calls, so worker logs showed
request_id:-(or anexecution_idthat didn't match the originating request). This threads the sameid through the whole chain.
Jira: UN-2123
How
The worker-side receiving/logging machinery already existed (UN-3435) and was
explicitly designed to accept a real
request_idfrom the producer — this PRsupplies the missing producer side and closes the remaining hops:
before_task_publish(backend) readsStateStore.get(Common.REQUEST_ID)andsets
headers["request_id"]. Guarded so it can never break task publishing._bind_task_context(task_prerun) readstask.request.request_id(Celery 5.6.2 exposes custom message headers on the task
Context; a rawrequest.headersmapping is used as a version-safe fallback). Resolutionorder: message-header
request_id→ payload-derived id → Celerytask_id._propagate_request_id_on_publish(before_task_publish) forwards thebound
request_idonto tasks the worker itself publishes.base_client._make_requestaddsX-Request-IDfrom the worker log context.create_app()wiressetup_logging+register_request_id_middlewarefrom a new self-containedapp/logging_util.py.unstract/core/flask/logging.py) is aligned tothe canonical backend/worker format.
Out of scope (recommended as separate tickets):
trace_idfields in every formatter, meta-deps) is currently inert — no service runs
under
opentelemetry-instrument, exporters are hardcoded tonone, and thereis no OTLP endpoint — so
trace_id/span_idalways render as-. Activatingreal distributed tracing is a larger DevOps/observability effort. This PR uses
request_id(which the frontend/backend already surface via theX-Request-IDresponse header) as the pragmatic correlation key.execution
request_idthreaded through several tool-side layers; belongs withthe tool-execution/OTel correlation follow-up. (x2text still logs any
X-Request-IDit receives after this PR.)Can this PR break any existing features. If yes, please list possible items. If no, please explain why. (PS: Admins do not merge the PR without this section filled)
Low risk — the change is additive and defensively guarded:
before_task_publishhandlers no-op safely (null-guarded,StateStoreread wrapped in try/except) so they cannot break task publishing even if
no
request_idis in scope (e.g. beat-scheduled tasks fall back to theexisting
execution_id/task_idbehaviour).task.request.request_idusesgetattr(..., None)and only overridesthe previously-working payload-derived id when a header is actually present —
so existing worker correlation is preserved when the header is absent.
the canonical backend/worker format, which means Flask-service log lines
(platform-service, runner) show
module:<source-module>instead of thelogger
name. Log content is otherwise unchanged; any dashboard keying onthe logger name string would need updating (grep-by-
request_id/trace_idisunaffected and is the point of the change).
Database Migrations
None.
Env Config
None. (x2text-service reads the existing
LOG_LEVELenv if set; defaults toINFO.)Relevant Docs
Related Issues or PRs
Dependencies Versions
None added. (x2text-service deliberately avoids taking on
unstract-core; ituses a small self-contained logging module instead.)
Notes on Testing
py_compile-clean; pre-commit (ruff, ruff-format,pycln, pyupgrade, secret-scan, test-selection) green.
exposed on the task
Contextvia both attribute and.get()access.trigger a workflow/API execution and confirm a single
X-Request-IDfrom theoriginating HTTP request appears in the worker log lines (
request_id:<id>)and on the worker→backend internal-API calls. If Celery header→Context
attribute promotion ever regressed, the
request.headersfallback covers it,but the smoke test is the real guarantee.
Screenshots
N/A (no UI surface).
Checklist
I have read and understood the Contribution Guidelines.