Skip to content

Instantly share code, notes, and snippets.

@pepoluan
Created December 14, 2020 16:55
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Save pepoluan/d4ffa7443f28962f870da8b41c70ea4f to your computer and use it in GitHub Desktop.
Simple RFC5424-over-UDP Logging Handler & Formatter
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
facility = 23 # local7; see RFC 5424
formatter = RFC5424Formatter(appname="my-awesome", facility=facility)
syslog_handler = UTF8DatagramHandler("127.0.0.1", 514)
syslog_handler.setFormatter(formatter)
syslog_handler.setLevel(logging.DEBUG)
logger.addHandler(syslog_handler)
import os
import pytz
import socket
import logging
import logging.handlers
import tzlocal
import datetime
from typing import Optional
class RFC5424Formatter(logging.Formatter):
PriorityMap = {
"DEBUG": 7,
"INFO": 6,
"NOTICE": 5,
"WARNING": 4,
"ERROR": 3,
"CRITICAL": 2,
"ALERT": 1,
"EMERGENCY": 0,
"EMERG": 0,
}
def __init__(self, appname: str, facility: int, machine_tz: Optional[str] = None):
fmt = " ".join(
[
"<{{pri}}>1", # version
"{asctime}", # timestamp
socket.gethostname(),
appname, # app-name
str(os.getpid()),
"{{msgid}}", # msg-id
"-", # structured-data
"{message}",
]
)
super().__init__(style="{", fmt=fmt)
self.facility = facility
self.machine_tz = machine_tz or tzlocal.get_localzone().zone
def converter(self, timestamp):
# From: https://stackoverflow.com/a/47104004/149900
dt = datetime.datetime.fromtimestamp(timestamp)
tzinfo = pytz.timezone(self.machine_tz)
return tzinfo.localize(dt)
def formatTime(
self, record: logging.LogRecord, datefmt: Optional[str] = ...
) -> str:
# From: https://stackoverflow.com/a/47104004/149900
dt = self.converter(record.created)
try:
s = dt.isoformat(timespec="milliseconds")
except TypeError:
s = dt.isoformat()
return s
def format(self, record: logging.LogRecord):
pri = self.facility * 8 + self.PriorityMap[record.levelname]
templ = super().format(record)
return templ.format(
pri=pri,
msgid=record.processName,
)
class UTF8DatagramHandler(logging.handlers.DatagramHandler):
def makePickle(self, record: logging.LogRecord) -> bytes:
msg = self.format(record)
return msg.encode()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment