python-3.x - 如何对部署在 AWS 上的 Flask 应用程序进行 50 次并发调用？

Question

我正在使用以下 python 脚本来测试在 AWS 实例上运行的应用程序，

import sys
import requests
import logging
import random
from datetime import datetime
import threading
import os
import time


logger = logging.getLogger('Intrudx')
handle = logging.FileHandler('Intrudx.log')
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
handle.setFormatter(formatter)
logger.addHandler(handle)
logger.setLevel(logging.INFO)


loop_count = int(sys.argv[1])
sleep_time = int(sys.argv[2])


# CHECKING THE HEARTBEAT
def heartbeat(SessionID, SiteID):
    logger.info("Starting heartbeat thread")
    try:
        heart_url = 'http://ec2-instance-address.com/license/heartbeat'
        heart_result = requests.post(heart_url, json={
                                                     "SessionID":str(SessionID),
                                                     "SiteID" : str(SiteID)
                                                      })
        if heart_result.status_code is 500:
            logger.error("Heartbeat Failed with 500")
            return "We Got 500"
        response_text = heart_result.json()["ResponseText"]
        logger.info("Heartbeat: "+str(response_text))
    except Exception as e:
        logger.error("Heartbeat Failed"+str(e))


# FINDING THE SERVER IP
def ip(SessionID):
    logger.info("Starting get server info thread")
    try:
        get_server_url = 'http://ec2-instance-address.com/server/getStreamingServer'
        get_server_result = requests.post(get_server_url, json={"SessionID": str(SessionID)})
        result_code = get_server_result.status_code
        if result_code is 500:
            logger.error("GetStreamingServerInfo: " + "Failed")
            return "We Got 500"
        response_text = get_server_result.json()["ResponseText"]
        logger.info("GetStreamingServerInfo: " + str(response_text))
    except Exception as e:
        logger.error("GetStreamingServerInfo: " + str(e))


def main():
    for i in range(loop_count):
        # LOGIN
        try:
            login_url = 'http://ec2-instance-address.com/user/login'
            login_result = requests.post(login_url, json={
                                                            "AccountName": "Account1",
                                                            "UserID": "user2",
                                                            "UserPassword": "test"
                                                         })
            result_code = login_result.status_code
            if result_code is 500:
                logger.error("Login: "+"Failed")
                return "We Got 500"
            SessionID = login_result.json()["SessionID"]
            response_text = login_result.json()["ResponseText"]
            logger.info("Login: "+str(response_text)+": "+ str(SessionID))
            print(str(SessionID)+str(response_text))

        except Exception as e:
            result_code = str(e)
            logger.error("Login: "+str(e))

        # GET NEW SITE
        try:
            get_new_site_url = 'http://ec2-instance-address.com/license/getNewSite'
            get_new_site_result = requests.post(get_new_site_url, json={"SessionID": str(SessionID)})
            result_code = get_new_site_result.status_code
            if result_code is 500:
                logger.error("Login: " + "Failed")
                return "We Got 500"
            response_text = get_new_site_result.json()["ResponseText"]
            site_id = get_new_site_result.json()["NewSiteID"]
            logger.info("getNewSite: "+str(response_text)+": "+str(site_id))

        except Exception as e:
            result_code = str(e)


             logger.error("getNewSite"+str(e))


        # STARTING HEARTBEAT THREAD
        try:
            threading.Thread(target=heartbeat(SessionID, site_id), args=(SessionID, site_id,)).start()
        except Exception as e:
            logger.error("Problem starting thread: "+str(e))

        # STARTING GET SERVER INFO THREAD
        try:
            threading.Thread(target=ip(SessionID), args=(SessionID)).start()
        except Exception as e:
            logger.error("Problem while starting Get Server Info Thread"+str(e))

此脚本仅使用一个用户，与服务器创建一个 session /连接以进行 API 调用。

以类似的方式，我想用 50 或 100 个不同的用户(使用不同的帐户/凭据)连接到进行 API 调用的服务器来测试应用程序。大约 50 或 100 个用户同时使用该应用程序。所以我可以确保应用程序正确处理 50 个用户。

如何使用脚本进行此类测试？

更新:大部分路由都是隐藏的，需要@login_required。

Answer 1

我建议你试试Bees With Machine Guns .它是一个 python 脚本，它将启动微型 EC2 实例并将这些实例的许多请求发送到您的应用程序。这将模拟性能测试的流量激增。

我是从AWS training videos on CBT Nuggets听说的.讲师有效地使用它来触发自动缩放并对他的配置进行负载测试。

祝你好运。