杰瑞科技汇

Python logging如何配置发送email告警?

Of course! Sending log messages via email is a powerful way to get notified about critical errors or important events, especially in server applications or long-running scripts. Here’s a comprehensive guide on how to set this up in Python, from a simple example to a more robust, production-ready configuration.

Python logging如何配置发送email告警?-图1
(图片来源网络,侵删)

The Core Concept

The standard Python logging module is highly flexible and extensible. Instead of just writing logs to a file or the console, you can configure a handler that sends log records to an email server. This handler is called SMTPHandler.

The general workflow is:

  1. Log an Event: Your application code calls a logging function (e.g., logger.error("Something went wrong!")).
  2. Log Record Creation: The logging module creates a LogRecord object containing all the information: the message, timestamp, logger name, level, etc.
  3. Handler Processing: The SMTPHandler intercepts this LogRecord.
  4. Email Formatting: The handler formats the record into a readable email body.
  5. Email Sending: The handler connects to an SMTP server and sends the email.

Simple Example: Sending an Error Email

This is the most basic approach. It's good for understanding the mechanics but has limitations (e.g., it sends an email for every single log message).

import logging
import logging.handlers
import smtplib
from email.message import EmailMessage
# 1. Get a logger instance
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)  # Set the lowest level of messages this logger will handle
# 2. Create an SMTPHandler
#    - mailhost: A tuple (host, port)
#    - fromaddr: The email address the message is sent from
#    - toaddrs: A list of email addresses to send to
#    - subject: The subject of the email
#    - credentials: A tuple (username, password) for SMTP authentication
#    - secure: A tuple used for SSL (e.g., ())
smtp_handler = logging.handlers.SMTPHandler(
    mailhost=('smtp.example.com', 587),
    fromaddr='app@example.com',
    toaddrs=['admin@example.com'],
    subject='My App: Log Alert',
    credentials=('your_username', 'your_password'),
    secure=()  # Use an empty tuple for STARTTLS
)
# 3. Set a formatter for the handler's output
#    This makes the email body readable.
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
smtp_handler.setFormatter(formatter)
# 4. Add the handler to the logger
logger.addHandler(smtp_handler)
# --- Now, let's generate some logs ---
logger.info("This is an informational message.")
logger.warning("This is a warning message.")
logger.error("This is an error message that will trigger an email.")
logger.critical("This is a critical message.")
# To see logs in the console as well (optional)
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
print("\nCheck your email for log messages from 'my_app'.")

How to Run This:

Python logging如何配置发送email告警?-图2
(图片来源网络,侵删)
  1. Replace 'smtp.example.com', 'your_username', 'your_password', and the email addresses with your actual SMTP server details (e.g., Gmail, Outlook, your company's mail server).
  2. Run the script. You will see output in your console, and you should receive an email for every log message at the INFO level and above.

Production-Ready Approach: Buffering and Criticality

Sending an email for every INFO message is a terrible idea. It will flood your inbox and likely get your email address flagged as spam. A much better strategy is to:

  1. Buffer Logs: Collect log messages in memory.
  2. Send on Critical Events: Only send an email when a message of a certain severity (e.g., ERROR or higher) occurs.
  3. Batch Messages: Include all the buffered messages in that single email, giving you context about what led up to the critical error.

The perfect tool for this is BufferingHandler.

import logging
import logging.handlers
import smtplib
from email.message import EmailMessage
# --- Configuration ---
SMTP_HOST = 'smtp.example.com'
SMTP_PORT = 587
SMTP_USER = 'your_username'
SMTP_PASSWORD = 'your_password'
FROM_ADDR = 'app@example.com'
TO_ADDRS = ['admin@example.com']
LOG_SUBJECT = 'My App: Critical Error Alert'
# --- Logger Setup ---
logger = logging.getLogger('my_app_prod')
logger.setLevel(logging.DEBUG) # Capture all levels, but only send some
# 1. Create a BufferingHandler to store logs in memory
buffer_handler = logging.handlers.BufferingHandler(capacity=100) # Store up to 100 records
# 2. Create an SMTPHandler (we'll attach this later)
smtp_handler = logging.handlers.SMTPHandler(
    mailhost=(SMTP_HOST, SMTP_PORT),
    fromaddr=FROM_ADDR,
    toaddrs=TO_ADDRS,
    subject=LOG_SUBJECT,
    credentials=(SMTP_USER, SMTP_PASSWORD),
    secure=() # Use () for STARTTLS
)
# 3. Set a formatter for both handlers
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
buffer_handler.setFormatter(formatter)
smtp_handler.setFormatter(formatter)
# 4. Add the buffer_handler to the logger NOW
logger.addHandler(buffer_handler)
# --- Application Logic ---
def process_data():
    logger.info("Starting data processing.")
    logger.debug("Loaded configuration from file.")
    logger.info("Fetching data from API...")
    # Simulate an error
    try:
        result = 1 / 0
    except ZeroDivisionError:
        logger.error("A critical error occurred during calculation!", exc_info=True)
    logger.info("Finished processing data.")
# --- The Trigger ---
# This function will be called when a critical error happens.
# It gets the buffered logs and sends them via email.
def send_error_log(record):
    # Get the records from the buffer
    records = buffer_handler.buffer
    # Create a detailed email body
    body = "A critical error has occurred. Here are the recent log messages:\n\n"
    for r in records:
        body += f"{r.levelname}: {r.getMessage()}\n"
    # Add exception info if available in the triggering record
    if record.exc_info:
        body += "\n--- Exception Traceback ---\n"
        body += logging.Formatter().formatException(record.exc_info)
    # Update the handler's subject to be more specific
    smtp_handler.subject = f"{LOG_SUBJECT}: {record.getMessage()}"
    # Clear the buffer after sending
    buffer_handler.buffer.clear()
    # Manually send the email
    smtp_handler.emit(record)
# --- Monkey Patching for the Trigger ---
# We override the SMTPHandler's emit method to only fire on errors
# and to include the buffered logs.
original_emit = smtp_handler.emit
def custom_emit(record):
    if record.levelno >= logging.ERROR:
        # Get the buffer contents before clearing it
        logs_to_send = list(buffer_handler.buffer)
        # Create a new, more detailed record for the email
        email_record = logging.LogRecord(
            name=record.name,
            level=record.levelno,
            pathname=record.pathname,
            lineno=record.lineno,
            msg="CRITICAL ERROR: See email for details.",
            args=record.args,
            exc_info=record.exc_info
        )
        # Format the full email body
        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        body = "A critical error has occurred. Here are the recent log messages:\n\n"
        for r in logs_to_send:
            body += f"{r.levelname}: {r.getMessage()}\n"
        if record.exc_info:
            body += "\n--- Exception Traceback ---\n"
            body += logging.Formatter().formatException(record.exc_info)
        # We need to manually construct and send the email here
        # because the standard SMTPHandler doesn't support this kind of batching.
        msg = EmailMessage()
        msg['Subject'] = f"{LOG_SUBJECT}: {record.getMessage()}"
        msg['From'] = FROM_ADDR
        msg['To'] = ', '.join(TO_ADDRS)
        msg.set_content(body)
        with smtplib.SMTP(SMTP_HOST, SMTP_PORT) as server:
            server.starttls()
            server.login(SMTP_USER, SMTP_PASSWORD)
            server.send_message(msg)
        # Clear the buffer after sending
        buffer_handler.buffer.clear()
    else:
        # For non-critical logs, just use the original handler (which does nothing in this setup)
        pass
# Apply our custom emit function
smtp_handler.emit = custom_emit
# Add the SMTP handler (it won't send on its own, only when our custom logic fires)
logger.addHandler(smtp_handler)
# --- Run the application ---
print("Running application...")
process_data()
print("Application finished. Check email if an error occurred.")

This second example is far more practical. It captures all logs but only sends a detailed, context-rich email when an ERROR or CRITICAL event occurs.


Using Third-Party Libraries (Simpler Alternative)

While the standard library is powerful

Python logging如何配置发送email告警?-图3
(图片来源网络,侵删)
分享:
扫描分享到社交APP
上一篇
下一篇