#!/usr/bin/env python3 """ veeam_exporter.py — Veeam Backup Prometheus Exporter 用法: python3 veeam_exporter.py --port 9199 --vbr 10.115.13.219 --user veeamuser01 --password 'Qazx1234!@#$qwer2' """ import argparse import json import ssl import time import urllib.request import urllib.parse from http.server import HTTPServer, BaseHTTPRequestHandler from datetime import datetime, timezone # ── Prometheus 指标定义 ── METRICS = {} def gauge(name, help_text): METRICS[name] = {"help": help_text, "type": "gauge", "samples": {}} # 任务级别指标 gauge("veeam_job_status", "Backup job last result: 0=unknown, 1=success, 2=warning, 3=failed") gauge("veeam_job_duration_seconds", "Last session duration in seconds") gauge("veeam_job_progress_percent", "Last session progress percent") gauge("veeam_job_processed_bytes", "Processed data size in bytes") gauge("veeam_job_read_bytes", "Read data size in bytes") gauge("veeam_job_transferred_bytes", "Transferred data size in bytes") gauge("veeam_job_processing_rate_bps", "Processing rate in bytes/sec") gauge("veeam_job_last_run_timestamp", "Last run end time as unix timestamp") # 会话级别指标 gauge("veeam_session_active", "Currently running sessions: 1=running, 0=stopped") gauge("veeam_session_status", "Session result: 0=unknown, 1=success, 2=warning, 3=failed") # 汇总指标 gauge("veeam_backup_jobs_total", "Total number of backup jobs") gauge("veeam_sessions_total_24h", "Total sessions in last 24 hours") gauge("veeam_sessions_failed_24h", "Failed sessions in last 24 hours") gauge("veeam_sessions_success_24h", "Successful sessions in last 24 hours") gauge("veeam_sessions_warning_24h", "Warning sessions in last 24 hours") gauge("veeam_api_up", "1 if Veeam API is reachable, 0 otherwise") gauge("veeam_collector_duration_seconds", "Time taken to collect metrics") # ── Veeam API Client ── class VeeamClient: def __init__(self, host, port, user, password, verify_ssl=False): self.base = f"https://{host}:{port}" self.user = user self.password = password self.token = None self.verify_ssl = verify_ssl self.ctx = ssl.create_default_context() if not verify_ssl: self.ctx.check_hostname = False self.ctx.verify_mode = ssl.CERT_NONE def _request(self, method, path, data=None, headers=None): url = f"{self.base}{path}" hdrs = headers or {} if self.token: hdrs["Authorization"] = f"Bearer {self.token}" body = None if data: body = urllib.parse.urlencode(data).encode() hdrs.setdefault("Content-Type", "application/x-www-form-urlencoded") req = urllib.request.Request(url, data=body, headers=hdrs, method=method) resp = urllib.request.urlopen(req, context=self.ctx, timeout=30) return json.loads(resp.read().decode()) def login(self): result = self._request("POST", "/api/oauth2/token", { "grant_type": "password", "username": self.user, "password": self.password }) self.token = result["access_token"] def get_jobs(self): return self._request("GET", "/api/v1/jobs") def get_sessions(self, limit=200): return self._request("GET", f"/api/v1/sessions?limit={limit}") # ── 指标收集 ── def collect(veeam): start = time.time() # 清空所有 samples for m in METRICS.values(): m["samples"] = {} try: veeam.login() set_metric("veeam_api_up", {}, 1) except Exception as e: set_metric("veeam_api_up", {}, 0) set_metric("veeam_collector_duration_seconds", {}, time.time() - start) return # ── 获取 Jobs ── try: jobs_data = veeam.get_jobs() jobs = jobs_data if isinstance(jobs_data, list) else jobs_data.get("jobs", jobs_data.get("data", [])) set_metric("veeam_backup_jobs_total", {}, len(jobs)) except Exception: jobs = [] # ── 获取 Sessions ── try: sess_data = veeam.get_sessions(200) sessions = sess_data if isinstance(sess_data, list) else sess_data.get("sessions", sess_data.get("data", [])) except Exception: sessions = [] # ── 处理每个 Session ── now = time.time() h24_ago = now - 86400 total_24h = 0 failed_24h = 0 success_24h = 0 warning_24h = 0 # 按 jobId 聚合最新 session latest_by_job = {} for s in sessions: stype = s.get("sessionType", "") name = s.get("name", "unknown") job_id = s.get("jobId", "") state = s.get("state", "") result_obj = s.get("result", {}) result = result_obj.get("result", "Unknown") end_time = s.get("endTime", "") progress = s.get("progress", {}) # 会话级别指标 labels = {"name": name, "type": stype, "job_id": job_id} set_metric("veeam_session_active", labels, 1 if state == "Working" else 0) set_metric("veeam_session_status", labels, result_to_num(result)) # 24小时统计 end_ts = parse_timestamp(end_time) if end_ts and end_ts >= h24_ago: if stype in ("BackupJob", "AgentBackupJob", "ReplicaJob", "ConfigurationBackup"): total_24h += 1 r = result.lower() if r == "success": success_24h += 1 elif r == "warning": warning_24h += 1 elif r == "failed": failed_24h += 1 # 只关注备份类任务的最新 session if stype in ("BackupJob", "AgentBackupJob", "ConfigurationBackup"): existing = latest_by_job.get(job_id) if not existing: latest_by_job[job_id] = s else: # 取最新的 existing_ts = parse_timestamp(existing.get("endTime", "")) or 0 current_ts = parse_timestamp(end_time) or 0 if current_ts > existing_ts: latest_by_job[job_id] = s # ── 输出每个 Job 的最新状态 ── for job_id, s in latest_by_job.items(): name = s.get("name", "unknown") stype = s.get("sessionType", "") result_obj = s.get("result", {}) result = result_obj.get("result", "Unknown") end_time = s.get("endTime", "") progress = s.get("progress", {}) platform = s.get("platformName", "unknown") labels = { "job_name": name, "job_id": job_id, "type": stype, "platform": platform } set_metric("veeam_job_status", labels, result_to_num(result)) dur = parse_duration(progress.get("duration", "")) if dur is not None: set_metric("veeam_job_duration_seconds", labels, dur) set_metric("veeam_job_progress_percent", labels, progress.get("progressPercent", 0)) set_metric("veeam_job_processed_bytes", labels, progress.get("processedSize", 0)) set_metric("veeam_job_read_bytes", labels, progress.get("readSize", 0)) set_metric("veeam_job_transferred_bytes", labels, progress.get("transferredSize", 0)) rate = parse_rate(progress.get("processingRate", "")) if rate is not None: set_metric("veeam_job_processing_rate_bps", labels, rate) end_ts = parse_timestamp(end_time) if end_ts: set_metric("veeam_job_last_run_timestamp", labels, end_ts) # ── 汇总 ── set_metric("veeam_sessions_total_24h", {}, total_24h) set_metric("veeam_sessions_failed_24h", {}, failed_24h) set_metric("veeam_sessions_success_24h", {}, success_24h) set_metric("veeam_sessions_warning_24h", {}, warning_24h) set_metric("veeam_collector_duration_seconds", {}, round(time.time() - start, 3)) def set_metric(name, labels, value): key = json.dumps(labels, sort_keys=True) METRICS[name]["samples"][key] = (labels, float(value)) def result_to_num(r): r = (r or "").lower() if r == "success": return 1 if r == "warning": return 2 if r == "failed": return 3 return 0 def parse_duration(dur_str): """解析 '09:12:53' 为秒数""" if not dur_str or dur_str == "N/A": return None try: parts = dur_str.split(":") if len(parts) == 3: return int(parts[0]) * 3600 + int(parts[1]) * 60 + int(parts[2]) except Exception: pass return None def parse_rate(rate_str): """解析 '195.1 MB' 为 bytes/sec""" if not rate_str or rate_str == "N/A": return None try: parts = rate_str.strip().split() val = float(parts[0]) unit = parts[1].upper() if len(parts) > 1 else "B" multipliers = {"B": 1, "KB": 1024, "MB": 1024**2, "GB": 1024**3, "TB": 1024**4} return val * multipliers.get(unit, 1) except Exception: return None def parse_timestamp(ts_str): """解析 ISO 时间为 unix timestamp""" if not ts_str: return None try: # 处理 '2026-08-23T07:13:08.563826-07:00' ts_str = ts_str.replace("Z", "+00:00") # 简单解析 from datetime import datetime # 去掉微秒后面的时区偏移处理 if "." in ts_str: main, frac = ts_str.split(".", 1) # frac: '563826-07:00' tz_idx = max(frac.find("+"), frac.find("-")) if tz_idx > 0: frac = frac[:tz_idx] ts_str_clean = main + "." + frac else: ts_str_clean = ts_str.split("+")[0].split("-0")[0] if "+" in ts_str or ts_str.count("-") > 2 else ts_str # 尝试多种格式 for fmt in ("%Y-%m-%dT%H:%M:%S.%f", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S"): try: dt = datetime.strptime(ts_str_clean[:26], fmt) return dt.timestamp() except ValueError: continue except Exception: pass return None # ── Prometheus 格式输出 ── def format_metrics(): lines = [] for name, m in sorted(METRICS.items()): lines.append(f"# HELP {name} {m['help']}") lines.append(f"# TYPE {name} {m['type']}") for key, (labels, value) in m["samples"].items(): if labels: lbl_str = ",".join(f'{k}="{escape_label(v)}"' for k, v in sorted(labels.items())) lines.append(f"{name}{{{lbl_str}}} {value}") else: lines.append(f"{name} {value}") return "\n".join(lines) + "\n" def escape_label(v): return str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") # ── HTTP Server ── class MetricsHandler(BaseHTTPRequestHandler): def do_GET(self): if self.path in ("/metrics", "/"): collect(self.server.veeam_client) output = format_metrics() self.send_response(200) self.send_header("Content-Type", "text/plain; charset=utf-8") self.end_headers() self.wfile.write(output.encode()) elif self.path == "/health": self.send_response(200) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(b"ok") else: self.send_response(404) self.end_headers() def log_message(self, format, *args): pass # 静默日志 def main(): parser = argparse.ArgumentParser(description="Veeam Backup Prometheus Exporter") parser.add_argument("--port", type=int, default=9199, help="监听端口 (default: 9199)") parser.add_argument("--vbr", required=True, help="VBR 服务器 IP") parser.add_argument("--vbr-port", type=int, default=9419, help="VBR API 端口 (default: 9419)") parser.add_argument("--user", required=True, help="VBR 用户名") parser.add_argument("--password", required=True, help="VBR 密码") args = parser.parse_args() client = VeeamClient(args.vbr, args.vbr_port, args.user, args.password) server = HTTPServer(("0.0.0.0", args.port), MetricsHandler) server.veeam_client = client print(f"[Veeam Exporter] Listening on :{args.port}, VBR={args.vbr}:{args.vbr_port}") try: server.serve_forever() except KeyboardInterrupt: print("\n[Veeam Exporter] Stopped") if __name__ == "__main__": main()