# Copyright (c) Microsoft. All rights reserved. # Licensed under the MIT license. See LICENSE file in the project root for # full license information. # # filename: docker_log_processor.py # author: v-greach@microsoft.com # created: 01/29/2019 # Rev: 03/05/2019 A from multiprocessing import Process, Queue, Event from threading import Thread from datetime import datetime, timedelta import traceback import docker import time import argparse import sys class DockerLogProcessor: def __init__(self, args): # Parse args parser = argparse.ArgumentParser(description="Docker Log Processor") group = parser.add_mutually_exclusive_group(required=True) group.add_argument( "-staticfile", action="append", nargs="+", help="filename to read from" ) group.add_argument( "-modulename", action="append", nargs="+", help="docker modulename to read from", ) parser.add_argument("-filterfile", nargs=1, help="filename of json filters") arguments = parser.parse_args(args) if arguments.staticfile: self.process_static_log(arguments.staticfile, arguments.filterfile) else: self.queue = Queue() self.logger_thread = Thread(target=self.process_queue) self.logger_thread.start() self.watcher_processes = [] for container_name in arguments.modulename: print("Getting Log for: " + container_name) new_process = Process( target=self.get_log_from_container, args=(container_name, self.queue), ) new_process.start() self.watcher_processes.append(new_process) @classmethod def format_date_and_time(self, date_in="", time_format="%Y-%m-%d %H:%M:%S.%f"): """ Formats a string into a datetime type. date_in comes in, if it's empty, set it to NOW, then using the format, convert the string to datetime type. Parameters ---------- date_in : string String to convert - can be null time_format : string format of string for datetime conversion Returns ------- datetime converted input string or NOW() if date_in is empty """ date_out = "" if not date_in: date_out = datetime.strftime(datetime.now(), time_format) return date_out date_in = date_in.replace("T", " ") if len(date_in) > 26: date_in = date_in[:26] date_out = datetime.strptime(date_in, time_format) return date_out @staticmethod def write_err(msg): """ write a string to stdout and stderr. """ print(msg, file=sys.stderr) print(msg) @staticmethod def get_log_from_container(container_name, queue): """ Gets log info from the Docker container then converts DateTime and puts each line in the queue. Parameters ---------- container_name : string Name of the Docker container queue : object queue to stuff the log object in """ client = docker.from_env() container = client.containers.get(container_name) for log_line in container.logs( stream=True, tail=0, follow=True, timestamps=True ): try: log_line = log_line.decode("utf8").strip() log_line_parts = log_line.split("Z ") if log_line_parts: log_data = "" num_parts = len(log_line_parts) # Handle case where more than one timestamp if num_parts > 2: for part in range(1, num_parts): log_data += log_line_parts[part] + " " else: log_data = log_line_parts[1] log_line_object = LogLineObject( DockerLogProcessor.format_date_and_time(log_line_parts[0]), container_name, log_data, ) queue.put(log_line_object) except Exception as e: DockerLogProcessor.write_err( "Exception getting container log_line from: " + container_name ) DockerLogProcessor.write_err(e) def split(self, string, delimiters): """ Split a string with multiple delimiters. """ import re regexPattern = "|".join(map(re.escape, delimiters)) return re.split(regexPattern, string) def get_timestamp_delta(self, date_one, date_two, line_count=0, line_mod=100): """ Diff date_one and date_two then format string for readability. Delta of the strings are converted by fields. line_count can be used to print a full timestamp every line_mod (%) lines """ if line_mod != 0 and line_count % line_mod == 0: return date_one time_delta_str = "" delimiters = (".", "-", " ", ":") field_count = 0 all_fields_one = self.split(date_one, delimiters) all_fields_two = self.split(date_two, delimiters) for field1 in all_fields_one: if field1 == all_fields_two[field_count]: for _ in field1: time_delta_str += " " else: time_delta_str += all_fields_one[field_count] if field_count < 2: time_delta_str += "-" elif field_count == 2: time_delta_str += " " elif field_count > 2 and field_count < 5: time_delta_str += ":" elif field_count == 5: time_delta_str += "." field_count += 1 return time_delta_str def process_static_log(self, static_filenames, filter_filenames): """ Static logs in args - set them up for processing. Static log(s) specified. static_filenames Optional filter_filename Path to JSON filter file read all log files and format each line sort and display to stdout """ import os import json import pathlib split_char = u"\u2588" loglines = [] max_name_len = 0 filter_list = "" pytest_owner = "" if filter_filenames: filter_filename = os.path.abspath(filter_filenames[0]) """ filter.json should have a format like this: { "filters": [ "Getting next batch", "Obtained next batch" ] } """ try: filter_json = open(filter_filename, "r", encoding="utf8").read() if filter_json: json_data = json.loads(filter_json) filter_list = json_data["filters"] except Exception: self.write_err("Exception processing JSON file: " + filter_filename) traceback.print_exc() # find the max_name_len of every staticfile filename basename for static_filename in static_filenames: if static_filename: base_filename = os.path.basename(static_filename[0]) name_len = len(base_filename) if name_len > max_name_len: max_name_len = name_len # read and process every static file for static_filename in static_filenames: if static_filename: static_filename = static_filename[0] module_name = os.path.basename(static_filename) print("Getting log from file: " + static_filename) # Pad the filename so that each is the same length for _ in range(len(module_name), max_name_len): module_name += " " try: read_file = ( open(static_filename, encoding="utf8").read().split("\n") ) except Exception as e: self.write_err("Exception opening LOG file: " + static_filename) self.write_err(e) return # Get and filter each line for log_line in read_file: ok_to_log = True if log_line: if "PYTEST" in log_line: if not pytest_owner: pytest_owner = module_name else: if pytest_owner != module_name: ok_to_log = False if ok_to_log: for filter in filter_list: if filter in log_line: ok_to_log = False if ok_to_log: # Made it past filters and PyTest, so Log the line log_line_parts = log_line.split("Z ") if log_line_parts: log_data = "" num_parts = len(log_line_parts) # Handle case where more than one timestamp if num_parts > 2: for part in range(1, num_parts): log_data += log_line_parts[part] + " " else: if num_parts == 2: log_data = log_line_parts[1] if num_parts >= 2: try: log_time = DockerLogProcessor.format_date_and_time( log_line_parts[0], "%Y-%m-%d %H:%M:%S.%f" ) log_line_object = LogLineObject( log_time, module_name, log_data ) loglines.append(log_line_object) except Exception: print( "INVALID_TIMESTAMP({}):{}".format( module_name, log_line ) ) else: print( "INVALID_LINE({}):{}".format( module_name, log_line ) ) # Sort the merged static file lines by timestamp loglines.sort(key=lambda x: x.timestamp) last_timestamp = datetime.now() + timedelta(days=-364) line_count = 0 # display the results to stdout for log_line in loglines: logline_timestamp = log_line.timestamp if ( "HORTON: Entering function" in log_line.log_data or "HORTON: Exiting function" in log_line.log_data ): date_delta = logline_timestamp.isoformat(timespec="microseconds") else: date_delta = self.get_timestamp_delta( logline_timestamp.isoformat(timespec="microseconds"), last_timestamp.isoformat(timespec="microseconds"), line_count, ) line_count += 1 out_line = ( log_line.module_name + " : " + date_delta + " " + split_char + " " + log_line.log_data ) last_timestamp = logline_timestamp try: print(out_line) except Exception: print("".join([i if ord(i) < 128 else "#" for i in out_line])) def process_queue(self): """ Process the line objects in the queue, and print as formatted. """ last_timestamp = datetime.now() + timedelta(days=-364) line_count = 0 split_char = u"\u2588" while True: log_line = self.queue.get() logline_timestamp = log_line.timestamp if ( "HORTON: Entering function" in log_line.log_data or "HORTON: Exiting function" in log_line.log_data ): date_delta = logline_timestamp.isoformat(timespec="microseconds") else: date_delta = self.get_timestamp_delta( logline_timestamp.isoformat(timespec="microseconds"), last_timestamp.isoformat(timespec="microseconds"), line_count, ) line_count += 1 last_timestamp = logline_timestamp out_line = ( log_line.module_name + " : " + date_delta + " " + split_char + " " + log_line.log_data ) try: print(out_line) except Exception: print("".join([i if ord(i) < 128 else "#" for i in out_line])) class LogLineObject: def __init__(self, timestamp, module_name="", log_data=""): self.timestamp = timestamp self.module_name = module_name self.log_data = log_data if __name__ == "__main__": log_processor = DockerLogProcessor(sys.argv[1:])