如何对部署在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)) 

这个脚本只使用一个用户，创建与服务器的一个会话/连接来进行API调用。

以类似的方式，我想用50或100个不同的用户（使用不同的帐户/凭证）连接到服务器进行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); 
}