Utilisation de Locust pour tester la charge d'une application FastAPI avec des users simultanées.

Je viens de terminer ma première application FastAPI . Cette application permet aux user d'avoir leurs propres éléments, ce qui signifie que les modèles de données user ont tous un champ user_id. Rien de spécial, mais comme FastAPI introduit certaines choses nouvelles pour moi, comme Dependency Injection, je n'étais pas sûr que mon application fonctionnerait comme je le voulais.

Ma question était la suivante : si je teste la charge de la API avec de nombreuses user simultanées, est-ce que cela fonctionne toujours ?

Avertissement. Ne lancez un test de charge sur votre ordinateur que si vous êtes sûr qu'il peut le supporter.

Définition du test

J'avais besoin d'un outil qui puisse :

• Initialiser plusieurs users
• Exécuter simultanément des tests pour chaque user.
• Faire cela aussi vite que possible

Pour initialiser une user , je dois enregistrer la user avec une adresse électronique unique. Ensuite, je récupère le jeton d'accès pour cette user, et je définis l'en-tête pour les opérations GET et POST .

Je veux le test suivant pour chaque user :

• Ajouter un élément avec un nom unique (POST)
  Cela renvoie l'élément ajouté, y compris son identifiant.
• Lire l'élément par son id (GET)
  Renvoie l'élément.
• Vérifier si le nom dans l'élément retourné est égal au nom unique

Sélection de Locust pour le test

Pour les applications web triviales, j'utilise Apache Bench (ab), un outil très facile à utiliser pour les tests de stress. Mais il n'est pas vraiment utile ici car nous devons effectuer une initialisation et écrire un test (simple).

J'ai déjà écrit de nombreux tests pour mon application FastAPI avec Pytest, et il n'y avait aucun problème. Je me suis renseigné sur pytest-xdist mais il semble qu'il ait des problèmes et/ou qu'il soit trop lent( ?).

J'ai décidé d'utiliser Locust dans ce cas. Il est facile à utiliser et vous pouvez coder vos tests dans Python. Merveilleux.

Le code

Voici le code du test. Quelques notes :

import logging
Je consigne également tout dans un fichier.

def unique_str()
Dans le test, j'ai besoin d'adresses e-mail et de noms de ville uniques. On pourrait aussi faire un UUID4 ici.

network_timeout, connection_timeout
Je ne veux pas que le test échoue mais que le client attende.

wait_time
Je veux que le test suivant commence immédiatement lorsque le test en cours se termine.

dummy task
J'active cette tâche et désactive toutes les autres tâches lors du développement de la méthode 'on_start'.

(POST/GET) catch_response=True
Avec la construction 'with', cela nous permet de marquer une requête comme ayant échoué en appelant response.failure().

(GET) name=<URL>
Locust montre toutes les URL mais ici GET a une URL différente à chaque fois à cause de l'ID. En utilisant 'name=', seule cette URL apparaît.

Vérification des codes d'état
Locust considère qu'une requête est réussie si le code de réponse est <400, mais je veux vérifier explicitement la valeur.

RescheduleTask()
Lorsqu'une erreur se produit, il n'est pas nécessaire de poursuivre une tâche, donc nous levons une exception. Cela pourrait être fait de manière plus avancée.

# 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()

Exécution du test

AVERTISSEMENT : Lorsque vous exécutez un test comme celui-ci, vous stressez votre CPU !

Pour exécuter le test, et enregistrer dans un fichier 'lc.log', ouvrez un terminal et entrez la commande :

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

Et ensuite dans votre navigateur tapez :

http://localhost:8089/

Puis par exemple entrez :

• Nombre total de user à simuler : 20
• Taux de spawn (users spawned/second) : 20
• Hôte : http:127.0.0.1

et appuyez sur le bouton 'Start swarming'. Immédiatement, vous devriez entendre les ventilateurs de votre ordinateur se mettre à tourner à la vitesse maximale...

Quel a été le résultat ?

J'ai effectivement trouvé un problème avec mon code. Ce problème n'est pas apparu avec Pytest. Il est apparu que quelque part dans ma AccessControlManager 4_6791_TNEMECALPER_4Q utilisée avec la Dependency Injection, une variable pouvait être écrasée par une autre requête. En conséquence, un mauvais user_id était renvoyé et le test échouait, pas tout le temps mais lorsque deux ou plusieurs user accédaient à la API au même moment.

Résumé

Je n'étais pas sûr que ma première application FastAPI fonctionnerait sous une charge avec de nombreux user simultanés. Locust m'a permis de simuler cette situation de manière très simple, en codant le test dans Python. Locust dispose de nombreuses options et peut également fonctionner sans tête, c'est-à-dire sans navigateur.

Ce test m'a permis non seulement de détecter le problème mais aussi de l'isoler. J'ai corrigé le problème, mais cela m'a également incité à en savoir plus sur Dependency Injection en FastAPI.

Liens / crédits

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