Network Checker

client.py

import getopt
import uuid
import sqlite3
import sys
from datetime import datetime

import os
import time
import socket

DEFAULT_HOST = "localhost"
DEFAULT_PORT = 6000

INTERVAL = 2
NODE_ID = uuid.uuid1()

ROOT_DIR = os.path.dirname(os.path.realpath(__file__))
HEARTBEAT_PATH = os.path.join(ROOT_DIR, 'heartbeat')
DATABASE_PATH = os.path.join(ROOT_DIR, 'records.{}.db'.format(NODE_ID))

socks = {}


def get_sock_or_connect(target_host, target_port):
    if NODE_ID not in socks:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.connect((target_host, target_port))
        socks[NODE_ID] = sock
    return socks[NODE_ID]


def connect_db():
    conn = sqlite3.connect(DATABASE_PATH)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS records
            (timestamp text, message text, exception text)
    ''')
    return conn


def record_exception(exc):
    print(exc)
    conn = connect_db()
    cursor = conn.cursor()
    query = """
        INSERT INTO records VALUES ('{}', '', '{}')
    """.format(
        now(), str(exc)
    )
    cursor.execute(query)
    conn.commit()


def record_success(res):
    conn = connect_db()
    cursor = conn.cursor()
    query = """
        INSERT INTO records VALUES ('{}','OK:{}', '')
    """.format(
        now(), res
    )
    cursor.execute(query)
    conn.commit()

    with open(HEARTBEAT_PATH, 'w') as fh:
        fh.write(now())


def now():
    return datetime.now().replace(microsecond=0).isoformat()


def build_msg():
    return "{}|{}|{}\n".format(NODE_ID, now(), socket.gethostname())


def run(host, port):

    while True:
        try:
            time.sleep(INTERVAL)

            try:
                sock = get_sock_or_connect(host, port)
            except OSError as exc:
                record_exception(exc)
                continue

            try:
                msg = build_msg()
                sent = sock.sendall(msg.encode('utf-8'))
                if sent == 0:
                    raise IOError("sent zero bytes")
                res = sock.recv(1024)
                if len(res) == 0:
                    raise IOError("recv zero bytes")
                record_success(res.decode('utf-8'))
            except IOError as exc:
                del socks[NODE_ID]
                record_exception(exc)

        except (SystemExit, KeyboardInterrupt):
            break


def main(argv):
    USAGE = 'client.py [-H <target_host>] [-p <target_port>]'
    try:
        opts, args = getopt.getopt(argv, "hH:p:", ["host=", "port="])
    except getopt.GetoptError:
        print(USAGE)
        sys.exit(2)

    host = DEFAULT_HOST
    port = DEFAULT_PORT

    for opt, arg in opts:
        if opt == '-h':
            print(USAGE)
            sys.exit()
        elif opt in ("-H", "--host"):
            host = arg
        elif opt in ("-p", "--port"):
            port = int(arg)

    print(host, port)
    run(host, port)


if __name__ == "__main__":
    main(sys.argv[1:])

healthcheck.py

import getopt
import os
import subprocess
import sys
from datetime import datetime, timedelta

ACCEPTABLE_DELTA = 3  # seconds

DEFAULT_HOST = "localhost"
DEFAULT_PORT = 6000

ROOT_DIR = os.path.dirname(os.path.realpath(__file__))
HEARTBEAT_PATH = os.path.join(ROOT_DIR, 'heartbeat')
SCRIPT_PATH = os.path.join(ROOT_DIR, 'client.py')
PID_PATH = os.path.join(ROOT_DIR, 'client.pid')
LOG_PATH = os.path.join(ROOT_DIR, 'client.out')


def now():
    return datetime.now().replace(microsecond=0)


def process_exists(pid):
    """ Check For the existence of a unix pid. """
    try:
        os.kill(pid, 0)
    except OSError:
        return False
    else:
        return True


def bootstrap_client(host, port):

    if os.path.isfile(PID_PATH):
        with open(PID_PATH, 'r') as fh:
            pid = int(fh.read().strip())

        if process_exists(pid):
            print("client already running")
            return

    args = ["python3", "-u", SCRIPT_PATH, "-H", host, "-p", str(port)]
    with open(LOG_PATH, 'w') as log_fh:
        process = subprocess.Popen(
            args, stdout=log_fh, stderr=subprocess.STDOUT, close_fds=True
        )

    with open(PID_PATH, 'w') as fh:
        fh.write(str(process.pid))


def run():

    last_beat = datetime.min

    if os.path.isfile(HEARTBEAT_PATH):
        with open(HEARTBEAT_PATH) as fh:
            heartbeat = fh.read()
        last_beat = datetime.strptime(heartbeat, "%Y-%m-%dT%H:%M:%S")

    delta = now() - last_beat
    if delta < timedelta(seconds=ACCEPTABLE_DELTA):
        print("OK: last beat {}s ago".format(delta.seconds))
    else:
        sys.exit("BAD: last beat {}s ago".format(delta.seconds))


def main(argv):

    USAGE = 'healthcheck.py [-H <target_host>] [-p <target_port>]'
    try:
        opts, args = getopt.getopt(argv, "hH:p:", ["host=", "port="])
    except getopt.GetoptError:
        print(USAGE)
        sys.exit(2)

    host = DEFAULT_HOST
    port = DEFAULT_PORT

    for opt, arg in opts:
        if opt == '-h':
            print(USAGE)
            sys.exit()
        elif opt in ("-H", "--host"):
            host = arg
        elif opt in ("-p", "--port"):
            port = int(arg)

    bootstrap_client(host, port)
    run()


if __name__ == "__main__":
    main(sys.argv[1:])

server.py

import sqlite3
import socket
import threading
import sys
import getopt
from datetime import datetime


DATABASE = "records.server.db"

DEFAULT_HOST = "0.0.0.0"
DEFAULT_PORT = 6000


def init_db():
    conn = sqlite3.connect(DATABASE)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS records
            (timestamp text, node_id text, node_hostname text, node_timetamp text)
    ''')
    conn.commit()
    conn.close()


def record(msg):
    conn = sqlite3.connect(DATABASE)
    cursor = conn.cursor()

    node_id, node_hostname, node_timestamp = msg.split("|")

    query = """
        INSERT INTO records VALUES ('{}', '{}', '{}', '{}')
    """.format(
        now(), node_id, node_hostname, node_timestamp
    )
    cursor.execute(query)
    conn.commit()


def handle(fd):
    while True:
        # pass through every non-eof line
        x = fd.readline()
        if not x:
            break
        record(x)
        fd.write(x)
        fd.flush()
        print("echoed", x, end='')
    print("client disconnected")


def now():
    return datetime.now().replace(microsecond=0).isoformat()


def run(host, port):

    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_sock.bind((host, port))
    server_sock.listen(100)

    print("server socket listening on {}:{}".format(host, port))

    while True:
        try:
            new_sock, address = server_sock.accept()
            print("accepted", address)

            fd = new_sock.makefile('rw')
            thread = threading.Thread(target=handle, args=(fd,))
            thread.start()

        except (SystemExit, KeyboardInterrupt):
            break

    server_sock.close()


def main(argv):
    USAGE = 'server.py [-H <bind_host>] [-p <bind_port>]'
    try:
        opts, args = getopt.getopt(argv, "hH:p:", ["host=", "port="])
    except getopt.GetoptError:
        print(USAGE)
        sys.exit(2)

    host = DEFAULT_HOST
    port = DEFAULT_PORT

    for opt, arg in opts:
        if opt == '-h':
            print(USAGE)
            sys.exit()
        elif opt in ("-H", "--host"):
            host = arg
        elif opt in ("-p", "--port"):
            port = int(arg)

    init_db()
    run(host, port)


if __name__ == "__main__":
    main(sys.argv[1:])

# install (inside container)
rm -r conncheck || true
rm master.tar.gz || true

curl -L https://gist.github.com/mattbennett/afc2fdd7133db255a2a9371235d42ff6/archive/master.tar.gz -o master.tar.gz
mkdir -p conncheck
tar xvfz master.tar.gz -C conncheck --strip-components=1

# start client (with screen)
screen -S conncheck
python3 healthcheck.py -H x.x.x.x

# start client (with docker)
sudo docker exec -it CONTAINER python3 /var/osl/conncheck/healthcheck.py -H x.x.x.x