#!/usr/bin/env python3
########################################################################
##
## Written by 6adminIT
## Based on LnxBil script on https://forum.proxmox.com/threads/commandline-to-show-status-of-last-backup.168175/#post-782193
##
## Licensed under GPL (see below)
##
## Python nagios/centreon script to monitor PVE server backups on PBS
## server.
##
## Need proxmoxer wrapper available here : https://pypi.org/project/proxmoxer/
##
## Tested PVE version : 8.4
## Tested PBS version : 3.4
##
## This program is free software: you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation, either version 3 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program. If not, see <http://www.gnu.org/licenses/>.
##
########################################################################
from proxmoxer import ProxmoxAPI
import argparse
import time
from datetime import datetime, timedelta
########### Filled from command line arguments if exists
##
excluded = [""]
time_range = 24
########### Parse command line args
##
parser = argparse.ArgumentParser(
prog='check_pve_backups_to_pbs',
description='Check Proxmox VE backups age on Proxmox Backup Server',
epilog='Note that user with token API must be set wih AUDIT permission on both servers')
parser.add_argument('--pveserver', help="PVE server IP or FQDN and port - eg: proxmoxserver.lan:8006", type=str, required=True)
parser.add_argument('--pveuser', help="PVE server user", type=str, required=True)
parser.add_argument('--pvetokenid', help="PVE API token ID", type=str, required=True)
parser.add_argument('--pvetoken', help="PVE API token value", type=str, required=True)
parser.add_argument('--pbsserver', help="PBS server IP or FQDN and port - eg: proxmoxbackup.lan:8007", type=str, required=True)
parser.add_argument('--pbsuser', help="PBS server user", type=str, required=True)
parser.add_argument('--pbstokenid', help="PBS API token ID", type=str, required=True)
parser.add_argument('--pbstoken', help="PBS API token value", type=str, required=True)
parser.add_argument('--datastore', help="name of the datastore of PBS", type=str, required=True)
parser.add_argument('--exclude', help="Optional - ID of excluded VM or CT, separate with commas - eg: 100,101", type=str, required=False)
parser.add_argument('--time', help="Optional - Range in hours to check backups (default 24)", type=int, required=False)
## Get args
args = parser.parse_args()
pve_server = args.pveserver
pve_user = args.pveuser
pve_tokenid = args.pvetokenid
pve_token = args.pvetoken
pbs_server = args.pbsserver
pbs_user = args.pbsuser
pbs_tokenid = args.pbstokenid
pbs_token = args.pbstoken
datastore = args.datastore
if args.exclude is not None:
excluded = args.exclude.split(",")
if args.time is not None:
time_range = args.time
########### OTHER VARS
##
# in hours
check_backups = True
msg_backup = ""
vm_count = 0
# return state for centreon/nagos: 0=OK, 1=WARNING, 2=CRITICAL, 3=UNKNOWN
stateNum = 0
########### BEGIN CHECKS
##
### Proxmox VE server
pve = ProxmoxAPI(
pve_server,
user=pve_user,
token_name=pve_tokenid,
token_value=pve_token,
verify_ssl=False
)
vmids = {}
for pve_node in pve.nodes.get():
for container in pve.nodes(pve_node['node']).lxc.get():
vmids[container['vmid']]=container['name']
for vm in pve.nodes(pve_node["node"]).qemu.get():
vmids[vm['vmid']]=vm['name']
### Proxmox Backup server
pbs = ProxmoxAPI(
pbs_server,
service="PBS",
user=pbs_user,
token_name=pbs_tokenid,
token_value=pbs_token,
verify_ssl=False
)
for vm in vmids:
if str(vm) not in excluded:
vm_count += 1
backup = pbs(f"admin/datastore/{datastore}/snapshots/?backup-id={vm}").get()
backups = len(backup)
if backups == 0:
msg_backup = msg_backup + f"{vm} no backup present - "
check_backups = False
stateNum = 2
else:
sorted_list = sorted(backup, key=lambda x: x["backup-time"])
ts = sorted_list[-1]["backup-time"]
ago = datetime.now() - datetime.fromtimestamp(ts)
if ago < timedelta(hours=time_range):
msg_backup = msg_backup + f"{vm} was backed up less than {time_range}h ago - "
else:
msg_backup = msg_backup + f"{vm} needs a new backup - "
check_backups = False
stateNum = 2
if check_backups:
print(f"OK - {vm_count} BACKUPS LESS THAN {time_range}h")
# print(f"OK - {msg_backup}")
else:
print(f"KO - {msg_backup}")
exit(stateNum)
