6a2db542c3
- GrafanaTab/PrometheusTab: use store.get_server(alias) to get server dict instead of passing (alias, store) to client constructors - Fix PrometheusTab: targets() and alerts() method names + parse response - Fix GrafanaTab: build dashboard URL from client.base_url Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
278 lines
11 KiB
Python
278 lines
11 KiB
Python
"""
|
|
Prometheus tab — PromQL query executor, targets overview, and alerts.
|
|
"""
|
|
|
|
import threading
|
|
from tkinter import ttk
|
|
|
|
import customtkinter as ctk
|
|
from core.prometheus_client import PrometheusClient
|
|
from core.i18n import t
|
|
from core.icons import icon_text, make_icon_button, reconfigure_icon_button
|
|
from gui.tabs.query_tab import apply_dark_scrollbar_style
|
|
|
|
|
|
class PrometheusTab(ctk.CTkFrame):
|
|
def __init__(self, master, store):
|
|
super().__init__(master, fg_color="transparent")
|
|
self.store = store
|
|
self._current_alias: str | None = None
|
|
self._client: PrometheusClient | None = None
|
|
|
|
self._build_ui()
|
|
|
|
def _build_ui(self):
|
|
apply_dark_scrollbar_style()
|
|
# ── PromQL query section ──
|
|
query_frame = ctk.CTkFrame(self, fg_color="transparent")
|
|
query_frame.pack(fill="x", padx=15, pady=(15, 5))
|
|
|
|
query_label = ctk.CTkLabel(query_frame, text=t("prom_query"),
|
|
font=ctk.CTkFont(size=14, weight="bold"), anchor="w")
|
|
query_label.pack(side="left", padx=(0, 10))
|
|
|
|
self._query_entry = ctk.CTkEntry(query_frame,
|
|
placeholder_text=t("prom_query_placeholder"),
|
|
font=ctk.CTkFont(family="Consolas", size=13))
|
|
self._query_entry.pack(side="left", fill="x", expand=True, padx=(0, 10))
|
|
self._query_entry.bind("<Return>", lambda e: self._execute_query())
|
|
|
|
self._exec_btn = make_icon_button(query_frame, "execute", t("prom_execute"), width=100,
|
|
command=self._execute_query)
|
|
self._exec_btn.pack(side="left")
|
|
|
|
# ── Query results ──
|
|
results_label = ctk.CTkLabel(self, text=t("prom_results"),
|
|
font=ctk.CTkFont(size=12, weight="bold"), anchor="w")
|
|
results_label.pack(fill="x", padx=15, pady=(10, 3))
|
|
|
|
self._results_box = ctk.CTkTextbox(self, height=150,
|
|
font=ctk.CTkFont(family="Consolas", size=12),
|
|
state="disabled")
|
|
self._results_box.pack(fill="x", padx=15, pady=(0, 5))
|
|
|
|
# ── Targets section ──
|
|
targets_header = ctk.CTkFrame(self, fg_color="transparent")
|
|
targets_header.pack(fill="x", padx=15, pady=(10, 3))
|
|
|
|
targets_label = ctk.CTkLabel(targets_header, text=t("prom_targets"),
|
|
font=ctk.CTkFont(size=14, weight="bold"), anchor="w")
|
|
targets_label.pack(side="left")
|
|
|
|
self._refresh_btn = make_icon_button(targets_header, "refresh", t("prom_refresh"), width=100,
|
|
command=self._refresh_all)
|
|
self._refresh_btn.pack(side="right")
|
|
|
|
targets_frame = ctk.CTkFrame(self, fg_color="transparent")
|
|
targets_frame.pack(fill="both", expand=True, padx=15, pady=(0, 5))
|
|
|
|
target_columns = ("job", "instance", "health", "last_scrape")
|
|
self._targets_tree = ttk.Treeview(targets_frame, columns=target_columns, show="headings",
|
|
selectmode="browse", height=6)
|
|
self._targets_tree.heading("job", text=t("prom_target_job"))
|
|
self._targets_tree.heading("instance", text=t("prom_target_instance"))
|
|
self._targets_tree.heading("health", text=t("prom_target_health"))
|
|
self._targets_tree.heading("last_scrape", text=t("prom_target_scrape"))
|
|
self._targets_tree.column("job", width=120, minwidth=80)
|
|
self._targets_tree.column("instance", width=200, minwidth=120)
|
|
self._targets_tree.column("health", width=80, minwidth=60)
|
|
self._targets_tree.column("last_scrape", width=150, minwidth=80)
|
|
self._targets_tree.pack(side="left", fill="both", expand=True)
|
|
|
|
targets_scroll = ttk.Scrollbar(targets_frame, orient="vertical",
|
|
command=self._targets_tree.yview,
|
|
style="Dark.Vertical.TScrollbar")
|
|
targets_scroll.pack(side="right", fill="y")
|
|
self._targets_tree.configure(yscrollcommand=targets_scroll.set)
|
|
|
|
# ── Alerts section ──
|
|
alerts_label = ctk.CTkLabel(self, text=t("prom_alerts"),
|
|
font=ctk.CTkFont(size=14, weight="bold"), anchor="w")
|
|
alerts_label.pack(fill="x", padx=15, pady=(10, 3))
|
|
|
|
self._alerts_box = ctk.CTkTextbox(self, height=100,
|
|
font=ctk.CTkFont(family="Consolas", size=12),
|
|
state="disabled")
|
|
self._alerts_box.pack(fill="x", padx=15, pady=(0, 5))
|
|
|
|
# ── Status bar ──
|
|
self._status_bar = ctk.CTkLabel(self, text=t("prom_no_server"), anchor="w",
|
|
font=ctk.CTkFont(size=11), text_color="#9ca3af")
|
|
self._status_bar.pack(fill="x", padx=15, pady=(5, 10))
|
|
|
|
# ── Public API ──
|
|
|
|
def set_server(self, alias: str | None):
|
|
"""Called when user selects a server in sidebar."""
|
|
self._current_alias = alias
|
|
self._client = None
|
|
self._clear_all()
|
|
|
|
if alias:
|
|
self._set_status(t("prom_connected").format(alias=alias), "#22c55e")
|
|
self._refresh_all()
|
|
else:
|
|
self._set_status(t("prom_no_server"), "#9ca3af")
|
|
|
|
# ── PromQL execution ──
|
|
|
|
def _execute_query(self):
|
|
query = self._query_entry.get().strip()
|
|
if not query:
|
|
return
|
|
if not self._current_alias:
|
|
self._set_results(t("no_server_selected"))
|
|
return
|
|
|
|
self._exec_btn.configure(state="disabled")
|
|
self._set_results(t("prom_executing"))
|
|
|
|
def _do():
|
|
try:
|
|
client = self._get_client()
|
|
result = client.query(query)
|
|
formatted = self._format_query_result(result)
|
|
self.after(0, lambda: self._set_results(formatted))
|
|
except Exception as e:
|
|
self.after(0, lambda: self._set_results(f"(error) {e}"))
|
|
finally:
|
|
self.after(0, lambda: self._exec_btn.configure(state="normal"))
|
|
|
|
threading.Thread(target=_do, daemon=True).start()
|
|
|
|
def _format_query_result(self, result: dict) -> str:
|
|
"""Format Prometheus query API response for display."""
|
|
status = result.get("status", "unknown")
|
|
if status != "success":
|
|
error = result.get("error", "Unknown error")
|
|
return f"Error: {error}"
|
|
|
|
data = result.get("data", {})
|
|
result_type = data.get("resultType", "")
|
|
results = data.get("result", [])
|
|
|
|
if not results:
|
|
return "(empty result)"
|
|
|
|
lines = [f"# Type: {result_type}", f"# Results: {len(results)}", ""]
|
|
|
|
for item in results:
|
|
metric = item.get("metric", {})
|
|
metric_str = ", ".join(f'{k}="{v}"' for k, v in metric.items())
|
|
|
|
if result_type == "vector":
|
|
value = item.get("value", [None, ""])[1]
|
|
lines.append(f"{{{metric_str}}} => {value}")
|
|
elif result_type == "matrix":
|
|
values = item.get("values", [])
|
|
lines.append(f"{{{metric_str}}}")
|
|
for ts, val in values[-10:]: # Show last 10 points
|
|
lines.append(f" @{ts} => {val}")
|
|
if len(values) > 10:
|
|
lines.append(f" ... ({len(values)} total points)")
|
|
else:
|
|
lines.append(f"{{{metric_str}}} => {item}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
# ── Refresh targets & alerts ──
|
|
|
|
def _refresh_all(self):
|
|
if not self._current_alias:
|
|
self._set_status(t("no_server_selected"), "#ef4444")
|
|
return
|
|
|
|
self._refresh_btn.configure(state="disabled", text=t("prom_loading"))
|
|
self._set_status(t("prom_loading"), "#ccaa00")
|
|
|
|
def _do():
|
|
try:
|
|
client = self._get_client()
|
|
|
|
targets_resp = client.targets()
|
|
targets = targets_resp.get("data", {}).get("activeTargets", [])
|
|
alerts_resp = client.alerts()
|
|
alerts = alerts_resp.get("data", {}).get("alerts", [])
|
|
|
|
self.after(0, lambda: self._populate_targets(targets))
|
|
self.after(0, lambda: self._populate_alerts(alerts))
|
|
self.after(0, lambda: self._set_status(
|
|
t("prom_loaded").format(
|
|
targets=len(targets), alerts=len(alerts)
|
|
), "#22c55e"))
|
|
except Exception as e:
|
|
self.after(0, lambda: self._set_status(f"(error) {e}", "#ef4444"))
|
|
finally:
|
|
self.after(0, lambda: (
|
|
self._refresh_btn.configure(state="normal"),
|
|
reconfigure_icon_button(self._refresh_btn, "refresh", t("prom_refresh")),
|
|
))
|
|
|
|
threading.Thread(target=_do, daemon=True).start()
|
|
|
|
def _get_client(self) -> PrometheusClient:
|
|
if self._client is None:
|
|
server = self.store.get_server(self._current_alias)
|
|
if not server:
|
|
raise ValueError(f"Server '{self._current_alias}' not found")
|
|
self._client = PrometheusClient(server)
|
|
return self._client
|
|
|
|
# ── Table population ──
|
|
|
|
def _populate_targets(self, targets: list[dict]):
|
|
self._targets_tree.delete(*self._targets_tree.get_children())
|
|
for target in targets:
|
|
job = target.get("labels", {}).get("job", "—")
|
|
instance = target.get("labels", {}).get("instance", "—")
|
|
health = target.get("health", "unknown")
|
|
last_scrape = target.get("lastScrape", "—")
|
|
|
|
tag = ""
|
|
if health == "up":
|
|
tag = "up"
|
|
elif health == "down":
|
|
tag = "down"
|
|
|
|
self._targets_tree.insert("", "end",
|
|
values=(job, instance, health, last_scrape),
|
|
tags=(tag,))
|
|
|
|
self._targets_tree.tag_configure("up", foreground="#22c55e")
|
|
self._targets_tree.tag_configure("down", foreground="#ef4444")
|
|
|
|
def _populate_alerts(self, alerts: list[dict]):
|
|
self._alerts_box.configure(state="normal")
|
|
self._alerts_box.delete("1.0", "end")
|
|
|
|
if not alerts:
|
|
self._alerts_box.insert("1.0", t("prom_no_alerts"))
|
|
else:
|
|
lines = []
|
|
for a in alerts:
|
|
name = a.get("labels", {}).get("alertname", a.get("name", "unknown"))
|
|
state = a.get("state", "unknown")
|
|
severity = a.get("labels", {}).get("severity", "—")
|
|
lines.append(f"[{state.upper()}] {name} (severity: {severity})")
|
|
self._alerts_box.insert("1.0", "\n".join(lines))
|
|
|
|
self._alerts_box.configure(state="disabled")
|
|
|
|
# ── Helpers ──
|
|
|
|
def _set_results(self, text: str):
|
|
self._results_box.configure(state="normal")
|
|
self._results_box.delete("1.0", "end")
|
|
self._results_box.insert("1.0", text)
|
|
self._results_box.configure(state="disabled")
|
|
|
|
def _clear_all(self):
|
|
self._targets_tree.delete(*self._targets_tree.get_children())
|
|
self._set_results("")
|
|
self._alerts_box.configure(state="normal")
|
|
self._alerts_box.delete("1.0", "end")
|
|
self._alerts_box.configure(state="disabled")
|
|
|
|
def _set_status(self, text: str, color: str = "#9ca3af"):
|
|
self._status_bar.configure(text=text, text_color=color)
|