Edit file

File name : check_jetbackup.py

Content :#!/usr/bin/env python3 """ JetBackup Health Check Plugin for Icinga Description: - Checks JetBackup MongoDB connection - Verifies JetBackup license status - Monitors backup job durations (--cdur / --wdur) - Monitors recent JetBackup alerts (--calert / --walert) - Prints backup destiantion(s) Usage: ./check_jetbackup.py [--cdur 24] [--wdur 12] [--calert 24] [--walert 12] Exit Codes: 0 OK 1 WARNING 2 CRITICAL 3 UNKNOWN """ import os import sys import subprocess import shutil import json import re from datetime import datetime, timedelta import argparse def run_command(command): try: return subprocess.run( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True, universal_newlines=True ) except subprocess.CalledProcessError: return None def check_jetmongo(): auth_file = "/usr/local/jetapps/etc/.mongod.auth" if not os.path.isfile(auth_file): print("OK: JetBackup is not installed on this server") return 0 try: with open(auth_file) as f: password = next((line.strip().split("=", 1)[1] for line in f if line.startswith("PASS=")), None) if not password: print("CRITICAL: Password not found in auth file") return 2 except Exception as e: print(f"CRITICAL: Failed to read auth file: {e}") return 2 mongo_paths = [ "/usr/local/jetapps/usr/bin/mongo", "/usr/local/jetapps/usr/bin/mongosh" ] mongo_cli = next((path for path in mongo_paths if os.access(path, os.X_OK)), None) if not mongo_cli: print("CRITICAL: mongo cli not found") return 2 result = run_command([ mongo_cli, "--port", "27217", "--username", "root", "--password", password, "--authenticationDatabase", "admin", "--quiet", "jetbackup", "--eval", "printjson(db.stats().ok)" ]) if result is None: print("CRITICAL: Failed to connect to JetMongo") return 2 if result.stdout.strip() != "1": print("CRITICAL: JetMongo is not working") return 2 print("OK: JetMongo is working") return 0 def check_jetbackup5_license(): result = run_command([ "sudo", "/usr/bin/jetbackup5api", "-F", "getDashboardDetails", "-O", "json" ]) if result is None: print("CRITICAL: Failed to run jetbackup5api") return 2 if '"licenseIssue":false' in result.stdout: print("OK: No licensing issues found (JetBackup 5)") return 0 else: print("CRITICAL: There are licensing issues with the JetBackup 5 instance") return 2 def check_jetbackup4_license(): result = run_command([ "sudo", "/usr/bin/jetapi", "backup", "-F", "licenseStatus" ]) if result is None: print("CRITICAL: Failed to run jetapi") return 2 if "licenseIssue: 1" in result.stdout: print("CRITICAL: There are licensing issues with the JetBackup 4 instance") return 2 else: print("OK: No licensing issues found (JetBackup 4)") return 0 def check_jetbackup_license(): if shutil.which("jetbackup5api"): return check_jetbackup5_license() else: return check_jetbackup4_license() def parse_time(time_str): if not time_str: return None time_str = re.sub(r"(\d{2}):(\d{2})$", r"\1\2", time_str) timezone_offset = timedelta(hours=int(time_str[-4:-2]), minutes=int(time_str[-2:])) return datetime.strptime(time_str[:-6], "%Y-%m-%dT%H:%M:%S") + timezone_offset def check_jetbackup_job_durations(): parser = argparse.ArgumentParser(add_help=False) parser.add_argument("--cdur", type=int, default=24, help="Critical duration threshold in hours (default: 24)") parser.add_argument("--wdur", type=int, default=12, help="Warning duration threshold in hours (default: 12)") args, _ = parser.parse_known_args() critical_hours = args.cdur warning_hours = args.wdur if shutil.which("jetbackup5api"): jobs_cmd = ["jetbackup5api", "-F", "listBackupJobs", "-O", "json"] elif shutil.which("jetapi"): jobs_cmd = ["jetapi", "backup", "-F", "listBackupJobs", "-O", "json"] else: print("UNKNOWN: No suitable JetBackup command found") return 3 jobs_output = run_command(jobs_cmd) if not jobs_output: print("UNKNOWN: Failed to get JetBackup job data") return 3 try: jobs_data = json.loads(jobs_output.stdout) except ValueError: print("UNKNOWN: Failed to parse JetBackup jobs JSON") return 3 jobs = jobs_data.get("data", {}).get("jobs", []) if not jobs: print("CRITICAL: No backup jobs found") return 2 now = datetime.now() critical_alerts = [] warning_alerts = [] for job in jobs: job_id = job.get("_id") running = job.get("running", False) last_run = parse_time(job.get("last_run")) if running and last_run: age = now - last_run if age > timedelta(hours=critical_hours): critical_alerts.append(job_id) elif age > timedelta(hours=warning_hours): warning_alerts.append(job_id) if critical_alerts: print(f"CRITICAL: BackupJob(s) running > {critical_hours}h: {', '.join(critical_alerts)}") return 2 elif warning_alerts: print(f"WARNING: BackupJob(s) running > {warning_hours}h: {', '.join(warning_alerts)}") return 1 print("OK: JetBackup Queue is ok") return 0 def check_jetbackup_alerts(): parser = argparse.ArgumentParser(add_help=False) parser.add_argument("--calert", type=int, default=24, help="Critical alert threshold in hours") parser.add_argument("--walert", type=int, default=12, help="Warning alert threshold in hours") args, _ = parser.parse_known_args() def is_command_available(command): try: subprocess.call([command], stdout=subprocess.PIPE, stderr=subprocess.PIPE) return True except OSError: return False def get_current_users_count(): try: if os.path.exists("/usr/local/cpanel/cpanel"): command = "whmapi1 --output=jsonpretty get_current_users_count" process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) output, _ = process.communicate() data = json.loads(output) return data['data']['users'] except: pass try: if os.path.exists("/usr/local/psa/version"): command = "plesk bin customer --list | wc -l" process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) output, _ = process.communicate() return int(output.strip()) except: pass try: if os.path.exists("/usr/local/directadmin/directadmin"): command = "/bin/ls -1 /usr/local/directadmin/data/users/ | /usr/bin/wc -l" process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True) output, _ = process.communicate() return int(output.decode('utf-8').strip()) except: pass def get_critical_titles(alerts, hours, source_cmd): titles = "" now = datetime.now() threshold = now - timedelta(hours=hours) exclude = ["Restore process for the account"] for alert in alerts: created = datetime.strptime(alert["created"][:-6], "%Y-%m-%dT%H:%M:%S") title = alert["title"] if any(ex in title for ex in exclude): continue if source_cmd == "jetapi" and alert.get("level", "0") == 128 and created > threshold: titles += (" | " if titles else "") + title elif source_cmd == "jetbackup5api" and alert.get("level", "0") == 4 and created > threshold: titles += (" | " if titles else "") + title return titles if is_command_available("jetbackup5api"): output = run_command(["jetbackup5api", "-F", "listAlerts", "-D", "sort[created]=-1", "-O", "json"]) command = "jetbackup5api" elif is_command_available("jetapi"): output = run_command(["jetapi", "backup", "-F", "listAlerts", "-O", "json"]) command = "jetapi" else: print("JetBackup not found on the server") return 0 if not output: print("UNKNOWN: Failed to retrieve alerts") return 3 try: data = json.loads(output.stdout) alerts = data["data"]["alerts"] except Exception: print("UNKNOWN: Failed to parse JetBackup alerts") return 3 users_count = get_current_users_count() critical_titles = get_critical_titles(alerts, args.calert, command) warning_titles = get_critical_titles(alerts, args.walert, command) found_no_accounts = any("No accounts found for backup" in a.get("message", "") for a in alerts) jb_config_export = any("JB configurations export Backup job Failed" in a.get("title", "") for a in alerts) jb_download_backup = any("Download process for the account" in a.get("title", "") for a in alerts) if critical_titles: if users_count > 0 and found_no_accounts and not jb_config_export: print(f"CRITICAL: {critical_titles} ({users_count} users)") return 2 elif users_count == 0 and found_no_accounts: print(f"OK: No accounts found for backup ({users_count} users)") return 0 elif jb_config_export and len(critical_titles.split("|")) == 1: print(f"WARNING: JB configurations export Backup job Failed ({users_count} users)") return 1 elif jb_download_backup and len(critical_titles.split("|")) == 1: print(f"WARNING: {critical_titles} ({users_count} users)") return 1 else: print(f"CRITICAL: {critical_titles} ({users_count} users)") return 2 elif warning_titles: if users_count > 0 and found_no_accounts: print(f"WARNING: {warning_titles} ({users_count} users)") return 1 elif users_count == 0 and found_no_accounts: print(f"OK: No accounts found for backup ({users_count} users)") return 0 else: print(f"WARNING: {warning_titles} ({users_count} users)") return 1 else: print(f"OK: JetBackup alerts clean ({users_count} users)") return 0 def get_backup_destinations(): binary = None for cmd in ['jetbackup5api', 'jetapi']: if os.path.exists(os.path.join('/usr/bin', cmd)): binary = cmd break try: subprocess.check_output(['which', cmd], stderr=subprocess.PIPE) binary = cmd break except subprocess.CalledProcessError: continue try: # Run the API command and get output output = subprocess.check_output( [binary, "-F", "listBackupJobs", "-O", "json"], stderr=subprocess.PIPE ).decode('utf-8') # Parse JSON and extract unique hosts data = json.loads(output) jet_destinations = set() jobs = data.get('data', {}).get('jobs', []) for job in jobs: destinations = job.get('destination_details', []) for dest in destinations: options = dest.get('options', {}) host = options.get('host') if host: jet_destinations.add(host) elif options.get('bucket'): bucket = options.get('bucket') region = options.get('region', '') endpoint = options.get('endpoint', '') formatted = f"bucket={bucket}/region={region}/endpoint={endpoint}" jet_destinations.add(formatted) backup_destinations = ",".join(jet_destinations) if jet_destinations else None if backup_destinations: print("OK: Backup destination(s):", backup_destinations) return 0 else: print("OK: Could not find a backup destination") return 0 except Exception as e: return None if __name__ == "__main__": if not (shutil.which("jetbackup5api") or shutil.which("jetbackup4api") or shutil.which("jetapi")): print("OK: JetBackup is not installed on this server") sys.exit(0) if shutil.which("backuply"): print("OK: Backuply is installed on this server") sys.exit(0) for check_fn in [check_jetmongo, get_backup_destinations, check_jetbackup_license, check_jetbackup_job_durations, check_jetbackup_alerts]: result = check_fn() if result != 0: sys.exit(result)

Save