events: initial log rework

Signed-off-by: Jens Langhammer <jens@goauthentik.io>

authentik/blueprints/v1/importer.py

@@ -19,8 +19,6 @@
 from rest_framework.exceptions import ValidationError
 from rest_framework.serializers import BaseSerializer, Serializer
 from structlog.stdlib import BoundLogger, get_logger
-from structlog.testing import capture_logs
-from structlog.types import EventDict
 from yaml import load
 
 from authentik.blueprints.v1.common import (
@@ -42,6 +40,7 @@
 from authentik.enterprise.license import LicenseKey
 from authentik.enterprise.models import LicenseUsage
 from authentik.enterprise.providers.rac.models import ConnectionToken
+from authentik.events.logs import LogEvent, capture_logs
 from authentik.events.models import SystemTask
 from authentik.events.utils import cleanse_dict
 from authentik.flows.models import FlowToken, Stage
@@ -376,7 +375,7 @@ def _apply_models(self, raise_errors=False) -> bool:
                 self.logger.debug("entry to delete with no instance, skipping")
         return True
 
-    def validate(self, raise_validation_errors=False) -> tuple[bool, list[EventDict]]:
+    def validate(self, raise_validation_errors=False) -> tuple[bool, list[LogEvent]]:
         """Validate loaded blueprint export, ensure all models are allowed
         and serializers have no errors"""
         self.logger.debug("Starting blueprint import validation")
@@ -390,9 +389,7 @@ def validate(self, raise_validation_errors=False) -> tuple[bool, list[EventDict]
         ):
             successful = self._apply_models(raise_errors=raise_validation_errors)
             if not successful:
-                self.logger.debug("Blueprint validation failed")
-        for log in logs:
-            getattr(self.logger, log.get("log_level"))(**log)
+                self.logger.warning("Blueprint validation failed")
         self.logger.debug("Finished blueprint import validation")
         self._import = orig_import
         return successful, logs

authentik/blueprints/v1/tasks.py

@@ -211,7 +211,7 @@ def apply_blueprint(self: SystemTask, instance_pk: str):
         if not valid:
             instance.status = BlueprintInstanceStatus.ERROR
             instance.save()
-            self.set_status(TaskStatus.ERROR, *[x["event"] for x in logs])
+            self.set_status(TaskStatus.ERROR, logs)
             return
         applied = importer.apply()
         if not applied:

authentik/core/api/applications.py

@@ -20,15 +20,14 @@
 from rest_framework.serializers import ModelSerializer
 from rest_framework.viewsets import ModelViewSet
 from structlog.stdlib import get_logger
-from structlog.testing import capture_logs
 
 from authentik.admin.api.metrics import CoordinateSerializer
 from authentik.blueprints.v1.importer import SERIALIZER_CONTEXT_BLUEPRINT
 from authentik.core.api.providers import ProviderSerializer
 from authentik.core.api.used_by import UsedByMixin
 from authentik.core.models import Application, User
+from authentik.events.logs import LogEventSerializer, capture_logs
 from authentik.events.models import EventAction
-from authentik.events.utils import sanitize_dict
 from authentik.lib.utils.file import (
     FilePathSerializer,
     FileUploadSerializer,
@@ -182,9 +181,9 @@ def check_access(self, request: Request, slug: str) -> Response:
         if request.user.is_superuser:
             log_messages = []
             for log in logs:
-                if log.get("process", "") == "PolicyProcess":
+                if log.attributes.get("process", "") == "PolicyProcess":
                     continue
-                log_messages.append(sanitize_dict(log))
+                log_messages.append(LogEventSerializer(log).data)
             result.log_messages = log_messages
             response = PolicyTestResultSerializer(result)
         return Response(response.data)

authentik/events/api/tasks.py

@@ -12,7 +12,6 @@
     ChoiceField,
     DateTimeField,
     FloatField,
-    ListField,
     SerializerMethodField,
 )
 from rest_framework.request import Request
@@ -21,6 +20,7 @@
 from rest_framework.viewsets import ReadOnlyModelViewSet
 from structlog.stdlib import get_logger
 
+from authentik.events.logs import LogEventSerializer
 from authentik.events.models import SystemTask, TaskStatus
 from authentik.rbac.decorators import permission_required
 
@@ -39,7 +39,7 @@ class SystemTaskSerializer(ModelSerializer):
     duration = FloatField(read_only=True)
 
     status = ChoiceField(choices=[(x.value, x.name) for x in TaskStatus])
-    messages = ListField(child=CharField())
+    messages = LogEventSerializer(many=True)
 
     def get_full_name(self, instance: SystemTask) -> str:
         """Get full name with UID"""

authentik/events/logs.py

@@ -0,0 +1,72 @@
+from collections.abc import Generator
+from contextlib import contextmanager
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any
+
+from django.utils.timezone import now
+from rest_framework.fields import CharField, ChoiceField, DateTimeField, DictField
+from structlog import configure, get_config
+from structlog.stdlib import NAME_TO_LEVEL, ProcessorFormatter
+from structlog.testing import LogCapture
+from structlog.types import EventDict
+
+from authentik.core.api.utils import PassiveSerializer
+from authentik.events.utils import sanitize_dict
+
+
+@dataclass()
+class LogEvent:
+
+    event: str
+    log_level: str
+    logger: str
+    timestamp: datetime = field(default_factory=now)
+    attributes: dict[str, Any] = field(default_factory=dict)
+
+    @staticmethod
+    def from_event_dict(item: EventDict) -> "LogEvent":
+        event = item.pop("event")
+        log_level = item.pop("level").lower()
+        timestamp = datetime.fromisoformat(item.pop("timestamp"))
+        item.pop("pid", None)
+        # Sometimes log entries have both `level` and `log_level` set, but `level` is always set
+        item.pop("log_level", None)
+        return LogEvent(
+            event, log_level, item.pop("logger"), timestamp, attributes=sanitize_dict(item)
+        )
+
+
+class LogEventSerializer(PassiveSerializer):
+    """Single log message with all context logged."""
+
+    timestamp = DateTimeField()
+    log_level = ChoiceField(choices=tuple((x, x) for x in NAME_TO_LEVEL.keys()))
+    logger = CharField()
+    event = CharField()
+    attributes = DictField()
+
+
+@contextmanager
+def capture_logs(log_default_output=True) -> Generator[list[LogEvent], None, None]:
+    """Capture log entries created"""
+    logs = []
+    cap = LogCapture()
+    # Modify `_Configuration.default_processors` set via `configure` but always
+    # keep the list instance intact to not break references held by bound
+    # loggers.
+    processors: list = get_config()["processors"]
+    old_processors = processors.copy()
+    try:
+        # clear processors list and use LogCapture for testing
+        processors.remove(ProcessorFormatter.wrap_for_formatter)
+        processors.append(cap)
+        configure(processors=processors)
+        yield logs
+        for raw_log in cap.entries:
+            logs.append(LogEvent.from_event_dict(raw_log))
+    finally:
+        # remove LogCapture and restore original processors
+        processors.clear()
+        processors.extend(old_processors)
+        configure(processors=processors)

authentik/events/system_tasks.py

@@ -9,6 +9,7 @@
 from structlog.stdlib import get_logger
 from tenant_schemas_celery.task import TenantTask
 
+from authentik.events.logs import LogEvent
 from authentik.events.models import Event, EventAction, TaskStatus
 from authentik.events.models import SystemTask as DBSystemTask
 from authentik.events.utils import sanitize_item
@@ -24,7 +25,7 @@ class SystemTask(TenantTask):
     save_on_success: bool
 
     _status: TaskStatus
-    _messages: list[str]
+    _messages: list[LogEvent]
 
     _uid: str | None
     # Precise start time from perf_counter
@@ -44,15 +45,20 @@ def set_uid(self, uid: str):
         """Set UID, so in the case of an unexpected error its saved correctly"""
         self._uid = uid
 
-    def set_status(self, status: TaskStatus, *messages: str):
+    def set_status(self, status: TaskStatus, *messages: LogEvent):
         """Set result for current run, will overwrite previous result."""
         self._status = status
-        self._messages = messages
+        self._messages = list(messages)
+        for idx, msg in enumerate(self._messages):
+            if not isinstance(msg, LogEvent):
+                self._messages[idx] = LogEvent(msg, logger=self.__name__, log_level="info")
 
     def set_error(self, exception: Exception):
         """Set result to error and save exception"""
         self._status = TaskStatus.ERROR
-        self._messages = [exception_to_string(exception)]
+        self._messages = [
+            LogEvent(exception_to_string(exception), logger=self.__name__, log_level="error")
+        ]
 
     def before_start(self, task_id, args, kwargs):
         self._start_precise = perf_counter()
@@ -98,8 +104,7 @@ def after_return(self, status, retval, task_id, args: list[Any], kwargs: dict[st
     def on_failure(self, exc, task_id, args, kwargs, einfo):
         super().on_failure(exc, task_id, args, kwargs, einfo=einfo)
         if not self._status:
-            self._status = TaskStatus.ERROR
-            self._messages = exception_to_string(exc)
+            self.set_error(exc)
         DBSystemTask.objects.update_or_create(
             name=self.__name__,
             uid=self._uid,

authentik/outposts/controllers/base.py

@@ -3,9 +3,9 @@
 from dataclasses import dataclass
 
 from structlog.stdlib import get_logger
-from structlog.testing import capture_logs
 
 from authentik import __version__, get_build_hash
+from authentik.events.logs import capture_logs
 from authentik.lib.config import CONFIG
 from authentik.lib.sentry import SentryIgnoredException
 from authentik.outposts.models import (

authentik/outposts/controllers/kubernetes.py

@@ -9,10 +9,10 @@
 from kubernetes.config.config_exception import ConfigException
 from kubernetes.config.incluster_config import load_incluster_config
 from kubernetes.config.kube_config import load_kube_config_from_dict
-from structlog.testing import capture_logs
 from urllib3.exceptions import HTTPError
 from yaml import dump_all
 
+from authentik.events.logs import capture_logs
 from authentik.outposts.controllers.base import BaseClient, BaseController, ControllerException
 from authentik.outposts.controllers.k8s.base import KubernetesObjectReconciler
 from authentik.outposts.controllers.k8s.deployment import DeploymentReconciler

authentik/policies/api/exec.py

@@ -1,10 +1,11 @@
 """Serializer for policy execution"""
 
-from rest_framework.fields import BooleanField, CharField, DictField, ListField
+from rest_framework.fields import BooleanField, CharField, ListField
 from rest_framework.relations import PrimaryKeyRelatedField
 
 from authentik.core.api.utils import JSONDictField, PassiveSerializer
 from authentik.core.models import User
+from authentik.events.logs import LogEventSerializer
 
 
 class PolicyTestSerializer(PassiveSerializer):
@@ -19,4 +20,4 @@ class PolicyTestResultSerializer(PassiveSerializer):
 
     passing = BooleanField()
     messages = ListField(child=CharField(), read_only=True)
-    log_messages = ListField(child=DictField(), read_only=True)
+    log_messages = LogEventSerializer(many=True, read_only=True)

authentik/policies/api/policies.py

@@ -11,12 +11,11 @@
 from rest_framework.serializers import ModelSerializer, SerializerMethodField
 from rest_framework.viewsets import GenericViewSet
 from structlog.stdlib import get_logger
-from structlog.testing import capture_logs
 
 from authentik.core.api.applications import user_app_cache_key
 from authentik.core.api.used_by import UsedByMixin
 from authentik.core.api.utils import CacheSerializer, MetaNameSerializer, TypeCreateSerializer
-from authentik.events.utils import sanitize_dict
+from authentik.events.logs import LogEventSerializer, capture_logs
 from authentik.lib.utils.reflection import all_subclasses
 from authentik.policies.api.exec import PolicyTestResultSerializer, PolicyTestSerializer
 from authentik.policies.models import Policy, PolicyBinding
@@ -166,9 +165,9 @@ def test(self, request: Request, pk: str) -> Response:
             result = proc.execute()
         log_messages = []
         for log in logs:
-            if log.get("process", "") == "PolicyProcess":
+            if log.attributes.get("process", "") == "PolicyProcess":
                 continue
-            log_messages.append(sanitize_dict(log))
+            log_messages.append(LogEventSerializer(log).data)
         result.log_messages = log_messages
         response = PolicyTestResultSerializer(result)
         return Response(response.data)

authentik/policies/types.py

@@ -13,6 +13,7 @@
 
 if TYPE_CHECKING:
     from authentik.core.models import User
+    from authentik.events.logs import LogEvent
     from authentik.policies.models import PolicyBinding
 
 LOGGER = get_logger()
@@ -74,7 +75,7 @@ class PolicyResult:
     source_binding: PolicyBinding | None
     source_results: list[PolicyResult] | None
 
-    log_messages: list[dict] | None
+    log_messages: list[LogEvent] | None
 
     def __init__(self, passing: bool, *messages: str):
         self.passing = passing

schema.yml

@@ -35515,6 +35515,48 @@ components:
           type: string
       required:
       - link
+    LogEvent:
+      type: object
+      description: Single log message with all context logged.
+      properties:
+        timestamp:
+          type: string
+          format: date-time
+        log_level:
+          $ref: '#/components/schemas/LogLevelEnum'
+        logger:
+          type: string
+        event:
+          type: string
+        attributes:
+          type: object
+          additionalProperties: {}
+      required:
+      - attributes
+      - event
+      - log_level
+      - logger
+      - timestamp
+    LogLevelEnum:
+      enum:
+      - critical
+      - exception
+      - error
+      - warn
+      - warning
+      - info
+      - debug
+      - notset
+      type: string
+      description: |-
+        * `critical` - critical
+        * `exception` - exception
+        * `error` - error
+        * `warn` - warn
+        * `warning` - warning
+        * `info` - info
+        * `debug` - debug
+        * `notset` - notset
     LoginChallengeTypes:
       oneOf:
       - $ref: '#/components/schemas/RedirectChallenge'
@@ -41309,8 +41351,7 @@ components:
         log_messages:
           type: array
           items:
-            type: object
-            additionalProperties: {}
+            $ref: '#/components/schemas/LogEvent'
           readOnly: true
       required:
       - log_messages
@@ -44324,7 +44365,7 @@ components:
         messages:
           type: array
           items:
-            type: string
+            $ref: '#/components/schemas/LogEvent'
       required:
       - description
       - duration