import json
import numpy as np
import pandas as pd

import pytz
import datetime

import requests
from requests.auth import HTTPBasicAuth

import wallaroo
import pyarrow as pa
import time

wl = wallaroo.Client()

workspace = wl.get_workspace(name="metric-retrieval-summarization-tutorial", create_if_not_exist=True)
wl.set_current_workspace(workspace)

{'name': 'metric-retrieval-summarization-tutorial', 'id': 1716, 'archived': False, 'created_by': '7d603858-88e0-472e-8f71-e41094afd7ec', 'created_at': '2025-08-11T16:56:50.306331+00:00', 'models': [], 'pipelines': []}

# upload the sentiment analyzer

input_schema = pa.schema([
     pa.field('inputs', pa.string()), # required
     pa.field('top_k', pa.int64()),  
])

output_schema = pa.schema([
     pa.field('label', pa.list_(pa.string(), list_size=6)), # list with a number of items same as top_k, list_size can be skipped but may lead in worse performance
     pa.field('score', pa.list_(pa.float64(), list_size=6)), # list with a number of items same as top_k, list_size can be skipped but may lead in worse performance
])

framework=wallaroo.framework.Framework.HUGGING_FACE_TEXT_CLASSIFICATION
model_name = "toxic-bert-analysis"
model_file_name = './models/unitary-toxic-bert.zip'

bert_model = wl.upload_model(model_name,
                         model_file_name,
                         framework=framework,
                         input_schema=input_schema,
                         output_schema=output_schema,
                         convert_wait=False)

while bert_model.status() != "ready" and bert_model.status() != "error":
    print(bert_model.status())
    time.sleep(10)
print(bert_model.status())

pending_load_container
pending_load_container
pending_load_container
pending_load_container
attempting_load_container
attempting_load_container
attempting_load_container
ready

# create the pipeline
pipeline_name = "metrics-retrieval-tutorial-summarization-pipeline"
pipeline = wl.build_pipeline(pipeline_name)

# add the model as a pipeline step

pipeline.add_model_step(bert_model)

# set the deployment configuration for 4 cpus, 1 replica, 8 Gi RAM
deployment_config = wallaroo.DeploymentConfigBuilder() \
    .cpus(0.25).memory('1Gi') \
    .sidekick_cpus(bert_model, 4) \
    .sidekick_memory(bert_model, "8Gi") \
    .build()

# deploy the pipeline
pipeline.deploy(deployment_config=deployment_config, wait_for_status=False)
# saved for later steps
deploy = pipeline._deployment

Deployment initiated for metrics-retrieval-tutorial-summarization-pipeline. Please check pipeline status.

# wait until deployment is complete before continuing
import time
time.sleep(15)

while pipeline.status()['status'] != 'Running':
    time.sleep(15)
    print("Waiting for deployment.")
    pipeline.status()['status']
pipeline.status()['status']

Waiting for deployment.
Waiting for deployment.
'Running'

import time
timeout = time.time() + 60   # 1 minutes from now
while True:
    if time.time() > timeout:
        break
    pipeline.infer_from_file("./data/sample_input.json")

pipeline.logs()

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

pipeline.undeploy()

# set prometheus requirements
pipeline_id = pipeline_name # the name of the pipeline
step = "1m" # the step of the calculation

# this will also format the timezone in the parsing section
timezone = "US/Central"

selected_timezone = pytz.timezone(timezone)

# Define the start and end times
# adjust these values based on your your historical data.
data_start = selected_timezone.localize(datetime.datetime(2025, 8, 10, 9, 0, 0))
data_end = selected_timezone.localize(datetime.datetime(2025, 8, 12, 9, 59, 59))

# this is the URL to get prometheus metrics
query_url = f"{wl.api_endpoint}/v1/metrics/api/v1/query_range"

# Retrieve the token 
headers = wl.auth.auth_header()

# Convert to UTC and get the Unix timestamps
start_timestamp = int(data_start.astimezone(pytz.UTC).timestamp())
end_timestamp = int(data_end.astimezone(pytz.UTC).timestamp())    

query_rps = f'sum by (pipeline_name) (rate(latency_histogram_ns_count{{pipeline_name="{pipeline_id}"}}[{step}]))'
#request parameters
params_rps = {
    'query': query_rps,
    'start': start_timestamp,
    'end': end_timestamp,
    'step': step
}

response_rps = requests.get(query_url, headers=headers, params=params_rps)

if response_rps.status_code == 200:
    print("Requests Per Second Data:")
    # display(response_rps.json()['data']['result'][0]['values'])
    display(list(filter(lambda x: x[1] != '0', response_rps.json()['data']['result'][0]['values'])))
else:
    print("Failed to fetch RPS data:", response_rps.status_code, response_rps.text)

Requests Per Second Data:

[[1754932200, '1.8166157575757578'],
 [1754935860, '1.0545454545454545'],
 [1754935920, '0.77652']]

query_inference_rate = f'sum by (pipeline_name) (rate(tensor_throughput_batch_count{{pipeline_name="{pipeline_id}"}}[{step}]))'

# inference rte
params_inference_rate = {
    'query': query_inference_rate,
    'start': start_timestamp,
    'end': end_timestamp,
    'step': step
}

response_inference_rate = requests.get(query_url, headers=headers, params=params_inference_rate)

if response_inference_rate.status_code == 200:
    print("Cluster Inference Rate Data:")
    # filter out non-zero
    display(list(filter(lambda x: x[1] != '0', response_rps.json()['data']['result'][0]['values'])))
else:
    print("Failed to fetch Inference Rate data:", response_inference_rate.status_code, response_inference_rate.text)

Cluster Inference Rate Data:

[[1754932200, '1.8166157575757578'],
 [1754935860, '1.0545454545454545'],
 [1754935920, '0.77652']]