Skip to content

Instantly share code, notes, and snippets.

@jamesmcdonald
Last active March 22, 2019 10:58
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save jamesmcdonald/2fff91232c3fc607f313e56acec34fd2 to your computer and use it in GitHub Desktop.
Save jamesmcdonald/2fff91232c3fc607f313e56acec34fd2 to your computer and use it in GitHub Desktop.
Format json logs as plain text
#!/usr/bin/env python3
"""Read a stream of one-per-line json events and print them in a more human-readable format"""
import json
import sys
import string
import argparse
class SkipFormatter(string.Formatter):
"""A string.Formatter that returns '-' for missing fields"""
def get_field(self, field_name, args, kwargs):
try:
val = super().get_field(field_name, args, kwargs)
except (KeyError, AttributeError):
val = None, field_name
return val
def format_field(self, value, format_spec):
if value is None:
return '-'
return super().format_field(value, format_spec)
DEFAULT_FORMAT = '{timestamp} {level} [{logger}] {message}'
def printlogs(logformat=DEFAULT_FORMAT, infile=sys.stdin, outfile=sys.stdout, flush=False):
"""Read logs and print them"""
if isinstance(infile, str):
infile = open(infile, 'r')
if isinstance(outfile, str):
outfile = open(outfile, 'w')
fmt = SkipFormatter()
for line in infile:
line = line.strip()
try:
data = json.loads(line)
log = fmt.format(logformat, **data)
print(log, file=outfile)
except json.decoder.JSONDecodeError:
print('[plain]', line, file=outfile)
if flush:
outfile.flush()
def main():
"""Maining the mains since 1976"""
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--format', '-f', dest='logformat', default=DEFAULT_FORMAT,
help='specify an output format (default: "' + DEFAULT_FORMAT + '")')
parser.add_argument('--flush', '-F', action='store_true',
help='flush output after every log line')
parser.add_argument('--input', '-i', dest='infile', default=sys.stdin,
help='file to write to (default: standard input)')
parser.add_argument('--output', '-o', dest='outfile', default=sys.stdout,
help='file to write to (default: standard output)')
args = parser.parse_args()
printlogs(**vars(args))
if __name__ == '__main__':
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment