350 lines
12 KiB
Python
350 lines
12 KiB
Python
#!/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() |