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 3 commits 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
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 "-".
There was a problem hiding this comment.
Good call — applied in 2fd885a. Switched to has_request_context() before reading g, so log records outside an active request context (e.g. gunicorn startup) deterministically get request_id:- rather than relying on LocalProxy.__bool__ semantics. Thanks!
|
| Filename | Overview |
|---|---|
| backend/backend/celery_signals.py | Adds producer and backend-worker signal handlers that propagate request IDs through Celery headers and task-local logging state. |
| workers/shared/infrastructure/logging/logger.py | Prefers message-header request IDs, tracks whether they are safe to propagate, and forwards them through worker task chains. |
| workers/shared/clients/base_client.py | Adds the active worker request ID to internal backend HTTP calls. |
| x2text-service/app/logging_util.py | Introduces self-contained request-ID middleware and canonical structured logging for x2text. |
| unstract/core/src/unstract/core/flask/middleware.py | Echoes the active request ID in Flask response headers. |
Sequence Diagram
sequenceDiagram
participant Client
participant Backend
participant Broker
participant WorkerA as Worker
participant WorkerB as Downstream Worker
participant X2Text
Client->>Backend: HTTP request + X-Request-ID
Backend->>Broker: Publish task + request_id header
Broker->>WorkerA: Deliver task
WorkerA->>WorkerA: Bind request_id to log context
WorkerA->>Broker: Publish child + request_id header
Broker->>WorkerB: Deliver child task
WorkerA->>Backend: Internal callback + X-Request-ID
Client->>X2Text: Request + optional X-Request-ID
X2Text-->>Client: Response + X-Request-ID
Reviews (3): Last reviewed commit: "UN-2123 [FIX] Address review: honor inco..." | Re-trigger Greptile
…has_request_context()
pk-zipstack
left a comment
pk-zipstack
left a comment
There was a problem hiding this comment.
Code review — 4 findings on the request_id propagation chain. The producer/consumer plumbing itself looks correct (verified against Celery 5.5.3 that a custom key in the v2 message headers is promoted onto Task.request both as an attribute and via request.headers, and that mutating the headers dict inside before_task_publish reaches producer.publish). The issues are at the two ends of the chain and in the precedence rule.
| # callbacks share the originating request's request_id in logs. | ||
| request_id = _current_request_id() | ||
| if request_id: | ||
| headers["X-Request-ID"] = request_id |
There was a problem hiding this comment.
The worker→backend hop does not actually correlate — the backend drops this header.
backend/backend/settings/base.py:279 sets LOG_REQUEST_ID_HEADER = "X-Request-ID", but django-log-request-id looks that value up in request.META, where WSGI maps HTTP headers to HTTP_X_REQUEST_ID:
# log_request_id/middleware.py::_get_request_id
return request.META.get(request_id_header, default_request_id)The lookup therefore always misses, and because GENERATE_REQUEST_ID_IF_NOT_IN_HEADER = True the backend mints a fresh uuid4 for every request instead. Verified against the installed package:
LOG_REQUEST_ID_HEADER='X-Request-ID', META={'HTTP_X_REQUEST_ID': 'abc-123-from-worker'} -> 98701742cad145069f9d1df02327af42
LOG_REQUEST_ID_HEADER='HTTP_X_REQUEST_ID', META={'HTTP_X_REQUEST_ID': 'abc-123-from-worker'} -> abc-123-from-worker
Concrete scenario: a file-processing worker calls the internal API with X-Request-ID: <originating id>; CustomRequestIDMiddleware ignores it, request.id becomes a new uuid4, CustomAuthMiddleware stores that in StateStore, and the backend log lines for the callback show an id unrelated to the worker's. So the "workers → backend" bullet in the PR description isn't achieved, and the same bug already silently discards the X-Request-ID the frontend forwards (frontend/src/helpers/requestId.js).
Fix is one line: LOG_REQUEST_ID_HEADER = "HTTP_X_REQUEST_ID" (keep REQUEST_ID_RESPONSE_HEADER = "X-Request-ID", which is a real header name and is correct as-is). Worth adding a test that posts X-Request-ID and asserts the response echoes the same value.
There was a problem hiding this comment.
Fixed in 713f9a4: LOG_REQUEST_ID_HEADER is now HTTP_X_REQUEST_ID. Spot on — and this also explains a red herring in our own dev smoke test, where the client X-Request-ID we sent was dropped and a fresh id minted (we had wrongly blamed the edge LB). The incoming header from the frontend and from workers is now actually honored. Thanks.
| ctx = WorkerLogger.get_context() | ||
| request_id = _coerce_id(getattr(ctx, "request_id", None)) if ctx else None | ||
| if request_id: | ||
| headers["request_id"] = request_id |
There was a problem hiding this comment.
This re-publish handler combined with the new header-first precedence regresses correlation for beat-scheduled pipelines.
_bind_task_context falls back to request_id or task_id, so a task with no extractable payload id binds its own opaque Celery task_id. This handler then stamps that task_id onto every task it publishes, and the child's _request_id_from_message gives the header top priority over its own payload (above file_execution_id / execution_id).
Concrete scenario — scheduled ETL/TASK pipelines, the most common production flow:
workers/scheduler/tasks.py:262execute_pipeline_task_v2(organization_id, pipeline_id, pipeline_name)— none of_REQUEST_ID_KEYS(request_id/file_execution_id/execution_id/run_id) is present, so it binds its Celerytask_id.- It then dispatches
async_execute_bin(workers/scheduler/tasks.py:172), whose payload does carry the realexecution_id. - Post-change that child — and every downstream file batch and callback — logs
request_id:<scheduler-celery-task-id>instead ofrequest_id:<execution_id>.
The PR's risk section says beat-scheduled tasks "fall back to the existing execution_id/task_id behaviour", but the header is now present on those chains, so they don't. Same shape applies to the per-file granularity _extract_request_id documents ("Workers bind file_execution_id … giving per-file granularity"): once any ancestor stamps a header, descendants can no longer surface their own id.
Suggest only propagating an id that is genuinely a cross-service correlation id — e.g. have _bind_task_context record whether the bound value came from a message header (or from a real request_id payload key) and have this handler forward it only in that case, leaving payload-derived / task_id fallbacks local to their own task.
There was a problem hiding this comment.
Fixed in 713f9a4. Added a request_id_propagatable marker to LogContext: _bind_task_context sets it True only when the id came from an upstream message header, and _propagate_request_id_on_publish now re-propagates only those — never a task_id or execution_id fallback. So beat/scheduler-originated child tasks keep deriving their own file_execution_id; only a genuine HTTP-origin id fans out across the chain.
| WorkerLogger.update_context(request_id=request_id, task_id=task_id) | ||
|
|
||
|
|
||
| def _propagate_request_id_on_publish(headers=None, **_): |
There was a problem hiding this comment.
The PG-queue transport bypasses this hop entirely, so correlation breaks there.
workers/queue_backend/dispatch.py::enqueue routes to _enqueue_pg when resolve_backend(...) is QueueBackend.PG, which serialises via to_payload() and writes straight to pg_queue_message — it never calls current_app.send_task, so before_task_publish never fires and no request_id is carried. On the consume side consumer.py:575 runs task.apply(headers={FAIRNESS_HEADER_NAME: ...}), i.e. the only headers reaching Context are the fairness ones, so _request_id_from_message returns None and the task falls back to its payload id.
Concrete scenario: an API-triggered execution whose task name is in WORKER_PG_QUEUE_ENABLED_TASKS (or that rides a backend= pipeline override) loses the HTTP request_id at the first PG hop and never regains it for the rest of the execution — exactly the chains the PG rollout is moving onto. Worth threading request_id into to_payload() / restoring it into the headers passed to task.apply(), or at minimum calling this gap out in the PR's out-of-scope list.
There was a problem hiding this comment.
Confirmed — the PG transport enqueues via PgQueueMessage.objects.create(), so before_task_publish never fires and TaskPayload carries no request_id. Keeping this out of scope for this PR (Celery is the live default; pg_queue_enabled is fail-closed). The PGMQ owner has a precise carrier spec (add request_id to TaskPayload, populate in the backend and worker PG producers, inject into the consumer apply(headers=...)). Added to the PR out-of-scope section.
| logger.debug("Unable to read request_id from StateStore", exc_info=True) | ||
| return | ||
| if request_id and not headers.get(Common.REQUEST_ID): | ||
| headers[Common.REQUEST_ID] = request_id |
There was a problem hiding this comment.
The header is injected for the Django Celery workers too, but nothing there consumes it — and those workers don't re-propagate.
The workers/ package gets its receiving side from _bind_task_context / _propagate_request_id_on_publish, but the backend Celery app has no task_prerun equivalent: its logging uses log_request_id.filters.RequestIDFilter, which reads log_request_id.local.request_id — only ever populated by the HTTP middleware. So for -A backend worker services (e.g. worker-metrics on dashboard_metric_events in docker/docker-compose.yaml:49, plus the celery / celery_api_deployments / celery_periodic_logs queues) the injected header is carried and then dropped: those task logs still show request_id:-.
Second-order effect: since StateStore is only written by CustomAuthMiddleware on an HTTP request, this handler is a no-op inside those worker processes, so any task they publish downstream loses the id too — the chain breaks at the first backend-worker hop.
Also a doc nit on the docstring above: StateStore[request_id] is populated by CustomAuthMiddleware (backend/account_v2/custom_auth_middleware.py:26), not by CustomRequestIDMiddleware (which only sets request.id). Worth correcting since the distinction matters for which requests actually have an id in scope.
There was a problem hiding this comment.
Fixed in 713f9a4. Added task_prerun/task_postrun handlers in celery_signals.py that bind the injected header onto log_request_id.local (so the backend RequestIDFilter emits it) and onto StateStore (so a task the backend worker itself publishes re-propagates). No longer a no-op on that side.
chandrasekharan-zipstack
left a comment
chandrasekharan-zipstack
left a comment
There was a problem hiding this comment.
A few points on hops and guarantees that I think are still open — details inline.
| # callbacks share the originating request's request_id in logs. | ||
| request_id = _current_request_id() | ||
| if request_id: | ||
| headers["X-Request-ID"] = request_id |
There was a problem hiding this comment.
The worker → runner hop stamps a different id, and it isn't in the out-of-scope list.
This closes the worker → backend hop, but the worker → runner hop already sends an X-Request-ID and it carries the wrong value:
# unstract/tool-sandbox/src/unstract/tool_sandbox/helper.py:439 (and :592)
headers = {
"X-Request-ID": file_execution_id,
}The runner reads that header via unstract.core.flask.register_request_id_middleware and renders it as request_id:<file_execution_id> — so after this PR the runner is the one service actively logging a different id under the same field name, which is exactly what breaks the single-request_id gcloud query the PR is built for.
This is called from worker context, where WorkerLogger.get_context().request_id is already bound by _bind_task_context, so the fix is the same shape as _current_request_id() here — prefer the propagated id and keep file_execution_id as the fallback.
Either close it or add it alongside the SDK1 x2text-adapter gap in the PR's out-of-scope list; right now it reads as covered.
There was a problem hiding this comment.
Good catch. tool-sandbox -> runner sends X-Request-ID = file_execution_id, which is inconsistent with standardizing on the HTTP request_id. Reconciling it means threading the request_id through tool-sandbox, which belongs with the tool-execution / OTel follow-up (same bucket as the SDK1 x2text-adapter forwarding). Added it explicitly to the PR out-of-scope list rather than leaving it silent.
|
|
||
| def create_app() -> Flask: | ||
| log_level = getattr(logging, env.get("LOG_LEVEL", "INFO").upper(), logging.INFO) | ||
| setup_logging(log_level) |
There was a problem hiding this comment.
The old logging config it replaces is still in the tree.
# x2text-service/app/controllers/controller.py:19
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)That runs at import of app.controllers (line 7 above), i.e. before create_app() is ever called. It happens to be harmless today only because dictConfig then clears and replaces the root handlers — but any path that imports the controller without going through create_app() (tests, an alternate entrypoint, a REPL) silently gets the pre-PR format with no request_id field at all.
Worth deleting as part of this change — it's the line setup_logging() is meant to supersede, and leaving both means the service has two competing logging configs whose behaviour depends on import order.
There was a problem hiding this comment.
Fixed in 713f9a4 — removed the stale logging.basicConfig at controller.py import. setup_logging (dictConfig) is the single source now.
|
|
||
| @app.before_request | ||
| def _assign_request_id() -> None: | ||
| g.request_id = request.headers.get("X-Request-ID", str(uuid.uuid4())) |
There was a problem hiding this comment.
x2text mints a request_id that nothing can ever learn.
There's no after_request echoing X-Request-ID back on the response — the Django backend does this via REQUEST_ID_RESPONSE_HEADER, and the frontend relies on it. Combined with the SDK1 x2text-adapter not forwarding the header (acknowledged as out of scope), the practical outcome today is: x2text generates a fresh uuid4 per request, receives it from nobody, and returns it to nobody. The id appears only in x2text's own log lines, where it correlates with nothing.
A three-line after_request at least makes it retrievable from a live call now, ahead of the SDK-side follow-up:
@app.after_request
def _echo_request_id(response):
response.headers["X-Request-ID"] = g.get("request_id", "-")
return responseSame gap exists in the shared unstract/core/src/unstract/core/flask/middleware.py:15, so fixing it in both keeps the two copies aligned.
There was a problem hiding this comment.
Fixed in 713f9a4 — added an after_request that echoes X-Request-ID, and put it in the shared unstract.core.flask middleware too, so platform-service and runner echo consistently, not just x2text.
| exc_info=True, | ||
| ) | ||
| request_id = task_id | ||
| request_id = _request_id_from_message(task) |
There was a problem hiding this comment.
This call sits outside the try, which breaks the guarantee the docstring three lines above still promises.
Catches any extraction failure so a malformed payload can never leave the previous task's id bound on the thread.
Before this PR the whole resolution — including the or task_id fallback — was inside the try, so update_context always ran. Now _request_id_from_message(task) runs unguarded: if it raises, WorkerLogger.update_context(...) on line 749 is never reached, and the previous task's request_id and task_id stay bound on the thread-local context. That is precisely the failure mode the guard was written for, and it's worse than the old one because a stale id is silently plausible where a missing one is obvious.
A raising task_prerun receiver also fails the task itself (send_prerun is inside the traced task body in celery/app/trace.py), so the blast radius isn't only logging.
Moving the call inside the existing try keeps the invariant with no other change:
try:
request_id = _request_id_from_message(task) or _extract_request_id(
args or (), kwargs or {}, task
)
except Exception:
...
request_id = NoneThere was a problem hiding this comment.
Fixed in 713f9a4 — moved the whole resolution (header read, payload extraction, and the or task_id fallback) inside the try, so update_context always runs and the docstring guarantee holds again.
…ion, backend reader - settings: LOG_REQUEST_ID_HEADER must be the WSGI META key HTTP_X_REQUEST_ID, else django-log-request-id never reads the incoming X-Request-ID and mints a fresh id every request (broke frontend + worker->backend correlation). - workers: run all of _bind_task_context's resolution inside the try (restores the 'never leave prior task's id bound' guarantee); mark only header-origin ids propagatable so _propagate_request_id_on_publish never stamps a local fallback (task_id/execution_id) onto child tasks and clobbers their own file_execution_id correlation (beat/scheduler pipelines). - backend celery: add task_prerun/postrun reader so the backend's OWN Celery workers (beat, dashboard tasks) consume the injected header instead of logging request_id:- (the injection had no reader on that side). - flask (core + x2text): echo X-Request-ID on responses so callers can learn a minted id (mirrors backend REQUEST_ID_RESPONSE_HEADER). - x2text: drop the stale module-import logging.basicConfig superseded by setup_logging's dictConfig.
|
Unstract test resultsPer-group results
Critical paths
|
3 participants
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.)tool-sandbox → runnercurrently sendsX-Request-ID = file_execution_id,inconsistent with the HTTP request_id standardized here; reconciling it means
threading the request_id through tool-sandbox — same tool-execution/OTel bucket.
pg_queue_enabled) — a separate, non-Celery transport(DB-INSERT enqueue + custom poll-loop consumer) that this PR's Celery-signal
propagation does not cover. Celery is the live default; a matching carrier
(
request_idonTaskPayload, populated in the PG producers, read into theconsumer's
apply(headers=…)) is handed to the PGMQ owner so correlationstays complete as
pg_queue_enabledramps.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.