in right_size_your_sagemaker_endpoints/load_test_helper.py [0:0]
def get_min_max_instances(results_df, min_requests, max_requests):
"""
Calculates recommendations for autoscaling
Based on the maximum requests handled by each endpoint, this function
calculates and returns the optimal instance count and type for an
autoscaling configuration.
Inputs:
results_df: pandas data frame with instance types and their maximum rps
min_requests: minimum number of requests per second required for the application
max_requests: maximum number of requests per second required for the application
Output:
Recommended instance type and count for optimal costs
"""
if max_requests < min_requests:
print("Minimum requests should be less than or equal to the maximum number of requests per second. Exiting..")
return
# calculate min and max number of instance required for each instance type
# to serve the min and max rps, and calculate the corresponding prices
results_df = results_df.copy(deep=True)
results_df['Min Instances'] = results_df['Max Requests per Second'].apply(lambda x: round(min_requests / x))
results_df['Pricing'] = results_df.apply(lambda x: x['Price per Hour'] * x['Min Instances'], axis=1)
results_df = results_df.sort_values(['Pricing'])
results_df = results_df[results_df['Min Instances'] > 0]
results_df['Max Instances'] = results_df['Max Requests per Second'].apply(lambda x: round(max_requests / x))
# recommended type is the top row of the sorted data frame
recommended_type = results_df.head(1).index.values[0]
recommended_type = re.sub(r'.x[0-9]', '', recommended_type)
recommended_min = results_df.head(1)['Min Instances'].values[0]
recommended_max = results_df.head(1)['Max Instances'].values[0]
recommended_dict = [
{"instance_type": recommended_type, "instance_count": int(recommended_min)},
{"instance_type": recommended_type, "instance_count": int(recommended_max)}
]
return recommended_dict