Prueba de carga
Load-testing a website, a web application or an API endpoint is always a good idea to ensure that your code scales.
Chris GPrueba de carga
Independientemente de si está creando un sitio web dinámico (CMS), una aplicación web o una API, siempre es una buena idea realizar pruebas de carga en su código para asegurarse de que no haya errores de concurrencia y que su infraestructura escala. Si bien hay muchas herramientas disponibles (algunas se enumeran en Herramientas de desarrollo web ), esto se puede hacer fácilmente en Python. Tenga en cuenta que si bien este script de ejemplo generará muchas solicitudes simultáneas, ejecutarlo en un solo sistema solo generará una cantidad moderada de tráfico, aún se requerirán pruebas a mayor escala para sitios web masivamente populares.
También se debe tener en cuenta que cada arquitectura tendrá su punto de quiebre, generando una gran cantidad de carga mientras que monitorear su infraestructura le permitirá identificar y resolver problemas con el eslabón más débil, pero esto solo le permitirá descubrir el siguiente eslabón más débil después eso, y así sucesivamente.
Load-testing the public endpoint is a great option, but you can also load-test individual back-end endpoints to see how they scale and behave under pressure.
Usando el script de prueba de carga
In the script below, replace <your_website_url> with the URL of your website, including the protocol 'https://www.mysite.com/'.
The default http_connection_timeout is 45 seconds, the expected HTTP Response Code is 200, and by default, it will create 10 threads that will each make 10 calls, for a total of 100 concurrent requests on the website. We purposely didn't use a Python session, we wanted each call to establish its own HTTP session.
La secuencia de comandos
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import (absolute_import, division,
print_function, unicode_literals)
'''
Load testing tool for HTTP requests.
'''
# I M P O R T S ###############################################################
import requests
import sys
import os
import logging
from logging.handlers import RotatingFileHandler
import traceback
from datetime import datetime, timedelta
import threading
import time
from functools import reduce
requests.packages.urllib3.disable_warnings()
logging.getLogger("requests").setLevel(logging.WARNING)
__author__ = "Videre Research, LLC"
__version__ = "1.0.0"
# G L O B A L S ###############################################################
url = '<your_website_url>'
http_connection_timeout = 45 # 45 seconds
headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
expectedResponse = 200
num_threads = 10
num_calls = 10
errors = {}
threadDuration = []
startTime = []
threadLock = threading.Lock()
intervals = (
('w', 604800), # 60 * 60 * 24 * 7
('d', 86400), # 60 * 60 * 24
('h', 3600), # 60 * 60
('m', 60),
('s', 1),
)
logger = logging.getLogger()
logger.setLevel(logging.INFO)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)
ch.setFormatter(formatter)
logger.addHandler(ch)
logger.info("-" * 80)
logger.info("Version: %s" % (__version__))
logger.info("Path: %s" % (os.path.realpath(__file__)))
try:
logFile = os.path.realpath(__file__) + ".log"
fh = RotatingFileHandler(logFile, maxBytes=(1048576 * 50), backupCount=7)
fh.setLevel(logging.DEBUG)
fh.setFormatter(formatter)
logger.addHandler(fh)
logger.info("Log file: %s" % logFile)
except Exception as e:
logger.warning("Unable to log to file: %s - %s" % (logFile, e))
# F U N C T I O N S ###########################################################
def time_this(original_function):
"""Wrapper to print the time it takes for a function to execute."""
def new_function(*args, **kwargs):
before = datetime.now()
x = original_function(*args, **kwargs)
after = datetime.now()
global logger
logger.debug('Duration of %s: %.4fs' % (original_function.__name__, (after - before).total_seconds()))
return x
return new_function
def display_time(seconds, granularity=2):
"""Display time with the appropriate unit."""
result = []
for name, count in intervals:
value = seconds // count
if value:
seconds -= value * count
if value == 1:
name = name.rstrip('s')
result.append("{0:.0f}{1}".format(value, name))
return ' '.join(result[:granularity])
def total_seconds(dt):
# Keep backward compatibility with Python 2.6 which doesn't have
# this method
if hasattr(datetime, 'total_seconds'):
return dt.total_seconds()
else:
return (dt.microseconds + (dt.seconds + dt.days * 24 * 3600) * 10**6) / 10**6
@time_this
def send_request(s, verb, url, rc, headers="", data="", terminate=False):
"""Generic method to send an http request."""
# logger.info("Sending request %s" % url)
try:
r = s.request(method=verb, url=url, data=data, headers=headers, timeout=http_connection_timeout)
# r = requests.request(method=verb, url=url, data=data, headers=headers, verify=False, timeout=http_connection_timeout, auth=auth)
#logger.debug(r.headers)
#logger.debug(r.text)
if r.status_code == rc:
return r.status_code, r.text
#logger.debug(r.text)
else:
logger.error(r.status_code)
logger.error(r.text)
logger.error(r.headers)
logger.error(sys.exc_info()[:2])
if terminate:
sys.exit(1)
else:
return r.status_code, r.text
except Exception as e:
logger.error("Error in http request: " + str(e))
if terminate:
sys.exit(1)
else:
return 0, 'ERROR'
def ProcessFileThread(i):
"""This is the worker thread function that will process files in the queue."""
s = requests.Session()
for j in range(num_calls):
start = time.time()
rc, response = send_request(s, "GET", url, expectedResponse, headers, False)
# print(response)
elapsed = str(time.time() - start)
with threadLock:
threadDuration.append(float(elapsed))
if rc != expectedResponse:
print(rc)
if rc in errors:
errors[rc] += 1
else:
errors[rc] = 1
@time_this
def main():
"""Main function."""
global auth
global url
before = datetime.now()
for i in range(num_threads):
worker = threading.Thread(target=ProcessFileThread, args=(i + 1,))
worker.setDaemon(True)
worker.start()
logger.info('*** Main thread waiting')
worker.join()
logger.info('*** Main thread Done')
time.sleep(1)
after = datetime.now()
logger.info('Duration %s' % display_time(total_seconds(after - before)))
logger.info(("Calls: ", len(threadDuration)))
logger.info(("Min Response time: ", min(threadDuration)))
logger.info(("Max Response time: ", max(threadDuration)))
logger.info(("Avg Response time: ", reduce(lambda x, y: x + y, threadDuration) / len(threadDuration)))
logger.info(errors)
sys.exit(0)
###############################################################################
if __name__ == "__main__":
main()
# E N D O F F I L E #######################################################Salida de secuencia de comandos
La secuencia de comandos no mostrará nada hasta que se haya completado el último hilo, y luego mostrará el resumen a continuación. Tenga en cuenta que también se crea un archivo de registro y que la última línea {} contendrá un recuento de cualquier error (códigos de retorno HTTP distintos al que especificó) encontrado.
Disfrute responsablemente, ¡no haga pruebas de carga de los sitios web de otras personas!
python3 load_test.py
2022-06-30 09:47:49,908 - INFO - --------------------------------------------------------------------------------
2022-06-30 09:47:49,908 - INFO - Version: 1.0.0
2022-06-30 09:47:49,909 - INFO - Path: /Users/me/load_test.py
2022-06-30 09:47:49,910 - INFO - Log file: /Users/me/load_test.py.log
2022-06-30 09:47:49,923 - INFO - *** Main thread waiting
2022-06-30 09:47:52,215 - INFO - *** Main thread Done
2022-06-30 09:47:53,221 - INFO - Duration 3s
('Calls: ', 100)
('Min Response time: ', 0.13335084915161133)
('Max Response time: ', 0.46799778938293457)
('Avg Response time: ', 0.22425265073776246)
{}Tagged with:
networking