rippled/bin/sidechain/python/log_analyzer.py

#!/usr/bin/env python3

import argparse
import json
import os
import re
import sys

from common import eprint
from typing import IO, Optional


class LogLine:
    UNSTRUCTURED_RE = re.compile(r'''(?x)
                # The x flag enables insignificant whitespace mode (allowing comments)
                ^(?P<timestamp>.*UTC)
                [\ ]
                (?P<module>[^:]*):(?P<level>[^\ ]*)
                [\ ]
                (?P<msg>.*$)
                ''')

    STRUCTURED_RE = re.compile(r'''(?x)
                # The x flag enables insignificant whitespace mode (allowing comments)
                ^(?P<timestamp>.*UTC)
                [\ ]
                (?P<module>[^:]*):(?P<level>[^\ ]*)
                [\ ]
                (?P<msg>[^{]*)
                [\ ]
                (?P<json_data>.*$)
                ''')

    def __init__(self, line: str):
        self.raw_line = line
        self.json_data = None

        try:
            if line.endswith('}'):
                m = self.STRUCTURED_RE.match(line)
                try:
                    self.json_data = json.loads(m.group('json_data'))
                except:
                    m = self.UNSTRUCTURED_RE.match(line)
            else:
                m = self.UNSTRUCTURED_RE.match(line)

            self.timestamp = m.group('timestamp')
            self.level = m.group('level')
            self.module = m.group('module')
            self.msg = m.group('msg')
        except Exception as e:
            eprint(f'init exception: {e} line: {line}')

    def to_mixed_json_str(self) -> str:
        '''
        return a pretty printed string as mixed json
        '''
        try:
            r = f'{self.timestamp} {self.module}:{self.level} {self.msg}'
            if self.json_data:
                r += '\n' + json.dumps(self.json_data, indent=1)
            return r
        except:
            eprint(f'Using raw line: {self.raw_line}')
            return self.raw_line

    def to_pure_json(self) -> dict:
        '''
        return a json dict
        '''
        dict = {}
        dict['t'] = self.timestamp
        dict['m'] = self.module
        dict['l'] = self.level
        dict['msg'] = self.msg
        if self.json_data:
            dict['data'] = self.json_data
        return dict

    def to_pure_json_str(self, f_id: Optional[str] = None) -> str:
        '''
        return a pretty printed string as pure json
        '''
        try:
            dict = self.to_pure_json(f_id)
            return json.dumps(dict, indent=1)
        except:
            return self.raw_line


def convert_log(in_file_name: str,
                out: str,
                *,
                as_list=False,
                pure_json=False,
                module: Optional[str] = 'SidechainFederator') -> list:
    result = []
    try:
        prev_lines = None
        with open(in_file_name) as input:
            for l in input:
                l = l.strip()
                if not l:
                    continue
                if LogLine.UNSTRUCTURED_RE.match(l):
                    if prev_lines:
                        log_line = LogLine(prev_lines)
                        if not module or log_line.module == module:
                            if as_list:
                                result.append(log_line.to_pure_json())
                            else:
                                if pure_json:
                                    print(log_line.to_pure_json_str(),
                                          file=out)
                                else:
                                    print(log_line.to_mixed_json_str(),
                                          file=out)
                    prev_lines = l
                else:
                    if not prev_lines:
                        eprint(f'Error: Expected prev_lines. Cur line: {l}')
                    assert prev_lines
                    prev_lines += f' {l}'
            if prev_lines:
                log_line = LogLine(prev_lines)
                if not module or log_line.module == module:
                    if as_list:
                        result.append(log_line.to_pure_json())
                    else:
                        if pure_json:
                            print(log_line.to_pure_json_str(f_id),
                                  file=out,
                                  flush=True)
                        else:
                            print(log_line.to_mixed_json_str(),
                                  file=out,
                                  flush=True)
    except Exception as e:
        eprint(f'Excption: {e}')
        raise e
    return result


def convert_all(in_dir_name: str, out: IO, *, pure_json=False):
    '''
    Convert all the "debug.log" log files in one directory level below the in_dir_name into a single json file.
    There will be a field called 'f' for the director name that the origional log file came from.
    This is useful when analyzing networks that run on the local machine.
    '''
    if not os.path.isdir(in_dir_name):
        print(f'Error: {in_dir_name} is not a directory')
    files = []
    f_ids = []
    for subdir in os.listdir(in_dir_name):
        file = f'{in_dir_name}/{subdir}/debug.log'
        if not os.path.isfile(file):
            continue
        files.append(file)
        f_ids.append(subdir)

    result = {}
    for f, f_id in zip(files, f_ids):
        l = convert_log(f, out, as_list=True, pure_json=pure_json, module=None)
        result[f_id] = l
    print(json.dumps(result, indent=1), file=out, flush=True)


def parse_args():
    parser = argparse.ArgumentParser(
        description=('python script to convert log files to json'))

    parser.add_argument(
        '--input',
        '-i',
        help=('input log file or sidechain config directory structure'),
    )

    parser.add_argument(
        '--output',
        '-o',
        help=('output log file'),
    )

    return parser.parse_known_args()[0]


if __name__ == '__main__':
    try:
        args = parse_args()
        with open(args.output, "w") as out:
            if os.path.isdir(args.input):
                convert_all(args.input, out, pure_json=True)
            else:
                convert_log(args.input, out, pure_json=True)
    except Exception as e:
        eprint(f'Excption: {e}')
        raise e