diff --git a/README.md b/README.md index fa9fd58c89..544b0f47ce 100644 --- a/README.md +++ b/README.md @@ -244,3 +244,8 @@ Made with [contrib.rocks](https://contrib.rocks) - Community Slack - Brand guidelines - Contact form + +## Diagnostics + +See [structured logging](docs/structured-logging.md) for defaults, troubleshooting +overrides, retained evidence streams and rollout requirements. diff --git a/docs/structured-logging.md b/docs/structured-logging.md new file mode 100644 index 0000000000..d70b277d9e --- /dev/null +++ b/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. diff --git a/mindsdb/api/common/middleware.py b/mindsdb/api/common/middleware.py index 6fb9338019..1d91bf9957 100644 --- a/mindsdb/api/common/middleware.py +++ b/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"}) diff --git a/mindsdb/api/http/start.py b/mindsdb/api/http/start.py index f2373ebb11..a4f5cffd3c 100644 --- a/mindsdb/api/http/start.py +++ b/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, diff --git a/mindsdb/integrations/handlers/snowflake_handler/README.md b/mindsdb/integrations/handlers/snowflake_handler/README.md index 2adbe736c3..3c1d36f520 100644 --- a/mindsdb/integrations/handlers/snowflake_handler/README.md +++ b/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. + **Important!** diff --git a/mindsdb/utilities/config.py b/mindsdb/utilities/config.py index e3fff3bf02..bbe1b35c6e 100644 --- a/mindsdb/utilities/config.py +++ b/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 diff --git a/mindsdb/utilities/diagnostics.py b/mindsdb/utilities/diagnostics.py new file mode 100644 index 0000000000..4f8193266b --- /dev/null +++ b/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) diff --git a/mindsdb/utilities/log.py b/mindsdb/utilities/log.py index 2ae311a61d..e025c130cf 100644 --- a/mindsdb/utilities/log.py +++ b/mindsdb/utilities/log.py @@ -1,6 +1,5 @@ import re import os -import json import logging import threading from typing import Any @@ -8,6 +7,7 @@ from logging.config import dictConfig from mindsdb.utilities.config import config as app_config +from mindsdb.utilities.diagnostics import DiagnosticFormatter, ServerDiagnosticFilter # Suppress Pydantic warnings for third-party libraries # TODO: Work on a better solution to this @@ -17,45 +17,18 @@ logging_initialized = False -class JsonFormatter(logging.Formatter): - def format(self, record): - record_message = super().format(record) - log_record = { - "process_name": record.processName, - "name": record.name, - "message": record_message, - "level": record.levelname, - "time": record.created, - } - return json.dumps(log_record) - - -class ColorFormatter(logging.Formatter): - green = "\x1b[32;20m" - default = "\x1b[39;20m" - yellow = "\x1b[33;20m" - red = "\x1b[31;20m" - bold_red = "\x1b[31;1m" - reset = "\x1b[0m" - format = "%(asctime)s %(processName)15s %(levelname)-8s %(name)s: %(message)s" - - FORMATS = { - logging.DEBUG: logging.Formatter(green + format + reset), - logging.INFO: logging.Formatter(default + format + reset), - logging.WARNING: logging.Formatter(yellow + format + reset), - logging.ERROR: logging.Formatter(red + format + reset), - logging.CRITICAL: logging.Formatter(bold_red + format + reset), - } +class JsonFormatter(DiagnosticFormatter): + pass + - def format(self, record): - log_fmt = self.FORMATS.get(record.levelno) - return log_fmt.format(record) +class ColorFormatter(DiagnosticFormatter): + """Legacy configuration name; diagnostics use the same safe envelope.""" FORMATTERS = { "default": {"()": ColorFormatter}, "json": {"()": JsonFormatter}, - "file": {"format": "%(asctime)s %(processName)15s %(levelname)-8s %(name)s: %(message)s"}, + "file": {"()": JsonFormatter}, } @@ -174,13 +147,15 @@ def sanitize_record(self, record): class StreamSanitizingHandler(SanitizingMixin, logging.StreamHandler): def emit(self, record): - record = self.sanitize_record(record) + if not isinstance(self.formatter, DiagnosticFormatter): + record = self.sanitize_record(record) super().emit(record) class FileSanitizingHandler(SanitizingMixin, logging.handlers.RotatingFileHandler): def emit(self, record): - record = self.sanitize_record(record) + if not isinstance(self.formatter, DiagnosticFormatter): + record = self.sanitize_record(record) super().emit(record) @@ -237,14 +212,26 @@ def get_handlers_config(process_name: str) -> dict: def configure_logging(process_name: str = None): handlers_config = get_handlers_config(process_name) mindsdb_log_level = get_mindsdb_log_level() + application_handlers = list(handlers_config) + handlers_config["access"] = {"class": "logging.StreamHandler", "formatter": "json", "level": "INFO"} logging_config = dict( version=1, + disable_existing_loggers=False, + filters={"server_diagnostic": {"()": ServerDiagnosticFilter}}, formatters=FORMATTERS, handlers=handlers_config, loggers={ + "uvicorn.access": {"handlers": [], "propagate": False}, + "uvicorn.error": { + "handlers": application_handlers, + "level": mindsdb_log_level, + "propagate": False, + "filters": ["server_diagnostic"], + }, + "mindsdb.http.access": {"handlers": ["access"], "level": "INFO", "propagate": False}, "": { # root logger - "handlers": list(handlers_config.keys()), + "handlers": application_handlers, "level": mindsdb_log_level, }, "__main__": { diff --git a/mindsdb/utilities/otel/logger.py b/mindsdb/utilities/otel/logger.py index 5836fb3179..9f6acf0ef3 100644 --- a/mindsdb/utilities/otel/logger.py +++ b/mindsdb/utilities/otel/logger.py @@ -7,6 +7,7 @@ from opentelemetry.sdk.resources import Resource from mindsdb.utilities.log import get_mindsdb_log_level +from mindsdb.utilities.diagnostics import SafeTelemetryFilter def setup_logger(resource: Resource, exporter: LogExporter) -> None: @@ -21,5 +22,7 @@ def setup_logger(resource: Resource, exporter: LogExporter) -> None: logger_provider.add_log_record_processor(BatchLogRecordProcessor(exporter)) handler = LoggingHandler(level=mindsdb_log_level, logger_provider=logger_provider) + handler.addFilter(SafeTelemetryFilter()) + # Attach OTLP handler to root logger logging.getLogger().addHandler(handler) diff --git a/requirements/requirements.txt b/requirements/requirements.txt index 3a1542cf0f..8db86c6b14 100644 --- a/requirements/requirements.txt +++ b/requirements/requirements.txt @@ -7,7 +7,7 @@ python-multipart == 0.0.26 cryptography>=46.0.5 psycopg[binary] psutil~=7.0 -sqlalchemy >= 2.0.0, < 3.0.0 +sqlalchemy >= 2.0.0, < 2.1.0 # Snowflake dialect 1.9 imports ORM interfaces removed in 2.1 psycopg2-binary # This is required for using sqlalchemy with postgres alembic >= 1.3.3 redis==6.4.0 diff --git a/tests/unit/utilities/test_diagnostics.py b/tests/unit/utilities/test_diagnostics.py new file mode 100644 index 0000000000..c08e11e5dc --- /dev/null +++ b/tests/unit/utilities/test_diagnostics.py @@ -0,0 +1,154 @@ +import asyncio +import io +import json +import logging + +import httpx +from a2wsgi import WSGIMiddleware +from flask import Flask +from starlette.applications import Starlette +from starlette.responses import Response, StreamingResponse +from starlette.routing import Mount, Route + +from mindsdb.utilities import log +from mindsdb.utilities.diagnostics import ( + DiagnosticFormatter, + RequestLoggingMiddleware, + SafeTelemetryFilter, + request_context, +) + + +def test_exception_and_telemetry_privacy(): + try: + try: + raise ValueError("private-query SELECT private-value") + except ValueError as cause: + raise RuntimeError("private-token") from cause + except RuntimeError: + record = logging.getLogger("mindsdb.test").makeRecord( + "mindsdb.test", + logging.ERROR, + __file__, + 12, + "private-provider-body %s", + ("private-key",), + __import__("sys").exc_info(), + extra={"private_extra": "private-body"}, + ) + output = DiagnosticFormatter().format(record) + event = json.loads(output) + assert "private-" not in output + assert event["exception"]["type"] == "RuntimeError" + assert event["exception"]["cause"]["type"] == "ValueError" + assert event["exception"]["frames"][0]["function"] == "test_exception_and_telemetry_privacy" + SafeTelemetryFilter().filter(record) + assert "private-" not in str(record.__dict__) + assert record.exc_info is None + assert json.loads(record.getMessage())["exception"]["type"] == "RuntimeError" + + +def test_installed_handlers_are_quiet_and_safe(capsys): + log.configure_logging() + logging.getLogger("mindsdb.test").info("ordinary_success") + assert capsys.readouterr().err == "" + logging.getLogger("mindsdb.test").warning("private-provider-reply") + event = json.loads(capsys.readouterr().err) + assert event["log_level"] == "WARNING" + assert "private-" not in str(event) + logging.getLogger("uvicorn.access").info("private-path?key=private-token") + assert capsys.readouterr().err == "" + + +def test_http_matrix_and_wsgi_context(monkeypatch): + async def exercise(): + output = io.StringIO() + handler = logging.StreamHandler(output) + handler.setFormatter(DiagnosticFormatter()) + names = ("mindsdb.http.request", "mindsdb.http.access", "mindsdb.flask") + saved = [] + for name in names: + logger = logging.getLogger(name) + saved.append((logger, logger.handlers[:], logger.level, logger.propagate)) + logger.handlers = [handler] + logger.setLevel(logging.INFO) + logger.propagate = False + + async def endpoint(request): + return Response(status_code=int(request.path_params.get("status", request.query_params.get("status", 200)))) + + async def stream(request): + async def body(): + yield b"partial" + raise RuntimeError("private-stream-body") + + return StreamingResponse(body()) + + flask_app = Flask(__name__) + + @flask_app.get("/context") + def context(): + logging.getLogger("mindsdb.flask").warning("private-provider-body") + return "ok" + + app = RequestLoggingMiddleware( + Starlette( + routes=[ + Route("/api/util/ping", endpoint, methods=["GET", "POST"]), + Route("/api/status", endpoint, methods=["GET", "POST"]), + Route("/result/{status}", endpoint), + Route("/stream", stream), + Mount("/wsgi", WSGIMiddleware(flask_app)), + ] + ) + ) + try: + async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client: + for mode, count in (("errors", 0), ("all", 1)): + monkeypatch.setenv("ACCESS_LOG_MODE", mode) + output.seek(0) + output.truncate() + for path in ("/api/util/ping", "/api/status"): + await client.get(f"{path}?secret=private-query") + await client.head(path) + await client.get("/result/200") + assert len(output.getvalue().splitlines()) == count + output.seek(0) + output.truncate() + for path in ("/api/util/ping", "/api/status"): + await client.post(path) + await client.get(f"{path}?status=503") + await client.head(f"{path}?status=503") + events = [json.loads(line) for line in output.getvalue().splitlines()] + assert sum(event["event"] == "http_access" for event in events) == 2 + assert sum(event["event"] == "request_failed" and event["status_code"] == 503 for event in events) == 4 + output.seek(0) + output.truncate() + ids = ["11111111-1111-4111-8111-111111111111", "22222222-2222-4222-8222-222222222222"] + results = await asyncio.gather( + *(client.get("/result/403", headers={"x-request-id": rid}) for rid in ids) + ) + assert [response.headers["x-request-id"] for response in results] == ids + events = [json.loads(line) for line in output.getvalue().splitlines()] + assert {event["request_id"] for event in events} == set(ids) + assert all(event["event"] == "request_refused" for event in events) + await client.get("/wsgi/context", headers={"x-request-id": ids[0]}) + events = [json.loads(line) for line in output.getvalue().splitlines()] + assert next(event for event in events if event["logger"] == "mindsdb.flask")["request_id"] == ids[0] + try: + await client.get("/stream") + except Exception: + pass + else: + raise AssertionError("stream failure must still reach the server") + events = [json.loads(line) for line in output.getvalue().splitlines()] + assert sum(event["event"] == "request_failed" for event in events) == 1 + assert "private-" not in output.getvalue() + assert request_context.get() == {} + finally: + for logger, handlers, level, propagate in saved: + logger.handlers = handlers + logger.setLevel(level) + logger.propagate = propagate + + asyncio.run(exercise())