J'utilise généralement Python principalement, mais les données structurelles sont plus pratiques lors du développement d'une plate-forme de collecte / analyse de journaux avec un service cloud, n'est-ce pas? Cependant, si vous utilisez le formateur normal de Logging, qui est une bibliothèque Python standard, vous devez le convertir une fois en données structurelles. Il ne sert à rien de créer une autre application uniquement pour la conversion. Par conséquent, j'ai développé un formateur qui génère des journaux en JSON.
--OS (fonctionnement garanti) - MacOS Catalina - Ubuntu 18.04 --Langue - Python ^3.7
--Directeur chargé d'emballage - Poetry
Bibliothèque
Bibliothèque standard uniquement
Bibliothèque de développement
Le formateur développé peut être trouvé sur homoluctus / json-pyformatter. Il est également publié sur PyPI-> json-pyformatter 0.1.0
pip install json-pyformatter
Le champ qui peut être affiché est logrecord-attributes de la bibliothèque standard Logging.
Les champs par défaut générés par le formateur développé sont ʻasctime, levelname et message`.
import logging
from json_pyformmatter import JsonFormatter
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
fields = ('levelname', 'filename', 'message')
formatter = JsonFormatter(fields=fields)
handler.setFormatter(formatter)
logger.addHandler(hander)
logger.info('hello')
Lorsque cela est exécuté, le journal JSON sera généré comme indiqué ci-dessous.
{"levelname": "INFO", "filename": "test_formatter.py", "message": "hello"}
De plus, il sera plus facile de voir si vous spécifiez ʻindent = 2` dans l'argument de JsonFormatter.
{
"levelname": "INFO",
"filename": "test_formatter.py",
"message": "hello"
}
Bien sûr, vous pouvez également générer un suivi. Puisqu'il est difficile de voir s'il s'agit d'une seule ligne, elle est organisée dans un tableau.
{
'asctime': '2019-12-01 13:58:34',
'levelname': 'ERROR',
'message': 'error occurred !!',
'traceback': [
'Traceback (most rec...ll last):',
'File "/example/test..._exc_info',
'raise TypeError(message)',
'TypeError: error occurred !!'
]
}
De là, je vais expliquer le code source.
J'écrirai tout le code source pour le moment.
import json
from collections import OrderedDict
from logging import Formatter
class JsonFormatter(Formatter):
default_fields = ('asctime', 'levelname', 'message')
def __init__(self, fields=None, datefmt=None, indent=None):
"""
Args:
fields (tuple, list)
datefmt (str)
indent (str, int)
"""
self.fields = (
self.get_or_none(fields, (list, tuple)) or self.default_fields
)
# default time format is %Y-%m-%d %H:%M:%S
self.datefmt = (
self.get_or_none(datefmt, str) or self.default_time_format
)
self._indent = self.get_or_none(indent, (str, int))
def get_or_none(self, target, types):
"""Check whether target value is expected type.
If target type does not match expected type, returns None.
Args:
target (any)
types (class, tuple)
Returns:
target or None
"""
if isinstance(target, types):
return target
return None
def getMessage(self, record):
if isinstance(record.msg, (list, tuple, dict)):
return record.msg
return record.getMessage()
def _format_json(self, record):
return json.dumps(record, ensure_ascii=False, indent=self._indent)
def _format(self, record):
log = OrderedDict()
try:
for field in self.fields:
log[field] = getattr(record, field)
return log
except AttributeError as err:
raise ValueError(f'Formatting field not found in log record {err}')
def format(self, record):
record.message = self.getMessage(record)
record.asctime = self.formatTime(record, self.datefmt)
formatted_record = self._format(record)
if record.exc_info:
if not record.exc_text:
record.exc_text = self.formatException(record.exc_info)
if record.exc_text:
formatted_record['traceback'] = [
msg.strip() for msg in record.exc_text.strip().split('\n')
]
if record.stack_info:
formatted_record['stack'] = record.stack_info.strip()
return self._format_json(formatted_record)
get_or_none Cette méthode est utilisée lors de la création d'une instance. S'il ne s'agit pas du type attendu, utilisez la valeur par défaut ou utilisez-la pour affecter None tel quel.
def get_or_none(self, target, types):
"""Check whether target value is expected type.
If target type does not match expected type, returns None.
Args:
target (any)
types (class, tuple)
Returns:
target or None
"""
if isinstance(target, types):
return target
return None
getMessage
Cette méthode est utilisée pour récupérer le message de l'argument de logger.info (). La valeur de retour de ce getMessage est définie dans le champ JSON message.
Si le type d'instance de record.msg est l'un des (list, tuple, dict), il est renvoyé tel quel, sinon, la valeur de retour de la méthode getMessage de l'instance d'enregistrement est renvoyée. Ce faisant, lorsque l'un des éléments (liste, tuple, dict) est passé, le journal peut être généré en tant que tableau / objet JSON.
def getMessage(self, record):
if isinstance(record.msg, (list, tuple, dict)):
return record.msg
return record.getMessage()
_format_json Cette méthode vide l'enregistrement d'argument en tant que JSON.
def _format_json(self, record):
return json.dumps(record, ensure_ascii=False, indent=self._indent)
_format Une méthode pour convertir un enregistrement en dictionnaire Python. Le champ spécifié par l'utilisateur n'existe pas nécessairement dans l'attribut d'enregistrement, il s'agit donc de "try-except". Le journal des valeurs de retour est défini sur ʻOrderedDict` pour garantir la commande.
def _format(self, record):
log = OrderedDict()
try:
for field in self.fields:
log[field] = getattr(record, field)
return log
except AttributeError as err:
raise ValueError(f'Formatting field not found in log record {err}')
format Cette méthode est appelée à partir de logging.Handler. Je ne pense pas que vous créerez votre propre Formatter et utiliserez le format de la classe parent (logging.Formatter), alors assurez-vous d'écraser cette méthode. Je formate record.exc_text pour afficher le traçage sous forme de tableau. Après avoir supprimé les espaces supplémentaires, il est divisé avec un saut de ligne pour créer un tableau. Enfin, je vide le dict Python en JSON.
def format(self, record):
record.message = self.getMessage(record)
record.asctime = self.formatTime(record, self.datefmt)
formatted_record = self._format(record)
if record.exc_info:
if not record.exc_text:
record.exc_text = self.formatException(record.exc_info)
if record.exc_text:
formatted_record['traceback'] = [
msg.strip() for msg in record.exc_text.strip().split('\n')
]
if record.stack_info:
formatted_record['stack'] = record.stack_info.strip()
return self._format_json(formatted_record)
Recommended Posts