Python3 has a very neat library in it that will save you a lot of unnecessary heavy lifting processing and re-engineering your average web 2.0 json output api. It’s pretty neat how if you know what your doing not only can you implement a really cool monitoring API (not that you’d need to there are plenty of 3rd party ones available like newrelic, datadog and many others), but perhaps you may have proprietary reason to develop your own internal monitoring, and various other reasons.

Also, when it comes to monitoring server side processes, states with application layers or microservices there is more going on in the middleware, but it’s exactly the same process except in your app.get you are going to define a function that checks for output or inputs of another program, and so on, be that a datastore, text file, api call, or conditional developing API with python is made really easy. We will use uvicorn to run the python-server.

import uvicorn
import os
import psutil
from fastapi import FastAPI

We also are using the fastapi library to generate our arbitrary server calls for the client. We use os,psutil to access certain os functions for the other calls.

/metric

app = FastAPI()
hits = 0

@app.get("/metric")
async def metric():
    global hits
    hits+=1
    return {"hits": hits}

We might be using the API call for our own custom service tracker, trigger alarm, alert, such as if bruteforcing etc is happening maybe the trigger that consumes this alarm can set an upper limit or range for its rate_of_change and so on. This simply provides output for the number of requests made. A seperate read_only function could be made that only reads the variable by removing hits+=1.

/health

A pretty standard thing for any monitoring APi is the health endpoint, normally an “true” or “OK” statement string suffices, but some people run much more conditional health that could check an array of resources or other api endpoints values before returning the client output. I haven’t done lots of development with Python but it seems like a pretty cool language worth time, and with Flash and uvicorn a pretty compelling usercase exists for rapid prototyping and development.

@app.get("/health")
async def health():
    return "ok"

/disk

I thought it would also be cool to try my hand at writing a basic disk metric. I am not sure if this was the best way to implement it and it is kind of hacky, you may not want to monitor /, but a specific path with a specific disk which is simple enough to do by altering the statement to reflect the path.

@app.get("/disk")
async def disk():
    disk_info = psutil.disk_usage("/")

    disk_alert_size = 1073741824
    current_disk_total = int(disk_info.total)

    print(current_disk_total)
    if disk_alert_size < current_disk_total:
        return "ok"
    else:
        return "bad"

/cpu and /memory

## while we're at it lets generate /memory and /cpu metrics too cause its really easy todo

@app.get("/memory")
async def memory():
    # Getting % usage of virtual_memory ( 3rd field)
    current_mem_used = psutil.virtual_memory()[2]
    print(current_mem_used)
    if current_mem_used < 90:
        return "ok"
    else:
        return "not ok"


@app.get("/cpu")
async def cpu():
    # should probably use aggregates for loadavg perhaps
    current_cpu_used =  psutil.cpu_percent(interval=10)
    max_percent_load = 90
    print(current_cpu_used)

    if current_cpu_used < max_percent_load:
        return "ok"
    else:
        return "not ok"

Example of Service Status endpoint for Nginx

Since I was at it I thought it worthwhile to spend a few minutes creating an arbitrary and hopelessly basic check of nginx using psutil. I am wondering if it is better to use systemctl status output or some other library instead, I don’t like using unnecessary libraries though unless it is making things a lot less convuluted by doing that.

# we need this helper function to get process matching for a given service as an argument
def is_process_running(name):
    for process in psutil.process_iter(['name']):
        if process.info['name'] == name:
            return True
            return False

# with the above helper function we can use this simple notation for all desired processes
@app.get("/nginx")
async def nginx():
    if is_process_running('nginx'):
        return "ok"
    else:
        return "bad"

Starting the API Service application

Much like with nodejs and scala its a pretty simple framework/lib to deploy to a server, just append this to the bottom of your python file and then execute as normal.

if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=8000)

python3 monitor-api-service.py

Let’s go ahead and open up this in a browser by navigating to http://ipofserver:8000

Pretty neat. Note this is running as http:// to run as https:// you will need a certificate and to define it like

if __name__ == '__main__':
    uvicorn.run("main:app", port=443, host='0.0.0.0', reload = True, reload_dirs = ["html_files"], ssl_keyfile="/etc/letsencrypt/live/my_domain/privkey.pem", ssl_certfile="/etc/letsencrypt/live/my_domain/fullchain.pem"

In this case because the HTTPS transport and cert are a seperate layer to the plaintext api they aren’t really anything to do with the implementing of the API itself and moreover to do with termination of the TLS certificate at the tld, in any case perhaps you want to implement properly with https. You can get free certs from letsencrypt using the certbot commandline utility.

The power of developing api rapidly like this is really eye-changing for anyone who had a complex server or rpc service backend, as perhaps they’ll see that they only need only translate the necessary command functions with such approach to provide nearly any data thru the api from any service or corresponding conditional test.

Producing something like this is easy but doing it correctly in the first place will always be the real challenge.