diff --git a/README.md b/README.md index 2e6b77d..48ffe24 100644 --- a/README.md +++ b/README.md @@ -61,35 +61,36 @@ To install any optional dependencies, such as development dependencies, use: pip install -e .[dev] ``` -### Filling Out Config File +## Making the executable(optional) + +Run this in your terminal to create the gecko executable: -## Logs -Logs will have the path to any and all logs you would like to comb upon execution. This includes the system logs on more . This tool with comb the last 24 hours for ERRORS or WARNINGS assuming the log timestamps are formatted as ```bash -time_pattern = r"^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})" +cd gecko +chmod +x gecko ``` -You can change this format the the source code if you are familiar with python. -Example Log Path: +### Filling Out Config File + +You can run the command below and you will be prompted with questions to fill out the config file. Read promps and answer accordingly: + ```bash -FAM_Logs = ~/LOGS/data/devices/FAM/FAM.log +python3 gecko -init +(or ./gecko -init) ``` -## System -This holds the information for your Linux system and version. --TODO: Implent cross platform utility legacy system compatability +### Make Executable avaiable anywhere in the system + +Symlink the executable file to bin: -## VNC -Your VNC credentials and all of the sessions the engineers may want screenshots of. ```bash -host = "hostname.outlook.com" -password = "Hello_Sky" -vnc_sessions = 1,2,3,4 +sudo ln -s /path/to/repo/Gecko/gecko /usr/local/bin/gecko ``` -## Report -Fill out contact information and path to report storage +### Generate a report -## Executing from anywhere -Make sure your global python enviornment has all dependices and move executable into bin +You can now generate gecko reports from any terminal on the machine! (assuming python is available globally) +```bash +./gecko -m "" +``` \ No newline at end of file diff --git a/gecko b/gecko index aea07fc..144a7f6 100755 --- a/gecko +++ b/gecko @@ -19,30 +19,65 @@ import argparse from triage_package.triage_tool import Triagetools #Fill out section then run like a CLI executable!############### -CONFIG_PATH = "/home/elijahab/general-dev/gecko/triage_package/triage.ini" +CONFIG_PATH = "triage_package/triage.ini" ################################################################ def run_triage(): '''Executable file for use on the Command line''' parser = argparse.ArgumentParser(description="Gecko command-line tool") - parser.add_argument( + + parser = argparse.ArgumentParser( + description=""" + Gecko Triage Tool + + A command-line utility to collect system and application logs, take screenshots, + grab science images, and optionally send a compressed report via email. + + You can either: + 1) Initialize the application (--initialize) to configure settings. + 2) Run a triage report (--message "description") to gather logs and send a report. + """, + formatter_class=argparse.RawTextHelpFormatter + ) + + # Mutually exclusive: either initialize or provide a message + group = parser.add_mutually_exclusive_group(required=True) + + group.add_argument( + "-init","--initialize", + action="store_true", + help="Initialize the application" + ) + + group.add_argument( "-m", "--message", type=str, - help="Commit message or description" + help="Run the triage workflow with a message describing the issue. " + "Example: --message 'GPU overheating issue observed today'" ) + args = parser.parse_args() - if args.message: - print(f"Message received: {args.message}") - gecko = Triagetools(config=CONFIG_PATH,message = args.message) - gecko.gather_system_info() + if args.initialize: + gecko = Triagetools(config=CONFIG_PATH, init=True) + print('\nInitialization complete!') + print('You can now run the GUI or use:\n gecko -m "your issue message" \n') + elif args.message: + gecko = Triagetools(config=CONFIG_PATH, message=args.message) + print(f"Running triage workflow for message: {args.message}\n") + gecko.gather_system_info() gecko.gather_logs() gecko.comb_logs() gecko.take_screenshots() + gecko.grab_science_image() gecko.compress_report() - #gecko.send_report() + + #if gecko.email_alerts: + # gecko.send_report() + print("\nTriage workflow complete!") + if __name__ == "__main__": run_triage() diff --git a/triage_package/triage.ini b/triage_package/triage.ini index 3e143f9..839b5ec 100644 --- a/triage_package/triage.ini +++ b/triage_package/triage.ini @@ -1,31 +1,66 @@ -# Configuration file for Triage Tool -# Only change valus not names of parameters [System] -#***Does Nothing with this info right now*** -os = +help_text = + This section will have ask you to provide your os and version to take not of in the report. + (EXAMPLES) + os: ubuntu + os_version: 24.04 + + +os = ubuntu os_version = +initialized = -#Path to log file locations [Logs] -#Example: system = /var/log/syslog -#system = /home/path/to/logging/blue_cal_gammavac.log +help_text = + Please provide the path to your logs and science directory + The logs and science directory should be composed of dated dir in UTC format for the program to iterate through. + (EXAMPLES) In this example latest is symliked to the most current date + logs_dir: /data/latest/logs/ + science_dir: /data/latest/ + + +logs_dir = +science_dir = [Report] -#Enable or disable email alerts -email_alerts=True #***Does Nothing with this info right now*** -#Email address to send alerts to -instrument_master_email="" #***Does Nothing with this info right now*** -#Where to store generated reports (full path) -report_path= #/home/user/dir/reports +help_text = + This section will ask for details on the reports. If you would not like email notifications, you can just provide a reports path. + (EXAMPLES) + report_path: /home/user/dir/reports + time_pattern: ^(\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}(?:\.\d+)?) + email_alerts: false + instrument_master_email: eng@observatory.edu + sender_email: hello@gmail.com + sender_password: password + + +report_path = +time_pattern = ^(\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}(?:\.\d+)?) +email_alerts = +instrument_master_email = +sender_email = +sender_password = [Machine] -#Threshold for alerting on high CPU usage (in percentage) -cpu_threshold=85 #***Does Nothing with this info right now*** -#Threshold for alerting on high Memory usage (in percentage) -memory_threshold=90 #***Does Nothing with this info right now*** +help_text = + Please provice percentage thresholds to monitor in the machine system. + (EXAMPLES) + cpu_threshold: 85 + memory_threshold: 90 + + +cpu_threshold = 75 +memory_threshold = 88 [VNC] -#IP address or host of session -host = #host.provider.com host.iden.edu -password = #Password1234 -vnc_sessions = #1,2,3,4,5,12 +help_text = + Please provide VNC details: + (EXAMPLES) + host: host.provider.com/host.iden.edu + password: Password1234 + vnc_sessions: 1,2,3,4,5,12 + + +host = +password = +vnc_sessions = diff --git a/triage_package/triage_tool.py b/triage_package/triage_tool.py index 34b9acc..33d89d5 100755 --- a/triage_package/triage_tool.py +++ b/triage_package/triage_tool.py @@ -17,16 +17,15 @@ -Elijah Anakalea-Buckley ''' import os +import sys import tarfile import shutil import configparser import smtplib -#import argparse -#from email.mime.text import MIMEText from email.message import EmailMessage import re from datetime import datetime, timezone, timedelta -#import glob +import glob import threading import socket import psutil @@ -35,7 +34,7 @@ class Triagetools(object): """Triage tool for bug catching and error reporting""" - def __init__(self, config: str, message:str = ""): + def __init__(self, config: str, message:str = "", init = False): # Grab UTC, and load config file self.config_file = config self.utc_date = datetime.now(timezone.utc) @@ -43,32 +42,86 @@ def __init__(self, config: str, message:str = ""): self.utc_time = str(self.utc_date.time()) self.cutoff = datetime.now().replace(tzinfo=None) - timedelta(hours=24) self.message = message - self.time_pattern = r"^(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})" + self.time_pattern = r"^(\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}(?:\.\d+)?)" - # Load config file - self.load_config(self.config_file) + # Create a ConfigParser object + self.config = configparser.ConfigParser() + + # Load config file if called + if init is True: + self.initialize(self.config_file) + else: + self.load_config(self.config_file) + + def initialize(self, config): + ''' Initialize and configure the application ''' + if not os.path.exists(config): + raise FileNotFoundError(f"Configuration file {config} not found") + self.config.read(config) + + #Iterate through prompts and populate ini file + for section in self.config.sections(): + print(f"\n\n====== Initializing Section: {section} ======") + for key, value in self.config[section].items(): + if key == "initialized": + continue + if key == "help_text": + print(value.rstrip()) + print() # extra blank line + continue + # Remove inline comments (anything after '#') + clean_value = value.split('#')[0].strip() + # Prompt user; show current value if it exists + prompt = f"Enter value for '{key}'" + if clean_value: + prompt += f" (current: {clean_value})" + prompt += ": " + user_input = input(prompt).strip() + + # If user presses enter, keep existing value + if user_input == "": + user_input = clean_value + + # Save updated value + self.config[section][key] = user_input + + #set initialized to true for later execution and write + self.config["System"]["initialized"] = "true" + with open(self.config_file, "w", encoding="utf-8") as config_file: + self.config.write(config_file) def load_config(self, config): ''' Load the configuration file ''' if not os.path.exists(config): raise FileNotFoundError(f"Configuration file {config} not found") - - # Create a ConfigParser object - self.config = configparser.ConfigParser() self.config.read(config) + initialized = config.getboolean("System", "initialized") + + if not initialized: + print("\nYou have not Initialized this application! Please run: 'gecko -init'\n") + sys.exit(0) + #Report Section(report file name and current UTCdate folder) - if self.config["Report"]["email_alerts"].lower().strip() == "true": - self.email_alerts = True + self.email_alerts = config.getboolean("Report", "email_alerts") + if self.email_alerts: self.target_email = self.config["Report"]["instrument_master_email"] - else: - self.email_alerts = False + self.sender_email = self.config["Report"]["sender_email"] + self.sender_password = self.config["Report"]["sender_password"] self.r_path = self.config["Report"]["report_path"] + self.log_dir = self.config["Logs"]["logs_dir"] + self.science_dir = self.config["Logs"]["science_dir"] self.reports_path = f"{self.r_path}/{self.current_utc_date}" if not os.path.exists(f"{self.reports_path}"): os.mkdir(f"{self.reports_path}") self.report_name = f"{self.reports_path}/gecko_report_{self.utc_time}.txt" self.regex_pattern = r"^.*error.*$|^.*warning.*$" + log_time_pattern = self.config["Reports"]["time_pattern"] + try: + # document the regex + self.time_pattern = re.compile(log_time_pattern) + except re.error as e: + raise ValueError(f"Invalid regex in config: {e}") #Machine Section self.cpu_threshold = self.config["Machine"]["cpu_threshold"] @@ -109,9 +162,11 @@ def gather_system_info(self): temps = psutil.sensors_temperatures() #Dict # Memory virtual_memory = psutil.virtual_memory() # + system_str = f"{self.os} :: {self.os_version}" with open(self.report_name, 'a', encoding='utf-8') as file: file.write("\n\n=====System Information=====\n") + file.write(f'{system_str}\n') file.write(f'Logical CPUs: {cpu_count_logical}\n') file.write(f'Physical CPUs: {cpu_count_physical}\n') file.write('Detailed CPU Usage:\n') @@ -187,62 +242,56 @@ def _worker(): def gather_logs(self): ''' Gathers Logs from all paths to dump into tar comp. ''' #Iterate through log paths in config file - for log in self.config["Logs"]: - log_path = self.config["Logs"][log] - - #Check that a log path exist - if os.path.exists(log_path): + for subdir, dirs, files in os.walk(self.log_dir): # pylint: disable = W0612 + for file in files: + log_file = os.path.join(subdir, file) #Copy over to logs folder try: - shutil.copy2(log_path,f"{self.reports_path}") + shutil.copy2(log_file,f"{self.reports_path}") except FileNotFoundError: - print(f"Error: The file at {log_path} was not found.") + print(f"Error: The file at {log_file} was not found.") except PermissionError: - print(f"Error: Permission denied to access the file at {log_path}.") + print(f"Error: Permission denied to access the file at {log_file}.") except Exception as e: # pylint: disable = W0718 print(f"An unexpected error occurred: {e}") - else: - print(f"Log path {log_path} does not exist.") - def comb_logs(self): ''' Comb logs for errors and warnings ''' - #Iterate through log paths in config file - for log in self.config["Logs"]: - path = self.config["Logs"][log] - #Edit this to format to LOG structure - log_path = f"{path}/{self.utc_date}/" - #Put log_path into report file - with open(self.report_name, 'a', encoding='utf-8') as file: - file.write(f"\n\n====={log_path}=====\n") - if os.path.exists(log_path): - print(f"Gathering log from: {log_path}") - try: - with open(log_path, 'r', encoding='utf-8') as log_file: - #Search for occurances of warnings and errors sequentially - full_log = log_file.read() - - #Use Regex - matches = (re.findall(self.regex_pattern, full_log, - re.IGNORECASE | re.MULTILINE)) - timeframe_matches = [ - match for match in matches - if (m := re.match(self.time_pattern, match)) - and datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S") >= self.cutoff - ] - - with open(self.report_name, 'a', encoding='utf-8') as report_file: - for match in timeframe_matches: - report_file.write(f"{match}\n") - except FileNotFoundError: - print(f"Error: The file at {log_path} was not found.") - except PermissionError: - print(f"Error: Permission denied to access the file at {log_path}.") - except Exception as e: # pylint: disable = W0718 - print(f"An unexpected error occurred: {e}") - else: - print(f"Log path {log_path} does not exist.") + for subdir, dirs, files in os.walk(self.log_dir): # pylint: disable = W0612 + for file in files: + if file.lower().endswith(".log"): + log_path = os.path.join(subdir, file) + with open(self.report_name, 'a', encoding='utf-8') as file: + file.write(f"\n\n====={log_path}=====\n") + print(f"Gathering log from: {log_path}") + try: + with open(log_path, 'r', encoding='utf-8') as log_file: + #Search for occurances of warnings and errors sequentially + full_log = log_file.read() + + #Use Regex + matches = (re.findall(self.regex_pattern, full_log, + re.IGNORECASE | re.MULTILINE)) + #timeframe_matches = [ + # match for match in matches + # if (m := re.match(self.time_pattern, match)) + # and datetime.strptime(m.group(1), "%Y-%m-%d %H:%M:%S") >= + # self.cutoff] + timeframe_matches = [ + match for match in matches + if (m := re.match(self.time_pattern, match)) + and datetime.fromisoformat(m.group(1)) >= self.cutoff] + + with open(self.report_name, 'a', encoding='utf-8') as report_file: + for match in timeframe_matches: + report_file.write(f"{match}\n") + except FileNotFoundError: + print(f"Error: The file at {log_path} was not found.") + except PermissionError: + print(f"Error: Permission denied to access the file at {log_path}.") + except Exception as e: # pylint: disable = W0718 + print(f"An unexpected error occurred: {e}") def compress_report(self): '''Compresses report file into a tar.gz format to be emailed''' @@ -250,23 +299,86 @@ def compress_report(self): with tarfile.open(f"{self.reports_path}/gecko_{self.utc_date}.tar.gz", "w:gz") as tar: tar.add(f"{self.reports_path}", arcname=os.path.basename(f"{self.reports_path}")) + def grab_science_image(self): + '''Grabs most recent science image(s) to include in triage''' + #Edit this section to fit instrument data schema + image_dirs = [ + f"{self.science_dir}/acam", + f"{self.science_dir}/slicecam", + self.science_dir, + ] + #Iterate through dirs and grab most recent modified file, + # which should be the most recent image taken + for i_dir in image_dirs: + try: + if not os.path.exists(i_dir): + with open(self.report_name, 'a', encoding='utf-8') as file: + file.write(f"\nScience Directory does not Exist: {dir}\n") + else: + files = [ + os.path.join(i_dir, f) + for f in os.listdir(i_dir) + if os.path.isfile(os.path.join(i_dir, f)) + ] + + if not files: + raise FileNotFoundError("No files found in source directory.") + + # Pick newest by modification time + newest_file = max(files, key=os.path.getmtime) + + # Copy to destination + shutil.copy(newest_file, self.reports_path) + except FileNotFoundError as e: + with open(self.report_name, 'a', encoding='utf-8') as file: + file.write(f"\nSource Directory: {i_dir}\n--{e}\n") + def send_report(self): - ''' Send the generated report to specified recipients ''' - #send message using smtplib - #format message + """Send the generated report to specified recipients.""" + + # Create email msg = EmailMessage() - msg ['Subject'] = f'' #pylint: disable = W1309 - msg['From'] = '' - msg['To'] = self.target_email + msg['Subject'] = f'Gecko Report {self.utc_date}' + msg['From'] = self.sender_email # replace with actual sender + msg['To'] = self.target_email # can be comma-separated string or list + + # Email body + msg.set_content("Please see attached report images.") + + #.txt file first + with open(self.report_name, 'rb') as f: + msg.add_attachment( + f.read(), + maintype='text', + subtype='plain', + filename=os.path.basename(self.report_name) + ) + + #tar.gz file next + tar_file = f"{self.reports_path}/gecko_{self.utc_date}.tar.gz" + with open(tar_file, 'rb') as f: + msg.add_attachment( + f.read(), + maintype='application', + subtype='gzip', + filename=os.path.basename(tar_file) + ) + + # Attach PNG images recursively from the reports_path + image_files = glob.glob(os.path.join(self.reports_path, '**', '*.png'), recursive=True) + for file in image_files: + with open(file, 'rb') as fp: + img_data = fp.read() + filename = os.path.basename(file) + msg.add_attachment(img_data, maintype='image', subtype='png', filename=filename) - #image_files = glob.glob(os.path.join(self.reports_path,'**', '*.png'), recursive=True) + ## Send email using local SMTP server + #with smtplib.SMTP('localhost') as sender: + # sender.send_message(msg) - # TODO: test add attatchment. - msg.add_attachment() - #for file in image_files: - # with open(file,'rb') as fp: - # img_data = fp.read() - # msg.add_attachment(img_data, maintype='image', subtype='png') + # Connect to Gmail SMTP server + with smtplib.SMTP_SSL('smtp.outlook.com', 465) as smtp: + smtp.login(self.sender_email, self.sender_password) # use an App Password + smtp.send_message(msg) - sender = smtplib.SMTP('localhost') - sender.quit() + print(f"Report sent to {self.target_email}")