""" Prometheus tab — PromQL query executor, targets overview, alerts, and rules. """ 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 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("", 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") # ── Quick query buttons ── quick_frame = ctk.CTkFrame(self, fg_color="transparent") quick_frame.pack(fill="x", padx=15, pady=(0, 5)) for label, query in [("up", "up"), ("CPU", "process_cpu_seconds_total"), ("Goroutines", "go_goroutines")]: btn = make_icon_button(quick_frame, "metrics", label, width=80, fg_color="#6b7280", hover_color="#4b5563", command=lambda q=query: self._run_quick(q)) btn.pack(side="left", padx=(0, 5)) self._metrics_btn = make_icon_button(quick_frame, "search", t("prom_metrics_browser"), width=100, fg_color="#6b7280", hover_color="#4b5563", command=self._open_metrics_browser) self._metrics_btn.pack(side="left", padx=(0, 5)) # ── 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=(5, 3)) self._results_box = ctk.CTkTextbox(self, height=120, 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=(5, 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=5) 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=(5, 3)) self._alerts_box = ctk.CTkTextbox(self, height=80, font=ctk.CTkFont(family="Consolas", size=12), state="disabled") self._alerts_box.pack(fill="x", padx=15, pady=(0, 5)) # ── Rules section ── rules_label = ctk.CTkLabel(self, text=t("prom_rules"), font=ctk.CTkFont(size=14, weight="bold"), anchor="w") rules_label.pack(fill="x", padx=15, pady=(5, 3)) rules_frame = ctk.CTkFrame(self, fg_color="transparent") rules_frame.pack(fill="both", expand=True, padx=15, pady=(0, 5)) rules_columns = ("type", "name", "group", "health") self._rules_tree = ttk.Treeview(rules_frame, columns=rules_columns, show="headings", selectmode="browse", height=5) self._rules_tree.heading("type", text=t("prom_rule_type")) self._rules_tree.heading("name", text=t("prom_rule_name")) self._rules_tree.heading("group", text=t("prom_rule_group")) self._rules_tree.heading("health", text=t("prom_rule_health")) self._rules_tree.column("type", width=80, minwidth=60) self._rules_tree.column("name", width=250, minwidth=120) self._rules_tree.column("group", width=150, minwidth=80) self._rules_tree.column("health", width=80, minwidth=60) self._rules_tree.pack(side="left", fill="both", expand=True) rules_scroll = ttk.Scrollbar(rules_frame, orient="vertical", command=self._rules_tree.yview, style="Dark.Vertical.TScrollbar") rules_scroll.pack(side="right", fill="y") self._rules_tree.configure(yscrollcommand=rules_scroll.set) # ── 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): 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") # ── Quick query ── def _run_quick(self, query: str): self._query_entry.delete(0, "end") self._query_entry.insert(0, query) self._execute_query() # ── 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("prom_no_server")) 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: status = result.get("status", "unknown") if status != "success": return f"Error: {result.get('error', 'Unknown 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:]: 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) # ── Metrics browser ── def _open_metrics_browser(self): if not self._current_alias: self._set_status(t("prom_no_server"), "#ef4444") return self._metrics_btn.configure(state="disabled") def _do(): try: client = self._get_client() resp = client._get("/api/v1/label/__name__/values") metrics = resp.get("data", []) self.after(0, lambda: self._show_metrics_popup(metrics)) except Exception as e: self.after(0, lambda: self._set_status(f"(error) {e}", "#ef4444")) finally: self.after(0, lambda: self._metrics_btn.configure(state="normal")) threading.Thread(target=_do, daemon=True).start() def _show_metrics_popup(self, metrics: list[str]): popup = ctk.CTkToplevel(self) popup.title(t("prom_metrics_browser")) popup.geometry("450x500") popup.transient(self.winfo_toplevel()) filter_entry = ctk.CTkEntry(popup, placeholder_text=t("prom_filter_metrics")) filter_entry.pack(fill="x", padx=10, pady=(10, 5)) listbox_frame = ctk.CTkFrame(popup, fg_color="transparent") listbox_frame.pack(fill="both", expand=True, padx=10, pady=(0, 10)) tree = ttk.Treeview(listbox_frame, columns=("metric",), show="headings", selectmode="browse") tree.heading("metric", text="Metric Name") tree.column("metric", width=400) tree.pack(side="left", fill="both", expand=True) scroll = ttk.Scrollbar(listbox_frame, orient="vertical", command=tree.yview, style="Dark.Vertical.TScrollbar") scroll.pack(side="right", fill="y") tree.configure(yscrollcommand=scroll.set) all_metrics = sorted(metrics) def populate(filter_text=""): tree.delete(*tree.get_children()) for m in all_metrics: if filter_text.lower() in m.lower(): tree.insert("", "end", values=(m,)) def on_filter(*_): populate(filter_entry.get()) filter_entry.bind("", on_filter) def on_select(event): sel = tree.selection() if sel: metric = tree.item(sel[0])["values"][0] self._query_entry.delete(0, "end") self._query_entry.insert(0, metric) popup.destroy() tree.bind("", on_select) populate() filter_entry.focus_set() # ── Refresh targets, alerts & rules ── def _refresh_all(self): if not self._current_alias: self._set_status(t("prom_no_server"), "#ef4444") return self._refresh_btn.configure(state="disabled") 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", []) rules_resp = client.rules() rule_groups = rules_resp.get("data", {}).get("groups", []) self.after(0, lambda: self._populate_targets(targets)) self.after(0, lambda: self._populate_alerts(alerts)) self.after(0, lambda: self._populate_rules(rule_groups)) rule_count = sum(len(g.get("rules", [])) for g in rule_groups) self.after(0, lambda: self._set_status( t("prom_loaded").format( targets=len(targets), alerts=len(alerts), rules=rule_count ), "#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 = "up" if health == "up" else ("down" if health == "down" else "") 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") def _populate_rules(self, groups: list[dict]): self._rules_tree.delete(*self._rules_tree.get_children()) for group in groups: group_name = group.get("name", "") for rule in group.get("rules", []): rtype = rule.get("type", "") name = rule.get("name", "") health = rule.get("health", "") tag = "up" if health == "ok" else ("down" if health == "err" else "") self._rules_tree.insert("", "end", values=(rtype, name, group_name, health), tags=(tag,)) self._rules_tree.tag_configure("up", foreground="#22c55e") self._rules_tree.tag_configure("down", foreground="#ef4444") # ── 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._rules_tree.delete(*self._rules_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)