Created
December 14, 2020 16:55
-
-
Save pepoluan/d4ffa7443f28962f870da8b41c70ea4f to your computer and use it in GitHub Desktop.
Simple RFC5424-over-UDP Logging Handler & Formatter
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
pytz | |
tzlocal |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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