Verwendung von Locust zum Lasttest einer FastAPI -App mit gleichzeitigen users

Ich habe gerade meine erste FastAPI -App fertiggestellt. Diese App erlaubt es users, ihre eigenen Elemente zu haben, was bedeutet, dass die user Datenmodelle alle ein user_id Feld haben. Nichts Besonderes, aber da FastAPI einige für mich neue Dinge einführt, wie z.B. Dependency Injection, war ich unsicher, ob meine App so funktionieren würde, wie ich es wollte.

Meine Frage war: Wenn ich den API mit vielen gleichzeitigen users lade, funktioniert er dann noch?

Achtung! Führen Sie nur dann einen Lasttest auf Ihrem Computer durch, wenn Sie sicher sind, dass er damit umgehen kann.

Definieren des Tests

Ich brauchte ein Werkzeug, das Folgendes kann:

• Mehrere users initialisieren
• Gleichzeitige Tests für jeden user ausführen
• Dies so schnell wie möglich tun

Um einen user zu initialisieren, muss ich den user mit einer eindeutigen E-Mail Adresse registrieren. Danach erhalte ich das Zugriffstoken für diese user, und setze den Header für die Operationen GET und POST .

Ich möchte den folgenden Test für jede user:

• Hinzufügen eines Elements mit einem eindeutigen Namen (POST)
  Dies gibt das hinzugefügte Element einschließlich seiner id zurück.
• Lesen Sie das Element nach id zurück (GET)
  Dies gibt das Element zurück.
• Prüfen, ob der Name im zurückgegebenen Element mit dem eindeutigen Namen übereinstimmt

Auswählen von Locust für den Test

Für triviale Webanwendungen verwende ich Apache Bench (ab), ein sehr einfach zu bedienendes Tool zum Stresstest. Aber hier nicht wirklich nützlich, weil wir eine Initialisierung durchführen und einen (einfachen) Test schreiben müssen.

Ich habe bereits viele Tests für meine FastAPI -App mit Pytest geschrieben, und es gab keine Probleme. Ich habe über pytest-xdist gelesen, aber es sieht so aus, als ob es Probleme hat und/oder zu langsam ist(?).

Ich habe mich in diesem Fall für Locust entschieden. Es ist einfach zu bedienen und Sie können Ihre Tests in Python codieren. Wunderbar.

Der Code

Unten ist der Code für den Test. Einige Hinweise:

import logging
Ich protokolliere alles auch in eine Datei.

def unique_str()
Im Test brauche ich eindeutige E-Mail-Adressen und Städtenamen. Wir könnten hier auch einen UUID4.

network_timeout, connection_timeout
Ich möchte nicht, dass der Test fehlschlägt, sondern dass der Client wartet.

wait_time
Ich möchte, dass der nächste Test sofort beginnt, wenn der aktuelle beendet ist.

Dummy-Task
Ich aktiviere diesen und deaktiviere alle anderen Tasks bei der Entwicklung der 'on_start'-Methode.

(POST/GET) catch_response=True
Zusammen mit dem 'with'-Konstrukt ermöglicht uns dies, eine Anfrage als fehlgeschlagen zu markieren, indem wir response.failure() aufrufen.

(GET) name=<URL>
Locust zeigt alle URLs, aber hier hat GET wegen der ID jedes Mal eine andere URL. Mit der 'name=' wird nur diese URL angezeigt.

Status_code-Prüfungen
Locust betrachtet eine Anfrage als erfolgreich, wenn der Antwortcode <400 ist, aber ich möchte den Wert explizit prüfen.

RescheduleTask()
Wenn ein Fehler auftritt, gibt es keinen Grund, eine Aufgabe fortzusetzen, also lösen wir eine Ausnahme aus. Könnte fortschrittlicher gemacht werden.

# test_cities_wrc.py

import logging
import random
import time

from locust import HttpUser, task, between
from locust.exception import RescheduleTask


test_email_host = 'my-local-test.test'

def unique_str():
    return '{:.10f}.{}'.format(time.time(), random.randint(1000, 9999))

def unique_email():
    return 'e.'  +  unique_str()  +  '@'  +  test_email_host

def unique_city_name():
    return 'c.'  +  unique_str()


class WebsiteTestUser(HttpUser):
     network_timeout  = 30.0
     connection_timeout  = 30.0
    #wait_time  = between(0.5, 3.0)

    def on_start(self):

        base_url = 'http://127.0.0.1:8020/api/v1'

        # set up urls
        register_url = base_url  +  '/auth/register'
        get_token_url = base_url  +  '/auth/token'
        # urls used in task
        self.cities_create_url = base_url  +  '/cities'
        self.cities_get_by_id_url = base_url  +  '/cities/'

        # get unique email
        email = unique_email()
        password = 'abcdefghi'

        # register
        response = self.client.post(
            register_url,
            json={'email': email, 'password': password},
        )
        if response.status_code != 201:
            error_msg = 'register: response.status_code = {}, expected 201'.format(response.status_code)
            logging.error(error_msg)

        # get_token
        # -  username instead of email
        # - x-www-form-urlencoded (instead of json)
        response = self.client.post(
            get_token_url,
            data={'username': email, 'password': password},
        )
        access_token = response.json()['access_token']
        logging.debug('get_token: for email = {}, access_token = {}'.format(email, access_token))

        # set headers with access token
        self.headers = {'Authorization': 'Bearer '  +  access_token}

    def on_stop(self):
        pass

    # enable this dummy task to develop 'on_start'
    #@task
    def dummy(self):
        pass

    @task
    def cities_write_read_check(self):
        # add city to api
        city_name = unique_city_name()
        logging.debug('cities_create: city_name = {}'.format(city_name))
        with self.client.post(
            self.cities_create_url,
            json={'name': city_name},
            headers=self.headers,
             catch_response=True,
        ) as response:

            if response.status_code != 201:
                error_msg = 'cities_create: response.status_code = {}, expected 201, city_name = {}'.format(response.status_code, city_name)
                logging.error(error_msg)
                response.failure(error_msg)
                raise  RescheduleTask()

            response_dict = response.json()
            if 'data' not in response_dict:
                error_msg = 'cities_create: data not in response_dict, city_name = {}'.format(city_name)
                logging.error(error_msg)
                response.failure(error_msg)
                raise  RescheduleTask()

            city_id = response_dict['data']['id']
            logging.debug('cities_create: for city_name = {}, city_id = {}'.format(city_name, city_id))

        # get city from api and check
        with self.client.get(
            self.cities_get_by_id_url  +  city_id,
            headers=self.headers,
             name=self.cities_get_by_id_url  +  'uuid', 
             catch_response=True,
        ) as response:

            if response.status_code != 200:
                error_msg = 'cities_get_by_id: response.status_code = {}, expected 200, city_name = {}'.format(response.status_code, city_name)
                logging.error(error_msg)
                response.failure(error_msg)
                raise  RescheduleTask()

            if 'data' not in response_dict:
                error_msg = 'cities_get_by_id: data not in response_dict, city_name = {}'.format(city_name)
                logging.error(error_msg)
                response.failure(error_msg)
                raise  RescheduleTask()

            city_name_returned = response_dict['data']['name']
            logging.debug('cities_get_by_id: for city_id = {}, city_name_returned = {}'.format(city_id, city_name_returned))
            
            if city_name_returned != city_name:
                error_msg = 'cities_get_by_id: city_name_returned = {} not equal city_name = {}'.format(city_name_returned, city_name)
                logging.error(error_msg)
                response.failure(error_msg)
                raise  RescheduleTask()

Ausführen des Tests

WARNUNG: Wenn Sie einen Test wie diesen ausführen, belasten Sie Ihren CPU!

Um den Test auszuführen und in einer Datei 'lc.log' zu protokollieren, öffnen Sie ein Terminal und geben Sie den Befehl ein:

locust  -L  DEBUG  --logfile  lc.log  -f ./test_cities_wrc.py

Und geben Sie dann in Ihrem Browser ein:

http://localhost:8089/

Dann geben Sie zum Beispiel ein:

• Anzahl der insgesamt zu simulierenden users: 20
• Spawn-Rate (users gespawnt/Sekunde): 20
• Host: http:127.0.0.1

und drücken Sie die Taste 'Start swarming'. Sofort sollten Sie hören, wie die Lüfter Ihres Computers beginnen, mit maximaler Geschwindigkeit zu laufen ...

Was war das Ergebnis?

Ich habe tatsächlich ein Problem mit meinem Code gefunden. Dieses Problem trat bei Pytest nicht auf. Es stellte sich heraus, dass irgendwo in meinem AccessControlManager Class mit Dependency Injection verwendet, eine Variable durch eine andere Anfrage überschrieben werden konnte. Als Konsequenz wurde eine falsche user_id zurückgegeben und der Test schlug fehl, nicht immer, aber wenn zwei oder mehr users gleichzeitig auf die API zugriffen.

Zusammenfassung

Ich war unsicher, ob meine erste FastAPI -App unter Last mit vielen gleichzeitigen users funktionieren würde. Mit Locust konnte ich diese Situation auf sehr einfache Weise simulieren, indem ich den Test in Python codierte. Locust hat viele Optionen und kann auch headless, d.h. ohne Browser, laufen.

Dieser Test hat mir nicht nur geholfen, das Problem zu erkennen, sondern es auch einzugrenzen. Ich habe das Problem behoben, was mich aber auch dazu brachte, mehr über Dependency Injection in FastAPI zu lesen.

Links / Impressum

Locust
https://docs.locust.io/en/stable/