""" 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("", 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 = client.get_targets() alerts = client.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: self._client = PrometheusClient(self._current_alias, self.store) 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)