如何限制Django与Celery日志量，保留错误日志并限制每秒日志条数？

你提出的日志规则完全可落地，无需侵入业务代码，通过Python标准日志库的自定义过滤器即可实现「ERROR及以上级别错误日志全量留存、INFO/DEBUG等非错误日志每秒最多存储5条」的效果，同时覆盖Django请求访问日志、Celery任务执行日志两类冗余日志场景。

Python logging原生支持过滤器扩展，我们只需要写一个通用过滤器，对错误级别日志直接放行，非错误日志按1秒时间窗口做计数限流即可。
在你的项目代码目录下新建utils/log_filters.py文件，写入以下代码：

import logging
import time
from collections import deque

class RateLimitFilter(logging.Filter):
    def __init__(self, max_per_second=5):
        super().__init__()
        self.max_per_second = max_per_second
        self.log_timestamps = deque()
        # ERROR及以上级别为错误日志，直接放行
        self.error_threshold = logging.ERROR

    def filter(self, record):
        if record.levelno >= self.error_threshold:
            return True
        # 非错误日志走滑动窗口限流
        now = time.time()
        # 清理1秒窗口外的过期时间戳
        while self.log_timestamps and self.log_timestamps[0] < now - 1:
            self.log_timestamps.popleft()
        # 未到阈值则放行，记录当前时间戳
        if len(self.log_timestamps) < self.max_per_second:
            self.log_timestamps.append(now)
            return True
        # 超过阈值直接丢弃日志
        return False

修改Django项目的settings.py文件，调整LOGGING配置，将限流过滤器绑定到日志输出处理器上，覆盖所有Django相关日志（包括你看到的POST格式访问请求日志）：

LOGGING = {
    "version": 1,
    "disable_existing_loggers": False,
    "filters": {
        "rate_limit": {
            "()": "backend.utils.log_filters.RateLimitFilter",
            "max_per_second": 5,
        }
    },
    "formatters": {
        "standard": {
            "format": "%(levelname)s %(asctime)s %(module)s %(message)s"
        },
    },
    "handlers": {
        "app_log": {
            "level": "INFO",
            "class": "logging.handlers.RotatingFileHandler",
            "filename": "/var/log/app/django.log",
            "formatter": "standard",
            "filters": ["rate_limit"],
            "maxBytes": 1024*1024*100,
            "backupCount": 10,
        },
    },
    "loggers": {
        "django": {
            "handlers": ["app_log"],
            "level": "INFO",
            "propagate": True,
        },
        # 单独绑定请求访问日志对应的logger
        "django.request": {
            "handlers": ["app_log"],
            "level": "INFO",
            "propagate": False,
        }
    },
}

如果你用Gunicorn、uWSGI等Web服务器托管Django，需要给Web服务器的访问日志处理器也加上同一个限流过滤器，不然服务器层面输出的access log不会被限流。

Celery默认复用Python标准日志体系，直接在Celery初始化文件（一般是项目根目录的celery.py）中，给Celery相关logger绑定同一个限流过滤器即可，覆盖你看到的任务接收、任务执行成功类日志：

import logging
from logging.handlers import RotatingFileHandler
from celery import Celery
from .utils.log_filters import RateLimitFilter

app = Celery("backend")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

@app.on_after_configure.connect
def setup_celery_logging(**kwargs):
    celery_logger = logging.getLogger("celery")
    log_handler = RotatingFileHandler(
        "/var/log/app/celery.log",
        maxBytes=1024*1024*100,
        backupCount=10
    )
    log_handler.setFormatter(logging.Formatter("%(levelname)s:%(name)s:%(message)s"))
    log_handler.addFilter(RateLimitFilter(max_per_second=5))
    celery_logger.addHandler(log_handler)
    celery_logger.setLevel(logging.INFO)

• 上述实现是单进程维度限流：如果你启动了多个Django服务进程、多个Celery worker进程，每个进程会独立按照每秒5条的阈值输出非错误日志，总日志量会随进程数同步上升。如果需要全局精准限流，把过滤器内的时间窗口计数逻辑替换为Redis共享计数即可，绝大多数场景下单进程限流已经能压减99%以上的冗余日志。
• 配置完成后可以做简单验证：循环打印100条INFO级别日志+20条ERROR级别日志，检查输出结果是否符合「INFO日志每秒仅保留5条、ERROR日志全部留存」的规则。
• 如果你已经接入了第三方日志采集服务，也可以不在应用层做限流，直接在日志采集规则侧配置采样策略：ERROR级别全量采集，INFO/DEBUG级别按每秒5条的阈值采样，效果完全一致，还不用修改应用代码。

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