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())
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