fix(logging): correlate safe server diagnostics (ENG-3579)

README.md

@@ -244,3 +244,8 @@ Made with [contrib.rocks](https://contrib.rocks)
 - <a href="https://mindsdb.com/joincommunity">Community Slack</a>
 - <a href="https://mindsdb.com/press-kit?utm_medium=community&utm_source=github&utm_campaign=mindsdb%20repo">Brand guidelines</a>
 - <a href="https://mindsdb.com/contact?utm_medium=community&utm_source=github&utm_campaign=mindsdb%20repo">Contact form</a>
+
+## Diagnostics
+
+See [structured logging](docs/structured-logging.md) for defaults, troubleshooting
+overrides, retained evidence streams and rollout requirements.

docs/structured-logging.md

@@ -0,0 +1,34 @@
+# Diagnostics
+
+Application console/file diagnostics default to WARNING and JSON. `LOG_LEVEL`
+sets their shared threshold; existing `MINDSDB_CONSOLE_LOG_LEVEL` and
+`MINDSDB_FILE_LOG_LEVEL` overrides retain precedence. Temporarily set a logger
+threshold to INFO/DEBUG, reproduce, then restore WARNING. These controls do not
+change the CLI/framework debug setting.
+
+The HTTP boundary supplies UUID/32-hex `x-request-id` correlation, retained across
+the WSGI thread bridge. It emits final 4xx/5xx and streamed exceptions. Successful
+GET/HEAD `/api/util/ping` and Docker `/api/status` probes are quiet, including
+with access logging enabled. Failed probes remain visible. `ACCESS_LOG_MODE=errors` is the default;
+`all` retains ordinary successes independently of the diagnostic threshold.
+Native Uvicorn access is disabled because it contains raw targets.
+
+SQL/provider message values and arbitrary extras are excluded. OTLP receives the
+same safe JSON body and cannot separately serialize raw exception values. Verify
+that the live collector parses that JSON body before rollout. Database audit,
+query results and other protocol outputs remain their own streams. Non-HTTP
+authentication failures retain their warning diagnostics.
+
+Diagnostics use `logged_at`, `logger`, `log_level`, and `message`. Explicit events,
+reasons, source locations and correlation fields identify failures. Exception
+values, source lines, locals, arbitrary extras, HTTP queries and provider bodies
+are excluded. Use the structured fields when querying logs; the previous text
+prefixes are no longer a parsing contract.
+
+Enable extra detail only for an owned troubleshooting session, repeat the failing
+operation, and restore the default afterward. Logging controls do not enable
+framework debug responses. Roll back a release by restoring its previous image
+or package version and its previous logging settings together. Before promotion,
+verify collector parsing, saved searches, alerts and retained operational streams
+against the emitted records. A passing unit suite does not establish deployment
+or collector compatibility.

mindsdb/api/common/middleware.py

@@ -3,6 +3,7 @@
 import hmac
 import secrets
 import hashlib
+import logging
 from collections import deque
 from http import HTTPStatus
 from typing import Optional
@@ -13,6 +14,7 @@
 
 from mindsdb.utilities import log
 from mindsdb.utilities.config import config
+from mindsdb.utilities.diagnostics import request_context
 
 logger = log.getLogger(__name__)
 
@@ -180,14 +182,22 @@ def check_auth(username, password, scramble_func, salt, company_id, user_id, con
             password = password.encode()
 
         if username != hardcoded_user:
-            logger.warning(f"Check auth, user={username}: user mismatch")
+            logger.log(
+                logging.DEBUG if request_context.get() else logging.WARNING,
+                "Authentication candidate rejected",
+                extra={"event": "authentication_candidate_rejected"},
+            )
             return {"success": False}
 
         if password != hardcoded_password and password != hardcoded_password_hash:
-            logger.warning(f"check auth, user={username}: password mismatch")
+            logger.log(
+                logging.DEBUG if request_context.get() else logging.WARNING,
+                "Authentication candidate rejected",
+                extra={"event": "authentication_candidate_rejected"},
+            )
             return {"success": False}
 
-        logger.info(f"Check auth, user={username}: Ok")
+        logger.debug("Authentication candidate accepted", extra={"event": "authentication_candidate_accepted"})
         return {"success": True, "username": username, "company_id": company_id, "user_id": user_id}
     except Exception:
-        logger.exception(f"Check auth, user={username}: ERROR")
+        logger.exception("Authentication dependency failed", extra={"event": "authentication_dependency_failed"})

mindsdb/api/http/start.py

@@ -14,6 +14,7 @@
 from mindsdb.api.http.initialize import initialize_app
 from mindsdb.interfaces.storage import db
 from mindsdb.utilities import log
+from mindsdb.utilities.diagnostics import RequestLoggingMiddleware
 from mindsdb.utilities.config import config
 from mindsdb.utilities.functions import init_lexer_parsers
 from mindsdb.integrations.libs.ml_exec_base import process_cache
@@ -115,7 +116,7 @@ async def lifespan(_):
 
     # Setting logging to None makes uvicorn use the existing logging configuration
     uvicorn.run(
-        Starlette(routes=routes, lifespan=lifespan, debug=verbose),
+        RequestLoggingMiddleware(Starlette(routes=routes, lifespan=lifespan, debug=verbose)),
         host=host,
         port=int(port),
         log_level=None,

mindsdb/integrations/handlers/snowflake_handler/README.md

@@ -6,6 +6,10 @@ sidebarTitle: Snowflake
 This documentation describes the integration of MindsDB with [Snowflake](https://www.snowflake.com/en/), a cloud data warehouse used to store and analyze data.
 The integration allows MindsDB to access data stored in the Snowflake database and enhance it with AI capabilities.
 
+The shared application requirements constrain SQLAlchemy to the 2.0 series. The pinned
+`snowflake-sqlalchemy==1.9.0` imports ORM interfaces removed in SQLAlchemy 2.1;
+install the application requirements together with this handler when updating an environment.
+
 <Warning>
 **Important!**
 

mindsdb/utilities/config.py

@@ -204,12 +204,12 @@ def __new__(cls, *args, **kwargs) -> "Config":
                 "handlers": {
                     "console": {
                         "enabled": True,
-                        "formatter": "default",
-                        "level": "INFO",  # MINDSDB_CONSOLE_LOG_LEVEL or MINDSDB_LOG_LEVEL (obsolete)
+                        "formatter": "json",
+                        "level": "WARNING",  # MINDSDB_CONSOLE_LOG_LEVEL or MINDSDB_LOG_LEVEL (obsolete)
                     },
                     "file": {
                         "enabled": False,
-                        "level": "INFO",  # MINDSDB_FILE_LOG_LEVEL
+                        "level": "WARNING",  # MINDSDB_FILE_LOG_LEVEL
                         "filename": "app.log",
                         "maxBytes": 1 << 19,  # 0.5 Mb
                         "backupCount": 3,
@@ -390,6 +390,9 @@ def prepare_env_config(self) -> None:
             self._env_config["auth"]["token"] = mindsdb_auth_token
 
         # region logging
+        if "LOG_LEVEL" in os.environ:
+            for handler in ("console", "file"):
+                self._env_config["logging"]["handlers"][handler]["level"] = os.environ["LOG_LEVEL"].upper()
         if os.environ.get("MINDSDB_LOG_LEVEL", "") != "":
             self._env_config["logging"]["handlers"]["console"]["level"] = os.environ["MINDSDB_LOG_LEVEL"]
             self._env_config["logging"]["handlers"]["console"]["enabled"] = True

mindsdb/utilities/diagnostics.py

@@ -0,0 +1,155 @@
+"""Safe native logging and one diagnostic per completed HTTP request."""
+
+import contextvars
+import json
+import logging
+import os
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from time import monotonic
+from uuid import uuid4
+
+request_context = contextvars.ContextVar("request_log_context", default={})
+
+
+def exception_frames(error, seen=None):
+    seen = set() if seen is None else seen
+    if id(error) in seen:
+        return {"type": type(error).__name__}
+    seen.add(id(error))
+    frames, tb = [], error.__traceback__
+    while tb:
+        frames.append(
+            {
+                "file": Path(tb.tb_frame.f_code.co_filename).name,
+                "line": tb.tb_lineno,
+                "function": tb.tb_frame.f_code.co_name,
+            }
+        )
+        tb = tb.tb_next
+    result = {"type": type(error).__name__, "frames": frames}
+    cause = error.__cause__ or (None if error.__suppress_context__ else error.__context__)
+    if cause:
+        result["cause"] = exception_frames(cause, seen)
+    children = getattr(error, "exceptions", ())
+    if children:
+        result["exceptions"] = [exception_frames(child, seen) for child in children]
+    return result
+
+
+class DiagnosticFormatter(logging.Formatter):
+    def format(self, record):
+        # Many handler messages contain SQL, provider replies or credentials.
+        # A source location and stable event identify the failure without them.
+        event = getattr(
+            record, "event", "operation_failed" if record.levelno >= logging.WARNING else "operation_progress"
+        )
+        payload = {
+            "logged_at": datetime.fromtimestamp(record.created, timezone.utc).isoformat(),
+            "logger": record.name,
+            "log_level": record.levelname,
+            "message": event,
+            "event": event,
+            "operation": record.funcName,
+            "source_file": Path(record.pathname).name,
+            "source_line": record.lineno,
+            **request_context.get(),
+        }
+        for key in ("status_code", "duration_ms", "reason"):
+            value = getattr(record, key, None)
+            if isinstance(value, (str, int, float)):
+                payload[key] = value
+        if record.exc_info and record.exc_info[1]:
+            payload["exception"] = exception_frames(record.exc_info[1])
+        return json.dumps(payload, separators=(",", ":"))
+
+
+class SafeTelemetryFilter(logging.Filter):
+    """The OTLP SDK otherwise adds raw exception values independently of format()."""
+
+    def filter(self, record):
+        record.msg = DiagnosticFormatter().format(record)
+        record.args = ()
+        record.exc_info = record.exc_text = record.stack_info = None
+        # OTLP exports arbitrary LogRecord extras by default.
+        standard = logging.makeLogRecord({}).__dict__
+        for key in list(record.__dict__):
+            if key not in standard:
+                del record.__dict__[key]
+        return True
+
+
+class ServerDiagnosticFilter(logging.Filter):
+    def filter(self, record):
+        return not (record.exc_info and getattr(record.exc_info[1], "_data_vault_logged", False))
+
+
+class RequestLoggingMiddleware:
+    def __init__(self, app):
+        self.app = app
+
+    async def __call__(self, scope, receive, send):
+        if scope["type"] != "http":
+            return await self.app(scope, receive, send)
+        supplied = dict(scope.get("headers", [])).get(b"x-request-id", b"").decode("latin-1")
+        request_id = (
+            supplied
+            if re.fullmatch(r"[0-9a-fA-F]{32}|[0-9a-fA-F]{8}(?:-[0-9a-fA-F]{4}){3}-[0-9a-fA-F]{12}", supplied)
+            else str(uuid4())
+        )
+        method = (
+            scope["method"]
+            if scope["method"] in {"GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"}
+            else "OTHER"
+        )
+        token = request_context.set({"request_id": request_id, "method": method})
+        status, failed, start = 500, False, monotonic()
+
+        async def send_logged(message):
+            nonlocal status
+            if message["type"] == "http.response.start":
+                status = message["status"]
+                message = {
+                    **message,
+                    "headers": [(k, v) for k, v in message.get("headers", []) if k.lower() != b"x-request-id"]
+                    + [(b"x-request-id", request_id.encode("ascii"))],
+                }
+            await send(message)
+
+        try:
+            await self.app(scope, receive, send_logged)
+        except Exception as error:
+            failed = True
+            logging.getLogger("mindsdb.http.request").exception(
+                "request_failed", extra={"event": "request_failed", "status_code": 500}
+            )
+            error._data_vault_logged = True
+            raise
+        finally:
+            try:
+                if not failed:
+                    probe = (
+                        method in {"GET", "HEAD"}
+                        and scope["path"] in {"/api/util/ping", "/api/status"}
+                        and 200 <= status < 300
+                    )
+                    fields = {"status_code": status, "duration_ms": round((monotonic() - start) * 1000, 2)}
+                    if status >= 400:
+                        fields.update(
+                            event="request_refused" if status < 500 else "request_failed",
+                            reason="authentication_required"
+                            if status == 401
+                            else "permission_denied"
+                            if status == 403
+                            else "http_error",
+                        )
+                        logging.getLogger("mindsdb.http.request").log(
+                            logging.ERROR if status >= 500 else logging.WARNING, fields["event"], extra=fields
+                        )
+                    elif not probe and os.getenv("ACCESS_LOG_MODE", "errors") == "all":
+                        logging.getLogger("mindsdb.http.access").info(
+                            "http_access", extra={**fields, "event": "http_access"}
+                        )
+            finally:
+                request_context.reset(token)