#!/usr/bin/env python3 """ Randomly writes from a set of messages simulating logging. """ import argparse import logging import random import time import os import re import hashlib def parse_args(): argp = argparse.ArgumentParser( description="Will generate fake log files specified by the 'file' argument. " "The --delay argment controls at what speed the file is written. " "File limits are controlled by the --size argument. The values are " "'random:N and consistent:N, where N is value in bytes indicated " "as 4096, 16384B, 10K, 250M, 10G, or 3T. Lowercase prefixes are " "accepted. Random will generate a file up to the limit specified. " "Consistent will make the file exactly the size detailed. " "Count will determine how how many log files are created." "The --count argument takes an integer. File past the first are " "given a short hash suffix before the final extension." ) argp.add_argument('file', help="Full path for log file") argp.add_argument('-d', '--delay', type=int, default=5, help="Number to delay log entries (1/n)") argp.add_argument( '-s', '--size', help="File size limits. Valid choices: 'random:N' (default) or 'consistent:N', where N is an upper limit" ) argp.add_argument( '-c', '--count', type=int, help="Number of log files to generate" ) return argp.parse_args() def random_message(log_level): debug_messages = [ "Values = [10, 20, 33]", "Cake ratio = 3.14", "Setting flat mode", "Back to normal mode", "Voltage: 1.21 Gigawatts", "Speed: 88 mph", "Temperature: 32.4C" ] info_messages = [ "My cat is blue", "All your base are belong to us", "I like turtles", "Chocolate rain...", "I hate prunes", "The cake is a lie", "I'm giving her all she's got!" ] warning_messages = [ "Numbers reaching critical levels", "Magnetic constrictors around the anti-matter containment fields are failing", "Approaching thermal thresholds", "Searching is taking too long", "There are monkeyboys in the facility", "Structural integrity approaching tolerance", "Process taking too long" ] error_messages = [ "Unable to process request", "Connection lost with host", "Buffer overflow", "Warp engines are offline", "No more coffee", "Server not responding", "Cannot divide by zero" ] message_level = [ debug_messages, info_messages, warning_messages, error_messages ] which_message = random.randint(0, 6) return message_level[log_level - 1][which_message] def main(): args = parse_args() _LEVEL = logging.NOTSET _FORMAT = logging.Formatter("%(asctime)-15s [%(levelname)-8s] : %(lineno)d : %(name)s.%(funcName)s : %(message)s") _file_handler = logging.FileHandler(args.file, mode='a') _file_handler.setFormatter(_FORMAT) _file_handler.setLevel(_LEVEL) log = logging.Logger('fake-logs') log.addHandler(_file_handler) file_complete = False file_count = 0 try: while file_count < args.count: log_path = os.path.dirname(args.file) log_filename = os.path.basename(args.file) new_filename, file_extension = os.path.splitext(log_filename) suffix = hashlib.md5(time.strftime('%s').encode('utf-8')).hexdigest()[-4:] log_filename = os.path.join(log_path, f"{new_filename}-{suffix}{file_extension}") while not file_complete: sleep_time = random.random() * float(f"0.{args.delay}") logging_level = random.randint(0, 40) % 4 log.log(logging_level * 10, random_message(logging_level)) time.sleep(sleep_time) # Determine file size if specified if args.size: size_type, size_value = args.size.split(':') size_value_parts = re.match(r'(\d+)([BbKkMmGgTt]?)', size_value).groups() if size_value_parts[1]: factor = ( (size_value_parts[1].lower() == 'k' and 1) or (size_value_parts[1].lower() == 'm' and 2) or (size_value_parts[1].lower() == 'g' and 3) or (size_value_parts[1].lower() == 't' and 4) ) size_limit = int(size_value_parts[0]) * 1024**factor else: size_limit - int(size_value_parts[0]) log_file_stats = os.stat(args.file) if log_file_stats.st_size > size_limit: if size_type == 'consistent': truncate_log_file(log_filename) file_complete = True file_count++ except KeyboardInterrupt: time.sleep(0.25) log.error(random_message(4)) if __name__ == "__main__": main()