Page MenuHomePhabricator
Paste P55025

[locust] article descriptions isvc load test script with lw_stats_history() and lw_stats_analysis() using pandas
ActivePublic
Actions

Authored by kevinbazira on Jan 19 2024, 1:14 PM.
import os
import pandas as pd
from datetime import datetime
from locust import FastHttpUser, task, between, events
def lw_stats_analysis():
df = pd.read_csv("lw_stats_history.csv")
df = df.sort_values(by=["Timestamp"], ascending=True)
deltas = df.diff().iloc[1:]
latest_deltas = deltas.iloc[-1]
is_worse = latest_deltas["Average Response Time"] > 0 or latest_deltas["Failure Count"] > 0
performance, difference = ("worse", "regression") if is_worse else ("better", "improvement")
analysis_data = latest_deltas[["Average Response Time", "Failure Count"]].to_string().replace("Failure Count", " Failure Count")
analysis_report = """
LIFTWING LOAD TEST COMPLETED
The most recent load test for this isvc has been analyzed and presents {difference} in performance:
{analysis_data}
Compared with the prior load test, the performance is {performance}.
Detailed results of the current and past load tests are located in: lw_stats_history.csv.
""".format(performance=performance, difference=difference, analysis_data=analysis_data)
print(analysis_report)
@events.test_stop.add_listener
def lw_stats_history(**kwargs):
df = pd.read_csv("article_descriptions_stats.csv")
aggregated_row = df[df["Name"] == "Aggregated"]
if not aggregated_row.empty:
current_time = datetime.now().strftime("%Y%m%d%H%M%S")
aggregated_data = aggregated_row.drop(columns=["Type", "Name"])
aggregated_data.insert(0, "Timestamp", current_time)
if not os.path.isfile("lw_stats_history.csv"):
header = ["Timestamp"] + [col for col in df.columns if col not in ["Type", "Name"]]
aggregated_data.to_csv("lw_stats_history.csv", mode="w", index=False, header=header)
first_report = """
LIFTWING LOAD TEST COMPLETED
The first load test for this isvc has been successfully executed and recorded.
At present, a comparative analysis is not available due to the absence of preceding test data.
Future tests will enable us to present a comparative performance analysis.
For load test results, please refer to: lw_stats_history.csv.
"""
print(first_report)
else:
aggregated_data.to_csv("lw_stats_history.csv", mode="a", index=False, header=False)
lw_stats_analysis()
class ArticleDescriptionsUser(FastHttpUser):
wait_time = between(1, 2) # Adjust wait time as needed
host = "https://inference-staging.svc.codfw.wmnet:30443"
@task
def predict(self):
headers = {
"Host": "article-descriptions.experimental.wikimedia.org",
"Content-Type": "application/json"
}
data = {"lang": "en", "title": "Clandonald", "num_beams": 3}
self.client.post("/v1/models/article-descriptions:predict",
json=data,
headers=headers)
Content licensed under Creative Commons Attribution-ShareAlike (CC BY-SA) 4.0 unless otherwise noted; code licensed under GNU General Public License (GPL) 2.0 or later and other open source licenses. By using this site, you agree to the Terms of Use, Privacy Policy, and Code of Conduct. · Wikimedia Foundation · Privacy Policy · Code of Conduct · Terms of Use · Disclaimer · CC-BY-SA · GPL