Beomagi's thoughts...: python

Showing posts with label python. Show all posts

Saturday, October 16, 2021

Add a multithreaded http stats service to an existing app

Quite often when we have a running app, we'd be curious about what's going on in it right now. Often we'd want to see some stats of some sort. Maybe have a way to trigger a test or see if the system is healthy. Whatever it is., we can reference the previous multithreaded http server example.

The only difference is we'll run the request loop in it's own thread. This will let the app carry out whatever it needs to, and in between heavy cpu operations, trigger the http request. We are restricted by the GIL, so if the app has long CPU bound operations, this isn't ideal obviously.

Here's a very basic sample:

'''    Add a threaded non blocking http response to an app    '''
#!/usr/bin/env python3
import socketserver
import http.server
import sys
import time
import urllib
import threading
HTPORT = 8000

class Handler(http.server.BaseHTTPRequestHandler):
    '''   use our own handlers functions '''

    def sendtextinfo(self, code, text):
        self.send_response(code)
        self.send_header('Content-type', 'text/html')
        self.end_headers()
        self.wfile.write((str(text)+"\n").encode())

    def do_GET(self):
        '''   handle get   '''
        tnow = time.time()
        gnow = time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(tnow)) #Formatted UTC time
        parsed_data = urllib.parse.urlparse(self.path)
        if parsed_data.geturl().lower() == "/time":
            message = gnow
        else : message="Secondary thread for http server does not block main loop<br/>{}".format(gnow)
        self.sendtextinfo(200,message)

class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
    '''    Basic threaded server class    '''
    http.server.HTTPServer.request_queue_size = 128


def backgroundfunction():
    HTSERVER = ThreadedHTTPServer(('', HTPORT), Handler)
    while 1:
        sys.stdout.flush()
        HTSERVER.handle_request()

threading.Thread(target=backgroundfunction).start()

#substitute this loop with main loop of your app
while 1:
    time.sleep(1)
    print("Other stuff here!")

Now I've mucked about with the "get" method to simplify the message return a little for expansion, but the handler can really just be a single function. ThreadedHTTPServer is the heart of this being able to handle multiple requests in separate threads, and this runs in a thread before the main loop starts.

How would we use this in a more practical example?

Lets have the app read files in one folder and manipulate them, dumping the result in another. We'll have the service respond with stats.

Example app to "encrypt" text files with rot 13. Takes a file from c:\tmp\a, and outputs the result in c:\tmp\b. File moved to c:\tmp\processed.

'''    Add a threaded non blocking http response to an app    '''
#!/usr/bin/env python3
import socketserver
import http.server
import sys
import time
import urllib
import threading
import glob
import codecs
import os
import shutil
import json
HTPORT = 8000
INPUT_FOLDER="c:\\tmp\\a\\"
PROCESSED_FOLDER="c:\\tmp\\processed\\"
OUTPUT_FOLDER="c:\\tmp\\b\\"
timetoleave=False
stats={}
stats["Errors"]=[]
stats["RecentRequests"]=[]

def gnow():
    tnow = time.time()
    return time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(tnow)) #Formatted UTC time


class Handler(http.server.BaseHTTPRequestHandler):
    '''   use our own handlers functions '''

    def sendtextinfo(self, code, text):
        self.send_response(code)
        self.send_header('Content-type', 'text/html')
        self.end_headers()
        self.wfile.write((str(text)+"\n").encode())

    def do_GET(self):
        global timetoleave
        global stats
        '''   handle get   '''
        parsed_data = urllib.parse.urlparse(self.path)
        if parsed_data.geturl().lower() == "/quit":
            message = "BYE!"
            timetoleave=True
        else :
            stats["time"]=gnow()
            tnow=time.time()
            if len(stats["RecentRequests"])>0:
                time_for_last_n=tnow-stats["RecentRequests"][0]["Time"]
                if time_for_last_n>0:
                    rate=len(stats["RecentRequests"])/time_for_last_n
                    stats["Rate"]=rate
            msgtext=json.dumps(stats,indent="  ")
            message=msgtext
            for chk in ["Chrome","Mozilla","Explorer","Safari"]:
                if chk in str(self.headers):
                    message=msgtext.replace("\n","<br/>").replace(" ","&nbsp;")
                    break
        self.sendtextinfo(200,message)

class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
    http.server.HTTPServer.request_queue_size = 128

def backgroundfunction():
    HTSERVER = ThreadedHTTPServer(('', HTPORT), Handler)
    HTSERVER.socket.settimeout(10)
    try:
        while not timetoleave:
            sys.stdout.flush()
            HTSERVER.handle_request()
    except KeyboardInterrupt:
        pass

threading.Thread(target=backgroundfunction).start()

def processfiles(filelist):
    for filename in filelist:
        reqnow={}
        reqnow["Time"]=time.time()
        reqnow["Request"]=filename
        stats["RecentRequests"].append(reqnow)
        stats["RecentRequests"]=stats["RecentRequests"][-10:]
        try:
            with open(filename,"r") as fh:
                inputtext=fh.read()
                outputtext=codecs.encode(inputtext,'rot_13')
                outfile=OUTPUT_FOLDER+os.path.basename(filename)
                with open(outfile,"w") as fh:
                    fh.write(outputtext)
            processedfile=PROCESSED_FOLDER+os.path.basename(filename)
            shutil.move(filename,processedfile)
        except Exception as err:
            print("Unable to open {}, {}".format(filename,err))
            errnow={}
            errnow["Time"]=time.time()
            errnow["Error"]=err
            stats["Errors"].append(errnow)
            stats["Errors"]=stats["Errors"][-10:]

#substitute this loop with main loop of your app
while not timetoleave:
    allinputfiles=glob.glob(INPUT_FOLDER+"*")
    processfiles(allinputfiles)
    time.sleep(1)

Here the threaded server can give stats while this app is running. Moving a few files to the input folder and pulling up the stats in a browser gives the result:

{
  "Errors": [],
  "RecentRequests": [
    {
      "Time": 1634412207.3869205,
      "Request": "c:\\tmp\\a\\a.txt"
    },
    {
      "Time": 1634412207.3869205,
      "Request": "c:\\tmp\\a\\q.txt"
    },
    {
      "Time": 1634412207.3869205,
      "Request": "c:\\tmp\\a\\test.txt"
    },
    {
      "Time": 1634412207.3869205,
      "Request": "c:\\tmp\\a\\test1.txt"
    }
  ],
  "time": "2021-10-16 19:23:29",
  "Rate": 1.8133347398556916
}

Friday, January 29, 2021

A Simple Multithreaded REST Service

I've talked about threading a little already.

https://beomagi.blogspot.com/2019/09/python-threading.html

https://beomagi.blogspot.com/2019/09/threading-continued-timing-thread.html

Next up is a more complete application. Lets look at a simple multithreaded web service. Now this can be used for basic webpage services, but I think it really shines for making a REST API for quick tools.

Source:

https://github.com/beomagi/BasicRestApp/blob/main/http_server_mt3.py

snapshot below:

'''    Basic threaded http server implementation    '''
#!/usr/bin/env python3
import socketserver
import http.server
import sys
import time
import urllib
import json

#-------------sample functions--------------------
keyval={}
def kput(a,b):
    keyval[a]=b
    return json.dumps({a:b})

def kget(a):
    return keyval.get(a)

def add2numbers(a,b):
    return int(a)+int(b)

def handlejob(func_and_params):
    func=func_and_params[0]
    params=func_and_params[1:]
    if func=="add": return add2numbers(params[0],params[1])
    if func=="put": return kput(params[0],params[1])
    if func=="get": return kget(params[0])
    return "Unknown Command"
#-^-^--^-^-^-^sample functions-^-^-^-^-^-^-^-^-^-^-

class Handler(http.server.BaseHTTPRequestHandler):
    '''   use our own handlers functions '''

    def sendtextinfo(self, code, text):
        self.send_response(code)
        self.send_header('Content-type', 'text/html')
        self.end_headers()
        if type(text)==type([]):
            for lines in text:
                self.wfile.write((str(lines)+"\n").encode())
        else:
            self.wfile.write((str(text)+"\n").encode())

    def do_GET(self):
        '''   handle get   '''
        tnow = time.time()
        gnow = time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(tnow)) #Formatted UTC time
        parsed_data = urllib.parse.urlparse(self.path)
        if parsed_data.geturl().lower() == "/time":
            message = gnow
        else : message="this is a test of the multithreaded webservice"
        self.sendtextinfo(200,message)

    def do_POST(self):
        '''   handle post like rest API   '''
        try: #try getting the bytestream of the request
            content_length = int(self.headers['Content-Length'])
        except Exception as err:
            print("malformed headers")
            self.sendtextinfo(200,str(err))
            return

        if content_length > 0:
            rawrequest = self.rfile.read(content_length).decode('utf-8')
            print("Received POST: {}".format(rawrequest))
            try:
                jrequest = json.loads(rawrequest)
            except BaseException as anError:
                self.sendtextinfo(200,"Error in JSON: {}".format(str(anError)))
                return

        if "cmd" in jrequest:
            commandandparams=jrequest['cmd']
            print("Command received: {}".format(commandandparams))
            result=handlejob(commandandparams)
            if result != None:
                self.sendtextinfo(200,result)
            else:
                self.sendtextinfo(200,"No Output")
        return

class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
    '''    Basic threaded server class    '''
    http.server.HTTPServer.request_queue_size = 128

if sys.argv[1:]:
    HTPORT = int(sys.argv[1])
else:
    HTPORT = 8000

HTSERVER = ThreadedHTTPServer(('', HTPORT), Handler)

try:
    while 1:
        sys.stdout.flush()
        HTSERVER.handle_request()
except KeyboardInterrupt:
    print("Server Stopped")

Discussion:

I've highlighted different sections of the code by altering the background of the div.

The bottom section with the dark purple background is our main area - could stick this in a construct like "if __name__ == '__main__':" but it's not necessary for our purposes.

The main here does a check for passed parameters to override the default port the application will be served on. Then we have an infinite loop that keep calling handle_request(). The "flush" before that forces any buffering in the io to be handled before the next handle. Useful for problems where buffered output isn't yet written at the time of a crash.

The main section makes a call to ThreadedHTTPServer. This manages threaded calls to the handler(3rd section). The handler implements calls for GET and POST.

I've added some simple functions on POST calls for adding 2 numbers, storing, and getting values in a key-value store sort of setup.

Interfacing with POST can be done from any other language. It's also nice to use CURL to test REST interfaces.

e.g.

beomagi@BeoBalthazar ~/gits/beomagi/BasicRestApp (main) 2021-01-29 13:35:09

└─ $ ∙ curl 127.0.0.1:8000

this is a test of the multithreaded webservice