· 7 years ago · Sep 07, 2018, 04:48 PM
1#! /usr/bin/env python3
2"""
3A simple job monitor for Python.
4usage: pycvl_monitor [-h] [--email EMAIL] [--command COMMAND | --job JOB]
5
6optional arguments:
7 -h, --help show this help message and exit
8 --email EMAIL If specified, an email will be sent to the provided email
9 address.
10 --command COMMAND The command to execute.
11 --job JOB Provide a YAML job file with the tasks to execute.
12 Specify one command per line.
13
14Example of typical job description file
15```
16email: "email@example" # comment out if you don't need email notifications
17
18jobs:
19 - python3 --version
20 - python3 --version
21```
22
23"""
24
25__author__ = "Sérgio Agostinho"
26__email__ = "sergio.agostinho@tecnico.ulisboa.pt"
27__license__ = "Apache v2"
28
29from argparse import ArgumentParser
30import getpass
31import re
32import shlex
33import smtplib
34import socket
35import subprocess
36import sys
37import tempfile
38import time
39
40import yaml
41
42_MAIL_INFO = {
43 "gmail.com": "smtp.gmail.com:587",
44 "tecnico.ulisboa.pt": "mail.tecnico.ulisboa.pt:587"
45}
46
47_messages = {
48 "failure-task" : ("Task Failed", "The requested command:\n{:s}\naborted "
49 "with the error code {:d}.\n"),
50 "failure-job" : ("Job Failed", lambda jobs: "The job completed with errors.\n" \
51 + report(jobs) + "\n"),
52 "success" : ("Job Completed", lambda jobs: "The job completed successfully.\n" \
53 + report(jobs) + "\n")
54}
55
56
57def report(jobs):
58 """Generates the final job listing with execution times and error codes."""
59
60 content = []
61 total_time = 0
62 for job in jobs:
63 msg = "[{}] {:d} {:s}".format(job["time"], job["code"], job["cmd"])
64 if job["code"]:
65 msg += "\n" + job["error"]
66 content.append(msg)
67
68 total_time += job["time"]
69
70 msg = "\n".join(content) + "\nTotal execution time {}.".format(total_time)
71 return msg
72
73
74def send_email(content, auth):
75 """Sends an email with the provided content and specified credentials."""
76
77 # Parse domain
78 email = auth["email"]
79 domain = email.split("@")[1]
80 if not domain:
81 print("Unable to find domain.", file=sys.stderr)
82 return
83
84 # Prepare message content
85 subject, body = content
86 subject = "[" + socket.getfqdn() + "] " + subject # append hostname
87 header = "From: {0}\r\nTo: {0}\r\n" \
88 "Subject: {1}\r\nX-Mailer: My-Mail\r\n\r\n".format(email, subject)
89
90 # Prepare session
91 smtp_url = _MAIL_INFO[domain]
92 smtp = smtplib.SMTP(smtp_url)
93 smtp.starttls()
94 smtp.login(auth["user"], auth["password"])
95 smtp.sendmail(email, email, header + body)
96 smtp.quit()
97
98
99def validate_email_credentials(auth):
100 """Validates SMTP credentials."""
101
102 # Parse domain
103 email = auth["email"]
104 domain = email.split("@")[1]
105 if not domain:
106 print("Unable to find domain.", file=sys.stderr)
107 return
108
109 # Prepare session
110 dirty = False
111 smtp_url = _MAIL_INFO[domain]
112 smtp = smtplib.SMTP(smtp_url)
113 smtp.starttls()
114 try:
115 smtp.login(auth["user"], auth["password"])
116 except smtplib.SMTPAuthenticationError as e:
117 dirty = True
118 print("Authentication error: {}".format(e), file=sys.stderr)
119 except:
120 dirty = True
121 print("Unhandled exception raised.", file=sys.stderr)
122 smtp.quit()
123 return not dirty
124
125
126# Set up argument parser
127parser = ArgumentParser()
128parser.add_argument(
129 "--email",
130 help="If specified, an email will be sent to the "
131 "provided email address.")
132meg = parser.add_mutually_exclusive_group()
133meg.add_argument("--command", help="The command to execute.")
134meg.add_argument(
135 "--job",
136 help="Provide a YAML job file with the tasks to "
137 "execute. Specify one command per line.")
138
139# parse arguments
140args = parser.parse_args()
141
142# Load command files
143cmds = []
144if args.command:
145 cmds.append(args.command)
146elif args.job:
147 configuration = None
148 with open(args.job) as f:
149 configuration = yaml.load(f.read())
150
151 # overwrite email if in configuration
152 if "email" in configuration:
153 args.email = configuration["email"]
154
155 cmds += configuration["jobs"]
156
157# If an email is privided ask for SMTP password
158auth = {"email": args.email, "user": args.email, "password": None}
159if args.email:
160 valid = False
161 while not valid:
162 print("SMTP username [" + args.email + "]: ", end="")
163 user = input()
164 if user:
165 auth["user"] = user
166 auth["password"] = getpass.getpass(
167 prompt="SMTP password for " + args.email + ": ")
168 valid = validate_email_credentials(auth)
169 print("Credentials accepted.")
170
171# Execute all requires tasks
172dirty = False
173jobs = []
174for cmd in cmds:
175
176 d = {"cmd": None, "time": None, "code": 0, "error": ""}
177
178 with tempfile.TemporaryFile() as fp:
179
180 # Launch the process
181 d["cmd"] = cmd
182 t_start = time.time()
183 info = subprocess.run(shlex.split(cmd), stderr=fp)
184 t_end = time.time()
185 d["time"] = t_end - t_start
186 d["code"] = info.returncode
187
188 # store error messages
189 fp.seek(0)
190 d["error"] = fp.read().decode("utf-8")
191
192 # Process results
193 if not dirty and info.returncode:
194 dirty = True
195 content = (_messages["failure-task"][0],
196 _messages["failure-task"][1].format(cmd, info.returncode) \
197 + d["error"])
198 print(content[1])
199 if args.email:
200 send_email(content=content, auth=auth)
201
202 jobs.append(d)
203
204# Print status and send notification if needed
205status = "failure-job" if dirty else "success"
206body = _messages[status][1](jobs)
207print(body)
208if args.email:
209 subject = _messages[status][0]
210 send_email(content=(subject, body), auth=auth)