Added rl-based test dashboard

test/dashboard.py

@@ -0,0 +1,296 @@
+import solara
+import pandas as pd
+import os
+import zipfile
+from ray.tune.registry import register_env
+from env import Teastore
+from ray.rllib.algorithms import ppo, sac, dqn
+from solara.components.file_drop import FileInfo
+import time
+
+test_plot_data = solara.reactive({'step': [], 'replica': [], 'cpu': [], "load": [], 
+                                  "num_request": [], "response_time": []})
+uploaded_algo = solara.reactive(None)
+error_state = solara.reactive(None)
+number_of_steps = solara.reactive(10)
+available_checkpoint_names = solara.reactive([])
+selected_checkpoint_name = solara.reactive(None)
+uploaded_algo_status = solara.reactive(False)
+
+
+
+@solara.component
+def status_plot(data):
+    options_replica = {
+        "xAxis": {
+            "type": "category",
+            "data": data["step"],
+        },
+        "yAxis": {
+            "type": "value",
+        },
+        "series": [ 
+            {
+                "name": "Replica",
+                "data": data['replica'],
+                "type": 'line'
+            },      
+        ],
+        "title": {
+            "text": 'Replica Number',
+            "left": "center"
+        },
+        "legend": {
+            "orient": 'vertical',
+            "right": 0,
+            # "top": 50,
+            # "bottom": 50,
+            "data": ["Replica"]
+        },
+
+
+    }
+
+    options_cpu= {
+        "xAxis": {
+            "type": "category",
+            "data": data["step"],
+        },
+        "yAxis": {
+            "type": "value",
+        },
+        "series": [ 
+            {
+                "name": "CPU",
+                "data": data['cpu'],
+                "type": 'line'
+            },         
+        ],
+        "title": {
+            "text": 'CPU Limit',
+            "left": "center"
+            },
+        "legend": {
+            "orient": 'vertical',
+            "right": 0,
+            # "top": 50,
+            # "bottom": 50,
+            "data": ["CPU"]
+        },
+    }
+    options_load= {
+        "xAxis": {
+            "type": "category",
+            "data": data["step"],
+        },
+        "yAxis": {
+            "type": "value",
+        },
+        "series": [ 
+            {
+                "name": "Processed req",
+                "data": data['num_request'],
+                "type": 'line'
+            },
+            {
+                "name": "Load",
+                "data": data['load'],
+                "type": 'line'
+            },
+
+        ],
+        "title": {
+            "text": 'Number of Request (Tps) and Load (Tps)',
+            "left": "center"
+            },
+        "legend": {
+            "orient": 'vertical',
+            "right": 0,
+            # "top": 50,
+            # "bottom": 50,
+            "data": ["Processed req", "Load"]
+        },
+    }
+    options_response_time= {
+        "xAxis": {
+            "type": "category",
+            "data": data["step"],
+        },
+        "yAxis": {
+            "type": "value",
+        },
+        "series": [ 
+            {
+                "name": "Response time",
+                "data": data['response_time'],
+                "type": 'line'
+            },         
+        ],
+        "title": {
+            "text": 'Response time (ms)',
+            "left": "center"
+            },
+        "legend": {
+            "orient": 'vertical',
+            "right": 0,
+            # "top": 50,
+            # "bottom": 50,
+            "data": ["Response time"]
+        },        
+    }
+
+
+
+    with solara.GridFixed(columns=1):
+        # with solara.Column():
+        solara.FigureEcharts(option=options_replica)
+        solara.FigureEcharts(option=options_cpu)
+        
+        
+        solara.FigureEcharts(option=options_load)
+        solara.FigureEcharts(option=options_response_time)
+
+
+
+
+@solara.component
+def CheckpointDrop():
+    zip_content, set_zip_content = solara.use_state("")
+    content, set_content = solara.use_state(b"")
+    filename, set_filename = solara.use_state("")
+    size, set_size = solara.use_state(0)
+    extract_path, set_extract_path = solara.use_state("")
+
+
+    def on_file(f: FileInfo):
+        set_filename(f["name"])
+        set_size(f["size"])
+        temp_path = os.path.join(f["name"])
+        with open(temp_path, "wb") as temp_file:
+            temp_file.write(f["file_obj"].read())
+
+        extracted_folder = os.path.join("extracted_files", os.path.splitext(f["name"])[0])
+        with zipfile.ZipFile(temp_path, 'r') as zip_ref:
+            zip_ref.extractall(extracted_folder) 
+
+        set_extract_path(extracted_folder)       
+        extracted_files = os.listdir(extracted_folder)
+        set_zip_content("\n".join(extracted_files))  
+
+        # find the names of the checkpoint folders
+        #existing_names = available_checkpoint_names.value 
+        # updated_names = set(new_names + existing_names)
+        available_checkpoint_names.set(['denee'])
+
+        os.remove(temp_path)
+
+
+    
+    solara.FileDrop(
+        label="Drag and drop a file here",
+        on_file=on_file,
+        lazy=True,  # We will only read the first 100 bytes
+    )
+
+@solara.component
+def ListAvailableCheckpoints():
+    list_subfolders_names = [f.name for f in os.scandir("extracted_files") if f.is_dir()]
+    available_checkpoint_names.set(list_subfolders_names)
+
+
+def load_agent():
+    if selected_checkpoint_name.value is None:
+        return None
+
+    register_env("teastore", lambda config: Teastore())
+
+    config_dqn = (
+        dqn.DQNConfig()
+        .environment(env="teastore")
+        .rollouts(num_rollout_workers=1, enable_connectors=False, num_envs_per_worker=1)
+        .resources(num_gpus=0, num_cpus_per_worker=1)
+        .training(train_batch_size=256, model={"fcnet_hiddens": [32, 32]})
+    )
+    algo = config_dqn.build()
+    checkpoint_dir = selected_checkpoint_name.value
+    checkpoint_path = os.path.join("extracted_files", checkpoint_dir)
+    algo.restore(checkpoint_path)
+
+    
+
+    return algo
+
+def start_test():
+
+    env = Teastore()
+    obs, info = env.reset()   
+    done = False
+    truncated = False
+    sum_reward = 0
+    step_list = []
+    replica_array = []
+    cpu_array = []
+    num_request_array = []
+    load_array = []
+    response_time_array = []
+
+
+
+    for i in range(number_of_steps.value):
+        step_list.append(i)
+        replica_array.append(obs[0])
+        cpu_array.append(obs[1])
+        load_array.append(env.load)
+        response_time_array.append(env.response_time)
+        num_request_array.append(env.num_request)
+
+        action = uploaded_algo.value.compute_single_action(obs, explore=False)
+        next_state, reward, _, _, _ = env.step(action)
+        obs = next_state
+
+
+        test_plot_data.set(
+                {'step':step_list.copy(),
+                 'replica': replica_array.copy(),
+                 'cpu': cpu_array.copy(),
+                 "load": load_array.copy(),
+                 "response_time": response_time_array.copy(),
+                 "num_request": num_request_array.copy()
+                 })
+        # time.sleep(2)
+
+def load_test():
+    algo = load_agent()
+    if algo is None:
+        error_state.set('Couldnt load checkpoint')
+    else:
+        uploaded_algo_status.set(True)
+    uploaded_algo.set(algo)
+
+
+@solara.component
+def Page():
+
+
+    with solara.Sidebar():
+        if error_state.value is not None:
+            solara.Error(label=f'{error_state.value}')
+        # CheckpointDrop()
+        ListAvailableCheckpoints()
+        solara.Select(label="Choose checkpoint", values=available_checkpoint_names.value, value=selected_checkpoint_name.value, 
+            on_value=selected_checkpoint_name.set
+        )
+        solara.Button(label="Run test", on_click=start_test, disabled=True if uploaded_algo.value is None else False)
+        solara.Button(label="Load agent", on_click=load_test, disabled=True if selected_checkpoint_name.value is None else False)
+        if uploaded_algo_status.value == False:
+            solara.Info("Agent is not uploaded")
+        else:
+            solara.Info("Agent is ready for test")
+        solara.SliderInt(label="choose number of steps", min=1, max=500, value=number_of_steps)
+    
+
+    status_plot(test_plot_data.value)
+    
+
+    
+

test/env.py

@@ -0,0 +1,133 @@
+#### No heap in state
+
+from ray.rllib.env.policy_client import PolicyClient
+import pandas as pd
+from prometheus_api_client import PrometheusConnect
+from kubernetes import client, config
+import time
+import numpy as np
+from collections import OrderedDict
+from gymnasium.spaces import Discrete, Dict, MultiDiscrete, Tuple, Box
+import ssl
+import random
+import logging
+ssl._create_default_https_context = ssl._create_unverified_context
+from itertools import product
+import time
+import gymnasium as gym
+import math
+
+
+
+class Teastore(gym.Env):
+    DATA_PATH = "./all_load_mpa_cpu_and_performance_without_average.csv"
+    MAX_STEPS = 500
+
+
+    def __init__(self) -> None:
+        self.data = pd.read_csv(self.DATA_PATH)
+        # drop_rows = (df["cpu_usage"] != 0) | (df["memory_usage"] != 0)
+        # self.data = df[drop_rows].reset_index(drop=True)
+        self.action_space = Discrete(5)
+        self.observation_space = Box(low=np.array([1, 4, 0, 0]), high=np.array([3, 9, 1000,1000]), dtype=np.float32)
+        self.count = 0
+        self.info = {}
+        self.previous_tps = 0
+        self.idx = 0
+        self.up = None
+        self.load = 0
+        self.response_time = 0
+        self.num_request = 0
+
+
+
+
+    def find_next_state(self, target, expected_tps):
+        if expected_tps == 144:
+            self.up = False
+        elif expected_tps == 24:
+            self.up = True
+        
+        if self.up == True:
+            new_expected_tps = expected_tps + 24
+        elif self.up == False:
+            new_expected_tps = expected_tps - 24
+
+        new_previous_tps = expected_tps
+        # new_expected_tps = 48
+        # new_previous_tps = 24
+        next = np.concatenate([target, [new_previous_tps, new_expected_tps]])
+        equal_rows = np.all(self.data.loc[:, ["replica", "cpu", "previous_tps", "expected_tps"]].values == next, axis=1)
+        matched_indexes = np.where(equal_rows)[0]
+        return matched_indexes.tolist(), new_previous_tps, new_expected_tps
+    
+
+    
+    def reset(self, *, seed=None, options=None):
+        self.idx = random.randint(0, len(self.data)-1)
+        self.state = np.array(self.data.loc[self.idx, ["replica", "cpu", 'previous_tps', "expected_tps"]])
+        # self.state = np.array([3,9,24,48])
+        self.previous_tps = self.state[2]
+        self.truncated = False
+        self.terminated = False
+        self.reward = 0
+        self.count = 0
+        self.info = {}
+        self.up = True if self.state[3] - self.state[2] > 0 else False 
+        self.load = self.state[-1]
+        self.response_time = self.data.loc[self.idx, "response_time"]
+        self.num_request = self.data.loc[self.idx, "num_request"]
+        return self.state, self.info
+    
+    def step(self, action):
+        selected_row_idx = 0
+        self.count += 1
+
+        if action == 0:
+            temp_state = self.state[0:2] + np.array([0, 0])
+        elif action == 1: # increase_replica
+            temp_state = self.state[0:2] + np.array([1, 0])
+        elif action == 2: # decrease_replica
+            temp_state = self.state[0:2] + np.array([-1, 0])
+        elif action == 3:
+            temp_state = self.state[0:2] + np.array([0, 1])
+        else:
+            temp_state = self.state[0:2] + np.array([0 , -1])
+        
+
+
+        idx, new_previous_tps, new_expected_tps  = self.find_next_state(temp_state, self.state[3])
+
+        if idx:
+            selected_row_idx = random.choice(idx)
+            selected_data = self.data.iloc[selected_row_idx]
+            self.state = np.array(selected_data[["replica", "cpu", 'previous_tps',"expected_tps"]])
+            self.reward = selected_data['reward']
+            # self.reward = 1
+            # print(f"state: {self.state} - previous_tps: {self.previous_tps}")
+            self.previous_tps = selected_data["expected_tps"]
+            self.num_request = self.data.loc[selected_row_idx, "num_request"]
+            self.response_time = self.data.loc[selected_row_idx, "response_time"]
+            
+        else:
+            self.state[2] = new_previous_tps
+            self.state[3] = new_expected_tps
+            self.previous_tps = new_expected_tps
+            self.reward = -5
+            self.num_request = 0
+            self.response_time = 200
+
+        self.load = self.state[-1]
+        # self.response_time = 20
+        # self.num_request = 20
+
+        self.terminated = (self.count >= self.MAX_STEPS)
+        self.truncated = self.terminated
+        return self.state, self.reward, self.terminated, self.truncated, self.info
+
+
+
+
+
+
+

test/extracted_files/case5/policies/default_policy/rllib_checkpoint.json

@@ -0,0 +1 @@
+{"type": "Policy", "checkpoint_version": "1.1", "format": "cloudpickle", "state_file": "policy_state.pkl", "ray_version": "2.4.0", "ray_commit": "4479f66d4db967d3c9dd0af2572061276ba926ba"}

test/extracted_files/case5/rllib_checkpoint.json

@@ -0,0 +1 @@
+{"type": "Algorithm", "checkpoint_version": "1.1", "format": "cloudpickle", "state_file": "/Users/hasan.nayir/ray_results/DQN_teastore_2024-03-27_13-22-15jpy5kzub/checkpoint_010000/algorithm_state.pkl", "policy_ids": ["default_policy"], "ray_version": "2.4.0", "ray_commit": "4479f66d4db967d3c9dd0af2572061276ba926ba"}

test/extracted_files/case6/policies/default_policy/rllib_checkpoint.json

@@ -0,0 +1 @@
+{"type": "Policy", "checkpoint_version": "1.1", "format": "cloudpickle", "state_file": "policy_state.pkl", "ray_version": "2.4.0", "ray_commit": "4479f66d4db967d3c9dd0af2572061276ba926ba"}

test/extracted_files/case6/rllib_checkpoint.json

@@ -0,0 +1 @@
+{"type": "Algorithm", "checkpoint_version": "1.1", "format": "cloudpickle", "state_file": "/Users/hasan.nayir/ray_results/DQN_teastore_2024-03-29_16-52-25n5fvwg_p/checkpoint_010000/algorithm_state.pkl", "policy_ids": ["default_policy"], "ray_version": "2.4.0", "ray_commit": "4479f66d4db967d3c9dd0af2572061276ba926ba"}