如何为Django构建自定义ELK JSON日志器以捕获请求、响应与性能数据

Hey there! Let's break down your problem step by step since you're looking to build a custom JSON logger for ELK that captures requests, responses, and performance data in Django + DRF. I've built similar audit logging setups before, so here's what I recommend:

First, let's clear up the middleware vs decorator question:

• Decorators work great for targeting individual views, but they can't provide global coverage. You'd have to manually apply them to every view (including DRF ViewSets, which requires extra work with @method_decorator), and they won't catch requests like static files or unhandled routes.
• Middleware runs for every incoming request and outgoing response in your Django app—perfect for audit logging, where you need full visibility into all traffic. It's also fully compatible with DRF once you account for a few edge cases.

Here's a complete, tested middleware implementation that captures all the data you need, handles DRF-specific behavior, and formats logs for ELK:

import time
import json
from django.utils.deprecation import MiddlewareMixin
from django.http import HttpResponse
import logging

# Initialize a dedicated logger for ELK audit logs
elk_audit_logger = logging.getLogger('elk_audit_logger')

class AuditLogMiddleware(MiddlewareMixin):
    def process_request(self, request):
        # Record start time for performance tracking
        request.audit_start_time = time.time()
        
        # Capture core request metadata
        self.request_data = {
            'http_method': request.method,
            'path': request.path,
            'query_params': dict(request.GET),
            'user': str(request.user) if request.user.is_authenticated else 'anonymous',
            'remote_ip': request.META.get('REMOTE_ADDR'),
            'user_agent': request.META.get('HTTP_USER_AGENT', 'unknown')
        }

        # Safely capture request body (critical for DRF JSON requests)
        # Note: request.body can only be read once, so we cache it if needed
        if request.content_type in ['application/json', 'application/x-www-form-urlencoded']:
            try:
                # Cache the body so DRF can still read it later
                request._cached_body = request.body
                self.request_data['request_body'] = json.loads(request._cached_body)
            except json.JSONDecodeError:
                self.request_data['request_body'] = 'Invalid JSON payload'
            except UnicodeDecodeError:
                self.request_data['request_body'] = 'Non-text payload (e.g., file upload)'

    def process_response(self, request, response):
        # Calculate request duration in milliseconds
        duration_ms = round((time.time() - request.audit_start_time) * 1000, 2)

        # Capture response data, with DRF-specific handling
        response_data = {
            'status_code': response.status_code,
            'content_type': response.get('Content-Type', 'unknown')
        }

        # Handle DRF Response objects (they have a .data attribute)
        if hasattr(response, 'data'):
            response_data['response_body'] = response.data
        else:
            # For regular Django HttpResponse, parse content if possible
            try:
                response_data['response_body'] = json.loads(response.content)
            except (json.JSONDecodeError, UnicodeDecodeError):
                response_data['response_body'] = 'Non-JSON response (truncated)'

        # Assemble final audit log in ELK-friendly JSON format
        audit_log_entry = {
            '@timestamp': time.strftime('%Y-%m-%dT%H:%M:%S.%fZ'),
            'request': self.request_data,
            'response': response_data,
            'performance': {
                'duration_ms': duration_ms
            }
        }

        # Log the entry (ensure your logger is configured for JSON output)
        elk_audit_logger.info(json.dumps(audit_log_entry))

        return response

Key DRF Compatibility Notes #

• Request Body Handling: DRF reads request.body internally, so we cache it to request._cached_body to avoid breaking DRF's parsing logic.
• Response Data: DRF's Response objects expose a .data attribute (the raw Python data before rendering), which is easier to log than parsing the rendered content.
• Edge Cases: We skip full body logging for non-text payloads (like file uploads) to keep logs manageable.

Add this to your settings.py to set up a JSON-formatted logger that outputs to a file (or directly to Logstash if you prefer):

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'formatters': {
        'elk_json': {
            '()': 'pythonjsonlogger.jsonlogger.JsonFormatter',
            'format': '%(asctime)s %(levelname)s %(message)s',
        },
    },
    'handlers': {
        'elk_file_handler': {
            'level': 'INFO',
            'class': 'logging.FileHandler',
            'filename': '/var/log/django/elk_audit.log',
            'formatter': 'elk_json',
        },
        # Optional: Send directly to Logstash using python-logstash
        # 'logstash_handler': {
        #     'level': 'INFO',
        #     'class': 'logstash.LogstashHandler',
        #     'host': 'your-elk-server-ip',
        #     'port': 5959,
        #     'version': 1,
        # },
    },
    'loggers': {
        'elk_audit_logger': {
            'handlers': ['elk_file_handler'],  # Add 'logstash_handler' here if using it
            'level': 'INFO',
            'propagate': False,
        },
    },
}

If you only need audit logs for specific views (not the entire app), a decorator works. Here's a quick example:

import time
import json
import logging
from functools import wraps
from django.utils.decorators import method_decorator

elk_audit_logger = logging.getLogger('elk_audit_logger')

def audit_log_view(view_func):
    @wraps(view_func)
    def wrapper(request, *args, **kwargs):
        start_time = time.time()
        # Capture request data (similar to middleware)
        request_data = {'http_method': request.method, 'path': request.path, ...}
        # Run the view
        response = view_func(request, *args, **kwargs)
        # Calculate duration and capture response
        duration_ms = round((time.time() - start_time)*1000,2)
        response_data = {'status_code': response.status_code, 'body': response.data if hasattr(response, 'data') else response.content}
        # Log the entry
        elk_audit_logger.info(json.dumps({'@timestamp': time.strftime('%Y-%m-%dT%H:%M:%S.%fZ'), 'request': request_data, 'response': response_data, 'duration_ms': duration_ms}))
        return response
    return wrapper

# Apply to a function-based view
@audit_log_view
def my_view(request):
    ...

# Apply to a DRF ViewSet method
class MyViewSet(viewsets.ModelViewSet):
    @method_decorator(audit_log_view)
    def create(self, request, *args, **kwargs):
        ...

But again, this is not ideal for full audit coverage—middleware is the way to go for your use case.

内容的提问来源于stack exchange，提问作者Grant Zukel