Astro-observe

.gitignore

@@ -327,6 +327,9 @@ cython_debug/
 # Visual Studio Code
 .vscode/
 
+# Weight & Biases
+wandb/
+
 # Project specific
 cache_dir
 demos

README.md

@@ -28,6 +28,7 @@ Moreover, AIOpsLab provides a built-in benchmark suite with a set of problems to
 
 ### Requirements
 - Python >= 3.11
+- [Helm](https://helm.sh/)
 
 Recommended installation:
 ```bash
@@ -64,10 +65,23 @@ kind create cluster --config kind/kind-config-arm.yaml
 
 If you're running into issues, consider building a Docker image for your machine by following this [README](kind/README.md). Please also open an issue.
 
+### [Tips]
+If you are running AIOpsLab using a proxy, beware of exporting the HTTP proxy as `172.17.0.1`. When creating the kind cluster, all the nodes in the cluster will inherit the proxy setting from the host environment and the Docker container. 
+
+The `172.17.0.1` address is used to communicate with the host machine. For more details, refer to the official guide: [Configure Kind to Use a Proxy](https://kind.sigs.k8s.io/docs/user/quick-start/#configure-kind-to-use-a-proxy).
+
+Additionally, Docker doesn't support SOCKS5 proxy directly. If you're using a SOCKS5 protocol to proxy, you may need to use [Privoxy](https://www.privoxy.org) to forward SOCKS5 to HTTP.
+
+If you're running VLLM and the LLM agent locally, Privoxy will by default proxy `localhost`, which will cause errors. To avoid this issue, you should set the following environment variable:
+
+```bash
+export no_proxy=localhost
+``` 
+
 After finishing cluster creation, proceed to the next "Update `config.yml`" step.
 
 ### b) Remote cluster
-AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks we have to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
+AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
 
 ### Update `config.yml`
 ```bash
@@ -89,12 +103,30 @@ python3 cli.py
 Run GPT-4 baseline agent:
 
 ```bash
-export OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>
+# Create a .env file in the project root (if not exists)
+echo "OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>" > .env
+# Add more API keys as needed:
+# echo "QWEN_API_KEY=<YOUR_QWEN_API_KEY>" >> .env
+# echo "DEEPSEEK_API_KEY=<YOUR_DEEPSEEK_API_KEY>" >> .env
+
 python3 clients/gpt.py # you can also change the problem to solve in the main() function
 ```
 
+The clients will automatically load API keys from your .env file.
+
 You can check the running status of the cluster using [k9s](https://k9scli.io/) or other cluster monitoring tools conveniently.
 
+To browse your logged `session_id` values in the W&B app as a table:
+
+1. Make sure you have W&B installed and configured.
+2. Set the USE_WANDB environment variable:
+ ```bash
+ # Add to your .env file
+ echo "USE_WANDB=true" >> .env
+ ```
+3. In the W&B web UI, open any run and click Tables → Add Query Panel.
+4. In the key field, type `runs.summary` and click `Run`, then you will see the results displayed in a table format.
+
 <h2 id="⚙️usage">⚙️ Usage</h2>
 
 AIOpsLab can be used in the following ways:

aiopslab/generators/fault/base.py

@@ -59,7 +59,6 @@ def _recover(
  self._invoke_method("recover", fault_type, microservices)
  elif fault_type:
  self._invoke_method("recover", fault_type)
- time.sleep(6)
 
  def _invoke_method(self, action_prefix, *args):
  """helper: injects/recovers faults based on name"""

aiopslab/generators/fault/inject_otel.py

@@ -8,7 +8,7 @@ class OtelFaultInjector(FaultInjector):
  def __init__(self, namespace: str):
  self.namespace = namespace
  self.kubectl = KubeCtl()
- self.configmap_name = f"{namespace}-flagd-config"
+ self.configmap_name = "flagd-config"
 
  def inject_fault(self, feature_flag: str):
  command = (
@@ -39,6 +39,11 @@ def inject_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
+
  print(f"Fault injected: Feature flag '{feature_flag}' set to 'on'.")
 
  def recover_fault(self, feature_flag: str):
@@ -70,6 +75,10 @@ def recover_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
  print(f"Fault recovered: Feature flag '{feature_flag}' set to 'off'.")
 
 

aiopslab/observer/__init__.py

@@ -12,7 +12,8 @@
 root_path = pathlib.Path(__file__).parent
 sys.path.append(root_path)
 # read the configuration file
-monitor_config = full_load(open(root_path / "monitor_config.yaml", "r"))
+with open(root_path / "monitor_config.yaml", "r") as f:
+ monitor_config = full_load(f)
 
 
 # root_config = full_load(open(root_path / "config.yaml", "r"))

aiopslab/observer/trace_api.py

@@ -22,13 +22,18 @@ def __init__(self, namespace: str):
  self.namespace = namespace
  self.stop_event = threading.Event()
 
- # NOTE: it may not be jaeger-out for other apps
- node_port = self.get_nodeport("jaeger", namespace)
- if node_port:
- self.base_url = f"http://localhost:{node_port}"
- else:
- self.base_url = "http://localhost:16686"
+ if self.namespace == "astronomy-shop":
+ # No NodePort in astronomy shop
+ self.base_url = "http://localhost:16686/jaeger/ui"
  self.start_port_forward()
+ else:
+ # Other namespaces may expose a NodePort
+ node_port = self.get_nodeport("jaeger", namespace)
+ if node_port:
+ self.base_url = f"http://localhost:{node_port}"
+ else:
+ self.base_url = "http://localhost:16686"
+ self.start_port_forward()
 
  def get_nodeport(self, service_name, namespace):
  """Fetch the NodePort for the given service."""
@@ -74,19 +79,38 @@ def print_output(self, stream):
  def is_port_in_use(self, port):
  with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
  return s.connect_ex(("127.0.0.1", port)) == 0
+
+ def get_jaeger_pod_name(self):
+ try:
+ result = subprocess.check_output(
+ ["kubectl", "get", "pods", "-n", self.namespace,
+ "-l", "app.kubernetes.io/name=jaeger",
+ "-o", "jsonpath={.items[0].metadata.name}"],
+ text=True
+ )
+ return result.strip()
+ except subprocess.CalledProcessError as e:
+ print("Error getting Jaeger pod name:", e)
+ raise
 
  def start_port_forward(self):
- """Starts kubectl port-forward command to access Jaeger service."""
+ """Starts kubectl port-forward command to access Jaeger service or pod."""
  for attempt in range(3):
  if self.is_port_in_use(16686):
  print(
- f"Port 16686 is already in use. Attempt {attempt + 1} of {3}. Retrying in {3} seconds..."
+ f"Port 16686 is already in use. Attempt {attempt + 1} of 3. Retrying in 3 seconds..."
  )
  time.sleep(3)
  continue
 
- # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
- command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+ # Use pod port-forwarding for astronomy-shop only
+ if self.namespace == "astronomy-shop":
+ pod_name = self.get_jaeger_pod_name()
+ command = f"kubectl port-forward pod/{pod_name} 16686:16686 -n {self.namespace}"
+ else:
+ command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+
+ print("Starting port-forward with command:", command)
  self.port_forward_process = subprocess.Popen(
  command,
  shell=True,
@@ -103,17 +127,17 @@ def start_port_forward(self):
  )
  thread_out.start()
  thread_err.start()
- time.sleep(3) # Wait a bit for the port-forward to establish
 
-if (
- self.port_forward_process.poll() is None
-): # Check if the process is still running
+time.sleep(3) # Let port-forward initialize
+
+ if self.port_forward_process.poll() is None:
  print("Port forwarding established successfully.")
  break
  else:
  print("Port forwarding failed. Retrying...")
  else:
- print("Failed to establish port forwarding after multiple attempts.")
+ print("Failed to establish port forwarding after 3 attempts.")
+
  # TODO: modify this command for other microservices
  # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
  # self.port_forward_process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
@@ -151,13 +175,14 @@ def cleanup(self):
  def get_services(self) -> list:
  """Fetch a list of services from the tracing API."""
  url = f"{self.base_url}/api/services"
- response = requests.get(url)
- if response.status_code == 200:
- data = response.json()
- # print(f"data: {response}")
- return data.get("data", [])
- else:
- print(f"Failed to get services: {response.status_code}")
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
+
+ try:
+ response = requests.get(url, headers=headers)
+ response.raise_for_status()
+ return response.json().get("data", [])
+ except Exception as e:
+ print(f"Failed to get services: {e}")
  return []
 
  def get_traces(
@@ -171,15 +196,14 @@ def get_traces(
  Fetch traces for a specific service between start_time and end_time.
  If limit is not specified, all available traces are fetched.
  """
- # Calculate the lookback in milliseconds.
  lookback = int((datetime.now() - start_time).total_seconds())
-
  url = f"{self.base_url}/api/traces?service={service_name}&lookback={lookback}s"
  if limit is not None:
  url += f"&limit={limit}"
 
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
  try:
- response = requests.get(url)
+ response = requests.get(url, headers=headers)
  response.raise_for_status()
  return response.json().get("data", [])
  except requests.RequestException as e:
@@ -211,9 +235,9 @@ def extract_traces(
  )
  for trace in traces:
  for span in trace["spans"]:
- span[
+ span["serviceName"] = trace["processes"][span["processID"]][
  "serviceName"
- ] = service # Directly associate service name with each span
+ ]
  all_traces.append(trace) # Collect the trace with service name included
  self.cleanup()
  print("Cleanup completed.")
@@ -223,22 +247,20 @@ def extract_traces(
  def process_traces(self, traces) -> pd.DataFrame:
  """Process raw traces data into a structured DataFrame."""
  trace_id_list = []
+ span_id_list = []
  service_name_list = []
  operation_name_list = []
  start_time_list = []
  duration_list = []
  parent_span_list = []
+ error_list = []
+ response_list = []
 
  for trace in traces:
  trace_id = trace["traceID"]
  for span in trace["spans"]:
  trace_id_list.append(trace_id)
- service_name_list.append(
- span["serviceName"]
- ) # Use the correct service name from the span
- operation_name_list.append(span["operationName"])
- start_time_list.append(span["startTime"])
- duration_list.append(span["duration"])
+ span_id_list.append(span["spanID"])
  parent_span = "ROOT"
  if "references" in span:
  for ref in span["references"]:
@@ -247,14 +269,34 @@ def process_traces(self, traces) -> pd.DataFrame:
  break
  parent_span_list.append(parent_span)
 
+ service_name_list.append(
+ span["serviceName"]
+ ) # Use the correct service name from the span
+ operation_name_list.append(span["operationName"])
+ start_time_list.append(span["startTime"])
+ duration_list.append(span["duration"])
+
+ has_error = False
+ response = "Unknown"
+ for tag in span.get("tags", []):
+ if tag["key"] == "error" and tag["value"] == True:
+ has_error = True
+ if tag["key"] == "http.status_code" or tag["key"] == "response_class":
+ response = tag["value"]
+ error_list.append(has_error)
+ response_list.append(response)
+
  df = pd.DataFrame(
  {
  "trace_id": trace_id_list,
+ "span_id": span_id_list,
+ "parent_span": parent_span_list,
  "service_name": service_name_list,
  "operation_name": operation_name_list,
  "start_time": start_time_list,
  "duration": duration_list,
- "parent_span": parent_span_list,
+ "has_error": error_list,
+ "response": response_list,
  }
  )
  return df

.gitignore

@@ -327,6 +327,9 @@ cython_debug/
 # Visual Studio Code
 .vscode/
 
+# Weight & Biases
+wandb/
+
 # Project specific
 cache_dir
 demos

README.md

@@ -28,6 +28,7 @@ Moreover, AIOpsLab provides a built-in benchmark suite with a set of problems to
 
 ### Requirements
 - Python >= 3.11
+- [Helm](https://helm.sh/)
 
 Recommended installation:
 ```bash
@@ -64,10 +65,23 @@ kind create cluster --config kind/kind-config-arm.yaml
 
 If you're running into issues, consider building a Docker image for your machine by following this [README](kind/README.md). Please also open an issue.
 
+### [Tips]
+If you are running AIOpsLab using a proxy, beware of exporting the HTTP proxy as `172.17.0.1`. When creating the kind cluster, all the nodes in the cluster will inherit the proxy setting from the host environment and the Docker container. 
+
+The `172.17.0.1` address is used to communicate with the host machine. For more details, refer to the official guide: [Configure Kind to Use a Proxy](https://kind.sigs.k8s.io/docs/user/quick-start/#configure-kind-to-use-a-proxy).
+
+Additionally, Docker doesn't support SOCKS5 proxy directly. If you're using a SOCKS5 protocol to proxy, you may need to use [Privoxy](https://www.privoxy.org) to forward SOCKS5 to HTTP.
+
+If you're running VLLM and the LLM agent locally, Privoxy will by default proxy `localhost`, which will cause errors. To avoid this issue, you should set the following environment variable:
+
+```bash
+export no_proxy=localhost
+``` 
+
 After finishing cluster creation, proceed to the next "Update `config.yml`" step.
 
 ### b) Remote cluster
-AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks we have to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
+AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
 
 ### Update `config.yml`
 ```bash
@@ -89,12 +103,30 @@ python3 cli.py
 Run GPT-4 baseline agent:
 
 ```bash
-export OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>
+# Create a .env file in the project root (if not exists)
+echo "OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>" > .env
+# Add more API keys as needed:
+# echo "QWEN_API_KEY=<YOUR_QWEN_API_KEY>" >> .env
+# echo "DEEPSEEK_API_KEY=<YOUR_DEEPSEEK_API_KEY>" >> .env
+
 python3 clients/gpt.py # you can also change the problem to solve in the main() function
 ```
 
+The clients will automatically load API keys from your .env file.
+
 You can check the running status of the cluster using [k9s](https://k9scli.io/) or other cluster monitoring tools conveniently.
 
+To browse your logged `session_id` values in the W&B app as a table:
+
+1. Make sure you have W&B installed and configured.
+2. Set the USE_WANDB environment variable:
+ ```bash
+ # Add to your .env file
+ echo "USE_WANDB=true" >> .env
+ ```
+3. In the W&B web UI, open any run and click Tables → Add Query Panel.
+4. In the key field, type `runs.summary` and click `Run`, then you will see the results displayed in a table format.
+
 <h2 id="⚙️usage">⚙️ Usage</h2>
 
 AIOpsLab can be used in the following ways:

aiopslab/generators/fault/base.py

@@ -59,7 +59,6 @@ def _recover(
  self._invoke_method("recover", fault_type, microservices)
  elif fault_type:
  self._invoke_method("recover", fault_type)
- time.sleep(6)
 
  def _invoke_method(self, action_prefix, *args):
  """helper: injects/recovers faults based on name"""

aiopslab/generators/fault/inject_otel.py

@@ -8,7 +8,7 @@ class OtelFaultInjector(FaultInjector):
  def __init__(self, namespace: str):
  self.namespace = namespace
  self.kubectl = KubeCtl()
- self.configmap_name = f"{namespace}-flagd-config"
+ self.configmap_name = "flagd-config"
 
  def inject_fault(self, feature_flag: str):
  command = (
@@ -39,6 +39,11 @@ def inject_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
+
  print(f"Fault injected: Feature flag '{feature_flag}' set to 'on'.")
 
  def recover_fault(self, feature_flag: str):
@@ -70,6 +75,10 @@ def recover_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
  print(f"Fault recovered: Feature flag '{feature_flag}' set to 'off'.")
 
 

aiopslab/observer/__init__.py

@@ -12,7 +12,8 @@
 root_path = pathlib.Path(__file__).parent
 sys.path.append(root_path)
 # read the configuration file
-monitor_config = full_load(open(root_path / "monitor_config.yaml", "r"))
+with open(root_path / "monitor_config.yaml", "r") as f:
+ monitor_config = full_load(f)
 
 
 # root_config = full_load(open(root_path / "config.yaml", "r"))

aiopslab/observer/trace_api.py

@@ -22,13 +22,18 @@ def __init__(self, namespace: str):
  self.namespace = namespace
  self.stop_event = threading.Event()
 
- # NOTE: it may not be jaeger-out for other apps
- node_port = self.get_nodeport("jaeger", namespace)
- if node_port:
- self.base_url = f"http://localhost:{node_port}"
- else:
- self.base_url = "http://localhost:16686"
+ if self.namespace == "astronomy-shop":
+ # No NodePort in astronomy shop
+ self.base_url = "http://localhost:16686/jaeger/ui"
  self.start_port_forward()
+ else:
+ # Other namespaces may expose a NodePort
+ node_port = self.get_nodeport("jaeger", namespace)
+ if node_port:
+ self.base_url = f"http://localhost:{node_port}"
+ else:
+ self.base_url = "http://localhost:16686"
+ self.start_port_forward()
 
  def get_nodeport(self, service_name, namespace):
  """Fetch the NodePort for the given service."""
@@ -74,19 +79,38 @@ def print_output(self, stream):
  def is_port_in_use(self, port):
  with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
  return s.connect_ex(("127.0.0.1", port)) == 0
+
+ def get_jaeger_pod_name(self):
+ try:
+ result = subprocess.check_output(
+ ["kubectl", "get", "pods", "-n", self.namespace,
+ "-l", "app.kubernetes.io/name=jaeger",
+ "-o", "jsonpath={.items[0].metadata.name}"],
+ text=True
+ )
+ return result.strip()
+ except subprocess.CalledProcessError as e:
+ print("Error getting Jaeger pod name:", e)
+ raise
 
  def start_port_forward(self):
- """Starts kubectl port-forward command to access Jaeger service."""
+ """Starts kubectl port-forward command to access Jaeger service or pod."""
  for attempt in range(3):
  if self.is_port_in_use(16686):
  print(
- f"Port 16686 is already in use. Attempt {attempt + 1} of {3}. Retrying in {3} seconds..."
+ f"Port 16686 is already in use. Attempt {attempt + 1} of 3. Retrying in 3 seconds..."
  )
  time.sleep(3)
  continue
 
- # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
- command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+ # Use pod port-forwarding for astronomy-shop only
+ if self.namespace == "astronomy-shop":
+ pod_name = self.get_jaeger_pod_name()
+ command = f"kubectl port-forward pod/{pod_name} 16686:16686 -n {self.namespace}"
+ else:
+ command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+
+ print("Starting port-forward with command:", command)
  self.port_forward_process = subprocess.Popen(
  command,
  shell=True,
@@ -103,17 +127,17 @@ def start_port_forward(self):
  )
  thread_out.start()
  thread_err.start()
- time.sleep(3) # Wait a bit for the port-forward to establish
 
-if (
- self.port_forward_process.poll() is None
-): # Check if the process is still running
+time.sleep(3) # Let port-forward initialize
+
+ if self.port_forward_process.poll() is None:
  print("Port forwarding established successfully.")
  break
  else:
  print("Port forwarding failed. Retrying...")
  else:
- print("Failed to establish port forwarding after multiple attempts.")
+ print("Failed to establish port forwarding after 3 attempts.")
+
  # TODO: modify this command for other microservices
  # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
  # self.port_forward_process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
@@ -151,13 +175,14 @@ def cleanup(self):
  def get_services(self) -> list:
  """Fetch a list of services from the tracing API."""
  url = f"{self.base_url}/api/services"
- response = requests.get(url)
- if response.status_code == 200:
- data = response.json()
- # print(f"data: {response}")
- return data.get("data", [])
- else:
- print(f"Failed to get services: {response.status_code}")
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
+
+ try:
+ response = requests.get(url, headers=headers)
+ response.raise_for_status()
+ return response.json().get("data", [])
+ except Exception as e:
+ print(f"Failed to get services: {e}")
  return []
 
  def get_traces(
@@ -171,15 +196,14 @@ def get_traces(
  Fetch traces for a specific service between start_time and end_time.
  If limit is not specified, all available traces are fetched.
  """
- # Calculate the lookback in milliseconds.
  lookback = int((datetime.now() - start_time).total_seconds())
-
  url = f"{self.base_url}/api/traces?service={service_name}&lookback={lookback}s"
  if limit is not None:
  url += f"&limit={limit}"
 
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
  try:
- response = requests.get(url)
+ response = requests.get(url, headers=headers)
  response.raise_for_status()
  return response.json().get("data", [])
  except requests.RequestException as e:
@@ -211,9 +235,9 @@ def extract_traces(
  )
  for trace in traces:
  for span in trace["spans"]:
- span[
+ span["serviceName"] = trace["processes"][span["processID"]][
  "serviceName"
- ] = service # Directly associate service name with each span
+ ]
  all_traces.append(trace) # Collect the trace with service name included
  self.cleanup()
  print("Cleanup completed.")
@@ -223,22 +247,20 @@ def extract_traces(
  def process_traces(self, traces) -> pd.DataFrame:
  """Process raw traces data into a structured DataFrame."""
  trace_id_list = []
+ span_id_list = []
  service_name_list = []
  operation_name_list = []
  start_time_list = []
  duration_list = []
  parent_span_list = []
+ error_list = []
+ response_list = []
 
  for trace in traces:
  trace_id = trace["traceID"]
  for span in trace["spans"]:
  trace_id_list.append(trace_id)
- service_name_list.append(
- span["serviceName"]
- ) # Use the correct service name from the span
- operation_name_list.append(span["operationName"])
- start_time_list.append(span["startTime"])
- duration_list.append(span["duration"])
+ span_id_list.append(span["spanID"])
  parent_span = "ROOT"
  if "references" in span:
  for ref in span["references"]:
@@ -247,14 +269,34 @@ def process_traces(self, traces) -> pd.DataFrame:
  break
  parent_span_list.append(parent_span)
 
+ service_name_list.append(
+ span["serviceName"]
+ ) # Use the correct service name from the span
+ operation_name_list.append(span["operationName"])
+ start_time_list.append(span["startTime"])
+ duration_list.append(span["duration"])
+
+ has_error = False
+ response = "Unknown"
+ for tag in span.get("tags", []):
+ if tag["key"] == "error" and tag["value"] == True:
+ has_error = True
+ if tag["key"] == "http.status_code" or tag["key"] == "response_class":
+ response = tag["value"]
+ error_list.append(has_error)
+ response_list.append(response)
+
  df = pd.DataFrame(
  {
  "trace_id": trace_id_list,
+ "span_id": span_id_list,
+ "parent_span": parent_span_list,
  "service_name": service_name_list,
  "operation_name": operation_name_list,
  "start_time": start_time_list,
  "duration": duration_list,
- "parent_span": parent_span_list,
+ "has_error": error_list,
+ "response": response_list,
  }
  )
  return df

.gitignore

@@ -327,6 +327,9 @@ cython_debug/
 # Visual Studio Code
 .vscode/
 
+# Weight & Biases
+wandb/
+
 # Project specific
 cache_dir
 demos

README.md

@@ -28,6 +28,7 @@ Moreover, AIOpsLab provides a built-in benchmark suite with a set of problems to
 
 ### Requirements
 - Python >= 3.11
+- [Helm](https://helm.sh/)
 
 Recommended installation:
 ```bash
@@ -64,10 +65,23 @@ kind create cluster --config kind/kind-config-arm.yaml
 
 If you're running into issues, consider building a Docker image for your machine by following this [README](kind/README.md). Please also open an issue.
 
+### [Tips]
+If you are running AIOpsLab using a proxy, beware of exporting the HTTP proxy as `172.17.0.1`. When creating the kind cluster, all the nodes in the cluster will inherit the proxy setting from the host environment and the Docker container. 
+
+The `172.17.0.1` address is used to communicate with the host machine. For more details, refer to the official guide: [Configure Kind to Use a Proxy](https://kind.sigs.k8s.io/docs/user/quick-start/#configure-kind-to-use-a-proxy).
+
+Additionally, Docker doesn't support SOCKS5 proxy directly. If you're using a SOCKS5 protocol to proxy, you may need to use [Privoxy](https://www.privoxy.org) to forward SOCKS5 to HTTP.
+
+If you're running VLLM and the LLM agent locally, Privoxy will by default proxy `localhost`, which will cause errors. To avoid this issue, you should set the following environment variable:
+
+```bash
+export no_proxy=localhost
+``` 
+
 After finishing cluster creation, proceed to the next "Update `config.yml`" step.
 
 ### b) Remote cluster
-AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks we have to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
+AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
 
 ### Update `config.yml`
 ```bash
@@ -89,12 +103,30 @@ python3 cli.py
 Run GPT-4 baseline agent:
 
 ```bash
-export OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>
+# Create a .env file in the project root (if not exists)
+echo "OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>" > .env
+# Add more API keys as needed:
+# echo "QWEN_API_KEY=<YOUR_QWEN_API_KEY>" >> .env
+# echo "DEEPSEEK_API_KEY=<YOUR_DEEPSEEK_API_KEY>" >> .env
+
 python3 clients/gpt.py # you can also change the problem to solve in the main() function
 ```
 
+The clients will automatically load API keys from your .env file.
+
 You can check the running status of the cluster using [k9s](https://k9scli.io/) or other cluster monitoring tools conveniently.
 
+To browse your logged `session_id` values in the W&B app as a table:
+
+1. Make sure you have W&B installed and configured.
+2. Set the USE_WANDB environment variable:
+ ```bash
+ # Add to your .env file
+ echo "USE_WANDB=true" >> .env
+ ```
+3. In the W&B web UI, open any run and click Tables → Add Query Panel.
+4. In the key field, type `runs.summary` and click `Run`, then you will see the results displayed in a table format.
+
 <h2 id="⚙️usage">⚙️ Usage</h2>
 
 AIOpsLab can be used in the following ways:

aiopslab/generators/fault/base.py

@@ -59,7 +59,6 @@ def _recover(
  self._invoke_method("recover", fault_type, microservices)
  elif fault_type:
  self._invoke_method("recover", fault_type)
- time.sleep(6)
 
  def _invoke_method(self, action_prefix, *args):
  """helper: injects/recovers faults based on name"""

aiopslab/generators/fault/inject_otel.py

@@ -8,7 +8,7 @@ class OtelFaultInjector(FaultInjector):
  def __init__(self, namespace: str):
  self.namespace = namespace
  self.kubectl = KubeCtl()
- self.configmap_name = f"{namespace}-flagd-config"
+ self.configmap_name = "flagd-config"
 
  def inject_fault(self, feature_flag: str):
  command = (
@@ -39,6 +39,11 @@ def inject_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
+
  print(f"Fault injected: Feature flag '{feature_flag}' set to 'on'.")
 
  def recover_fault(self, feature_flag: str):
@@ -70,6 +75,10 @@ def recover_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
  print(f"Fault recovered: Feature flag '{feature_flag}' set to 'off'.")
 
 

aiopslab/observer/__init__.py

@@ -12,7 +12,8 @@
 root_path = pathlib.Path(__file__).parent
 sys.path.append(root_path)
 # read the configuration file
-monitor_config = full_load(open(root_path / "monitor_config.yaml", "r"))
+with open(root_path / "monitor_config.yaml", "r") as f:
+ monitor_config = full_load(f)
 
 
 # root_config = full_load(open(root_path / "config.yaml", "r"))

aiopslab/observer/trace_api.py

@@ -22,13 +22,18 @@ def __init__(self, namespace: str):
  self.namespace = namespace
  self.stop_event = threading.Event()
 
- # NOTE: it may not be jaeger-out for other apps
- node_port = self.get_nodeport("jaeger", namespace)
- if node_port:
- self.base_url = f"http://localhost:{node_port}"
- else:
- self.base_url = "http://localhost:16686"
+ if self.namespace == "astronomy-shop":
+ # No NodePort in astronomy shop
+ self.base_url = "http://localhost:16686/jaeger/ui"
  self.start_port_forward()
+ else:
+ # Other namespaces may expose a NodePort
+ node_port = self.get_nodeport("jaeger", namespace)
+ if node_port:
+ self.base_url = f"http://localhost:{node_port}"
+ else:
+ self.base_url = "http://localhost:16686"
+ self.start_port_forward()
 
  def get_nodeport(self, service_name, namespace):
  """Fetch the NodePort for the given service."""
@@ -74,19 +79,38 @@ def print_output(self, stream):
  def is_port_in_use(self, port):
  with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
  return s.connect_ex(("127.0.0.1", port)) == 0
+
+ def get_jaeger_pod_name(self):
+ try:
+ result = subprocess.check_output(
+ ["kubectl", "get", "pods", "-n", self.namespace,
+ "-l", "app.kubernetes.io/name=jaeger",
+ "-o", "jsonpath={.items[0].metadata.name}"],
+ text=True
+ )
+ return result.strip()
+ except subprocess.CalledProcessError as e:
+ print("Error getting Jaeger pod name:", e)
+ raise
 
  def start_port_forward(self):
- """Starts kubectl port-forward command to access Jaeger service."""
+ """Starts kubectl port-forward command to access Jaeger service or pod."""
  for attempt in range(3):
  if self.is_port_in_use(16686):
  print(
- f"Port 16686 is already in use. Attempt {attempt + 1} of {3}. Retrying in {3} seconds..."
+ f"Port 16686 is already in use. Attempt {attempt + 1} of 3. Retrying in 3 seconds..."
  )
  time.sleep(3)
  continue
 
- # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
- command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+ # Use pod port-forwarding for astronomy-shop only
+ if self.namespace == "astronomy-shop":
+ pod_name = self.get_jaeger_pod_name()
+ command = f"kubectl port-forward pod/{pod_name} 16686:16686 -n {self.namespace}"
+ else:
+ command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+
+ print("Starting port-forward with command:", command)
  self.port_forward_process = subprocess.Popen(
  command,
  shell=True,
@@ -103,17 +127,17 @@ def start_port_forward(self):
  )
  thread_out.start()
  thread_err.start()
- time.sleep(3) # Wait a bit for the port-forward to establish
 
-if (
- self.port_forward_process.poll() is None
-): # Check if the process is still running
+time.sleep(3) # Let port-forward initialize
+
+ if self.port_forward_process.poll() is None:
  print("Port forwarding established successfully.")
  break
  else:
  print("Port forwarding failed. Retrying...")
  else:
- print("Failed to establish port forwarding after multiple attempts.")
+ print("Failed to establish port forwarding after 3 attempts.")
+
  # TODO: modify this command for other microservices
  # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
  # self.port_forward_process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
@@ -151,13 +175,14 @@ def cleanup(self):
  def get_services(self) -> list:
  """Fetch a list of services from the tracing API."""
  url = f"{self.base_url}/api/services"
- response = requests.get(url)
- if response.status_code == 200:
- data = response.json()
- # print(f"data: {response}")
- return data.get("data", [])
- else:
- print(f"Failed to get services: {response.status_code}")
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
+
+ try:
+ response = requests.get(url, headers=headers)
+ response.raise_for_status()
+ return response.json().get("data", [])
+ except Exception as e:
+ print(f"Failed to get services: {e}")
  return []
 
  def get_traces(
@@ -171,15 +196,14 @@ def get_traces(
  Fetch traces for a specific service between start_time and end_time.
  If limit is not specified, all available traces are fetched.
  """
- # Calculate the lookback in milliseconds.
  lookback = int((datetime.now() - start_time).total_seconds())
-
  url = f"{self.base_url}/api/traces?service={service_name}&lookback={lookback}s"
  if limit is not None:
  url += f"&limit={limit}"
 
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
  try:
- response = requests.get(url)
+ response = requests.get(url, headers=headers)
  response.raise_for_status()
  return response.json().get("data", [])
  except requests.RequestException as e:
@@ -211,9 +235,9 @@ def extract_traces(
  )
  for trace in traces:
  for span in trace["spans"]:
- span[
+ span["serviceName"] = trace["processes"][span["processID"]][
  "serviceName"
- ] = service # Directly associate service name with each span
+ ]
  all_traces.append(trace) # Collect the trace with service name included
  self.cleanup()
  print("Cleanup completed.")
@@ -223,22 +247,20 @@ def extract_traces(
  def process_traces(self, traces) -> pd.DataFrame:
  """Process raw traces data into a structured DataFrame."""
  trace_id_list = []
+ span_id_list = []
  service_name_list = []
  operation_name_list = []
  start_time_list = []
  duration_list = []
  parent_span_list = []
+ error_list = []
+ response_list = []
 
  for trace in traces:
  trace_id = trace["traceID"]
  for span in trace["spans"]:
  trace_id_list.append(trace_id)
- service_name_list.append(
- span["serviceName"]
- ) # Use the correct service name from the span
- operation_name_list.append(span["operationName"])
- start_time_list.append(span["startTime"])
- duration_list.append(span["duration"])
+ span_id_list.append(span["spanID"])
  parent_span = "ROOT"
  if "references" in span:
  for ref in span["references"]:
@@ -247,14 +269,34 @@ def process_traces(self, traces) -> pd.DataFrame:
  break
  parent_span_list.append(parent_span)
 
+ service_name_list.append(
+ span["serviceName"]
+ ) # Use the correct service name from the span
+ operation_name_list.append(span["operationName"])
+ start_time_list.append(span["startTime"])
+ duration_list.append(span["duration"])
+
+ has_error = False
+ response = "Unknown"
+ for tag in span.get("tags", []):
+ if tag["key"] == "error" and tag["value"] == True:
+ has_error = True
+ if tag["key"] == "http.status_code" or tag["key"] == "response_class":
+ response = tag["value"]
+ error_list.append(has_error)
+ response_list.append(response)
+
  df = pd.DataFrame(
  {
  "trace_id": trace_id_list,
+ "span_id": span_id_list,
+ "parent_span": parent_span_list,
  "service_name": service_name_list,
  "operation_name": operation_name_list,
  "start_time": start_time_list,
  "duration": duration_list,
- "parent_span": parent_span_list,
+ "has_error": error_list,
+ "response": response_list,
  }
  )
  return df

.gitignore

@@ -327,6 +327,9 @@ cython_debug/
 # Visual Studio Code
 .vscode/
 
+# Weight & Biases
+wandb/
+
 # Project specific
 cache_dir
 demos

README.md

@@ -28,6 +28,7 @@ Moreover, AIOpsLab provides a built-in benchmark suite with a set of problems to
 
 ### Requirements
 - Python >= 3.11
+- [Helm](https://helm.sh/)
 
 Recommended installation:
 ```bash
@@ -64,10 +65,23 @@ kind create cluster --config kind/kind-config-arm.yaml
 
 If you're running into issues, consider building a Docker image for your machine by following this [README](kind/README.md). Please also open an issue.
 
+### [Tips]
+If you are running AIOpsLab using a proxy, beware of exporting the HTTP proxy as `172.17.0.1`. When creating the kind cluster, all the nodes in the cluster will inherit the proxy setting from the host environment and the Docker container. 
+
+The `172.17.0.1` address is used to communicate with the host machine. For more details, refer to the official guide: [Configure Kind to Use a Proxy](https://kind.sigs.k8s.io/docs/user/quick-start/#configure-kind-to-use-a-proxy).
+
+Additionally, Docker doesn't support SOCKS5 proxy directly. If you're using a SOCKS5 protocol to proxy, you may need to use [Privoxy](https://www.privoxy.org) to forward SOCKS5 to HTTP.
+
+If you're running VLLM and the LLM agent locally, Privoxy will by default proxy `localhost`, which will cause errors. To avoid this issue, you should set the following environment variable:
+
+```bash
+export no_proxy=localhost
+``` 
+
 After finishing cluster creation, proceed to the next "Update `config.yml`" step.
 
 ### b) Remote cluster
-AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks we have to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
+AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
 
 ### Update `config.yml`
 ```bash
@@ -89,12 +103,30 @@ python3 cli.py
 Run GPT-4 baseline agent:
 
 ```bash
-export OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>
+# Create a .env file in the project root (if not exists)
+echo "OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>" > .env
+# Add more API keys as needed:
+# echo "QWEN_API_KEY=<YOUR_QWEN_API_KEY>" >> .env
+# echo "DEEPSEEK_API_KEY=<YOUR_DEEPSEEK_API_KEY>" >> .env
+
 python3 clients/gpt.py # you can also change the problem to solve in the main() function
 ```
 
+The clients will automatically load API keys from your .env file.
+
 You can check the running status of the cluster using [k9s](https://k9scli.io/) or other cluster monitoring tools conveniently.
 
+To browse your logged `session_id` values in the W&B app as a table:
+
+1. Make sure you have W&B installed and configured.
+2. Set the USE_WANDB environment variable:
+ ```bash
+ # Add to your .env file
+ echo "USE_WANDB=true" >> .env
+ ```
+3. In the W&B web UI, open any run and click Tables → Add Query Panel.
+4. In the key field, type `runs.summary` and click `Run`, then you will see the results displayed in a table format.
+
 <h2 id="⚙️usage">⚙️ Usage</h2>
 
 AIOpsLab can be used in the following ways:

aiopslab/generators/fault/base.py

@@ -59,7 +59,6 @@ def _recover(
  self._invoke_method("recover", fault_type, microservices)
  elif fault_type:
  self._invoke_method("recover", fault_type)
- time.sleep(6)
 
  def _invoke_method(self, action_prefix, *args):
  """helper: injects/recovers faults based on name"""

aiopslab/generators/fault/inject_otel.py

@@ -8,7 +8,7 @@ class OtelFaultInjector(FaultInjector):
  def __init__(self, namespace: str):
  self.namespace = namespace
  self.kubectl = KubeCtl()
- self.configmap_name = f"{namespace}-flagd-config"
+ self.configmap_name = "flagd-config"
 
  def inject_fault(self, feature_flag: str):
  command = (
@@ -39,6 +39,11 @@ def inject_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
+
  print(f"Fault injected: Feature flag '{feature_flag}' set to 'on'.")
 
  def recover_fault(self, feature_flag: str):
@@ -70,6 +75,10 @@ def recover_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
  print(f"Fault recovered: Feature flag '{feature_flag}' set to 'off'.")
 
 

aiopslab/observer/__init__.py

@@ -12,7 +12,8 @@
 root_path = pathlib.Path(__file__).parent
 sys.path.append(root_path)
 # read the configuration file
-monitor_config = full_load(open(root_path / "monitor_config.yaml", "r"))
+with open(root_path / "monitor_config.yaml", "r") as f:
+ monitor_config = full_load(f)
 
 
 # root_config = full_load(open(root_path / "config.yaml", "r"))

aiopslab/observer/trace_api.py

@@ -22,13 +22,18 @@ def __init__(self, namespace: str):
  self.namespace = namespace
  self.stop_event = threading.Event()
 
- # NOTE: it may not be jaeger-out for other apps
- node_port = self.get_nodeport("jaeger", namespace)
- if node_port:
- self.base_url = f"http://localhost:{node_port}"
- else:
- self.base_url = "http://localhost:16686"
+ if self.namespace == "astronomy-shop":
+ # No NodePort in astronomy shop
+ self.base_url = "http://localhost:16686/jaeger/ui"
  self.start_port_forward()
+ else:
+ # Other namespaces may expose a NodePort
+ node_port = self.get_nodeport("jaeger", namespace)
+ if node_port:
+ self.base_url = f"http://localhost:{node_port}"
+ else:
+ self.base_url = "http://localhost:16686"
+ self.start_port_forward()
 
  def get_nodeport(self, service_name, namespace):
  """Fetch the NodePort for the given service."""
@@ -74,19 +79,38 @@ def print_output(self, stream):
  def is_port_in_use(self, port):
  with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
  return s.connect_ex(("127.0.0.1", port)) == 0
+
+ def get_jaeger_pod_name(self):
+ try:
+ result = subprocess.check_output(
+ ["kubectl", "get", "pods", "-n", self.namespace,
+ "-l", "app.kubernetes.io/name=jaeger",
+ "-o", "jsonpath={.items[0].metadata.name}"],
+ text=True
+ )
+ return result.strip()
+ except subprocess.CalledProcessError as e:
+ print("Error getting Jaeger pod name:", e)
+ raise
 
  def start_port_forward(self):
- """Starts kubectl port-forward command to access Jaeger service."""
+ """Starts kubectl port-forward command to access Jaeger service or pod."""
  for attempt in range(3):
  if self.is_port_in_use(16686):
  print(
- f"Port 16686 is already in use. Attempt {attempt + 1} of {3}. Retrying in {3} seconds..."
+ f"Port 16686 is already in use. Attempt {attempt + 1} of 3. Retrying in 3 seconds..."
  )
  time.sleep(3)
  continue
 
- # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
- command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+ # Use pod port-forwarding for astronomy-shop only
+ if self.namespace == "astronomy-shop":
+ pod_name = self.get_jaeger_pod_name()
+ command = f"kubectl port-forward pod/{pod_name} 16686:16686 -n {self.namespace}"
+ else:
+ command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+
+ print("Starting port-forward with command:", command)
  self.port_forward_process = subprocess.Popen(
  command,
  shell=True,
@@ -103,17 +127,17 @@ def start_port_forward(self):
  )
  thread_out.start()
  thread_err.start()
- time.sleep(3) # Wait a bit for the port-forward to establish
 
-if (
- self.port_forward_process.poll() is None
-): # Check if the process is still running
+time.sleep(3) # Let port-forward initialize
+
+ if self.port_forward_process.poll() is None:
  print("Port forwarding established successfully.")
  break
  else:
  print("Port forwarding failed. Retrying...")
  else:
- print("Failed to establish port forwarding after multiple attempts.")
+ print("Failed to establish port forwarding after 3 attempts.")
+
  # TODO: modify this command for other microservices
  # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
  # self.port_forward_process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
@@ -151,13 +175,14 @@ def cleanup(self):
  def get_services(self) -> list:
  """Fetch a list of services from the tracing API."""
  url = f"{self.base_url}/api/services"
- response = requests.get(url)
- if response.status_code == 200:
- data = response.json()
- # print(f"data: {response}")
- return data.get("data", [])
- else:
- print(f"Failed to get services: {response.status_code}")
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
+
+ try:
+ response = requests.get(url, headers=headers)
+ response.raise_for_status()
+ return response.json().get("data", [])
+ except Exception as e:
+ print(f"Failed to get services: {e}")
  return []
 
  def get_traces(
@@ -171,15 +196,14 @@ def get_traces(
  Fetch traces for a specific service between start_time and end_time.
  If limit is not specified, all available traces are fetched.
  """
- # Calculate the lookback in milliseconds.
  lookback = int((datetime.now() - start_time).total_seconds())
-
  url = f"{self.base_url}/api/traces?service={service_name}&lookback={lookback}s"
  if limit is not None:
  url += f"&limit={limit}"
 
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
  try:
- response = requests.get(url)
+ response = requests.get(url, headers=headers)
  response.raise_for_status()
  return response.json().get("data", [])
  except requests.RequestException as e:
@@ -211,9 +235,9 @@ def extract_traces(
  )
  for trace in traces:
  for span in trace["spans"]:
- span[
+ span["serviceName"] = trace["processes"][span["processID"]][
  "serviceName"
- ] = service # Directly associate service name with each span
+ ]
  all_traces.append(trace) # Collect the trace with service name included
  self.cleanup()
  print("Cleanup completed.")
@@ -223,22 +247,20 @@ def extract_traces(
  def process_traces(self, traces) -> pd.DataFrame:
  """Process raw traces data into a structured DataFrame."""
  trace_id_list = []
+ span_id_list = []
  service_name_list = []
  operation_name_list = []
  start_time_list = []
  duration_list = []
  parent_span_list = []
+ error_list = []
+ response_list = []
 
  for trace in traces:
  trace_id = trace["traceID"]
  for span in trace["spans"]:
  trace_id_list.append(trace_id)
- service_name_list.append(
- span["serviceName"]
- ) # Use the correct service name from the span
- operation_name_list.append(span["operationName"])
- start_time_list.append(span["startTime"])
- duration_list.append(span["duration"])
+ span_id_list.append(span["spanID"])
  parent_span = "ROOT"
  if "references" in span:
  for ref in span["references"]:
@@ -247,14 +269,34 @@ def process_traces(self, traces) -> pd.DataFrame:
  break
  parent_span_list.append(parent_span)
 
+ service_name_list.append(
+ span["serviceName"]
+ ) # Use the correct service name from the span
+ operation_name_list.append(span["operationName"])
+ start_time_list.append(span["startTime"])
+ duration_list.append(span["duration"])
+
+ has_error = False
+ response = "Unknown"
+ for tag in span.get("tags", []):
+ if tag["key"] == "error" and tag["value"] == True:
+ has_error = True
+ if tag["key"] == "http.status_code" or tag["key"] == "response_class":
+ response = tag["value"]
+ error_list.append(has_error)
+ response_list.append(response)
+
  df = pd.DataFrame(
  {
  "trace_id": trace_id_list,
+ "span_id": span_id_list,
+ "parent_span": parent_span_list,
  "service_name": service_name_list,
  "operation_name": operation_name_list,
  "start_time": start_time_list,
  "duration": duration_list,
- "parent_span": parent_span_list,
+ "has_error": error_list,
+ "response": response_list,
  }
  )
  return df

.gitignore

@@ -327,6 +327,9 @@ cython_debug/
 # Visual Studio Code
 .vscode/
 
+# Weight & Biases
+wandb/
+
 # Project specific
 cache_dir
 demos

README.md

@@ -28,6 +28,7 @@ Moreover, AIOpsLab provides a built-in benchmark suite with a set of problems to
 
 ### Requirements
 - Python >= 3.11
+- [Helm](https://helm.sh/)
 
 Recommended installation:
 ```bash
@@ -64,10 +65,23 @@ kind create cluster --config kind/kind-config-arm.yaml
 
 If you're running into issues, consider building a Docker image for your machine by following this [README](kind/README.md). Please also open an issue.
 
+### [Tips]
+If you are running AIOpsLab using a proxy, beware of exporting the HTTP proxy as `172.17.0.1`. When creating the kind cluster, all the nodes in the cluster will inherit the proxy setting from the host environment and the Docker container. 
+
+The `172.17.0.1` address is used to communicate with the host machine. For more details, refer to the official guide: [Configure Kind to Use a Proxy](https://kind.sigs.k8s.io/docs/user/quick-start/#configure-kind-to-use-a-proxy).
+
+Additionally, Docker doesn't support SOCKS5 proxy directly. If you're using a SOCKS5 protocol to proxy, you may need to use [Privoxy](https://www.privoxy.org) to forward SOCKS5 to HTTP.
+
+If you're running VLLM and the LLM agent locally, Privoxy will by default proxy `localhost`, which will cause errors. To avoid this issue, you should set the following environment variable:
+
+```bash
+export no_proxy=localhost
+``` 
+
 After finishing cluster creation, proceed to the next "Update `config.yml`" step.
 
 ### b) Remote cluster
-AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks we have to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
+AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
 
 ### Update `config.yml`
 ```bash
@@ -89,12 +103,30 @@ python3 cli.py
 Run GPT-4 baseline agent:
 
 ```bash
-export OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>
+# Create a .env file in the project root (if not exists)
+echo "OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>" > .env
+# Add more API keys as needed:
+# echo "QWEN_API_KEY=<YOUR_QWEN_API_KEY>" >> .env
+# echo "DEEPSEEK_API_KEY=<YOUR_DEEPSEEK_API_KEY>" >> .env
+
 python3 clients/gpt.py # you can also change the problem to solve in the main() function
 ```
 
+The clients will automatically load API keys from your .env file.
+
 You can check the running status of the cluster using [k9s](https://k9scli.io/) or other cluster monitoring tools conveniently.
 
+To browse your logged `session_id` values in the W&B app as a table:
+
+1. Make sure you have W&B installed and configured.
+2. Set the USE_WANDB environment variable:
+ ```bash
+ # Add to your .env file
+ echo "USE_WANDB=true" >> .env
+ ```
+3. In the W&B web UI, open any run and click Tables → Add Query Panel.
+4. In the key field, type `runs.summary` and click `Run`, then you will see the results displayed in a table format.
+
 <h2 id="⚙️usage">⚙️ Usage</h2>
 
 AIOpsLab can be used in the following ways:

aiopslab/generators/fault/base.py

@@ -59,7 +59,6 @@ def _recover(
  self._invoke_method("recover", fault_type, microservices)
  elif fault_type:
  self._invoke_method("recover", fault_type)
- time.sleep(6)
 
  def _invoke_method(self, action_prefix, *args):
  """helper: injects/recovers faults based on name"""

aiopslab/generators/fault/inject_otel.py

@@ -8,7 +8,7 @@ class OtelFaultInjector(FaultInjector):
  def __init__(self, namespace: str):
  self.namespace = namespace
  self.kubectl = KubeCtl()
- self.configmap_name = f"{namespace}-flagd-config"
+ self.configmap_name = "flagd-config"
 
  def inject_fault(self, feature_flag: str):
  command = (
@@ -39,6 +39,11 @@ def inject_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
+
  print(f"Fault injected: Feature flag '{feature_flag}' set to 'on'.")
 
  def recover_fault(self, feature_flag: str):
@@ -70,6 +75,10 @@ def recover_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
  print(f"Fault recovered: Feature flag '{feature_flag}' set to 'off'.")
 
 

aiopslab/observer/__init__.py

@@ -12,7 +12,8 @@
 root_path = pathlib.Path(__file__).parent
 sys.path.append(root_path)
 # read the configuration file
-monitor_config = full_load(open(root_path / "monitor_config.yaml", "r"))
+with open(root_path / "monitor_config.yaml", "r") as f:
+ monitor_config = full_load(f)
 
 
 # root_config = full_load(open(root_path / "config.yaml", "r"))

aiopslab/observer/trace_api.py

@@ -22,13 +22,18 @@ def __init__(self, namespace: str):
  self.namespace = namespace
  self.stop_event = threading.Event()
 
- # NOTE: it may not be jaeger-out for other apps
- node_port = self.get_nodeport("jaeger", namespace)
- if node_port:
- self.base_url = f"http://localhost:{node_port}"
- else:
- self.base_url = "http://localhost:16686"
+ if self.namespace == "astronomy-shop":
+ # No NodePort in astronomy shop
+ self.base_url = "http://localhost:16686/jaeger/ui"
  self.start_port_forward()
+ else:
+ # Other namespaces may expose a NodePort
+ node_port = self.get_nodeport("jaeger", namespace)
+ if node_port:
+ self.base_url = f"http://localhost:{node_port}"
+ else:
+ self.base_url = "http://localhost:16686"
+ self.start_port_forward()
 
  def get_nodeport(self, service_name, namespace):
  """Fetch the NodePort for the given service."""
@@ -74,19 +79,38 @@ def print_output(self, stream):
  def is_port_in_use(self, port):
  with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
  return s.connect_ex(("127.0.0.1", port)) == 0
+
+ def get_jaeger_pod_name(self):
+ try:
+ result = subprocess.check_output(
+ ["kubectl", "get", "pods", "-n", self.namespace,
+ "-l", "app.kubernetes.io/name=jaeger",
+ "-o", "jsonpath={.items[0].metadata.name}"],
+ text=True
+ )
+ return result.strip()
+ except subprocess.CalledProcessError as e:
+ print("Error getting Jaeger pod name:", e)
+ raise
 
  def start_port_forward(self):
- """Starts kubectl port-forward command to access Jaeger service."""
+ """Starts kubectl port-forward command to access Jaeger service or pod."""
  for attempt in range(3):
  if self.is_port_in_use(16686):
  print(
- f"Port 16686 is already in use. Attempt {attempt + 1} of {3}. Retrying in {3} seconds..."
+ f"Port 16686 is already in use. Attempt {attempt + 1} of 3. Retrying in 3 seconds..."
  )
  time.sleep(3)
  continue
 
- # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
- command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+ # Use pod port-forwarding for astronomy-shop only
+ if self.namespace == "astronomy-shop":
+ pod_name = self.get_jaeger_pod_name()
+ command = f"kubectl port-forward pod/{pod_name} 16686:16686 -n {self.namespace}"
+ else:
+ command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+
+ print("Starting port-forward with command:", command)
  self.port_forward_process = subprocess.Popen(
  command,
  shell=True,
@@ -103,17 +127,17 @@ def start_port_forward(self):
  )
  thread_out.start()
  thread_err.start()
- time.sleep(3) # Wait a bit for the port-forward to establish
 
-if (
- self.port_forward_process.poll() is None
-): # Check if the process is still running
+time.sleep(3) # Let port-forward initialize
+
+ if self.port_forward_process.poll() is None:
  print("Port forwarding established successfully.")
  break
  else:
  print("Port forwarding failed. Retrying...")
  else:
- print("Failed to establish port forwarding after multiple attempts.")
+ print("Failed to establish port forwarding after 3 attempts.")
+
  # TODO: modify this command for other microservices
  # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
  # self.port_forward_process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
@@ -151,13 +175,14 @@ def cleanup(self):
  def get_services(self) -> list:
  """Fetch a list of services from the tracing API."""
  url = f"{self.base_url}/api/services"
- response = requests.get(url)
- if response.status_code == 200:
- data = response.json()
- # print(f"data: {response}")
- return data.get("data", [])
- else:
- print(f"Failed to get services: {response.status_code}")
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
+
+ try:
+ response = requests.get(url, headers=headers)
+ response.raise_for_status()
+ return response.json().get("data", [])
+ except Exception as e:
+ print(f"Failed to get services: {e}")
  return []
 
  def get_traces(
@@ -171,15 +196,14 @@ def get_traces(
  Fetch traces for a specific service between start_time and end_time.
  If limit is not specified, all available traces are fetched.
  """
- # Calculate the lookback in milliseconds.
  lookback = int((datetime.now() - start_time).total_seconds())
-
  url = f"{self.base_url}/api/traces?service={service_name}&lookback={lookback}s"
  if limit is not None:
  url += f"&limit={limit}"
 
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
  try:
- response = requests.get(url)
+ response = requests.get(url, headers=headers)
  response.raise_for_status()
  return response.json().get("data", [])
  except requests.RequestException as e:
@@ -211,9 +235,9 @@ def extract_traces(
  )
  for trace in traces:
  for span in trace["spans"]:
- span[
+ span["serviceName"] = trace["processes"][span["processID"]][
  "serviceName"
- ] = service # Directly associate service name with each span
+ ]
  all_traces.append(trace) # Collect the trace with service name included
  self.cleanup()
  print("Cleanup completed.")
@@ -223,22 +247,20 @@ def extract_traces(
  def process_traces(self, traces) -> pd.DataFrame:
  """Process raw traces data into a structured DataFrame."""
  trace_id_list = []
+ span_id_list = []
  service_name_list = []
  operation_name_list = []
  start_time_list = []
  duration_list = []
  parent_span_list = []
+ error_list = []
+ response_list = []
 
  for trace in traces:
  trace_id = trace["traceID"]
  for span in trace["spans"]:
  trace_id_list.append(trace_id)
- service_name_list.append(
- span["serviceName"]
- ) # Use the correct service name from the span
- operation_name_list.append(span["operationName"])
- start_time_list.append(span["startTime"])
- duration_list.append(span["duration"])
+ span_id_list.append(span["spanID"])
  parent_span = "ROOT"
  if "references" in span:
  for ref in span["references"]:
@@ -247,14 +269,34 @@ def process_traces(self, traces) -> pd.DataFrame:
  break
  parent_span_list.append(parent_span)
 
+ service_name_list.append(
+ span["serviceName"]
+ ) # Use the correct service name from the span
+ operation_name_list.append(span["operationName"])
+ start_time_list.append(span["startTime"])
+ duration_list.append(span["duration"])
+
+ has_error = False
+ response = "Unknown"
+ for tag in span.get("tags", []):
+ if tag["key"] == "error" and tag["value"] == True:
+ has_error = True
+ if tag["key"] == "http.status_code" or tag["key"] == "response_class":
+ response = tag["value"]
+ error_list.append(has_error)
+ response_list.append(response)
+
  df = pd.DataFrame(
  {
  "trace_id": trace_id_list,
+ "span_id": span_id_list,
+ "parent_span": parent_span_list,
  "service_name": service_name_list,
  "operation_name": operation_name_list,
  "start_time": start_time_list,
  "duration": duration_list,
- "parent_span": parent_span_list,
+ "has_error": error_list,
+ "response": response_list,
  }
  )
  return df

.gitignore

@@ -327,6 +327,9 @@ cython_debug/
 # Visual Studio Code
 .vscode/
 
+# Weight & Biases
+wandb/
+
 # Project specific
 cache_dir
 demos

README.md

@@ -28,6 +28,7 @@ Moreover, AIOpsLab provides a built-in benchmark suite with a set of problems to
 
 ### Requirements
 - Python >= 3.11
+- [Helm](https://helm.sh/)
 
 Recommended installation:
 ```bash
@@ -64,10 +65,23 @@ kind create cluster --config kind/kind-config-arm.yaml
 
 If you're running into issues, consider building a Docker image for your machine by following this [README](kind/README.md). Please also open an issue.
 
+### [Tips]
+If you are running AIOpsLab using a proxy, beware of exporting the HTTP proxy as `172.17.0.1`. When creating the kind cluster, all the nodes in the cluster will inherit the proxy setting from the host environment and the Docker container. 
+
+The `172.17.0.1` address is used to communicate with the host machine. For more details, refer to the official guide: [Configure Kind to Use a Proxy](https://kind.sigs.k8s.io/docs/user/quick-start/#configure-kind-to-use-a-proxy).
+
+Additionally, Docker doesn't support SOCKS5 proxy directly. If you're using a SOCKS5 protocol to proxy, you may need to use [Privoxy](https://www.privoxy.org) to forward SOCKS5 to HTTP.
+
+If you're running VLLM and the LLM agent locally, Privoxy will by default proxy `localhost`, which will cause errors. To avoid this issue, you should set the following environment variable:
+
+```bash
+export no_proxy=localhost
+``` 
+
 After finishing cluster creation, proceed to the next "Update `config.yml`" step.
 
 ### b) Remote cluster
-AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks we have to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
+AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
 
 ### Update `config.yml`
 ```bash
@@ -89,12 +103,30 @@ python3 cli.py
 Run GPT-4 baseline agent:
 
 ```bash
-export OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>
+# Create a .env file in the project root (if not exists)
+echo "OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>" > .env
+# Add more API keys as needed:
+# echo "QWEN_API_KEY=<YOUR_QWEN_API_KEY>" >> .env
+# echo "DEEPSEEK_API_KEY=<YOUR_DEEPSEEK_API_KEY>" >> .env
+
 python3 clients/gpt.py # you can also change the problem to solve in the main() function
 ```
 
+The clients will automatically load API keys from your .env file.
+
 You can check the running status of the cluster using [k9s](https://k9scli.io/) or other cluster monitoring tools conveniently.
 
+To browse your logged `session_id` values in the W&B app as a table:
+
+1. Make sure you have W&B installed and configured.
+2. Set the USE_WANDB environment variable:
+ ```bash
+ # Add to your .env file
+ echo "USE_WANDB=true" >> .env
+ ```
+3. In the W&B web UI, open any run and click Tables → Add Query Panel.
+4. In the key field, type `runs.summary` and click `Run`, then you will see the results displayed in a table format.
+
 <h2 id="⚙️usage">⚙️ Usage</h2>
 
 AIOpsLab can be used in the following ways:

aiopslab/generators/fault/base.py

@@ -59,7 +59,6 @@ def _recover(
  self._invoke_method("recover", fault_type, microservices)
  elif fault_type:
  self._invoke_method("recover", fault_type)
- time.sleep(6)
 
  def _invoke_method(self, action_prefix, *args):
  """helper: injects/recovers faults based on name"""

aiopslab/generators/fault/inject_otel.py

@@ -8,7 +8,7 @@ class OtelFaultInjector(FaultInjector):
  def __init__(self, namespace: str):
  self.namespace = namespace
  self.kubectl = KubeCtl()
- self.configmap_name = f"{namespace}-flagd-config"
+ self.configmap_name = "flagd-config"
 
  def inject_fault(self, feature_flag: str):
  command = (
@@ -39,6 +39,11 @@ def inject_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
+
  print(f"Fault injected: Feature flag '{feature_flag}' set to 'on'.")
 
  def recover_fault(self, feature_flag: str):
@@ -70,6 +75,10 @@ def recover_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
  print(f"Fault recovered: Feature flag '{feature_flag}' set to 'off'.")
 
 

aiopslab/observer/__init__.py

@@ -12,7 +12,8 @@
 root_path = pathlib.Path(__file__).parent
 sys.path.append(root_path)
 # read the configuration file
-monitor_config = full_load(open(root_path / "monitor_config.yaml", "r"))
+with open(root_path / "monitor_config.yaml", "r") as f:
+ monitor_config = full_load(f)
 
 
 # root_config = full_load(open(root_path / "config.yaml", "r"))

aiopslab/observer/trace_api.py

@@ -22,13 +22,18 @@ def __init__(self, namespace: str):
  self.namespace = namespace
  self.stop_event = threading.Event()
 
- # NOTE: it may not be jaeger-out for other apps
- node_port = self.get_nodeport("jaeger", namespace)
- if node_port:
- self.base_url = f"http://localhost:{node_port}"
- else:
- self.base_url = "http://localhost:16686"
+ if self.namespace == "astronomy-shop":
+ # No NodePort in astronomy shop
+ self.base_url = "http://localhost:16686/jaeger/ui"
  self.start_port_forward()
+ else:
+ # Other namespaces may expose a NodePort
+ node_port = self.get_nodeport("jaeger", namespace)
+ if node_port:
+ self.base_url = f"http://localhost:{node_port}"
+ else:
+ self.base_url = "http://localhost:16686"
+ self.start_port_forward()
 
  def get_nodeport(self, service_name, namespace):
  """Fetch the NodePort for the given service."""
@@ -74,19 +79,38 @@ def print_output(self, stream):
  def is_port_in_use(self, port):
  with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
  return s.connect_ex(("127.0.0.1", port)) == 0
+
+ def get_jaeger_pod_name(self):
+ try:
+ result = subprocess.check_output(
+ ["kubectl", "get", "pods", "-n", self.namespace,
+ "-l", "app.kubernetes.io/name=jaeger",
+ "-o", "jsonpath={.items[0].metadata.name}"],
+ text=True
+ )
+ return result.strip()
+ except subprocess.CalledProcessError as e:
+ print("Error getting Jaeger pod name:", e)
+ raise
 
  def start_port_forward(self):
- """Starts kubectl port-forward command to access Jaeger service."""
+ """Starts kubectl port-forward command to access Jaeger service or pod."""
  for attempt in range(3):
  if self.is_port_in_use(16686):
  print(
- f"Port 16686 is already in use. Attempt {attempt + 1} of {3}. Retrying in {3} seconds..."
+ f"Port 16686 is already in use. Attempt {attempt + 1} of 3. Retrying in 3 seconds..."
  )
  time.sleep(3)
  continue
 
- # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
- command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+ # Use pod port-forwarding for astronomy-shop only
+ if self.namespace == "astronomy-shop":
+ pod_name = self.get_jaeger_pod_name()
+ command = f"kubectl port-forward pod/{pod_name} 16686:16686 -n {self.namespace}"
+ else:
+ command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+
+ print("Starting port-forward with command:", command)
  self.port_forward_process = subprocess.Popen(
  command,
  shell=True,
@@ -103,17 +127,17 @@ def start_port_forward(self):
  )
  thread_out.start()
  thread_err.start()
- time.sleep(3) # Wait a bit for the port-forward to establish
 
-if (
- self.port_forward_process.poll() is None
-): # Check if the process is still running
+time.sleep(3) # Let port-forward initialize
+
+ if self.port_forward_process.poll() is None:
  print("Port forwarding established successfully.")
  break
  else:
  print("Port forwarding failed. Retrying...")
  else:
- print("Failed to establish port forwarding after multiple attempts.")
+ print("Failed to establish port forwarding after 3 attempts.")
+
  # TODO: modify this command for other microservices
  # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
  # self.port_forward_process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
@@ -151,13 +175,14 @@ def cleanup(self):
  def get_services(self) -> list:
  """Fetch a list of services from the tracing API."""
  url = f"{self.base_url}/api/services"
- response = requests.get(url)
- if response.status_code == 200:
- data = response.json()
- # print(f"data: {response}")
- return data.get("data", [])
- else:
- print(f"Failed to get services: {response.status_code}")
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
+
+ try:
+ response = requests.get(url, headers=headers)
+ response.raise_for_status()
+ return response.json().get("data", [])
+ except Exception as e:
+ print(f"Failed to get services: {e}")
  return []
 
  def get_traces(
@@ -171,15 +196,14 @@ def get_traces(
  Fetch traces for a specific service between start_time and end_time.
  If limit is not specified, all available traces are fetched.
  """
- # Calculate the lookback in milliseconds.
  lookback = int((datetime.now() - start_time).total_seconds())
-
  url = f"{self.base_url}/api/traces?service={service_name}&lookback={lookback}s"
  if limit is not None:
  url += f"&limit={limit}"
 
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
  try:
- response = requests.get(url)
+ response = requests.get(url, headers=headers)
  response.raise_for_status()
  return response.json().get("data", [])
  except requests.RequestException as e:
@@ -211,9 +235,9 @@ def extract_traces(
  )
  for trace in traces:
  for span in trace["spans"]:
- span[
+ span["serviceName"] = trace["processes"][span["processID"]][
  "serviceName"
- ] = service # Directly associate service name with each span
+ ]
  all_traces.append(trace) # Collect the trace with service name included
  self.cleanup()
  print("Cleanup completed.")
@@ -223,22 +247,20 @@ def extract_traces(
  def process_traces(self, traces) -> pd.DataFrame:
  """Process raw traces data into a structured DataFrame."""
  trace_id_list = []
+ span_id_list = []
  service_name_list = []
  operation_name_list = []
  start_time_list = []
  duration_list = []
  parent_span_list = []
+ error_list = []
+ response_list = []
 
  for trace in traces:
  trace_id = trace["traceID"]
  for span in trace["spans"]:
  trace_id_list.append(trace_id)
- service_name_list.append(
- span["serviceName"]
- ) # Use the correct service name from the span
- operation_name_list.append(span["operationName"])
- start_time_list.append(span["startTime"])
- duration_list.append(span["duration"])
+ span_id_list.append(span["spanID"])
  parent_span = "ROOT"
  if "references" in span:
  for ref in span["references"]:
@@ -247,14 +269,34 @@ def process_traces(self, traces) -> pd.DataFrame:
  break
  parent_span_list.append(parent_span)
 
+ service_name_list.append(
+ span["serviceName"]
+ ) # Use the correct service name from the span
+ operation_name_list.append(span["operationName"])
+ start_time_list.append(span["startTime"])
+ duration_list.append(span["duration"])
+
+ has_error = False
+ response = "Unknown"
+ for tag in span.get("tags", []):
+ if tag["key"] == "error" and tag["value"] == True:
+ has_error = True
+ if tag["key"] == "http.status_code" or tag["key"] == "response_class":
+ response = tag["value"]
+ error_list.append(has_error)
+ response_list.append(response)
+
  df = pd.DataFrame(
  {
  "trace_id": trace_id_list,
+ "span_id": span_id_list,
+ "parent_span": parent_span_list,
  "service_name": service_name_list,
  "operation_name": operation_name_list,
  "start_time": start_time_list,
  "duration": duration_list,
- "parent_span": parent_span_list,
+ "has_error": error_list,
+ "response": response_list,
  }
  )
  return df

.gitignore

@@ -327,6 +327,9 @@ cython_debug/
 # Visual Studio Code
 .vscode/
 
+# Weight & Biases
+wandb/
+
 # Project specific
 cache_dir
 demos

README.md

@@ -28,6 +28,7 @@ Moreover, AIOpsLab provides a built-in benchmark suite with a set of problems to
 
 ### Requirements
 - Python >= 3.11
+- [Helm](https://helm.sh/)
 
 Recommended installation:
 ```bash
@@ -64,10 +65,23 @@ kind create cluster --config kind/kind-config-arm.yaml
 
 If you're running into issues, consider building a Docker image for your machine by following this [README](kind/README.md). Please also open an issue.
 
+### [Tips]
+If you are running AIOpsLab using a proxy, beware of exporting the HTTP proxy as `172.17.0.1`. When creating the kind cluster, all the nodes in the cluster will inherit the proxy setting from the host environment and the Docker container. 
+
+The `172.17.0.1` address is used to communicate with the host machine. For more details, refer to the official guide: [Configure Kind to Use a Proxy](https://kind.sigs.k8s.io/docs/user/quick-start/#configure-kind-to-use-a-proxy).
+
+Additionally, Docker doesn't support SOCKS5 proxy directly. If you're using a SOCKS5 protocol to proxy, you may need to use [Privoxy](https://www.privoxy.org) to forward SOCKS5 to HTTP.
+
+If you're running VLLM and the LLM agent locally, Privoxy will by default proxy `localhost`, which will cause errors. To avoid this issue, you should set the following environment variable:
+
+```bash
+export no_proxy=localhost
+``` 
+
 After finishing cluster creation, proceed to the next "Update `config.yml`" step.
 
 ### b) Remote cluster
-AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks we have to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
+AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
 
 ### Update `config.yml`
 ```bash
@@ -89,12 +103,30 @@ python3 cli.py
 Run GPT-4 baseline agent:
 
 ```bash
-export OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>
+# Create a .env file in the project root (if not exists)
+echo "OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>" > .env
+# Add more API keys as needed:
+# echo "QWEN_API_KEY=<YOUR_QWEN_API_KEY>" >> .env
+# echo "DEEPSEEK_API_KEY=<YOUR_DEEPSEEK_API_KEY>" >> .env
+
 python3 clients/gpt.py # you can also change the problem to solve in the main() function
 ```
 
+The clients will automatically load API keys from your .env file.
+
 You can check the running status of the cluster using [k9s](https://k9scli.io/) or other cluster monitoring tools conveniently.
 
+To browse your logged `session_id` values in the W&B app as a table:
+
+1. Make sure you have W&B installed and configured.
+2. Set the USE_WANDB environment variable:
+ ```bash
+ # Add to your .env file
+ echo "USE_WANDB=true" >> .env
+ ```
+3. In the W&B web UI, open any run and click Tables → Add Query Panel.
+4. In the key field, type `runs.summary` and click `Run`, then you will see the results displayed in a table format.
+
 <h2 id="⚙️usage">⚙️ Usage</h2>
 
 AIOpsLab can be used in the following ways:

aiopslab/generators/fault/base.py

@@ -59,7 +59,6 @@ def _recover(
  self._invoke_method("recover", fault_type, microservices)
  elif fault_type:
  self._invoke_method("recover", fault_type)
- time.sleep(6)
 
  def _invoke_method(self, action_prefix, *args):
  """helper: injects/recovers faults based on name"""

aiopslab/generators/fault/inject_otel.py

@@ -8,7 +8,7 @@ class OtelFaultInjector(FaultInjector):
  def __init__(self, namespace: str):
  self.namespace = namespace
  self.kubectl = KubeCtl()
- self.configmap_name = f"{namespace}-flagd-config"
+ self.configmap_name = "flagd-config"
 
  def inject_fault(self, feature_flag: str):
  command = (
@@ -39,6 +39,11 @@ def inject_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
+
  print(f"Fault injected: Feature flag '{feature_flag}' set to 'on'.")
 
  def recover_fault(self, feature_flag: str):
@@ -70,6 +75,10 @@ def recover_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
  print(f"Fault recovered: Feature flag '{feature_flag}' set to 'off'.")
 
 

aiopslab/observer/__init__.py

@@ -12,7 +12,8 @@
 root_path = pathlib.Path(__file__).parent
 sys.path.append(root_path)
 # read the configuration file
-monitor_config = full_load(open(root_path / "monitor_config.yaml", "r"))
+with open(root_path / "monitor_config.yaml", "r") as f:
+ monitor_config = full_load(f)
 
 
 # root_config = full_load(open(root_path / "config.yaml", "r"))

aiopslab/observer/trace_api.py

@@ -22,13 +22,18 @@ def __init__(self, namespace: str):
  self.namespace = namespace
  self.stop_event = threading.Event()
 
- # NOTE: it may not be jaeger-out for other apps
- node_port = self.get_nodeport("jaeger", namespace)
- if node_port:
- self.base_url = f"http://localhost:{node_port}"
- else:
- self.base_url = "http://localhost:16686"
+ if self.namespace == "astronomy-shop":
+ # No NodePort in astronomy shop
+ self.base_url = "http://localhost:16686/jaeger/ui"
  self.start_port_forward()
+ else:
+ # Other namespaces may expose a NodePort
+ node_port = self.get_nodeport("jaeger", namespace)
+ if node_port:
+ self.base_url = f"http://localhost:{node_port}"
+ else:
+ self.base_url = "http://localhost:16686"
+ self.start_port_forward()
 
  def get_nodeport(self, service_name, namespace):
  """Fetch the NodePort for the given service."""
@@ -74,19 +79,38 @@ def print_output(self, stream):
  def is_port_in_use(self, port):
  with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
  return s.connect_ex(("127.0.0.1", port)) == 0
+
+ def get_jaeger_pod_name(self):
+ try:
+ result = subprocess.check_output(
+ ["kubectl", "get", "pods", "-n", self.namespace,
+ "-l", "app.kubernetes.io/name=jaeger",
+ "-o", "jsonpath={.items[0].metadata.name}"],
+ text=True
+ )
+ return result.strip()
+ except subprocess.CalledProcessError as e:
+ print("Error getting Jaeger pod name:", e)
+ raise
 
  def start_port_forward(self):
- """Starts kubectl port-forward command to access Jaeger service."""
+ """Starts kubectl port-forward command to access Jaeger service or pod."""
  for attempt in range(3):
  if self.is_port_in_use(16686):
  print(
- f"Port 16686 is already in use. Attempt {attempt + 1} of {3}. Retrying in {3} seconds..."
+ f"Port 16686 is already in use. Attempt {attempt + 1} of 3. Retrying in 3 seconds..."
  )
  time.sleep(3)
  continue
 
- # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
- command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+ # Use pod port-forwarding for astronomy-shop only
+ if self.namespace == "astronomy-shop":
+ pod_name = self.get_jaeger_pod_name()
+ command = f"kubectl port-forward pod/{pod_name} 16686:16686 -n {self.namespace}"
+ else:
+ command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+
+ print("Starting port-forward with command:", command)
  self.port_forward_process = subprocess.Popen(
  command,
  shell=True,
@@ -103,17 +127,17 @@ def start_port_forward(self):
  )
  thread_out.start()
  thread_err.start()
- time.sleep(3) # Wait a bit for the port-forward to establish
 
-if (
- self.port_forward_process.poll() is None
-): # Check if the process is still running
+time.sleep(3) # Let port-forward initialize
+
+ if self.port_forward_process.poll() is None:
  print("Port forwarding established successfully.")
  break
  else:
  print("Port forwarding failed. Retrying...")
  else:
- print("Failed to establish port forwarding after multiple attempts.")
+ print("Failed to establish port forwarding after 3 attempts.")
+
  # TODO: modify this command for other microservices
  # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
  # self.port_forward_process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
@@ -151,13 +175,14 @@ def cleanup(self):
  def get_services(self) -> list:
  """Fetch a list of services from the tracing API."""
  url = f"{self.base_url}/api/services"
- response = requests.get(url)
- if response.status_code == 200:
- data = response.json()
- # print(f"data: {response}")
- return data.get("data", [])
- else:
- print(f"Failed to get services: {response.status_code}")
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
+
+ try:
+ response = requests.get(url, headers=headers)
+ response.raise_for_status()
+ return response.json().get("data", [])
+ except Exception as e:
+ print(f"Failed to get services: {e}")
  return []
 
  def get_traces(
@@ -171,15 +196,14 @@ def get_traces(
  Fetch traces for a specific service between start_time and end_time.
  If limit is not specified, all available traces are fetched.
  """
- # Calculate the lookback in milliseconds.
  lookback = int((datetime.now() - start_time).total_seconds())
-
  url = f"{self.base_url}/api/traces?service={service_name}&lookback={lookback}s"
  if limit is not None:
  url += f"&limit={limit}"
 
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
  try:
- response = requests.get(url)
+ response = requests.get(url, headers=headers)
  response.raise_for_status()
  return response.json().get("data", [])
  except requests.RequestException as e:
@@ -211,9 +235,9 @@ def extract_traces(
  )
  for trace in traces:
  for span in trace["spans"]:
- span[
+ span["serviceName"] = trace["processes"][span["processID"]][
  "serviceName"
- ] = service # Directly associate service name with each span
+ ]
  all_traces.append(trace) # Collect the trace with service name included
  self.cleanup()
  print("Cleanup completed.")
@@ -223,22 +247,20 @@ def extract_traces(
  def process_traces(self, traces) -> pd.DataFrame:
  """Process raw traces data into a structured DataFrame."""
  trace_id_list = []
+ span_id_list = []
  service_name_list = []
  operation_name_list = []
  start_time_list = []
  duration_list = []
  parent_span_list = []
+ error_list = []
+ response_list = []
 
  for trace in traces:
  trace_id = trace["traceID"]
  for span in trace["spans"]:
  trace_id_list.append(trace_id)
- service_name_list.append(
- span["serviceName"]
- ) # Use the correct service name from the span
- operation_name_list.append(span["operationName"])
- start_time_list.append(span["startTime"])
- duration_list.append(span["duration"])
+ span_id_list.append(span["spanID"])
  parent_span = "ROOT"
  if "references" in span:
  for ref in span["references"]:
@@ -247,14 +269,34 @@ def process_traces(self, traces) -> pd.DataFrame:
  break
  parent_span_list.append(parent_span)
 
+ service_name_list.append(
+ span["serviceName"]
+ ) # Use the correct service name from the span
+ operation_name_list.append(span["operationName"])
+ start_time_list.append(span["startTime"])
+ duration_list.append(span["duration"])
+
+ has_error = False
+ response = "Unknown"
+ for tag in span.get("tags", []):
+ if tag["key"] == "error" and tag["value"] == True:
+ has_error = True
+ if tag["key"] == "http.status_code" or tag["key"] == "response_class":
+ response = tag["value"]
+ error_list.append(has_error)
+ response_list.append(response)
+
  df = pd.DataFrame(
  {
  "trace_id": trace_id_list,
+ "span_id": span_id_list,
+ "parent_span": parent_span_list,
  "service_name": service_name_list,
  "operation_name": operation_name_list,
  "start_time": start_time_list,
  "duration": duration_list,
- "parent_span": parent_span_list,
+ "has_error": error_list,
+ "response": response_list,
  }
  )
  return df

.gitignore

@@ -327,6 +327,9 @@ cython_debug/
 # Visual Studio Code
 .vscode/
 
+# Weight & Biases
+wandb/
+
 # Project specific
 cache_dir
 demos

README.md

@@ -28,6 +28,7 @@ Moreover, AIOpsLab provides a built-in benchmark suite with a set of problems to
 
 ### Requirements
 - Python >= 3.11
+- [Helm](https://helm.sh/)
 
 Recommended installation:
 ```bash
@@ -64,10 +65,23 @@ kind create cluster --config kind/kind-config-arm.yaml
 
 If you're running into issues, consider building a Docker image for your machine by following this [README](kind/README.md). Please also open an issue.
 
+### [Tips]
+If you are running AIOpsLab using a proxy, beware of exporting the HTTP proxy as `172.17.0.1`. When creating the kind cluster, all the nodes in the cluster will inherit the proxy setting from the host environment and the Docker container. 
+
+The `172.17.0.1` address is used to communicate with the host machine. For more details, refer to the official guide: [Configure Kind to Use a Proxy](https://kind.sigs.k8s.io/docs/user/quick-start/#configure-kind-to-use-a-proxy).
+
+Additionally, Docker doesn't support SOCKS5 proxy directly. If you're using a SOCKS5 protocol to proxy, you may need to use [Privoxy](https://www.privoxy.org) to forward SOCKS5 to HTTP.
+
+If you're running VLLM and the LLM agent locally, Privoxy will by default proxy `localhost`, which will cause errors. To avoid this issue, you should set the following environment variable:
+
+```bash
+export no_proxy=localhost
+``` 
+
 After finishing cluster creation, proceed to the next "Update `config.yml`" step.
 
 ### b) Remote cluster
-AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks we have to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
+AIOpsLab supports any remote kubernetes cluster that your `kubectl` context is set to, whether it's a cluster from a cloud provider or one you build yourself. We have some Ansible playbooks to setup clusters on providers like [CloudLab](https://www.cloudlab.us/) and our own machines. Follow this [README](./scripts/ansible/README.md) to set up your own cluster, and then proceed to the next "Update `config.yml`" step.
 
 ### Update `config.yml`
 ```bash
@@ -89,12 +103,30 @@ python3 cli.py
 Run GPT-4 baseline agent:
 
 ```bash
-export OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>
+# Create a .env file in the project root (if not exists)
+echo "OPENAI_API_KEY=<YOUR_OPENAI_API_KEY>" > .env
+# Add more API keys as needed:
+# echo "QWEN_API_KEY=<YOUR_QWEN_API_KEY>" >> .env
+# echo "DEEPSEEK_API_KEY=<YOUR_DEEPSEEK_API_KEY>" >> .env
+
 python3 clients/gpt.py # you can also change the problem to solve in the main() function
 ```
 
+The clients will automatically load API keys from your .env file.
+
 You can check the running status of the cluster using [k9s](https://k9scli.io/) or other cluster monitoring tools conveniently.
 
+To browse your logged `session_id` values in the W&B app as a table:
+
+1. Make sure you have W&B installed and configured.
+2. Set the USE_WANDB environment variable:
+ ```bash
+ # Add to your .env file
+ echo "USE_WANDB=true" >> .env
+ ```
+3. In the W&B web UI, open any run and click Tables → Add Query Panel.
+4. In the key field, type `runs.summary` and click `Run`, then you will see the results displayed in a table format.
+
 <h2 id="⚙️usage">⚙️ Usage</h2>
 
 AIOpsLab can be used in the following ways:

aiopslab/generators/fault/base.py

@@ -59,7 +59,6 @@ def _recover(
  self._invoke_method("recover", fault_type, microservices)
  elif fault_type:
  self._invoke_method("recover", fault_type)
- time.sleep(6)
 
  def _invoke_method(self, action_prefix, *args):
  """helper: injects/recovers faults based on name"""

aiopslab/generators/fault/inject_otel.py

@@ -8,7 +8,7 @@ class OtelFaultInjector(FaultInjector):
  def __init__(self, namespace: str):
  self.namespace = namespace
  self.kubectl = KubeCtl()
- self.configmap_name = f"{namespace}-flagd-config"
+ self.configmap_name = "flagd-config"
 
  def inject_fault(self, feature_flag: str):
  command = (
@@ -39,6 +39,11 @@ def inject_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
+
  print(f"Fault injected: Feature flag '{feature_flag}' set to 'on'.")
 
  def recover_fault(self, feature_flag: str):
@@ -70,6 +75,10 @@ def recover_fault(self, feature_flag: str):
  self.kubectl.create_or_update_configmap(
  self.configmap_name, self.namespace, updated_data
  )
+
+ self.kubectl.exec_command(
+ f"kubectl rollout restart deployment flagd -n {self.namespace}"
+ )
  print(f"Fault recovered: Feature flag '{feature_flag}' set to 'off'.")
 
 

aiopslab/observer/__init__.py

@@ -12,7 +12,8 @@
 root_path = pathlib.Path(__file__).parent
 sys.path.append(root_path)
 # read the configuration file
-monitor_config = full_load(open(root_path / "monitor_config.yaml", "r"))
+with open(root_path / "monitor_config.yaml", "r") as f:
+ monitor_config = full_load(f)
 
 
 # root_config = full_load(open(root_path / "config.yaml", "r"))

aiopslab/observer/trace_api.py

@@ -22,13 +22,18 @@ def __init__(self, namespace: str):
  self.namespace = namespace
  self.stop_event = threading.Event()
 
- # NOTE: it may not be jaeger-out for other apps
- node_port = self.get_nodeport("jaeger", namespace)
- if node_port:
- self.base_url = f"http://localhost:{node_port}"
- else:
- self.base_url = "http://localhost:16686"
+ if self.namespace == "astronomy-shop":
+ # No NodePort in astronomy shop
+ self.base_url = "http://localhost:16686/jaeger/ui"
  self.start_port_forward()
+ else:
+ # Other namespaces may expose a NodePort
+ node_port = self.get_nodeport("jaeger", namespace)
+ if node_port:
+ self.base_url = f"http://localhost:{node_port}"
+ else:
+ self.base_url = "http://localhost:16686"
+ self.start_port_forward()
 
  def get_nodeport(self, service_name, namespace):
  """Fetch the NodePort for the given service."""
@@ -74,19 +79,38 @@ def print_output(self, stream):
  def is_port_in_use(self, port):
  with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
  return s.connect_ex(("127.0.0.1", port)) == 0
+
+ def get_jaeger_pod_name(self):
+ try:
+ result = subprocess.check_output(
+ ["kubectl", "get", "pods", "-n", self.namespace,
+ "-l", "app.kubernetes.io/name=jaeger",
+ "-o", "jsonpath={.items[0].metadata.name}"],
+ text=True
+ )
+ return result.strip()
+ except subprocess.CalledProcessError as e:
+ print("Error getting Jaeger pod name:", e)
+ raise
 
  def start_port_forward(self):
- """Starts kubectl port-forward command to access Jaeger service."""
+ """Starts kubectl port-forward command to access Jaeger service or pod."""
  for attempt in range(3):
  if self.is_port_in_use(16686):
  print(
- f"Port 16686 is already in use. Attempt {attempt + 1} of {3}. Retrying in {3} seconds..."
+ f"Port 16686 is already in use. Attempt {attempt + 1} of 3. Retrying in 3 seconds..."
  )
  time.sleep(3)
  continue
 
- # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
- command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+ # Use pod port-forwarding for astronomy-shop only
+ if self.namespace == "astronomy-shop":
+ pod_name = self.get_jaeger_pod_name()
+ command = f"kubectl port-forward pod/{pod_name} 16686:16686 -n {self.namespace}"
+ else:
+ command = f"kubectl port-forward svc/jaeger 16686:16686 -n {self.namespace}"
+
+ print("Starting port-forward with command:", command)
  self.port_forward_process = subprocess.Popen(
  command,
  shell=True,
@@ -103,17 +127,17 @@ def start_port_forward(self):
  )
  thread_out.start()
  thread_err.start()
- time.sleep(3) # Wait a bit for the port-forward to establish
 
-if (
- self.port_forward_process.poll() is None
-): # Check if the process is still running
+time.sleep(3) # Let port-forward initialize
+
+ if self.port_forward_process.poll() is None:
  print("Port forwarding established successfully.")
  break
  else:
  print("Port forwarding failed. Retrying...")
  else:
- print("Failed to establish port forwarding after multiple attempts.")
+ print("Failed to establish port forwarding after 3 attempts.")
+
  # TODO: modify this command for other microservices
  # command = "kubectl port-forward svc/jaeger 16686:16686 -n hotel-reservation"
  # self.port_forward_process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
@@ -151,13 +175,14 @@ def cleanup(self):
  def get_services(self) -> list:
  """Fetch a list of services from the tracing API."""
  url = f"{self.base_url}/api/services"
- response = requests.get(url)
- if response.status_code == 200:
- data = response.json()
- # print(f"data: {response}")
- return data.get("data", [])
- else:
- print(f"Failed to get services: {response.status_code}")
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
+
+ try:
+ response = requests.get(url, headers=headers)
+ response.raise_for_status()
+ return response.json().get("data", [])
+ except Exception as e:
+ print(f"Failed to get services: {e}")
  return []
 
  def get_traces(
@@ -171,15 +196,14 @@ def get_traces(
  Fetch traces for a specific service between start_time and end_time.
  If limit is not specified, all available traces are fetched.
  """
- # Calculate the lookback in milliseconds.
  lookback = int((datetime.now() - start_time).total_seconds())
-
  url = f"{self.base_url}/api/traces?service={service_name}&lookback={lookback}s"
  if limit is not None:
  url += f"&limit={limit}"
 
+ headers = {"Accept": "application/json"} if self.namespace == "astronomy-shop" else {}
  try:
- response = requests.get(url)
+ response = requests.get(url, headers=headers)
  response.raise_for_status()
  return response.json().get("data", [])
  except requests.RequestException as e:
@@ -211,9 +235,9 @@ def extract_traces(
  )
  for trace in traces:
  for span in trace["spans"]:
- span[
+ span["serviceName"] = trace["processes"][span["processID"]][
  "serviceName"
- ] = service # Directly associate service name with each span
+ ]
  all_traces.append(trace) # Collect the trace with service name included
  self.cleanup()
  print("Cleanup completed.")
@@ -223,22 +247,20 @@ def extract_traces(
  def process_traces(self, traces) -> pd.DataFrame:
  """Process raw traces data into a structured DataFrame."""
  trace_id_list = []
+ span_id_list = []
  service_name_list = []
  operation_name_list = []
  start_time_list = []
  duration_list = []
  parent_span_list = []
+ error_list = []
+ response_list = []
 
  for trace in traces:
  trace_id = trace["traceID"]
  for span in trace["spans"]:
  trace_id_list.append(trace_id)
- service_name_list.append(
- span["serviceName"]
- ) # Use the correct service name from the span
- operation_name_list.append(span["operationName"])
- start_time_list.append(span["startTime"])
- duration_list.append(span["duration"])
+ span_id_list.append(span["spanID"])
  parent_span = "ROOT"
  if "references" in span:
  for ref in span["references"]:
@@ -247,14 +269,34 @@ def process_traces(self, traces) -> pd.DataFrame:
  break
  parent_span_list.append(parent_span)
 
+ service_name_list.append(
+ span["serviceName"]
+ ) # Use the correct service name from the span
+ operation_name_list.append(span["operationName"])
+ start_time_list.append(span["startTime"])
+ duration_list.append(span["duration"])
+
+ has_error = False
+ response = "Unknown"
+ for tag in span.get("tags", []):
+ if tag["key"] == "error" and tag["value"] == True:
+ has_error = True
+ if tag["key"] == "http.status_code" or tag["key"] == "response_class":
+ response = tag["value"]
+ error_list.append(has_error)
+ response_list.append(response)
+
  df = pd.DataFrame(
  {
  "trace_id": trace_id_list,
+ "span_id": span_id_list,
+ "parent_span": parent_span_list,
  "service_name": service_name_list,
  "operation_name": operation_name_list,
  "start_time": start_time_list,
  "duration": duration_list,
- "parent_span": parent_span_list,
+ "has_error": error_list,
+ "response": response_list,
  }
  )
  return df