<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Analyzing Locust Load Testing Results
This Notebook demonstrates how to analyze AI Platform Prediction load testing runs using metrics captured in Cloud Monitoring. 
This Notebook build on the 02-perf-testing.ipynb notebook that shows how to configure and run load tests against AI Platform Prediction using Locust.io. The outlined testing process results in a Pandas dataframe that aggregates the standard AI Platform Prediction metrics with a set of custom, log-based metrics generated from log entries captured by the Locust testing script.
The Notebook covers the following steps
Step1:  Configure GCP environment settings
Step2:  1. Retrieve and consolidate test results
Locust's web interface along with a Cloud Monitoring dashboard provide a cursory view into performance of a tested AI Platform Prediction model version. A more thorough analysis can be performed by consolidating metrics collected during a test and using data analytics and visualization tools.
In this section, you will retrieve the metrics captured in Cloud Monitoring and consolidate them into a single Pandas dataframe.
1.1 List available AI Platform Prediction metrics
Step3:  1.2. List custom log based metrics
Step7:  1.3. Retrieve test metrics
Define a helper function that retrieves test metrics from Cloud Monitoring
Step8:  The retrieved dataframe uses hierarchical indexing for column names. The reason is that some metrics contain multiple time series. For example, the GPU duty_cycle metric includes a time series of measures per each GPU used in the deployment (denoted as replica_id). The top level of the column index is a metric name. The second level is a replica_id. The third level is a signature of a model.
All metrics are aligned on the same timeline. 
2. Analyzing and Visualizing test results
In the context of our scenario the key concern is GPU utilization at various levels of throughput and latency. The primary metric exposed by AI Platform Prediction to monitor GPU utilization is duty cycle. This metric captures an average fraction of time over the 60 second period during which the accelerator(s) were actively processing.
2.1. GPU utilization
Step9:  2.2. CPU utilization
Step10:  2.3. Latency
Step11:  2.4. Request throughput
We are going to use the response_rate metric, which tracks a number of responses returned by AI Platform Prediction over a 1 minute interval.
Step12:  Cleaning up
<ASSISTANT_TASK:>
Python Code:
import time
from datetime import datetime
from typing import List
import numpy as np
import pandas as pd
import google.auth
from google.cloud import logging_v2
from google.cloud.monitoring_dashboard.v1 import DashboardsServiceClient
from google.cloud.logging_v2 import MetricsServiceV2Client
from google.cloud.monitoring_v3.query import Query
from google.cloud.monitoring_v3 import MetricServiceClient
import matplotlib.pyplot as plt
Explanation: Analyzing Locust Load Testing Results
This Notebook demonstrates how to analyze AI Platform Prediction load testing runs using metrics captured in Cloud Monitoring. 
This Notebook build on the 02-perf-testing.ipynb notebook that shows how to configure and run load tests against AI Platform Prediction using Locust.io. The outlined testing process results in a Pandas dataframe that aggregates the standard AI Platform Prediction metrics with a set of custom, log-based metrics generated from log entries captured by the Locust testing script.
The Notebook covers the following steps:
1. Retrieve and consolidate test results from Cloud Monitoring
2. Analyze and visualize utilization and latency results
Setup
This notebook was tested on AI Platform Notebooks using the standard TF 2.2 image.
Import libraries
End of explanation
PROJECT_ID = '[your-project-id]' # Set your project Id
MODEL_NAME = 'resnet_classifier'
MODEL_VERSION = 'v1'
LOG_NAME = 'locust' # Set your log name
TEST_ID = 'test-20200829-190943' # Set your test Id
TEST_START_TIME = datetime.fromisoformat('2020-08-28T21:30:00-00:00') # Set your test start time
TEST_END_TIME = datetime.fromisoformat('2020-08-29T22:00:00-00:00') # Set your test end time
Explanation: Configure GCP environment settings
End of explanation
creds , _ = google.auth.default()
client = MetricServiceClient(credentials=creds)
project_path = client.project_path(PROJECT_ID)
filter = 'metric.type=starts_with("ml.googleapis.com/prediction")'
for descriptor in client.list_metric_descriptors(project_path, filter_=filter):
    print(descriptor.type)
Explanation: 1. Retrieve and consolidate test results
Locust's web interface along with a Cloud Monitoring dashboard provide a cursory view into performance of a tested AI Platform Prediction model version. A more thorough analysis can be performed by consolidating metrics collected during a test and using data analytics and visualization tools.
In this section, you will retrieve the metrics captured in Cloud Monitoring and consolidate them into a single Pandas dataframe.
1.1 List available AI Platform Prediction metrics
End of explanation
filter = 'metric.type=starts_with("logging.googleapis.com/user")'
for descriptor in client.list_metric_descriptors(project_path, filter_=filter):
    print(descriptor.type)
Explanation: 1.2. List custom log based metrics
End of explanation
def retrieve_metrics(client, project_id, start_time, end_time, model, model_version, test_id, log_name):
    
    Retrieves test metrics from Cloud Monitoring.
    
    def _get_aipp_metric(metric_type: str, labels: List[str]=[], metric_name=None)-> pd.DataFrame:
        
        Retrieves a specified AIPP metric.
        
        query = Query(client, project_id, metric_type=metric_type)
        query = query.select_interval(end_time, start_time)
        query = query.select_resources(model_id=model)
        query = query.select_resources(version_id=model_version)
        
        if metric_name:
            labels = ['metric'] + labels 
        df = query.as_dataframe(labels=labels)
        
        if not df.empty:
            if metric_name:
                df.columns.set_levels([metric_name], level=0, inplace=True)
            df = df.set_index(df.index.round('T'))
        
        return df
    
    def _get_locust_metric(metric_type: str, labels: List[str]=[], metric_name=None)-> pd.DataFrame:
        
        Retrieves a specified custom log-based metric.
        
        query = Query(client, project_id, metric_type=metric_type)
        query = query.select_interval(end_time, start_time)
        query = query.select_metrics(log=log_name)
        query = query.select_metrics(test_id=test_id)
        
        if metric_name:
            labels = ['metric'] + labels 
        df = query.as_dataframe(labels=labels)
        
        if not df.empty: 
            if metric_name:
                df.columns.set_levels([metric_name], level=0, inplace=True)
            df = df.apply(lambda row: [metric.mean for metric in row])
            df = df.set_index(df.index.round('T'))
        
        return df
    
    # Retrieve GPU duty cycle
    metric_type = 'ml.googleapis.com/prediction/online/accelerator/duty_cycle'
    metric = _get_aipp_metric(metric_type, ['replica_id', 'signature'], 'duty_cycle')
    df = metric
    # Retrieve CPU utilization
    metric_type = 'ml.googleapis.com/prediction/online/cpu/utilization'
    metric = _get_aipp_metric(metric_type, ['replica_id', 'signature'], 'cpu_utilization')
    if not metric.empty:
        df = df.merge(metric, how='outer', right_index=True, left_index=True)
    
    # Retrieve prediction count
    metric_type = 'ml.googleapis.com/prediction/prediction_count'
    metric = _get_aipp_metric(metric_type, ['replica_id', 'signature'], 'prediction_count')
    if not metric.empty:
        df = df.merge(metric, how='outer', right_index=True, left_index=True)
    
    # Retrieve responses per second
    metric_type = 'ml.googleapis.com/prediction/response_count'
    metric = _get_aipp_metric(metric_type, ['replica_id', 'signature'], 'response_rate')
    if not metric.empty:
        metric = (metric/60).round(2)
        df = df.merge(metric, how='outer', right_index=True, left_index=True)
    
    # Retrieve backend latencies
    metric_type = 'ml.googleapis.com/prediction/latencies'
    metric = _get_aipp_metric(metric_type, ['latency_type', 'replica_id', 'signature'])
    if not metric.empty:
        metric = metric.apply(lambda row: [round(latency.mean/1000,1) for latency in row])
        metric.columns.set_names(['metric', 'replica_id', 'signature'], inplace=True)
        level_values = ['Latency: ' + value for value in metric.columns.get_level_values(level=0)]
        metric.columns.set_levels(level_values, level=0, inplace=True)
        df = df.merge(metric, how='outer', right_index=True, left_index=True)
    
    # Retrieve Locust latency
    metric_type = 'logging.googleapis.com/user/locust_latency'
    metric = _get_locust_metric(metric_type, ['replica_id', 'signature'], 'Latency: client')
    if not metric.empty:
        metric = metric.round(2).replace([0], np.nan)
        df = df.merge(metric, how='outer', right_index=True, left_index=True)
    
    # Retrieve Locust user count
    metric_type = 'logging.googleapis.com/user/locust_users'
    metric = _get_locust_metric(metric_type, ['replica_id', 'signature'], 'User count')
    if not metric.empty:
        metric = metric.round()
        df = df.merge(metric, how='outer', right_index=True, left_index=True)
    
    # Retrieve Locust num_failures
    metric_type = 'logging.googleapis.com/user/num_failures'
    metric = _get_locust_metric(metric_type, ['replica_id', 'signature'], 'Num of failures')
    if not metric.empty:
        metric = metric.round()
        df = df.merge(metric, how='outer', right_index=True, left_index=True)
    
    # Retrieve Locust num_failures
    metric_type = 'logging.googleapis.com/user/num_requests'
    metric = _get_locust_metric(metric_type, ['replica_id', 'signature'], 'Num of requests')
    if not metric.empty:
        metric = metric.round()
        df = df.merge(metric, how='outer', right_index=True, left_index=True)
    return df
    
test_result = retrieve_metrics(
    client, 
    PROJECT_ID, 
    TEST_START_TIME, 
    TEST_END_TIME, 
    MODEL_NAME, 
    MODEL_VERSION,
    TEST_ID, 
    LOG_NAME
)
test_result.head().T
Explanation: 1.3. Retrieve test metrics
Define a helper function that retrieves test metrics from Cloud Monitoring
End of explanation
gpu_utilization_results = test_result['duty_cycle']
gpu_utilization_results.columns = gpu_utilization_results.columns.get_level_values(0)
ax = gpu_utilization_results.plot(figsize=(14, 9), legend=True)
ax.set_xlabel('Time', fontsize=16)
ax.set_ylabel('Utilization ratio', fontsize=16)
_ = ax.set_title("GPU Utilization", fontsize=20)
Explanation: The retrieved dataframe uses hierarchical indexing for column names. The reason is that some metrics contain multiple time series. For example, the GPU duty_cycle metric includes a time series of measures per each GPU used in the deployment (denoted as replica_id). The top level of the column index is a metric name. The second level is a replica_id. The third level is a signature of a model.
All metrics are aligned on the same timeline. 
2. Analyzing and Visualizing test results
In the context of our scenario the key concern is GPU utilization at various levels of throughput and latency. The primary metric exposed by AI Platform Prediction to monitor GPU utilization is duty cycle. This metric captures an average fraction of time over the 60 second period during which the accelerator(s) were actively processing.
2.1. GPU utilization
End of explanation
cpu_utilization_results = test_result['cpu_utilization']
cpu_utilization_results.columns = cpu_utilization_results.columns.get_level_values(0)
ax = cpu_utilization_results.plot(figsize=(14, 9), legend=True)
ax.set_xlabel('Time', fontsize=16)
ax.set_ylabel('Utilization ratio', fontsize=16)
_ = ax.set_title("CPU Utilization", fontsize=20)
Explanation: 2.2. CPU utilization
End of explanation
latency_results = test_result[['Latency: model', 'Latency: client']]
latency_results.columns = latency_results.columns.get_level_values(0)
ax = latency_results.plot(figsize=(14, 9), legend=True)
ax.set_xlabel('Time', fontsize=16)
ax.set_ylabel('milisecond', fontsize=16)
_ = ax.set_title("Latency", fontsize=20)
Explanation: 2.3. Latency
End of explanation
throughput_results = test_result[['response_rate', 'User count']]
throughput_results.columns = throughput_results.columns.get_level_values(0)
ax = throughput_results.plot(figsize=(14, 9), legend=True)
ax.set_xlabel('Time', fontsize=16)
ax.set_ylabel('Count', fontsize=16)
_ = ax.set_title("Response Rate vs User Count", fontsize=20)
Explanation: 2.4. Request throughput
We are going to use the response_rate metric, which tracks a number of responses returned by AI Platform Prediction over a 1 minute interval.
End of explanation
logging_client = MetricsServiceV2Client(credentials=creds)
parent = logging_client.project_path(PROJECT_ID)
for element in logging_client.list_log_metrics(parent):
    metric_path = logging_client.metric_path(PROJECT_ID, element.name)
    logging_client.delete_log_metric(metric_path)
    print("Deleted metric: ", metric_path)
display_name = 'AI Platform Prediction and Locust'
dashboard_service_client = DashboardsServiceClient(credentials=creds)
parent = 'projects/{}'.format(PROJECT_ID)
for dashboard in dashboard_service_client.list_dashboards(parent):
    if dashboard.display_name == display_name:
        dashboard_service_client.delete_dashboard(dashboard.name)
        print("Deleted dashboard:", dashboard.name)
Explanation: Cleaning up: delete the log-based metrics and dasboard
End of explanation
<END_TASK>