Custom script for successful NS8 backup notifications

20notify

#!/usr/bin/env python3

#
# Copyright (C) 2026 Nethesis S.r.l.
# SPDX-License-Identifier: GPL-3.0-or-later
#

import agent
import sys
import os
import json
import subprocess
from datetime import datetime, timezone

#
# Installation (leader node only)
#
# mkdir -vp /var/lib/nethserver/cluster/events/backup-status-changed/
# cp 20notify /var/lib/nethserver/cluster/events/backup-status-changed/
# chmod 755 /var/lib/nethserver/cluster/events/backup-status-changed/20notify
#
# Configure cluster notifications:
# https://docs.nethserver.org/projects/ns8/en/latest/email_notifications.html
#

#
# Configure this script
#

# 1. Adjust sender and recipient address as wanted:
mail_from = "no-reply@example.com"
mail_to = "backup-admin@example.com"

# 2. Change to False, to not receive successful backup notifications:
notify_success = True

# 3. Change to True, to receive a notification if backup fails:
notify_failed = False

# 4. Change mail_subject and mail_message to tweak the mail contents:
def main():
    evdata = json.load(sys.stdin)
    rdb = agent.redis_connect()
    backup_id = evdata['backup_id']
    node_id = str(evdata['node_id'])
    module_id = evdata['module_id']
    module_ui_name = rdb.get(f"module/{module_id}/ui_name") or ""
    hbackup_status = json.loads(rdb.hget(f"node/{node_id}/backup_status/{backup_id}", module_id))
    backup_status = "SUCCESS" if hbackup_status['errors'] == 0 else "FAIL"
    hbackup = rdb.hgetall(f"cluster/backup/{backup_id}")
    backup_name = hbackup['name']
    dest_uuid = hbackup['repository']
    dest_name = rdb.hget(f"cluster/backup_repository/{dest_uuid}", "name")
    cluster_admin_url = f"https://{agent.get_hostname()}/cluster-admin"
    start_time = datetime.fromtimestamp(hbackup_status['start'], tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
    end_time = datetime.fromtimestamp(hbackup_status['end'], tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
    duration_secs = hbackup_status['end'] - hbackup_status['start']
    total_size = format_size(hbackup_status['total_size'])
    mail_subject = f"Backup {module_ui_name or module_id}: {backup_status}"
    mail_message = (
        f"The backup for {module_id} ({module_ui_name or '-'}) has completed with status: {backup_status}.\n"
        "\n"
        f"  Schedule:     {backup_name} ({backup_id})\n"
        f"  Destination:  {dest_name} ({dest_uuid})\n"
        f"  Start time:   {start_time}\n"
        f"  End time:     {end_time}\n"
        f"  Duration:     {duration_secs}s\n"
        f"  Total size:   {total_size}\n"
        f"  Total files:  {hbackup_status['total_file_count']}\n"
        f"  Snapshots:    {hbackup_status['snapshots_count']}\n"
        f"  Errors:       {hbackup_status['errors']}\n"
        "\n"
        f"Please check the system for details: {cluster_admin_url}\n"
    )
    if (backup_status == "SUCCESS" and notify_success) or (backup_status == "FAIL" and notify_failed):
        send_message(mail_subject, mail_message)

def format_size(size_bytes):
    """Format bytes into a human-readable string."""
    for unit in ('B', 'KiB', 'MiB', 'GiB', 'TiB'):
        if abs(size_bytes) < 1024:
            return f"{size_bytes:.1f} {unit}"
        size_bytes /= 1024
    return f"{size_bytes:.1f} PiB"

def send_message(mail_subject, mail_message):
    subprocess.run(
        ["ns8-sendmail", "-s", mail_subject, "-f", mail_from, mail_to],
        input=mail_message,
        text=True,
    )

if __name__ == "__main__":
    main()