AWSにFlaskアプリケーションを50回並行してデプロイするにはどうすればよいですか？

Question

私はAWSインスタンス上で実行されているアプリケーションをテストするには、次のPythonスクリプトを使用しています

、

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呼び出しを行うために、サーバーとの一つのセッション/接続を作成し、ただ一人のユーザーを使用しています。

同様の方法で、API呼び出しを行うサーバーに接続された50人または100人の異なるユーザー（異なるアカウントと資格情報を持つユーザー）でアプリケーションをテストしたいとします。 50人または100人のユーザーが同時にアプリケーションを使用しています。したがって、アプリケーションが50人のユーザーを適切に処理していることを確認できます。

スクリプトを使用してこのようなテストを行うにはどうすればよいですか？

更新：ほとんどのルートは非表示で、@login_requiredが必要です。

Answer 1

Bees With Machine Gunsをお試しください。マイクロEC2インスタンスを起動し、これらのインスタンスからアプリケーションに多くのリクエストを送信するPythonスクリプトです。これは、パフォーマンステストのためのトラフィックの急増をシミュレートします。

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); 
}