mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-27 09:09:02 +08:00
feat(logging): production-grade structured logs for Grafana/Loki (#13164)
* feat(logging): production-grade structured logs for Grafana/Loki Make langflow and lfx log output viable for ingestion by Grafana/Loki and other observability tools when run in JSON mode (LANGFLOW_LOG_ENV=container). Core changes in src/lfx/src/lfx/log/logger.py: - Preserve exceptions in JSON output via structlog.processors.ExceptionRenderer with ExceptionDictTransformer. Tracebacks now emit as a structured exception array (exc_type, exc_value, frames) instead of being dropped. - show_locals defaults to OFF; opt-in via LANGFLOW_LOG_TRACE_LOCALS=true so frame locals can't leak API keys, env, or request bodies in shipped logs. - Add service metadata (service / version / environment) from LANGFLOW_SERVICE_NAME / LANGFLOW_VERSION / LANGFLOW_ENVIRONMENT. - Add logger name to every record so Grafana can filter by source. - Add optional OpenTelemetry trace_id / span_id correlation. Import is resolved once at module load; runtime calls are wrapped so a flaky tracer SDK can never break logging. - Add default-on PII redaction for password, token, api_key, authorization, cookie, etc. Walks nested dicts, lists, and tuples up to depth 4. Extra keys via LANGFLOW_LOG_REDACT_KEYS. - Add per-logger level overrides via LANGFLOW_LOG_LEVELS="name=LEVEL,...". Malformed entries (typos like WARN instead of WARNING) raise UserWarning instead of silently dropping. - Use ISO 8601 UTC timestamps. - Install a stdlib InterceptHandler on the root logger in JSON modes so uvicorn, sqlalchemy, httpx, langchain, asyncio etc. emit a single unified JSON stream. Forwards exc_info and stack_info. emit() is wrapped to route any error through handleError so a malformed third-party log call cannot raise into the request path. Install is idempotent: re-running configure() updates the level instead of stacking handlers. Not installed in pretty mode so dev terminals don't get duplicate lines. - Reset cached loggers at the start of configure() so modules that captured a logger before configure() ran pick up the new processor chain. - Preserve the get_logger() name through PrintLogger so add_logger_name can attach it. Tests in src/backend/tests/unit/test_logger.py: - Cover structured tracebacks, PII redaction (top-level, nested, list, tuple, depth limit), logger name, stdlib intercept forwarding exc_info and stack_info, intercept idempotency, intercept-not-installed in pretty mode, malformed-args safety net, service info defaults and env overrides, malformed LANGFLOW_LOG_LEVELS warning, container_csv exception text, show_locals default off (verified by absence of the secret value, not just the key) and opt-in. * docs(observability): Grafana + Loki reference stack and env-var docs Adds a self-contained Loki + Promtail + Grafana docker-compose stack under deploy/observability/grafana-loki/ with a pre-provisioned dashboard that demonstrates the production logging features: structured tracebacks, PII redaction, service/version/environment labels, and the stdlib intercept path. Anyone running Langflow in JSON mode can point Promtail at their log file and get a working board on first run. Also documents the new env vars (LANGFLOW_SERVICE_NAME, LANGFLOW_VERSION, LANGFLOW_ENVIRONMENT, LANGFLOW_LOG_LEVELS, LANGFLOW_LOG_REDACT_KEYS, LANGFLOW_LOG_TRACE_LOCALS) in docs/docs/Develop/logging.mdx and adds a new page docs/docs/Develop/observability-grafana-loki.mdx covering JSON output shape, structured exceptions, stdlib routing, and OpenTelemetry trace correlation. Registered in the Observability sidebar category. * docs(observability): rename dashboard to 'Langflow Logs' * docs(observability): fix broken link from logging guide to Grafana/Loki page The logging guide linked to the new Grafana/Loki page with an absolute path (/observability-grafana-loki). Since the page is new and only exists in the next docs version, the absolute link resolved to a non-existent root-version route and failed the Docusaurus broken-link check. Use a version-aware relative .mdx link instead. * docs(observability): clarify stdout requirement for unified JSON logs The Grafana/Loki guide told users to set LANGFLOW_LOG_FILE, but the stdlib intercept that routes uvicorn/sqlalchemy/httpx/langchain into the JSON stream is only installed on the stdout path, so those library logs landed in the file as plain text and the json parse stage could not label them. Point the guide at stdout (redirected to the scraped file) and note the file-mode limitation. Also document that the JSON format is platform-agnostic and works with any JSON-ingesting backend including IBM Instana, whose OpenTelemetry-based Python tracer (3.0+) correlates logs to traces via trace_id/span_id. * fix(logging): render stdlib logs as redacted JSON in file mode In JSON mode with LANGFLOW_LOG_FILE set, the stdlib intercept was skipped to avoid a recursion loop, so third-party loggers (uvicorn, sqlalchemy, httpx, asyncio) wrote plain text straight to the file. Those lines bypassed both JSON rendering and PII redaction, so secrets in their structured fields landed in the file verbatim and Loki could not parse or label them. Route JSON file output through a structlog ProcessorFormatter on the rotating handler: foreign stdlib records are enriched via foreign_pre_chain (ExtraAdder + redaction) and rendered as JSON alongside application logs, while the RotatingFileHandler keeps log rotation. The stdout path is unchanged. Also forward stdlib extra fields through InterceptHandler so the stdout path redacts them too, keeping both paths consistent. Update the Grafana/Loki deploy guide to reflect that LANGFLOW_LOG_FILE now produces a single redacted JSON stream. * fix(logging): retrieval buffer captures the message text add_serialized stored the rendered message under the 'message' key, but SizedLogBuffer.write only read 'event'/'msg'/'text', so every entry returned by the /logs and /logs-stream endpoints had an empty message. Read 'message' first and keep the other keys as fallbacks for records written in other shapes.
This commit is contained in:
Gabriel Luiz Freitas Almeida
committed by
GitHub
GitHub
parent
fb046c02df
commit
e2d17a6d84
69
deploy/observability/grafana-loki/README.md
Normal file
69
deploy/observability/grafana-loki/README.md
Normal file
@ -0,0 +1,69 @@
|
||||
# Langflow on Grafana + Loki
|
||||
|
||||
Reference stack that ingests Langflow's structured JSON logs into [Loki](https://grafana.com/oss/loki/) and visualizes them with a pre-provisioned Grafana dashboard.
|
||||
|
||||
Use this as a starting point. The compose file, Promtail config, and dashboard JSON are independent of the rest of `deploy/` and can be lifted into any environment.
|
||||
|
||||
## What you get
|
||||
|
||||
- **Loki 3.2** on `:3100`
|
||||
- **Promtail 3.2** scraping a directory of Langflow log files
|
||||
- **Grafana 11.3** on `:3000` with the Loki datasource and the `Langflow Logs` dashboard already provisioned
|
||||
|
||||
## Prerequisites on the Langflow side
|
||||
|
||||
The dashboard expects Langflow to be running in JSON mode with service metadata set. At minimum:
|
||||
|
||||
```bash
|
||||
LANGFLOW_LOG_ENV=container
|
||||
LANGFLOW_LOG_FILE=/var/log/langflow/langflow.log
|
||||
LANGFLOW_SERVICE_NAME=langflow
|
||||
LANGFLOW_VERSION=1.10.0
|
||||
LANGFLOW_ENVIRONMENT=production
|
||||
```
|
||||
|
||||
In JSON mode the file is a single JSON stream: application logs and third-party stdlib loggers
|
||||
(`uvicorn`, `sqlalchemy`, `httpx`, `langchain`) are all rendered as JSON and run through PII
|
||||
redaction, so the `json` parse stage and the **Stdlib intercept routing** panel work against it
|
||||
directly. If you'd rather not write a file, drop `LANGFLOW_LOG_FILE` and scrape the container's
|
||||
stdout instead (same JSON, same labels).
|
||||
|
||||
See [Logs and observability](../../../docs/docs/Develop/observability-grafana-loki.mdx) for the full list of environment variables (per-logger overrides, extra PII redaction keys, trace correlation, etc.).
|
||||
|
||||
## Run
|
||||
|
||||
From this directory:
|
||||
|
||||
```bash
|
||||
# Point Promtail at the directory containing your langflow*.log file(s).
|
||||
# Defaults to ./logs/ if unset (useful for a quick local smoke test).
|
||||
export LANGFLOW_LOG_DIR=/path/to/langflow/logs
|
||||
|
||||
docker compose up -d
|
||||
```
|
||||
|
||||
Then open [http://localhost:3000/d/langflow-prod-logs](http://localhost:3000/d/langflow-prod-logs). Default credentials are `admin` / `admin` (override with `GF_ADMIN_USER` and `GF_ADMIN_PASSWORD`).
|
||||
|
||||
To stop:
|
||||
|
||||
```bash
|
||||
docker compose down -v
|
||||
```
|
||||
|
||||
## What each dashboard panel proves
|
||||
|
||||
| Panel | LogQL it runs |
|
||||
|---|---|
|
||||
| **PII leak count (must be 0)** | `sum(count_over_time({job="langflow"} \|~ "sk-do-not-leak\|hunter2\|Bearer xyz" [$__range]))` |
|
||||
| **Errors with structured tracebacks** | `{job="langflow", level=~"error\|critical"} \|= "exception" \| json` |
|
||||
| **Redaction proof** | `{job="langflow"} \|~ "\\*\\*\\*"` |
|
||||
| **Stdlib intercept routing** | `{job="langflow", logger=~"uvicorn.*\|sqlalchemy.*\|httpx.*\|langchain.*"}` |
|
||||
| **Service / environment / version coverage** | `sum by (service, environment, version) (count_over_time({job="langflow"}[$__range]))` |
|
||||
| **Log rate by level** | `sum by (level) (rate({job="langflow"}[1m]))` |
|
||||
| **Log rate by logger** | `topk(10, sum by (logger) (rate({job="langflow"}[1m])))` |
|
||||
|
||||
## Notes
|
||||
|
||||
- Promtail only promotes `level`, `service`, `environment`, `version`, `logger` to labels. High-cardinality fields (`user_id`, `flow_id`, `trace_id`) stay in the log body — query them with `| json` in LogQL.
|
||||
- Replace Promtail with [Grafana Alloy](https://grafana.com/oss/alloy/) if you already standardize on it; the JSON parse stage maps 1:1.
|
||||
- If your runtime ships logs through a different transport (Fluent Bit, Vector, OTLP), only the scrape side changes — the dashboard and label schema stay the same.
|
||||
41
deploy/observability/grafana-loki/docker-compose.yml
Normal file
41
deploy/observability/grafana-loki/docker-compose.yml
Normal file
@ -0,0 +1,41 @@
|
||||
services:
|
||||
loki:
|
||||
image: grafana/loki:3.2.0
|
||||
container_name: lf-loki
|
||||
ports:
|
||||
- "3100:3100"
|
||||
command: -config.file=/etc/loki/local-config.yaml
|
||||
healthcheck:
|
||||
test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://localhost:3100/ready"]
|
||||
interval: 5s
|
||||
timeout: 3s
|
||||
retries: 20
|
||||
|
||||
promtail:
|
||||
image: grafana/promtail:3.2.0
|
||||
container_name: lf-promtail
|
||||
volumes:
|
||||
- ./promtail/config.yml:/etc/promtail/config.yml
|
||||
# Mount the directory that holds your langflow JSON log file.
|
||||
# Override LANGFLOW_LOG_DIR to point at your real log location.
|
||||
- ${LANGFLOW_LOG_DIR:-./logs}:/var/log/langflow:ro
|
||||
command: -config.file=/etc/promtail/config.yml
|
||||
depends_on:
|
||||
loki:
|
||||
condition: service_healthy
|
||||
|
||||
grafana:
|
||||
image: grafana/grafana:11.3.0
|
||||
container_name: lf-grafana
|
||||
ports:
|
||||
- "3000:3000"
|
||||
environment:
|
||||
- GF_SECURITY_ADMIN_USER=${GF_ADMIN_USER:-admin}
|
||||
- GF_SECURITY_ADMIN_PASSWORD=${GF_ADMIN_PASSWORD:-admin}
|
||||
- GF_USERS_DEFAULT_THEME=dark
|
||||
volumes:
|
||||
- ./grafana/provisioning:/etc/grafana/provisioning
|
||||
- ./grafana/dashboards:/var/lib/grafana/dashboards
|
||||
depends_on:
|
||||
loki:
|
||||
condition: service_healthy
|
||||
@ -0,0 +1,163 @@
|
||||
{
|
||||
"uid": "langflow-prod-logs",
|
||||
"title": "Langflow Logs",
|
||||
"description": "Structured logs from Langflow and lfx: tracebacks, PII redaction, service labels, stdlib intercept, trace correlation.",
|
||||
"tags": ["langflow", "lfx", "structured-logs"],
|
||||
"timezone": "browser",
|
||||
"schemaVersion": 39,
|
||||
"version": 1,
|
||||
"refresh": "5s",
|
||||
"time": {
|
||||
"from": "now-15m",
|
||||
"to": "now"
|
||||
},
|
||||
"templating": {
|
||||
"list": [
|
||||
{
|
||||
"name": "environment",
|
||||
"type": "query",
|
||||
"datasource": "Loki",
|
||||
"query": "label_values({job=\"langflow\"}, environment)",
|
||||
"current": {"text": "All", "value": "$__all"},
|
||||
"includeAll": true,
|
||||
"multi": true,
|
||||
"refresh": 2
|
||||
},
|
||||
{
|
||||
"name": "service",
|
||||
"type": "query",
|
||||
"datasource": "Loki",
|
||||
"query": "label_values({job=\"langflow\"}, service)",
|
||||
"current": {"text": "All", "value": "$__all"},
|
||||
"includeAll": true,
|
||||
"multi": true,
|
||||
"refresh": 2
|
||||
}
|
||||
]
|
||||
},
|
||||
"panels": [
|
||||
{
|
||||
"id": 1,
|
||||
"type": "stat",
|
||||
"title": "Total log lines (window)",
|
||||
"gridPos": {"h": 4, "w": 6, "x": 0, "y": 0},
|
||||
"datasource": "Loki",
|
||||
"targets": [
|
||||
{"expr": "sum(count_over_time({job=\"langflow\", service=~\"$service\", environment=~\"$environment\"}[$__range]))", "refId": "A"}
|
||||
],
|
||||
"options": {"reduceOptions": {"calcs": ["lastNotNull"]}}
|
||||
},
|
||||
{
|
||||
"id": 2,
|
||||
"type": "stat",
|
||||
"title": "Errors (window)",
|
||||
"gridPos": {"h": 4, "w": 6, "x": 6, "y": 0},
|
||||
"datasource": "Loki",
|
||||
"fieldConfig": {"defaults": {"color": {"mode": "thresholds"}, "thresholds": {"steps": [{"color": "green", "value": null}, {"color": "red", "value": 1}]}}},
|
||||
"targets": [
|
||||
{"expr": "sum(count_over_time({job=\"langflow\", level=~\"error|critical\", service=~\"$service\", environment=~\"$environment\"}[$__range]))", "refId": "A"}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": 3,
|
||||
"type": "stat",
|
||||
"title": "Distinct loggers",
|
||||
"gridPos": {"h": 4, "w": 6, "x": 12, "y": 0},
|
||||
"datasource": "Loki",
|
||||
"targets": [
|
||||
{"expr": "count(count by (logger)({job=\"langflow\", service=~\"$service\", environment=~\"$environment\"}))", "refId": "A"}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": 4,
|
||||
"type": "stat",
|
||||
"title": "PII leak count (must be 0)",
|
||||
"description": "Counts log lines that contain literal sensitive values that should have been redacted. Anything non-zero means redaction failed.",
|
||||
"gridPos": {"h": 4, "w": 6, "x": 18, "y": 0},
|
||||
"datasource": "Loki",
|
||||
"fieldConfig": {"defaults": {"color": {"mode": "thresholds"}, "thresholds": {"steps": [{"color": "green", "value": null}, {"color": "red", "value": 1}]}}},
|
||||
"targets": [
|
||||
{"expr": "sum(count_over_time({job=\"langflow\"} |~ \"sk-do-not-leak|hunter2|Bearer xyz\"[$__range]))", "refId": "A"}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": 5,
|
||||
"type": "timeseries",
|
||||
"title": "Log rate by level",
|
||||
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 4},
|
||||
"datasource": "Loki",
|
||||
"targets": [
|
||||
{"expr": "sum by (level) (rate({job=\"langflow\", service=~\"$service\", environment=~\"$environment\"}[1m]))", "refId": "A", "legendFormat": "{{level}}"}
|
||||
],
|
||||
"options": {"legend": {"displayMode": "table", "placement": "right", "calcs": ["sum"]}}
|
||||
},
|
||||
{
|
||||
"id": 6,
|
||||
"type": "timeseries",
|
||||
"title": "Log rate by logger (top 10)",
|
||||
"gridPos": {"h": 8, "w": 12, "x": 12, "y": 4},
|
||||
"datasource": "Loki",
|
||||
"targets": [
|
||||
{"expr": "topk(10, sum by (logger) (rate({job=\"langflow\", service=~\"$service\", environment=~\"$environment\"}[1m])))", "refId": "A", "legendFormat": "{{logger}}"}
|
||||
],
|
||||
"options": {"legend": {"displayMode": "table", "placement": "right", "calcs": ["sum"]}}
|
||||
},
|
||||
{
|
||||
"id": 7,
|
||||
"type": "logs",
|
||||
"title": "Structured tracebacks (errors with exception array)",
|
||||
"description": "Every error in JSON mode now ships a structured exception array (exc_type, exc_value, frames) instead of being silently dropped. Expand a row to see the structure.",
|
||||
"gridPos": {"h": 12, "w": 24, "x": 0, "y": 12},
|
||||
"datasource": "Loki",
|
||||
"targets": [
|
||||
{"expr": "{job=\"langflow\", level=~\"error|critical\", service=~\"$service\", environment=~\"$environment\"} |= \"exception\" | json", "refId": "A"}
|
||||
],
|
||||
"options": {"showLabels": true, "showTime": true, "wrapLogMessage": true, "prettifyLogMessage": true, "dedupStrategy": "none"}
|
||||
},
|
||||
{
|
||||
"id": 8,
|
||||
"type": "logs",
|
||||
"title": "PII redaction proof (look for *** sentinels, not the raw value)",
|
||||
"description": "Sensitive keys (password, api_key, token, authorization, cookie) appear as *** before the JSON ever leaves the process.",
|
||||
"gridPos": {"h": 10, "w": 12, "x": 0, "y": 24},
|
||||
"datasource": "Loki",
|
||||
"targets": [
|
||||
{"expr": "{job=\"langflow\", service=~\"$service\"} |~ \"\\\\*\\\\*\\\\*\"", "refId": "A"}
|
||||
],
|
||||
"options": {"showLabels": false, "showTime": true, "wrapLogMessage": true, "prettifyLogMessage": true}
|
||||
},
|
||||
{
|
||||
"id": 9,
|
||||
"type": "logs",
|
||||
"title": "Stdlib intercept: third-party loggers routed into JSON",
|
||||
"description": "uvicorn, sqlalchemy, httpx, langchain, asyncio used to emit unstructured text. With LANGFLOW_LOG_ENV=container they come through the same JSON stream with logger=<name>.",
|
||||
"gridPos": {"h": 10, "w": 12, "x": 12, "y": 24},
|
||||
"datasource": "Loki",
|
||||
"targets": [
|
||||
{"expr": "{job=\"langflow\", logger=~\"uvicorn.*|sqlalchemy.*|httpx.*|langchain.*|third_party.*\", service=~\"$service\"}", "refId": "A"}
|
||||
],
|
||||
"options": {"showLabels": true, "showTime": true, "wrapLogMessage": true, "prettifyLogMessage": true}
|
||||
},
|
||||
{
|
||||
"id": 10,
|
||||
"type": "table",
|
||||
"title": "Top error sources",
|
||||
"gridPos": {"h": 8, "w": 12, "x": 0, "y": 34},
|
||||
"datasource": "Loki",
|
||||
"targets": [
|
||||
{"expr": "topk(10, sum by (logger) (count_over_time({job=\"langflow\", level=~\"error|critical\"}[$__range])))", "refId": "A", "instant": true, "format": "table"}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": 11,
|
||||
"type": "table",
|
||||
"title": "Service / environment / version coverage",
|
||||
"description": "Confirms every record carries the service metadata Grafana dashboards key off of.",
|
||||
"gridPos": {"h": 8, "w": 12, "x": 12, "y": 34},
|
||||
"datasource": "Loki",
|
||||
"targets": [
|
||||
{"expr": "sum by (service, environment, version) (count_over_time({job=\"langflow\"}[$__range]))", "refId": "A", "instant": true, "format": "table"}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
@ -0,0 +1,10 @@
|
||||
apiVersion: 1
|
||||
providers:
|
||||
- name: langflow
|
||||
folder: ''
|
||||
type: file
|
||||
disableDeletion: false
|
||||
updateIntervalSeconds: 10
|
||||
allowUiUpdates: true
|
||||
options:
|
||||
path: /var/lib/grafana/dashboards
|
||||
@ -0,0 +1,10 @@
|
||||
apiVersion: 1
|
||||
datasources:
|
||||
- name: Loki
|
||||
type: loki
|
||||
access: proxy
|
||||
url: http://loki:3100
|
||||
isDefault: true
|
||||
editable: false
|
||||
jsonData:
|
||||
maxLines: 1000
|
||||
35
deploy/observability/grafana-loki/promtail/config.yml
Normal file
35
deploy/observability/grafana-loki/promtail/config.yml
Normal file
@ -0,0 +1,35 @@
|
||||
server:
|
||||
http_listen_port: 9080
|
||||
grpc_listen_port: 0
|
||||
|
||||
positions:
|
||||
filename: /tmp/positions.yaml
|
||||
|
||||
clients:
|
||||
- url: http://loki:3100/loki/api/v1/push
|
||||
|
||||
scrape_configs:
|
||||
- job_name: langflow
|
||||
static_configs:
|
||||
- targets:
|
||||
- localhost
|
||||
labels:
|
||||
job: langflow
|
||||
__path__: /var/log/langflow/*.log
|
||||
pipeline_stages:
|
||||
# Parse the structlog JSON line and promote useful fields to labels.
|
||||
# Keep this set small; high-cardinality fields (user_id, flow_id,
|
||||
# trace_id) should stay in the log body and be queried with `| json`.
|
||||
- json:
|
||||
expressions:
|
||||
level: level
|
||||
service: service
|
||||
environment: environment
|
||||
version: version
|
||||
logger: logger
|
||||
- labels:
|
||||
level:
|
||||
service:
|
||||
environment:
|
||||
version:
|
||||
logger:
|
||||
@ -44,6 +44,12 @@ To customize log storage locations and behaviors, set the following [Langflow en
|
||||
| `LANGFLOW_ENABLE_LOG_RETRIEVAL` | Boolean | `False` | Enables retrieval of logs from your Langflow instance with [Logs endpoints](/api-logs). |
|
||||
| `LANGFLOW_LOG_RETRIEVER_BUFFER_SIZE` | Integer | `10000` | Set the buffer size for log retrieval if `LANGFLOW_ENABLE_LOG_RETRIEVAL=True`. Must be greater than `0` for log retrieval to function. |
|
||||
| `LANGFLOW_NATIVE_TRACING` | Boolean | `true` | Enables the tracer to record execution traces directly in the Langflow database for use in Trace View. Set to `false` to disable tracing. |
|
||||
| `LANGFLOW_SERVICE_NAME` | String | `langflow` | Value of the `service` field on every JSON log record. Used as a label by Grafana/Loki. See [Grafana and Loki](./observability-grafana-loki.mdx). |
|
||||
| `LANGFLOW_VERSION` | String | Not set | Adds a `version` field when set. Omitted otherwise. |
|
||||
| `LANGFLOW_ENVIRONMENT` | String | Not set | Adds an `environment` field when set (`staging`, `production`, etc.). Omitted otherwise. |
|
||||
| `LANGFLOW_LOG_LEVELS` | String | Not set | Per-logger level overrides as `name=LEVEL,name=LEVEL`. Quiets noisy libraries without changing global defaults. Malformed entries raise a `UserWarning` at startup. Example: `sqlalchemy.engine=WARNING,httpx=INFO`. |
|
||||
| `LANGFLOW_LOG_REDACT_KEYS` | String | Not set | Comma-separated extra keys to scrub from JSON log records, on top of the defaults (`password`, `api_key`, `token`, `authorization`, `cookie`, etc.). |
|
||||
| `LANGFLOW_LOG_TRACE_LOCALS` | Boolean | `false` | Include frame locals in structured tracebacks. Off by default because locals can leak secrets; enable only for local debugging. |
|
||||
|
||||
## View logs in real-time
|
||||
|
||||
@ -181,6 +187,7 @@ The log file is only created when Langflow Desktop runs. If you don't see a log
|
||||
|
||||
## See also
|
||||
|
||||
* [Grafana and Loki](./observability-grafana-loki.mdx)
|
||||
* [Logs endpoints](/api-logs)
|
||||
* [Memory management options](/memory)
|
||||
* [Configure an external PostgreSQL database](/configuration-custom-database)
|
||||
102
docs/docs/Develop/observability-grafana-loki.mdx
Normal file
102
docs/docs/Develop/observability-grafana-loki.mdx
Normal file
@ -0,0 +1,102 @@
|
||||
---
|
||||
title: Grafana and Loki
|
||||
slug: /observability-grafana-loki
|
||||
---
|
||||
|
||||
Langflow emits structured JSON logs designed to be ingested by [Grafana Loki](https://grafana.com/oss/loki/) or any other log aggregator that understands one JSON object per line. This page covers the environment variables that control the JSON output and the reference Grafana dashboard shipped with the repository.
|
||||
|
||||
## Enable JSON mode
|
||||
|
||||
Set `LANGFLOW_LOG_ENV=container` (or `container_json`). Every line written to stdout becomes a JSON object that includes the event message, level, timestamp, logger name, exception structure, and the service metadata listed below.
|
||||
|
||||
```bash
|
||||
LANGFLOW_LOG_ENV=container
|
||||
LANGFLOW_LOG_LEVEL=INFO
|
||||
LANGFLOW_SERVICE_NAME=langflow
|
||||
LANGFLOW_VERSION=1.10.0
|
||||
LANGFLOW_ENVIRONMENT=production
|
||||
```
|
||||
|
||||
## Environment variables
|
||||
|
||||
The variables below are evaluated once at startup. They are read by `lfx.log.logger.configure()` and apply to both `langflow` and `lfx` log output.
|
||||
|
||||
| Variable | Format | Default | Description |
|
||||
|---|---|---|---|
|
||||
| `LANGFLOW_SERVICE_NAME` | String | `langflow` | Value of the `service` field on every log record. Surfaces as a Loki label for filtering. |
|
||||
| `LANGFLOW_VERSION` | String | unset | Adds a `version` field when set. Omitted otherwise. |
|
||||
| `LANGFLOW_ENVIRONMENT` | String | unset | Adds an `environment` field when set (`staging`, `production`, etc.). Omitted otherwise. |
|
||||
| `LANGFLOW_LOG_LEVELS` | String | unset | Per-logger level overrides as `name=LEVEL,name=LEVEL`. Used to quiet noisy libraries in production without changing global defaults. Example: `LANGFLOW_LOG_LEVELS=sqlalchemy.engine=WARNING,httpx=INFO`. Malformed entries raise a `UserWarning` at startup so typos like `WARN` surface instead of being silently dropped. |
|
||||
| `LANGFLOW_LOG_REDACT_KEYS` | String | unset | Comma-separated extra keys to scrub before serialization. Added on top of the default set: `password`, `passwd`, `secret`, `api_key`, `apikey`, `token`, `access_token`, `refresh_token`, `authorization`, `auth`, `cookie`, `set-cookie`, `x-api-key`, `x-auth-token`. Matching is case-insensitive and walks nested dicts, lists, and tuples up to depth 4. |
|
||||
| `LANGFLOW_LOG_TRACE_LOCALS` | Boolean | `false` | If `true`, includes frame locals in structured tracebacks. Off by default because locals can contain API keys, request bodies, and environment values. Enable only for local debugging. |
|
||||
|
||||
## What the JSON looks like
|
||||
|
||||
A typical record in JSON mode:
|
||||
|
||||
```json
|
||||
{
|
||||
"event": "incoming request",
|
||||
"level": "info",
|
||||
"logger": "langflow.api.run",
|
||||
"timestamp": "2026-05-17T18:18:29.100798Z",
|
||||
"service": "langflow",
|
||||
"version": "1.10.0",
|
||||
"environment": "production",
|
||||
"user_id": "user-4823",
|
||||
"flow_id": "flow-718",
|
||||
"authorization": "***",
|
||||
"cookie": "***",
|
||||
"request_body": {
|
||||
"input": "Tell me a joke",
|
||||
"api_key": "***",
|
||||
"session_id": "***"
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Errors include the full structured exception:
|
||||
|
||||
```json
|
||||
{
|
||||
"event": "flow run failed",
|
||||
"level": "error",
|
||||
"logger": "langflow.api.run",
|
||||
"timestamp": "2026-05-17T18:18:32.412Z",
|
||||
"service": "langflow",
|
||||
"exception": [
|
||||
{
|
||||
"exc_type": "ConnectionError",
|
||||
"exc_value": "upstream returned 503",
|
||||
"frames": [
|
||||
{"filename": "/app/.../runner.py", "lineno": 142, "name": "run_flow"}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
Frame locals are intentionally absent unless `LANGFLOW_LOG_TRACE_LOCALS=true`.
|
||||
|
||||
## Stdlib loggers
|
||||
|
||||
Third-party libraries that use Python's standard logging module (`uvicorn`, `sqlalchemy`, `httpx`, `langchain`, `asyncio`) are routed through the same JSON stream in container mode, whether you log to stdout or to `LANGFLOW_LOG_FILE`. Their original logger name is preserved on the `logger` field, so the same Grafana queries that work for application loggers also work for library output, and the same PII redaction is applied to their structured fields.
|
||||
|
||||
## OpenTelemetry trace correlation
|
||||
|
||||
If `opentelemetry-api` is installed and an active span is present, every log record automatically picks up `trace_id` and `span_id`. Use those fields to jump between logs and traces in Grafana. OpenTelemetry is not a hard dependency; if it is not installed, the processor is a no-op and nothing else changes.
|
||||
|
||||
## Other backends
|
||||
|
||||
The JSON format is not Loki-specific. It is one JSON object per line with standard fields (`timestamp`, `level`, `logger`, `service`, structured `exception`, optional `trace_id`/`span_id`), so any backend that ingests JSON logs works: Elasticsearch, Splunk, Datadog, and IBM Instana among them. Only the scrape side (Promtail, here) changes.
|
||||
|
||||
For IBM Instana, point the Instana host agent or an OTLP log exporter at the same stdout stream. Trace correlation works through the OpenTelemetry hook above: Instana's Python tracer is OpenTelemetry-based from version 3.0, so when it is active the `trace_id` and `span_id` on each record line up with the traces Instana already collects.
|
||||
|
||||
## Reference Grafana stack
|
||||
|
||||
A pre-provisioned Loki + Promtail + Grafana stack lives in `deploy/observability/grafana-loki/`. It includes the dashboard `Langflow Logs` with panels for structured tracebacks, PII redaction proof, the stdlib intercept path, and service/environment/version coverage. See the [README in that directory](https://github.com/langflow-ai/langflow/tree/main/deploy/observability/grafana-loki) for run instructions.
|
||||
|
||||
## See also
|
||||
|
||||
* [Logs](/logging)
|
||||
* [Logs endpoints](/api-logs)
|
||||
@ -154,6 +154,7 @@ module.exports = {
|
||||
label: "Observability",
|
||||
items: [
|
||||
"Develop/logging",
|
||||
"Develop/observability-grafana-loki",
|
||||
"Develop/traces",
|
||||
{
|
||||
type: "category",
|
||||
|
||||
@ -29,7 +29,6 @@ from lfx.log.logger import (
|
||||
buffer_writer,
|
||||
configure,
|
||||
log_buffer,
|
||||
remove_exception_in_production,
|
||||
setup_gunicorn_logger,
|
||||
setup_uvicorn_logger,
|
||||
)
|
||||
@ -523,37 +522,27 @@ class TestLogProcessors:
|
||||
assert serialized_data["level"] == "INFO"
|
||||
assert serialized_data["module"] == "test_module"
|
||||
|
||||
def test_remove_exception_in_production(self):
|
||||
"""Test remove_exception_in_production() removes exception info in prod."""
|
||||
event_dict = {"event": "Test message", "exception": "Some exception", "exc_info": "Some exc info"}
|
||||
def test_container_json_includes_structured_traceback(self, capsys):
|
||||
"""JSON renderer must serialize exc_info as a structured traceback for Grafana."""
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("traceback-test")
|
||||
try:
|
||||
msg = "boom"
|
||||
raise ValueError(msg)
|
||||
except ValueError as exc:
|
||||
log.error("connection failed", exc_info=exc) # noqa: TRY400 - exc_info kwarg path under test
|
||||
|
||||
# Import the actual module to access DEV
|
||||
import sys
|
||||
|
||||
logger_module = sys.modules["lfx.log.logger"]
|
||||
with patch.object(logger_module, "DEV", False): # noqa: FBT003
|
||||
result = remove_exception_in_production(None, "error", event_dict)
|
||||
|
||||
# Should remove exception info in production
|
||||
assert "exception" not in result
|
||||
assert "exc_info" not in result
|
||||
assert result["event"] == "Test message"
|
||||
|
||||
def test_remove_exception_in_development(self):
|
||||
"""Test remove_exception_in_production() keeps exception info in dev."""
|
||||
event_dict = {"event": "Test message", "exception": "Some exception", "exc_info": "Some exc info"}
|
||||
|
||||
# Import the actual module to access DEV
|
||||
import sys
|
||||
|
||||
logger_module = sys.modules["lfx.log.logger"]
|
||||
with patch.object(logger_module, "DEV", True): # noqa: FBT003
|
||||
result = remove_exception_in_production(None, "error", event_dict)
|
||||
|
||||
# Should keep exception info in development
|
||||
assert result["exception"] == "Some exception"
|
||||
assert result["exc_info"] == "Some exc info"
|
||||
assert result["event"] == "Test message"
|
||||
out = capsys.readouterr().out.strip().splitlines()[-1]
|
||||
record = json.loads(out)
|
||||
assert record["event"] == "connection failed"
|
||||
assert record["level"] == "error"
|
||||
# dict_tracebacks emits an "exception" list with at least one frame.
|
||||
assert isinstance(record.get("exception"), list)
|
||||
assert record["exception"]
|
||||
first = record["exception"][0]
|
||||
assert first["exc_type"] == "ValueError"
|
||||
assert first["exc_value"] == "boom"
|
||||
assert first.get("frames")
|
||||
|
||||
def test_buffer_writer_with_buffer_disabled(self):
|
||||
"""Test buffer_writer() when log buffer is disabled."""
|
||||
@ -622,6 +611,271 @@ class TestConstants:
|
||||
assert isinstance(level_value, int), f"Level {level_name} value {level_value} is not an integer"
|
||||
|
||||
|
||||
class TestProductionObservability:
|
||||
"""Best-practice production logging: redaction, logger name, stdlib intercept, per-logger levels."""
|
||||
|
||||
def setup_method(self):
|
||||
# Fully reset structlog so a previous test's cached factory (which may
|
||||
# hold a now-closed capsys pipe) cannot bleed into this test.
|
||||
structlog.reset_defaults()
|
||||
|
||||
def teardown_method(self):
|
||||
structlog.reset_defaults()
|
||||
# Drop any InterceptHandler we installed so other tests aren't affected.
|
||||
logging.root.handlers = [h for h in logging.root.handlers if not isinstance(h, InterceptHandler)]
|
||||
|
||||
def _emit_and_parse(self, capsys, fn):
|
||||
fn()
|
||||
out = capsys.readouterr().out.strip().splitlines()
|
||||
return [json.loads(line) for line in out if line.startswith("{")]
|
||||
|
||||
def test_pii_redaction_top_level_and_nested(self, capsys):
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("redact.test")
|
||||
records = self._emit_and_parse(
|
||||
capsys,
|
||||
lambda: log.info(
|
||||
"login",
|
||||
user="alice",
|
||||
password="hunter2", # noqa: S106 - test fixture for redaction # pragma: allowlist secret
|
||||
api_key="sk-leak", # pragma: allowlist secret
|
||||
nested={"authorization": "Bearer xyz", "safe": "ok"}, # pragma: allowlist secret
|
||||
),
|
||||
)
|
||||
rec = records[-1]
|
||||
assert rec["password"] == "***" # noqa: S105
|
||||
assert rec["api_key"] == "***"
|
||||
assert rec["nested"]["authorization"] == "***"
|
||||
assert rec["nested"]["safe"] == "ok"
|
||||
assert rec["user"] == "alice"
|
||||
|
||||
def test_logger_name_in_json_output(self, capsys):
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("my.service.module")
|
||||
records = self._emit_and_parse(capsys, lambda: log.info("hello"))
|
||||
assert records[-1]["logger"] == "my.service.module"
|
||||
|
||||
def test_stdlib_intercept_forwards_exception(self, capsys):
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
stdlib = logging.getLogger("third_party_lib")
|
||||
|
||||
def emit():
|
||||
try:
|
||||
msg = "upstream"
|
||||
raise ConnectionError(msg)
|
||||
except ConnectionError:
|
||||
stdlib.error("call failed", exc_info=True)
|
||||
|
||||
records = self._emit_and_parse(capsys, emit)
|
||||
rec = records[-1]
|
||||
assert rec["event"] == "call failed"
|
||||
assert rec["logger"] == "third_party_lib"
|
||||
assert isinstance(rec.get("exception"), list)
|
||||
assert rec["exception"][0]["exc_type"] == "ConnectionError"
|
||||
|
||||
def test_per_logger_level_overrides_via_env(self, monkeypatch):
|
||||
monkeypatch.setenv("LANGFLOW_LOG_LEVELS", "noisy.lib=WARNING,other=ERROR")
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
assert logging.getLogger("noisy.lib").level == logging.WARNING
|
||||
assert logging.getLogger("other").level == logging.ERROR
|
||||
|
||||
def test_extra_redact_keys_via_env(self, capsys, monkeypatch):
|
||||
monkeypatch.setenv("LANGFLOW_LOG_REDACT_KEYS", "session_id,internal_key")
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("redact.extra")
|
||||
records = self._emit_and_parse(
|
||||
capsys,
|
||||
lambda: log.info("hi", session_id="abc", internal_key="xyz", safe="ok"),
|
||||
)
|
||||
rec = records[-1]
|
||||
assert rec["session_id"] == "***"
|
||||
assert rec["internal_key"] == "***"
|
||||
assert rec["safe"] == "ok"
|
||||
|
||||
def test_traceback_locals_disabled_by_default(self, capsys):
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("locals.test")
|
||||
secret_var = "sk-do-not-leak" # pragma: allowlist secret # noqa: F841, S105
|
||||
|
||||
def emit():
|
||||
try:
|
||||
msg = "boom"
|
||||
raise RuntimeError(msg)
|
||||
except RuntimeError as e:
|
||||
log.error("trace", exc_info=e) # noqa: TRY400 - exc_info kwarg path under test
|
||||
|
||||
records = self._emit_and_parse(capsys, emit)
|
||||
# The secret string must not appear anywhere in the rendered JSON.
|
||||
# That is the real security property; field renames in
|
||||
# ExceptionDictTransformer must not cause this test to pass by accident.
|
||||
rendered = json.dumps(records[-1])
|
||||
assert "sk-do-not-leak" not in rendered # pragma: allowlist secret
|
||||
|
||||
def test_traceback_locals_enabled_via_opt_in(self, capsys, monkeypatch):
|
||||
# LANGFLOW_LOG_TRACE_LOCALS=true is the explicit opt-in for local
|
||||
# debugging. Verifies the opt-in actually flips the safe default.
|
||||
monkeypatch.setenv("LANGFLOW_LOG_TRACE_LOCALS", "true")
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("locals.optin")
|
||||
|
||||
def emit():
|
||||
traceable_marker = "marker-locals-on" # noqa: F841 - must appear in frame locals
|
||||
try:
|
||||
msg = "boom"
|
||||
raise RuntimeError(msg)
|
||||
except RuntimeError as e:
|
||||
log.error("trace", exc_info=e) # noqa: TRY400 - exc_info kwarg path under test
|
||||
|
||||
records = self._emit_and_parse(capsys, emit)
|
||||
rendered = json.dumps(records[-1])
|
||||
assert "marker-locals-on" in rendered
|
||||
|
||||
def test_intercept_handler_is_idempotent(self):
|
||||
# Two configure() calls must leave exactly one InterceptHandler
|
||||
# attached, with the second call's level winning. A regression here
|
||||
# multiplies log volume in production.
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
configure(log_env="container", log_level="WARNING", cache=False)
|
||||
handlers = [h for h in logging.root.handlers if isinstance(h, InterceptHandler)]
|
||||
assert len(handlers) == 1
|
||||
assert handlers[0].level == logging.WARNING
|
||||
|
||||
def test_intercept_handler_not_installed_in_pretty_mode(self, monkeypatch):
|
||||
# Pretty/console mode must NOT route stdlib through structlog,
|
||||
# otherwise dev terminals get duplicated lines.
|
||||
monkeypatch.setenv("LANGFLOW_PRETTY_LOGS", "true")
|
||||
configure(log_env="", log_level="DEBUG", cache=False)
|
||||
handlers = [h for h in logging.root.handlers if isinstance(h, InterceptHandler)]
|
||||
assert handlers == []
|
||||
|
||||
def test_intercept_handler_forwards_stack_info(self, capsys):
|
||||
# stack_info is the new behavior added to InterceptHandler.emit -
|
||||
# locks in that stdlib `logger.error("...", stack_info=True)` survives.
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
stdlib = logging.getLogger("third_party.stack")
|
||||
records = self._emit_and_parse(
|
||||
capsys,
|
||||
lambda: stdlib.error("with stack", stack_info=True),
|
||||
)
|
||||
rec = records[-1]
|
||||
assert rec["event"] == "with stack"
|
||||
# structlog renders stack_info under the `stack` key.
|
||||
assert "stack" in rec
|
||||
assert "Stack (most recent call last)" in rec["stack"]
|
||||
|
||||
def test_intercept_handler_swallows_broken_format_args(self):
|
||||
# A buggy library calling `logger.info("user %s", a, b)` must not
|
||||
# raise into the caller from our handler. Exercise emit() directly
|
||||
# to avoid pytest's caplog also handling the same record (which
|
||||
# would re-raise the formatting error from its own handler).
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
handler = InterceptHandler()
|
||||
record = logging.LogRecord(
|
||||
name="broken.lib",
|
||||
level=logging.ERROR,
|
||||
pathname=__file__,
|
||||
lineno=1,
|
||||
msg="user %s",
|
||||
args=("alice", "extra-arg-not-allowed"),
|
||||
exc_info=None,
|
||||
)
|
||||
# Must not raise; stdlib contract is to route through handleError.
|
||||
with Path(os.devnull).open("w") as devnull, contextlib.redirect_stderr(devnull):
|
||||
handler.emit(record)
|
||||
|
||||
def test_service_info_defaults_to_langflow(self, capsys):
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("svc.default")
|
||||
records = self._emit_and_parse(capsys, lambda: log.info("hi"))
|
||||
rec = records[-1]
|
||||
assert rec["service"] == "langflow"
|
||||
# version/environment are omitted when unset.
|
||||
assert "version" not in rec
|
||||
assert "environment" not in rec
|
||||
|
||||
def test_service_info_from_env_appears_in_records(self, capsys, monkeypatch):
|
||||
monkeypatch.setenv("LANGFLOW_SERVICE_NAME", "lfx-runner")
|
||||
monkeypatch.setenv("LANGFLOW_VERSION", "1.2.3")
|
||||
monkeypatch.setenv("LANGFLOW_ENVIRONMENT", "staging")
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("svc.env")
|
||||
records = self._emit_and_parse(capsys, lambda: log.info("hi"))
|
||||
rec = records[-1]
|
||||
assert rec["service"] == "lfx-runner"
|
||||
assert rec["version"] == "1.2.3"
|
||||
assert rec["environment"] == "staging"
|
||||
|
||||
def test_malformed_log_levels_emits_warning(self, monkeypatch):
|
||||
# Typos like `WARN` instead of `WARNING` must surface, not silently
|
||||
# drop. Operators need feedback that their config didn't apply.
|
||||
monkeypatch.setenv(
|
||||
"LANGFLOW_LOG_LEVELS",
|
||||
"sqlalchemy.engine=WARN,good=INFO,broken,=NOLEVEL,empty=,a=NOTALEVEL",
|
||||
)
|
||||
with pytest.warns(UserWarning, match="LANGFLOW_LOG_LEVELS"):
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
# Valid entry still applied past the bad ones.
|
||||
assert logging.getLogger("good").level == logging.INFO
|
||||
|
||||
def test_container_csv_preserves_exception_text(self, capsys):
|
||||
# Before the fix, container_csv silently dropped exceptions. Lock in
|
||||
# that the traceback is rendered into the CSV-style row.
|
||||
configure(log_env="container_csv", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("csv.test")
|
||||
try:
|
||||
msg = "boom-csv"
|
||||
raise ValueError(msg)
|
||||
except ValueError as e:
|
||||
log.error("oh no", exc_info=e) # noqa: TRY400 - exc_info kwarg path under test
|
||||
out = capsys.readouterr().out
|
||||
assert "ValueError" in out
|
||||
assert "boom-csv" in out
|
||||
|
||||
def test_redaction_walks_lists_and_tuples(self, capsys):
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("redact.collections")
|
||||
records = self._emit_and_parse(
|
||||
capsys,
|
||||
lambda: log.info(
|
||||
"audit",
|
||||
# pragma: allowlist secret
|
||||
users_list=[{"password": "x", "name": "alice"}],
|
||||
# pragma: allowlist secret
|
||||
users_tuple=({"token": "y", "name": "bob"},),
|
||||
),
|
||||
)
|
||||
rec = records[-1]
|
||||
assert rec["users_list"][0]["password"] == "***" # noqa: S105
|
||||
assert rec["users_list"][0]["name"] == "alice"
|
||||
assert rec["users_tuple"][0]["token"] == "***" # noqa: S105
|
||||
assert rec["users_tuple"][0]["name"] == "bob"
|
||||
|
||||
def test_redaction_depth_limit_documents_contract(self, capsys):
|
||||
# The 4-level walk is a deliberate trade-off; this test pins it down
|
||||
# so a future change to the depth constant has to be intentional.
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("redact.depth")
|
||||
marker = "leak-at-5" # pragma: allowlist secret
|
||||
deep = {"l1": {"l2": {"l3": {"l4": {"password": marker}}}}}
|
||||
records = self._emit_and_parse(capsys, lambda: log.info("deep", payload=deep))
|
||||
rec = records[-1]
|
||||
# At depth 5 (beyond the limit) the password value is passed through.
|
||||
assert rec["payload"]["l1"]["l2"]["l3"]["l4"]["password"] == marker
|
||||
|
||||
def test_otel_processor_no_op_when_no_span(self, capsys):
|
||||
# OTel SDK may be installed but no span is active. Processor must
|
||||
# produce records without trace_id/span_id rather than failing.
|
||||
configure(log_env="container", log_level="DEBUG", cache=False)
|
||||
log = structlog.get_logger("otel.nospan")
|
||||
records = self._emit_and_parse(capsys, lambda: log.info("hi"))
|
||||
rec = records[-1]
|
||||
# Either keys are absent or both are present (if a real span is
|
||||
# somehow active in this process); they must never half-render.
|
||||
if "trace_id" in rec or "span_id" in rec:
|
||||
assert "trace_id" in rec
|
||||
assert "span_id" in rec
|
||||
|
||||
|
||||
class TestEdgeCasesAndErrorConditions:
|
||||
"""Test suite for edge cases and error conditions."""
|
||||
|
||||
@ -1327,3 +1581,128 @@ class TestBufferWriterBytesSerializationFix:
|
||||
finally:
|
||||
# Restore buffer state
|
||||
log_buffer.max = original_max
|
||||
|
||||
|
||||
class TestFileModeStdlibUnification:
|
||||
"""JSON file mode must route third-party stdlib logs through structlog too.
|
||||
|
||||
Regression coverage for the bug where ``LANGFLOW_LOG_ENV=container`` plus
|
||||
``LANGFLOW_LOG_FILE`` skipped the stdlib path entirely: uvicorn, sqlalchemy,
|
||||
httpx, asyncio wrote plain text straight to the file, bypassing both JSON
|
||||
rendering and PII redaction.
|
||||
"""
|
||||
|
||||
def teardown_method(self):
|
||||
for handler in logging.root.handlers[:]:
|
||||
if isinstance(handler, logging.handlers.RotatingFileHandler):
|
||||
logging.root.removeHandler(handler)
|
||||
handler.close()
|
||||
structlog.reset_defaults()
|
||||
structlog.configure()
|
||||
|
||||
@staticmethod
|
||||
def _read_records(log_file_path):
|
||||
for handler in logging.root.handlers:
|
||||
if hasattr(handler, "flush"):
|
||||
handler.flush()
|
||||
lines = [ln for ln in log_file_path.read_text().splitlines() if ln.strip()]
|
||||
# Every line must be valid JSON. Plain-text stdlib output raises here.
|
||||
return [json.loads(ln) for ln in lines]
|
||||
|
||||
def test_container_file_mode_stdlib_logs_are_json_with_logger_name(self):
|
||||
"""A third-party stdlib log lands in the file as JSON carrying its logger name."""
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
log_file_path = Path(tmp_dir) / "langflow.log"
|
||||
configure(log_env="container", log_level="INFO", log_file=log_file_path, cache=False)
|
||||
|
||||
logging.getLogger("sqlalchemy.engine").warning("connecting to pool")
|
||||
|
||||
records = self._read_records(log_file_path)
|
||||
assert records, "expected at least one JSON log line"
|
||||
sa = [r for r in records if r.get("logger") == "sqlalchemy.engine"]
|
||||
assert sa, f"sqlalchemy.engine record missing; loggers seen: {[r.get('logger') for r in records]}"
|
||||
assert sa[0]["event"] == "connecting to pool"
|
||||
assert sa[0]["level"] == "warning"
|
||||
# Service metadata is attached to stdlib records too.
|
||||
assert sa[0]["service"] == "langflow"
|
||||
|
||||
def test_container_file_mode_redacts_stdlib_extra(self):
|
||||
"""PII redaction applies to structured fields on stdlib records in file mode."""
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
log_file_path = Path(tmp_dir) / "langflow.log"
|
||||
configure(log_env="container", log_level="INFO", log_file=log_file_path, cache=False)
|
||||
|
||||
logging.getLogger("httpx").warning(
|
||||
"request sent",
|
||||
extra={"authorization": "Bearer xyz"}, # pragma: allowlist secret
|
||||
)
|
||||
|
||||
records = self._read_records(log_file_path)
|
||||
hx = [r for r in records if r.get("logger") == "httpx"]
|
||||
assert hx, f"httpx record missing; loggers seen: {[r.get('logger') for r in records]}"
|
||||
assert hx[0].get("authorization") == "***"
|
||||
|
||||
def test_container_file_mode_app_logs_still_json_and_redacted(self):
|
||||
"""Application logs keep their JSON + redaction in file mode (structlog path)."""
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
log_file_path = Path(tmp_dir) / "langflow.log"
|
||||
configure(log_env="container", log_level="INFO", log_file=log_file_path, cache=False)
|
||||
|
||||
structlog.get_logger("langflow.api").info(
|
||||
"incoming",
|
||||
api_key="sk-do-not-leak", # pragma: allowlist secret
|
||||
)
|
||||
|
||||
records = self._read_records(log_file_path)
|
||||
app = [r for r in records if r.get("logger") == "langflow.api"]
|
||||
assert app, f"app record missing; loggers seen: {[r.get('logger') for r in records]}"
|
||||
assert app[0]["event"] == "incoming"
|
||||
assert app[0].get("api_key") == "***"
|
||||
|
||||
|
||||
class TestInterceptExtraForwarding:
|
||||
"""The stdout InterceptHandler forwards stdlib `extra` fields and redacts them."""
|
||||
|
||||
def teardown_method(self):
|
||||
for handler in logging.root.handlers[:]:
|
||||
if isinstance(handler, InterceptHandler):
|
||||
logging.root.removeHandler(handler)
|
||||
structlog.reset_defaults()
|
||||
structlog.configure()
|
||||
|
||||
def test_intercept_forwards_and_redacts_stdlib_extra(self, capsys):
|
||||
"""A stdlib `extra` lands as a structured field; sensitive keys are redacted, others kept."""
|
||||
configure(log_env="container", log_level="INFO", cache=False)
|
||||
|
||||
logging.getLogger("sqlalchemy.engine").warning(
|
||||
"checking out connection",
|
||||
extra={"password": "hunter2", "pool_size": 5}, # pragma: allowlist secret
|
||||
)
|
||||
|
||||
lines = [ln for ln in capsys.readouterr().out.strip().splitlines() if ln.strip().startswith("{")]
|
||||
records = [json.loads(ln) for ln in lines]
|
||||
sa = [r for r in records if r.get("logger") == "sqlalchemy.engine"]
|
||||
assert sa, f"sqlalchemy.engine record missing; loggers seen: {[r.get('logger') for r in records]}"
|
||||
assert sa[0]["password"] == "***" # noqa: S105 - sensitive extra redacted
|
||||
assert sa[0]["pool_size"] == 5 # non-sensitive extra preserved
|
||||
|
||||
|
||||
class TestRetrievalBufferMessage:
|
||||
"""The /logs retrieval buffer must capture the actual message text, not an empty string."""
|
||||
|
||||
def teardown_method(self):
|
||||
log_buffer.max = 0
|
||||
structlog.reset_defaults()
|
||||
structlog.configure()
|
||||
|
||||
def test_buffer_captures_message_text(self):
|
||||
"""add_serialized writes the message under `message`; the buffer must read it back."""
|
||||
log_buffer.max = 100
|
||||
try:
|
||||
configure(log_env="container", log_level="INFO", cache=False)
|
||||
structlog.get_logger("demo").info("hello buffer world")
|
||||
values = list(log_buffer.get_last_n(5).values())
|
||||
assert values, "expected a buffered entry"
|
||||
assert "hello buffer world" in values
|
||||
finally:
|
||||
log_buffer.max = 0
|
||||
|
||||
@ -6,6 +6,7 @@ import logging
|
||||
import logging.handlers
|
||||
import os
|
||||
import sys
|
||||
import warnings
|
||||
from collections import deque
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
@ -20,6 +21,13 @@ from typing_extensions import NotRequired
|
||||
|
||||
from lfx.settings import DEV
|
||||
|
||||
# OpenTelemetry is optional. Resolve once at import time so the per-record
|
||||
# processor is a simple attribute check, not a repeated import attempt.
|
||||
try:
|
||||
from opentelemetry import trace as _otel_trace # type: ignore[import-not-found]
|
||||
except ImportError:
|
||||
_otel_trace = None
|
||||
|
||||
VALID_LOG_LEVELS = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
|
||||
|
||||
# Map log level names to integers
|
||||
@ -58,7 +66,9 @@ class SizedLogBuffer:
|
||||
def write(self, message: str) -> None:
|
||||
"""Write a message to the buffer."""
|
||||
record = json.loads(message)
|
||||
log_entry = record.get("event", record.get("msg", record.get("text", "")))
|
||||
# ``add_serialized`` stores the rendered text under ``message``; fall back to
|
||||
# ``event`` / ``msg`` / ``text`` for records written directly in other shapes.
|
||||
log_entry = record.get("message") or record.get("event", record.get("msg", record.get("text", "")))
|
||||
|
||||
# Extract timestamp - support both direct timestamp and nested record.time.timestamp
|
||||
timestamp = record.get("timestamp", 0)
|
||||
@ -178,14 +188,168 @@ def add_serialized(_logger: Any, _method_name: str, event_dict: dict[str, Any])
|
||||
return event_dict
|
||||
|
||||
|
||||
def remove_exception_in_production(_logger: Any, _method_name: str, event_dict: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Remove exception details in production."""
|
||||
if DEV is False:
|
||||
event_dict.pop("exception", None)
|
||||
event_dict.pop("exc_info", None)
|
||||
def _get_service_info() -> dict[str, str]:
|
||||
"""Read service metadata once so it can be injected into every log record."""
|
||||
service = os.getenv("LANGFLOW_SERVICE_NAME", "langflow")
|
||||
version = os.getenv("LANGFLOW_VERSION", "")
|
||||
environment = os.getenv("LANGFLOW_ENVIRONMENT", "")
|
||||
info = {"service": service}
|
||||
if version:
|
||||
info["version"] = version
|
||||
if environment:
|
||||
info["environment"] = environment
|
||||
return info
|
||||
|
||||
|
||||
# Default keys whose values are redacted before rendering. Production logs leak
|
||||
# auth tokens, cookies, and API keys with surprising regularity (third-party
|
||||
# clients log request bodies, dict reprs, kwargs, etc.); a cheap, default-on
|
||||
# redactor is the only thing that survives.
|
||||
DEFAULT_REDACT_KEYS: frozenset[str] = frozenset(
|
||||
{
|
||||
"password",
|
||||
"passwd",
|
||||
"secret",
|
||||
"api_key",
|
||||
"apikey",
|
||||
"token",
|
||||
"access_token",
|
||||
"refresh_token",
|
||||
"authorization",
|
||||
"auth",
|
||||
"cookie",
|
||||
"set-cookie",
|
||||
"x-api-key",
|
||||
"x-auth-token",
|
||||
}
|
||||
)
|
||||
_REDACTED = "***"
|
||||
_REDACT_MAX_DEPTH = 4
|
||||
|
||||
|
||||
def _build_redact_processor(extra_keys: frozenset[str]) -> Any:
|
||||
"""Build a structlog processor that scrubs sensitive keys.
|
||||
|
||||
Matches case-insensitively, walks nested dicts and lists up to a small
|
||||
depth, and replaces values with a fixed sentinel so logs still show the
|
||||
shape of the data without leaking the value.
|
||||
"""
|
||||
sensitive = {k.lower() for k in DEFAULT_REDACT_KEYS | extra_keys}
|
||||
|
||||
def _scrub(value: Any, depth: int) -> Any:
|
||||
if depth >= _REDACT_MAX_DEPTH:
|
||||
return value
|
||||
if isinstance(value, dict):
|
||||
return {
|
||||
k: (_REDACTED if isinstance(k, str) and k.lower() in sensitive else _scrub(v, depth + 1))
|
||||
for k, v in value.items()
|
||||
}
|
||||
if isinstance(value, list):
|
||||
return [_scrub(item, depth + 1) for item in value]
|
||||
if isinstance(value, tuple):
|
||||
return tuple(_scrub(item, depth + 1) for item in value)
|
||||
return value
|
||||
|
||||
def redact(_logger: Any, _method_name: str, event_dict: dict[str, Any]) -> dict[str, Any]:
|
||||
for key in list(event_dict.keys()):
|
||||
if isinstance(key, str) and key.lower() in sensitive:
|
||||
event_dict[key] = _REDACTED
|
||||
else:
|
||||
event_dict[key] = _scrub(event_dict[key], 1)
|
||||
return event_dict
|
||||
|
||||
return redact
|
||||
|
||||
|
||||
def add_logger_name(logger: Any, _method_name: str, event_dict: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Attach the bound logger's name as ``logger`` so Grafana can filter on it."""
|
||||
name = getattr(logger, "name", None)
|
||||
if name:
|
||||
event_dict.setdefault("logger", name)
|
||||
return event_dict
|
||||
|
||||
|
||||
class _NamedPrintLoggerFactory:
|
||||
"""Logger factory that preserves the logger name across calls.
|
||||
|
||||
structlog's default ``PrintLoggerFactory`` drops the name passed to
|
||||
``get_logger("x")``. We keep it so the ``add_logger_name`` processor can
|
||||
set the ``logger`` field on every record.
|
||||
"""
|
||||
|
||||
def __init__(self, file: Any) -> None:
|
||||
self._file = file
|
||||
|
||||
def __call__(self, *args: Any) -> structlog.PrintLogger:
|
||||
logger = structlog.PrintLogger(file=self._file)
|
||||
logger.name = args[0] if args else None
|
||||
return logger
|
||||
|
||||
|
||||
def add_otel_trace_context(_logger: Any, _method_name: str, event_dict: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Inject OpenTelemetry trace_id / span_id when a span is active.
|
||||
|
||||
OpenTelemetry is optional in lfx, so the import is resolved once at module
|
||||
load. Runtime calls are wrapped in a broad except: a misbehaving tracer
|
||||
SDK must never break logging, which is the only signal an operator has
|
||||
when the tracer itself is broken.
|
||||
"""
|
||||
if _otel_trace is None:
|
||||
return event_dict
|
||||
try:
|
||||
ctx = _otel_trace.get_current_span().get_span_context()
|
||||
except Exception: # noqa: BLE001 - logger must never break on a flaky tracer
|
||||
return event_dict
|
||||
if not ctx.is_valid:
|
||||
return event_dict
|
||||
event_dict.setdefault("trace_id", format(ctx.trace_id, "032x"))
|
||||
event_dict.setdefault("span_id", format(ctx.span_id, "016x"))
|
||||
return event_dict
|
||||
|
||||
|
||||
def _apply_logger_level_overrides() -> None:
|
||||
"""Apply ``LANGFLOW_LOG_LEVELS`` env var: ``name=LEVEL,name=LEVEL,...``.
|
||||
|
||||
Used to quiet noisy third-party loggers (``sqlalchemy.engine``, ``httpx``,
|
||||
``httpcore``, ``urllib3``) in production without changing global defaults.
|
||||
|
||||
Malformed entries (missing ``=``, unknown level, empty name) raise a
|
||||
warning instead of being silently dropped so operators see typos like
|
||||
``WARN`` instead of ``WARNING``.
|
||||
"""
|
||||
raw = os.getenv("LANGFLOW_LOG_LEVELS", "").strip()
|
||||
if not raw:
|
||||
return
|
||||
for pair in raw.split(","):
|
||||
entry = pair.strip()
|
||||
if not entry:
|
||||
continue
|
||||
if "=" not in entry:
|
||||
warnings.warn(
|
||||
f"LANGFLOW_LOG_LEVELS: ignoring {entry!r} (expected 'name=LEVEL')",
|
||||
stacklevel=2,
|
||||
)
|
||||
continue
|
||||
name, _, level = entry.partition("=")
|
||||
name = name.strip()
|
||||
level_str = level.strip().upper()
|
||||
if not name:
|
||||
warnings.warn(
|
||||
f"LANGFLOW_LOG_LEVELS: ignoring {entry!r} (empty logger name)",
|
||||
stacklevel=2,
|
||||
)
|
||||
continue
|
||||
numeric = LOG_LEVEL_MAP.get(level_str)
|
||||
if numeric is None:
|
||||
warnings.warn(
|
||||
f"LANGFLOW_LOG_LEVELS: ignoring {entry!r} (unknown level {level_str!r}, "
|
||||
f"expected one of {sorted(LOG_LEVEL_MAP)})",
|
||||
stacklevel=2,
|
||||
)
|
||||
continue
|
||||
logging.getLogger(name).setLevel(numeric)
|
||||
|
||||
|
||||
def buffer_writer(_logger: Any, _method_name: str, event_dict: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Write to log buffer if enabled."""
|
||||
if log_buffer.enabled() and "serialized" in event_dict:
|
||||
@ -227,8 +391,14 @@ def setup_loguru_logger(log_level: str, *, enqueue: bool = False) -> None:
|
||||
)
|
||||
|
||||
|
||||
def setup_log_file(log_file: Path, *, max_bytes: int) -> None:
|
||||
"""Set up Langflow's rotating file handler."""
|
||||
def setup_log_file(log_file: Path, *, max_bytes: int, formatter: logging.Formatter | None = None) -> None:
|
||||
"""Set up Langflow's rotating file handler.
|
||||
|
||||
``formatter`` lets JSON modes attach a ``structlog.stdlib.ProcessorFormatter``
|
||||
so third-party stdlib records (uvicorn, sqlalchemy, httpx, ...) are rendered
|
||||
as JSON through the same processor chain as application logs. When omitted,
|
||||
the handler writes the message verbatim (structlog has already rendered it).
|
||||
"""
|
||||
global _file_handler # noqa: PLW0603
|
||||
|
||||
if _file_handler is not None:
|
||||
@ -240,7 +410,7 @@ def setup_log_file(log_file: Path, *, max_bytes: int) -> None:
|
||||
maxBytes=max_bytes,
|
||||
backupCount=5,
|
||||
)
|
||||
_file_handler.setFormatter(logging.Formatter("%(message)s"))
|
||||
_file_handler.setFormatter(formatter if formatter is not None else logging.Formatter("%(message)s"))
|
||||
logging.root.addHandler(_file_handler)
|
||||
|
||||
|
||||
@ -296,10 +466,25 @@ def configure(
|
||||
log_format = os.getenv("LANGFLOW_LOG_FORMAT")
|
||||
|
||||
# Configure processors based on environment
|
||||
processors = [
|
||||
service_info = _get_service_info()
|
||||
|
||||
def _add_service_info(_logger: Any, _method_name: str, event_dict: dict[str, Any]) -> dict[str, Any]:
|
||||
for key, value in service_info.items():
|
||||
event_dict.setdefault(key, value)
|
||||
return event_dict
|
||||
|
||||
extra_redact = frozenset(
|
||||
k.strip().lower() for k in os.getenv("LANGFLOW_LOG_REDACT_KEYS", "").split(",") if k.strip()
|
||||
)
|
||||
redact_processor = _build_redact_processor(extra_redact)
|
||||
|
||||
processors: list[Any] = [
|
||||
structlog.contextvars.merge_contextvars,
|
||||
structlog.processors.add_log_level,
|
||||
structlog.processors.TimeStamper(fmt="iso"),
|
||||
add_logger_name,
|
||||
add_otel_trace_context,
|
||||
structlog.processors.TimeStamper(fmt="iso", utc=True),
|
||||
_add_service_info,
|
||||
]
|
||||
|
||||
# Add callsite information only when LANGFLOW_DEV is set
|
||||
@ -316,16 +501,66 @@ def configure(
|
||||
|
||||
processors.extend(
|
||||
[
|
||||
redact_processor,
|
||||
add_serialized,
|
||||
remove_exception_in_production,
|
||||
buffer_writer,
|
||||
]
|
||||
)
|
||||
|
||||
# Configure output based on environment
|
||||
if log_env.lower() == "container" or log_env.lower() == "container_json":
|
||||
processors.append(structlog.processors.JSONRenderer())
|
||||
# Configure output based on environment.
|
||||
# For machine-parseable renderers, serialize exc_info as structured tracebacks
|
||||
# so Grafana/Loki see a complete stack trace (type, value, frames) instead of
|
||||
# dropping the exception or rendering its repr. ConsoleRenderer formats
|
||||
# exc_info itself, so we don't add a tracebacks processor on that path.
|
||||
#
|
||||
# `show_locals` is OFF by default in JSON output because frame locals can
|
||||
# leak secrets (API keys, env, request bodies). Opt in with
|
||||
# LANGFLOW_LOG_TRACE_LOCALS=true when you need it for local debugging.
|
||||
show_locals = os.getenv("LANGFLOW_LOG_TRACE_LOCALS", "false").lower() == "true"
|
||||
json_traceback = structlog.processors.ExceptionRenderer(
|
||||
structlog.tracebacks.ExceptionDictTransformer(show_locals=show_locals, max_frames=50)
|
||||
)
|
||||
|
||||
# When JSON output is written to a file, render through a stdlib
|
||||
# ProcessorFormatter on the rotating handler instead of an inline
|
||||
# JSONRenderer. That routes foreign stdlib records (uvicorn, sqlalchemy,
|
||||
# httpx, asyncio) through the same renderer and the same redaction, so the
|
||||
# file is a single JSON stream and PII redaction is not bypassed -- while the
|
||||
# stdlib RotatingFileHandler keeps log rotation. Foreign records are enriched
|
||||
# by ``foreign_pre_chain``; structlog records carry the context built above
|
||||
# and are handed off via ``wrap_for_formatter``.
|
||||
file_json_formatter: logging.Formatter | None = None
|
||||
|
||||
def _append_json_tail() -> None:
|
||||
nonlocal file_json_formatter
|
||||
if log_file:
|
||||
processors.append(structlog.stdlib.ProcessorFormatter.wrap_for_formatter)
|
||||
foreign_pre_chain = [
|
||||
structlog.contextvars.merge_contextvars,
|
||||
structlog.stdlib.ExtraAdder(),
|
||||
structlog.stdlib.add_log_level,
|
||||
structlog.stdlib.add_logger_name,
|
||||
add_otel_trace_context,
|
||||
structlog.processors.TimeStamper(fmt="iso", utc=True),
|
||||
_add_service_info,
|
||||
redact_processor,
|
||||
]
|
||||
file_json_formatter = structlog.stdlib.ProcessorFormatter(
|
||||
foreign_pre_chain=foreign_pre_chain,
|
||||
processors=[
|
||||
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
|
||||
json_traceback,
|
||||
structlog.processors.JSONRenderer(),
|
||||
],
|
||||
)
|
||||
else:
|
||||
processors.append(json_traceback)
|
||||
processors.append(structlog.processors.JSONRenderer())
|
||||
|
||||
if log_env.lower() in ("container", "container_json"):
|
||||
_append_json_tail()
|
||||
elif log_env.lower() == "container_csv":
|
||||
processors.append(structlog.processors.format_exc_info)
|
||||
# Include callsite fields in key order when DEV is enabled
|
||||
key_order = ["timestamp", "level", "event"]
|
||||
if DEV:
|
||||
@ -338,11 +573,12 @@ def configure(
|
||||
if log_stdout_pretty:
|
||||
# If custom format is provided, use KeyValueRenderer with custom format
|
||||
if log_format:
|
||||
processors.append(structlog.processors.format_exc_info)
|
||||
processors.append(structlog.processors.KeyValueRenderer())
|
||||
else:
|
||||
processors.append(structlog.dev.ConsoleRenderer(colors=True))
|
||||
else:
|
||||
processors.append(structlog.processors.JSONRenderer())
|
||||
_append_json_tail()
|
||||
|
||||
# Get numeric log level
|
||||
numeric_level = LOG_LEVEL_MAP.get(log_level.upper(), logging.ERROR)
|
||||
@ -355,11 +591,17 @@ def configure(
|
||||
# Default to stdout for backward compatibility, unless output_file is specified
|
||||
log_output_file = output_file if output_file is not None else sys.stdout
|
||||
|
||||
# Wipe cached loggers before reconfiguring so any module that captured a
|
||||
# logger before the real configure() call picks up the new processor chain
|
||||
# (otherwise cache_logger_on_first_use=True binds the bootstrap chain
|
||||
# permanently to that reference).
|
||||
structlog.reset_defaults()
|
||||
|
||||
structlog.configure(
|
||||
processors=processors,
|
||||
wrapper_class=wrapper_class,
|
||||
context_class=dict,
|
||||
logger_factory=structlog.PrintLoggerFactory(file=log_output_file)
|
||||
logger_factory=_NamedPrintLoggerFactory(file=log_output_file)
|
||||
if not log_file
|
||||
else structlog.stdlib.LoggerFactory(),
|
||||
cache_logger_on_first_use=cache if cache is not None else True,
|
||||
@ -389,14 +631,29 @@ def configure(
|
||||
else:
|
||||
max_bytes = 10 * 1024 * 1024 # Default 10MB
|
||||
|
||||
# Since structlog doesn't have built-in rotation, we'll use stdlib logging for file output
|
||||
setup_log_file(log_file, max_bytes=max_bytes)
|
||||
# Since structlog doesn't have built-in rotation, we'll use stdlib logging for file output.
|
||||
# In JSON file mode the formatter renders both structlog and foreign stdlib records as JSON.
|
||||
setup_log_file(log_file, max_bytes=max_bytes, formatter=file_json_formatter)
|
||||
logging.root.setLevel(numeric_level)
|
||||
|
||||
# Set up interceptors for uvicorn and gunicorn
|
||||
setup_uvicorn_logger()
|
||||
setup_gunicorn_logger()
|
||||
|
||||
# In JSON modes we want a single unified stdout stream: every stdlib log
|
||||
# record (uvicorn access logs, sqlalchemy, httpx, langchain, asyncio)
|
||||
# routed into structlog so it comes out as JSON instead of unstructured
|
||||
# text. In non-JSON modes leave stdlib alone so dev console output stays
|
||||
# readable.
|
||||
json_mode = log_env.lower() in ("container", "container_json") or (
|
||||
not log_env and os.getenv("LANGFLOW_PRETTY_LOGS", "true").lower() != "true"
|
||||
)
|
||||
if json_mode and not log_file:
|
||||
_install_stdlib_intercept(numeric_level)
|
||||
|
||||
# Apply per-logger level overrides last so user env beats library defaults.
|
||||
_apply_logger_level_overrides()
|
||||
|
||||
# Create the global logger instance
|
||||
global logger # noqa: PLW0603
|
||||
logger = structlog.get_logger()
|
||||
@ -427,27 +684,72 @@ def setup_gunicorn_logger() -> None:
|
||||
logging.getLogger("gunicorn.access").propagate = True
|
||||
|
||||
|
||||
_STDLIB_LEVEL_TO_STRUCTLOG = (
|
||||
(logging.CRITICAL, "critical"),
|
||||
(logging.ERROR, "error"),
|
||||
(logging.WARNING, "warning"),
|
||||
(logging.INFO, "info"),
|
||||
)
|
||||
|
||||
# Attributes present on a vanilla LogRecord. Anything else in record.__dict__ was
|
||||
# attached via ``logging.*(..., extra={...})`` and is forwarded to structlog so it
|
||||
# lands as a structured field (and is therefore subject to PII redaction), mirroring
|
||||
# the ExtraAdder used on the file-mode ProcessorFormatter path.
|
||||
_RESERVED_LOGRECORD_ATTRS = frozenset(logging.makeLogRecord({}).__dict__) | {"message", "asctime"}
|
||||
|
||||
|
||||
class InterceptHandler(logging.Handler):
|
||||
"""Intercept standard logging messages and route them to structlog."""
|
||||
"""Route stdlib logging records into structlog.
|
||||
|
||||
Forwards ``exc_info`` and ``stack_info`` so library tracebacks (httpx,
|
||||
sqlalchemy, langchain, uvicorn) survive into the JSON output. Without
|
||||
this, errors raised inside third-party libraries log a one-line message
|
||||
with no stack trace.
|
||||
"""
|
||||
|
||||
def emit(self, record: logging.LogRecord) -> None:
|
||||
"""Emit a log record by passing it to structlog."""
|
||||
# Get corresponding structlog logger
|
||||
logger_name = record.name
|
||||
structlog_logger = structlog.get_logger(logger_name)
|
||||
# Mirrors the stdlib Handler.emit safety net: a malformed third-party
|
||||
# log call (e.g. mismatched %-format args) must not propagate up and
|
||||
# crash the request path. Anything that raises here is routed to
|
||||
# handleError, which is the documented contract callers expect.
|
||||
try:
|
||||
structlog_logger = structlog.get_logger(record.name)
|
||||
kwargs: dict[str, Any] = {}
|
||||
if record.exc_info:
|
||||
kwargs["exc_info"] = record.exc_info
|
||||
if record.stack_info:
|
||||
# stdlib already formats stack_info as a string. Pass it as
|
||||
# the rendered ``stack`` field directly so it survives without
|
||||
# needing StackInfoRenderer to recompute from a different frame.
|
||||
kwargs["stack"] = record.stack_info
|
||||
for key, value in record.__dict__.items():
|
||||
if key not in _RESERVED_LOGRECORD_ATTRS and not key.startswith("_") and key not in kwargs:
|
||||
kwargs[key] = value
|
||||
method_name = "debug"
|
||||
for threshold, name in _STDLIB_LEVEL_TO_STRUCTLOG:
|
||||
if record.levelno >= threshold:
|
||||
method_name = name
|
||||
break
|
||||
getattr(structlog_logger, method_name)(record.getMessage(), **kwargs)
|
||||
except Exception: # noqa: BLE001 - logging must never break the caller
|
||||
self.handleError(record)
|
||||
|
||||
# Map log levels
|
||||
level = record.levelno
|
||||
if level >= logging.CRITICAL:
|
||||
structlog_logger.critical(record.getMessage())
|
||||
elif level >= logging.ERROR:
|
||||
structlog_logger.error(record.getMessage())
|
||||
elif level >= logging.WARNING:
|
||||
structlog_logger.warning(record.getMessage())
|
||||
elif level >= logging.INFO:
|
||||
structlog_logger.info(record.getMessage())
|
||||
else:
|
||||
structlog_logger.debug(record.getMessage())
|
||||
|
||||
def _install_stdlib_intercept(numeric_level: int) -> None:
|
||||
"""Install (or refresh) the InterceptHandler on the stdlib root logger.
|
||||
|
||||
Routes every stdlib log record (uvicorn, sqlalchemy, httpx, langchain,
|
||||
asyncio, ...) into structlog so the entire process emits a single JSON
|
||||
stream. Re-runnable: a second call updates the level rather than stacking
|
||||
handlers.
|
||||
"""
|
||||
root = logging.root
|
||||
handler = next((h for h in root.handlers if isinstance(h, InterceptHandler)), None)
|
||||
if handler is None:
|
||||
handler = InterceptHandler()
|
||||
root.addHandler(handler)
|
||||
handler.setLevel(numeric_level)
|
||||
root.setLevel(numeric_level)
|
||||
|
||||
|
||||
# Initialize logger - will be reconfigured when configure() is called
|
||||
|
||||
Reference in New Issue
Block a user