hat.syslog.generator

Syslog test message generator

  1"""Syslog test message generator"""
  2
  3import argparse
  4import asyncio
  5import contextlib
  6import logging.config
  7import sys
  8
  9from hat import aio
 10from hat.syslog import common
 11
 12
 13mlog: logging.Logger = logging.getLogger('hat.syslog.generator')
 14"""Module logger"""
 15
 16
 17def create_argument_parser() -> argparse.ArgumentParser:
 18    """Create argument parser"""
 19    parser = argparse.ArgumentParser()
 20    parser.add_argument(
 21        '--comm-type', choices=[i.name for i in common.CommType],
 22        default='TCP', help="syslog server host name (default TCP)")
 23    parser.add_argument(
 24        '--host', metavar='HOST', default='127.0.0.1',
 25        help="syslog server host name (default 127.0.0.1)")
 26    parser.add_argument(
 27        '--port', metavar='PORT', type=int, default=6514,
 28        help="syslog server port (default 6514)")
 29    parser.add_argument(
 30        '--count', metavar='N', type=int, default=1,
 31        help="number of log messages (default 1)")
 32    parser.add_argument(
 33        '--text', metavar='TEXT', default='syslog generator test',
 34        help="log message text")
 35    parser.add_argument(
 36        '--queue-size', metavar='N', type=int, default=1024,
 37        help="client's log message queue size (default 1024)")
 38    parser.add_argument(
 39        '--msg-delay', metavar='T', type=float, default=0.01,
 40        help="time delay between message generation in seconds "
 41             "(default 0.01)")
 42    parser.add_argument(
 43        '--end-delay', metavar='T', type=float, default=0.5,
 44        help="time delay affter all messages have been generated in seconds "
 45             "(default 0.5)")
 46    return parser
 47
 48
 49def main():
 50    """Syslog test message generator"""
 51    parser = create_argument_parser()
 52    args = parser.parse_args()
 53
 54    aio.init_asyncio()
 55
 56    logging.config.dictConfig({
 57        'version': 1,
 58        'formatters': {
 59            'default': {}},
 60        'handlers': {
 61            'syslog': {
 62                'class': 'hat.syslog.handler.SysLogHandler',
 63                'host': args.host,
 64                'port': args.port,
 65                'comm_type': 'TCP',
 66                'level': 'DEBUG',
 67                'formatter': 'default',
 68                'queue_size': args.queue_size}},
 69        'root': {
 70            'level': 'INFO',
 71            'handlers': ['syslog']},
 72        'disable_existing_loggers': False})
 73
 74    with contextlib.suppress(asyncio.CancelledError):
 75        aio.run_asyncio(async_main(count=args.count,
 76                                   text=args.text,
 77                                   msg_delay=args.msg_delay,
 78                                   end_delay=args.end_delay))
 79
 80
 81async def async_main(count: int,
 82                     text: str,
 83                     msg_delay: float,
 84                     end_delay: float):
 85    """Async main"""
 86    for i in range(count):
 87        mlog.info(
 88            ('{} {:0' + str(_number_of_digits(count)) + '}').format(
 89                text, i))
 90        await asyncio.sleep(msg_delay)
 91    await asyncio.sleep(end_delay)
 92
 93
 94def _number_of_digits(x):
 95    if x < 10:
 96        return 1
 97    return 1 + _number_of_digits(x // 10)
 98
 99
100if __name__ == '__main__':
101    sys.argv[0] = 'hat-syslog-generator'
102    sys.exit(main())
mlog: logging.Logger = <Logger hat.syslog.generator (WARNING)>

Module logger

def create_argument_parser() -> argparse.ArgumentParser:
18def create_argument_parser() -> argparse.ArgumentParser:
19    """Create argument parser"""
20    parser = argparse.ArgumentParser()
21    parser.add_argument(
22        '--comm-type', choices=[i.name for i in common.CommType],
23        default='TCP', help="syslog server host name (default TCP)")
24    parser.add_argument(
25        '--host', metavar='HOST', default='127.0.0.1',
26        help="syslog server host name (default 127.0.0.1)")
27    parser.add_argument(
28        '--port', metavar='PORT', type=int, default=6514,
29        help="syslog server port (default 6514)")
30    parser.add_argument(
31        '--count', metavar='N', type=int, default=1,
32        help="number of log messages (default 1)")
33    parser.add_argument(
34        '--text', metavar='TEXT', default='syslog generator test',
35        help="log message text")
36    parser.add_argument(
37        '--queue-size', metavar='N', type=int, default=1024,
38        help="client's log message queue size (default 1024)")
39    parser.add_argument(
40        '--msg-delay', metavar='T', type=float, default=0.01,
41        help="time delay between message generation in seconds "
42             "(default 0.01)")
43    parser.add_argument(
44        '--end-delay', metavar='T', type=float, default=0.5,
45        help="time delay affter all messages have been generated in seconds "
46             "(default 0.5)")
47    return parser

Create argument parser

def main():
50def main():
51    """Syslog test message generator"""
52    parser = create_argument_parser()
53    args = parser.parse_args()
54
55    aio.init_asyncio()
56
57    logging.config.dictConfig({
58        'version': 1,
59        'formatters': {
60            'default': {}},
61        'handlers': {
62            'syslog': {
63                'class': 'hat.syslog.handler.SysLogHandler',
64                'host': args.host,
65                'port': args.port,
66                'comm_type': 'TCP',
67                'level': 'DEBUG',
68                'formatter': 'default',
69                'queue_size': args.queue_size}},
70        'root': {
71            'level': 'INFO',
72            'handlers': ['syslog']},
73        'disable_existing_loggers': False})
74
75    with contextlib.suppress(asyncio.CancelledError):
76        aio.run_asyncio(async_main(count=args.count,
77                                   text=args.text,
78                                   msg_delay=args.msg_delay,
79                                   end_delay=args.end_delay))

Syslog test message generator

async def async_main(count: int, text: str, msg_delay: float, end_delay: float):
82async def async_main(count: int,
83                     text: str,
84                     msg_delay: float,
85                     end_delay: float):
86    """Async main"""
87    for i in range(count):
88        mlog.info(
89            ('{} {:0' + str(_number_of_digits(count)) + '}').format(
90                text, i))
91        await asyncio.sleep(msg_delay)
92    await asyncio.sleep(end_delay)

Async main