如何对部署在 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/connection 与服务器进行 API 调用。

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

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

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

Answer 1

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

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

祝你好运。

Answer 2

您也可以尝试我们的小工具 k6：https://github.com/loadimpact/k6

您使用 JavaScript 为虚拟用户的行为编写脚本，因此很容易让 50 个不同的用户使用不同的凭据登录。看起来像这样（虽然这段代码需要调试:)

import http from "k6/http";

let login_url = "http://ec2-instance-address.com/user/login";
let get_new_site_url = "http://ec2-instance-address.com/license/getNewSite";
let credentials = [
    { "account": "Account1", "username": "joe", "password": "secret" },
    { "account": "Account2", "username": "jane", "password": "verysecret" }
];

export default function() {
  let session_id = doLogin();
  let response = doGetNewSite(session_id);
  let response_text = response["ResponseText"];
  let new_site_id = response["NewSiteID"];
  for (i = 0; i < loop_count; i++) {
    // do heartbeat stuff?
  }
}

function doLogin() {
  let index = Math.floor(Math.random() * credentials.length);
  let post_body = {
    "AccountName": credentials[index]["account"],
    "UserID": credentials[index]["username"],
    "UserPassword": credentials[index]["password"]
  };
  let http_headers = { "Content-Type": "application/json" };
  let res = http.post(login_url, JSON.stringify(post_body), { headers: http_headers });
  check(res, {
    "Response code is 200": (r) => r.status == 200,
    "Login successful": (r) => JSON.parse(r.body).hasOwnProperty("SessionID")
  });
  return JSON.parse(res.body)["SessionID"];
}

function doGetNewSite(session_id) {
  let http_headers = { "Content-Type": "application/json" };
  let post_body = { "SessionID": session_id };
  let res = http.post(get_new_site_url, JSON.strjngify(post_body), { headers: http_headers });
  check(res, {
    "Status code was 200": (r) => r.status == 200,
    "Got response text": (r) => JSON.parse(r.body).hasOwnProperty("ResponseText"),
    "Got new site id": (r) => JSON.parse(r.body).hasOwnProperty("NewSiteID")
  });
  return JSON.parse(res.body);
}

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

What to do to make 50 concurrent calls to Flask app deployed on AWS?

python-3.x

amazon-web-services

amazon-ec2

flask

load-testing