Custom Python 'Server Status Checker'

Question

I've written a home-grown solution for checking the status of my servers. This is both an exercise for me in learning further about Python, but it's also critical to status-checking critical things on my "network" of servers - both running against both an internal ("home") network against 18 or 19 VMs internally I need to check are up or not, but also against 17 external servers (VPSes) (all are defined by either hostname or IP in a list of triples, as explained in the comments on the code). It's ideally being executed on a box via cron or some type of automatic scheduler and then email me a notification at a specified address of which 'servers' are not up (if and only if there's no 'down' servers and every server in the list is 'up') by checking the primary service running to an externally-bound port through a TCP socket (either SSL or Plain text, defined in the triples per-server).

This definitely works fine, but I have a few concerns with the way I've written this, because it seems to me to be a hackish approach to achieve things at times:

1. Correct Handling of Connections. I think that the way I'm doing 'works', but could be much better handled than the way I've written it. I'm also fearful of 'unexpected' cases where it could fail, though I've not seen any thus far.

2. Email handling. I've run into this in other times at the workplace where I've had to "modernize" python code from a 2.4 or 2.5 codebase to a 2.7 codebase, as well as to put it on a new server. The cases I'm afraid of are timeouts, or any type of unhandled cases where emailing can just flat out die with unexpected errors. In such cases, the process should just die off, but I am fairly certain I'm handling those cases poorly.

   Note that because this runs inside my home network on a dynamic residential ISP connection, I have to use an external SMTP, and Google works because I have Google Apps (grandfathered in from the "Free" tier of old) handling mail on some of my domains (and in cases like this one). I'm not concerned with that going down, but I'm concerned the SMTP connection is not being handled properly either (edge cases that'll trigger uncaught exceptions)

3. Any other performance concerns. I've not run into any, but on the rare 'slow connection' on my network due to my installing servers or creating datastores on my hypervisors, I'm sure I'll hit some kind of performance bottleneck - any recommendations on improvement are welcome.

4. Recommended practices which I'm not following. I come from a SysAdmin background, not a programming background, so I'm probably not following some of the 'recommended practices'. If there's recommended practices I'm not following, please let me know, I'm happy to use recommended practices instead of hackish ones!

What I don't want commented upon is the use of the "Print" statements. I keep a run/debug log for each run (courtesy of cron and piping data to a file for storage), and the print statements currently handle this. I'm going to be using a logging type logger in the future, as I have for other projects, but for now I'm content with the 'print' statements and a text file dumped to of console-level output.

---

The Code (server_checker.py):

#!/usr/bin/env python

import socket
from datetime import datetime
from email.mime.text import MIMEText
import smtplib
import atexit
import ssl

# #### VARIABLES #### #

# list of servers to check with the following items in the
# definitions per-server: ('hostname', 'ssl or plain', portnumber)
SERVER_LIST = [
    ('server1', 'plain', 80),
    ('server2', 'ssl', 443),
    ('server3', 'ssl', 8443),
    ]

# Globally define these lists as 'empty' for population later.
SRV_DOWN = []
SRV_UP = []

# Email handling items - email addresses
ADMIN_NOTIFY_LIST = ["an_email_address"]
FROM_ADDRESS = "no-reply@foo-bar-baz"

# Valid Priorities for Mail
LOW = 1
NORMAL = 2
HIGH = 3


# Begin Execution Here

@atexit.register
def _exit():
    print "%s  Server Status Checker Now Exiting." % (current_timestamp())


def current_timestamp():
    return datetime.now().strftime("[%Y-%m-%d %H:%M:%S]")


def send_server_status_report():
    # Init priority - should be NORMAL for most cases, so init it to that.
    priority = NORMAL

    # Init the send_mail flag.  Ideally, we would be sending mail if this function is
    # called, but we need to make sure that there are cases where it's not necessary
    # such as when there are no offline servers.
    send_mail = True

    if len(SRV_UP) == 0:
        up_str = "Servers online: None!  ***THIS IS REALLY BAD!!!***"
        priority = HIGH
    else:
        up_str = "Servers online: " + ", ".join(SRV_UP)

    if len(SRV_DOWN) == 0:
        down_str = "Servers down: None!"
        send_mail = False
    else:
        down_str = "Servers down: " + ", ".join(SRV_DOWN) + "   ***CHECK IF SERVERS LISTED ARE REALLY DOWN!***"
        priority = HIGH

    if len(SRV_UP) == len(SERVER_LIST) and len(SRV_DOWN) == 0:
        priority = LOW

    if send_mail:
        body = """Server Status Report - %s

    %s

    %s""" % (current_timestamp(), down_str, up_str)

        # craft msg base
        msg = MIMEText(body)
        msg['Subject'] = "Server Status Report - %s" % (current_timestamp())
        msg['From'] = FROM_ADDRESS
        msg['Sender'] = FROM_ADDRESS  # This is sort of important...

        if priority == LOW:
            # ThunderBird "Lowest", works with Exchange.
            msg['X-Priority'] = '5'
        elif priority == NORMAL:
            # Plain old "Normal". Works with Exchange.
            msg['X-Priority'] = 'Normal'
        elif priority == HIGH:
            # ThunderBird "Highest", works with Exchange.
            msg['X-Priority'] = '1'

        # Initialize SMTP session variable so it has the correct scope
        # within this function, rather than just inside the 'try' statement.
        smtp = None

        try:
            # SMTP is important, so configure it via Google Mail.
            smtp = smtplib.SMTP('smtp.gmail.com', 587)
            smtp.starttls()
            smtp.login(FROM_ADDRESS, 'ThePassword')
        except Exception as e:
            print "Could not correctly establish SMTP connection with Google, error was: %s" % (e.__str__())
            exit()

        for destaddr in ADMIN_NOTIFY_LIST:
            # Change 'to' field, so only one shows up in 'To' headers.
            msg['To'] = destaddr

            try:
                # Actually send the email.
                smtp.sendmail(FROM_ADDRESS, destaddr, msg.as_string())
                print "%s  Status email sent to [%s]." % (current_timestamp(), destaddr)
            except Exception as e:
                print "Could not send message, error was: %s" % (e.__str__())
                continue

        # No more emails, so close the SMTP connection!
        smtp.close()
    else:
        print "%s  All's good, do nothing." % (current_timestamp())


def main():
    for (srv, mechanism, port) in sorted(SERVER_LIST):
        # [ 'serverhost' , 'ssl' or 'plain' ]
        print srv, ", ", mechanism, ", ", port

        try:
            if mechanism == 'plain':
                # Use a plain text connector for this.
                print "%s  Using Plain for [%s]..." % (current_timestamp(), srv)
                socket.create_connection(("%s.layerbnc.org" % srv, port), timeout=10)
            elif mechanism == 'ssl':
                # We're going to use an SSL connector for this.
                print "%s  Using SSL for [%s]..." % (current_timestamp(), srv)
                ssl.wrap_socket(socket.create_connection(("%s" % srv, port), timeout=10))
            else:
                print "%s  Invalid mechanism defined for [%s], skipping..." % (current_timestamp(), srv)
                continue
            SRV_UP.append(srv)
            print "%s  %s: UP" % (current_timestamp(), srv)
        except socket.timeout:
            SRV_DOWN.append(srv)
            print "%s  %s: DOWN" % (current_timestamp(), srv)
            continue
        except Exception as err:
            print "An error occurred: %s" % (err.__str__())
            exit()

    send_server_status_report()  # Create email to send the status notices.

    exit()  # Exit when done


if __name__ == "__main__":
    print "%s  Server Status Checker Running...." % (current_timestamp())
    main()

Answer 1

Nice script. Here are a few points that can be improved.

---

You could use a dictionnary to handle the conversion from priority to x-priority.

PRIORITY_TO_XPRIORITY = {
    LOW: '5',  # ThunderBird "Lowest", works with Exchange.
    NORMAL: 'Normal',  # Plain old "Normal". Works with Exchange.
    HIGH: '1',  # ThunderBird "Highest", works with Exchange.
}
...
msg['X-Priority'] = PRIORITY_TO_XPRIORITY[priority]

---

You don't need the continue in your for loop after the socket.timeout.

---

You probably should fail in a more explicit way when given an invalid connection mechanism.

More generally, having the exit() function called from anywhere makes things a bit hard to track.

---

You do not need global variable to propagate the list of servers up/down : a list populated in the main and fed to send_server_status_report should do the trick.

---

You could define a function to check a server status.

Then you could call it to know if the server is up without bothering about the way it is done.

You main function simply becomes :

def server_is_up(server):
    srv, mechanism, port = server
    try:
        if mechanism == 'plain':
            # Use a plain text connector for this.
            print "%s  Using Plain for [%s]..." % (current_timestamp(), srv)
            socket.create_connection(("%s.layerbnc.org" % srv, port), timeout=10)
        elif mechanism == 'ssl':
            # We're going to use an SSL connector for this.
            print "%s  Using SSL for [%s]..." % (current_timestamp(), srv)
            ssl.wrap_socket(socket.create_connection(("%s" % srv, port), timeout=10))
        else:
            print "%s  Invalid mechanism defined for [%s], skipping..." % (current_timestamp(), srv)
            raise ValueError("Invalid mechanism")
        print "%s  %s: UP" % (current_timestamp(), srv)
        return True
    except socket.timeout:
        print "%s  %s: DOWN" % (current_timestamp(), srv)
        return False
    except Exception as err:
        print "An error occurred: %s" % (err.__str__())
        exit()


def main():
    srv_up = []
    svr_down = []
    for server in sorted(SERVER_LIST):
        (srv_up if server_is_up(server) else svr_down).append(server[0])
    send_server_status_report(svr_up, svr_down)  # Create email to send the status notices.

    exit()  # Exit when done

---

I do not think you need to define smtp being None before the try/catch. Scopes in Python do not quite work like in other languages.

---

I know you do not want to talk about this but your current_timestamp function could be removed if you were to use a proper logging function. If you really do not want to do this, you could at least define a log function doing exactly what you are doing but calling the current_timestamp function automatically so that you do not have to do this every where in your code.