Posted on
• Artificial Intelligence

Artificial Intelligence Backup Automation Projects

Author
  • User
    linuxbash
    Posts by this author
    Posts by this author

AI-Powered Bash: Backup Automation Projects You Can Ship This Week

Ever kicked off a backup only to watch it slow your system to a crawl—or worse, discover days later that it silently failed? Backups are mission-critical, but traditional “set-and-forget” cron jobs don’t adapt to changing workloads, data growth, or subtle failures. This post shows how to bring practical artificial intelligence to your Linux backup stack—without abandoning Bash—so your backups run at the right time, catch anomalies early, and optimize where your data lands.

You’ll get 3 hands-on projects with copy-pasteable code, cross-distro install commands, and a clean separation between Bash orchestration and small Python utilities for the “AI” bits.

Why put AI in your backup pipeline?

  • Workloads are dynamic. Yesterday’s 3 AM quiet window might be today’s batch job peak.

  • Silent failures are costly. Catching unusual backup sizes or durations can save you from discovering corruption only when you need to restore.

  • Storage isn’t free. Smarter placement (fast local vs. cheap cloud) can tame costs while keeping RTO/RPO goals.

AI here doesn’t mean complicated deep learning. We’ll start with lightweight models (clustering, anomaly detection, logistic regression) that run in seconds and integrate neatly with Bash.

Prerequisites and installation

You’ll need Python for the ML helpers, plus common backup tools. Use your package manager:

  • On Debian/Ubuntu (apt):
sudo apt update
sudo apt install -y \
  python3 python3-pip python3-venv \
  rsync rclone restic borgbackup \
  at sysstat cron
sudo systemctl enable --now atd
sudo systemctl enable --now cron
  • On Fedora/RHEL/CentOS Stream (dnf):
sudo dnf install -y \
  python3 python3-pip python3-virtualenv \
  rsync rclone restic borgbackup \
  at sysstat cronie
sudo systemctl enable --now atd
sudo systemctl enable --now crond
  • On openSUSE (zypper):
sudo zypper install -y \
  python3 python3-pip python3-virtualenv \
  rsync rclone restic borgbackup \
  at sysstat cron
sudo systemctl enable --now atd
sudo systemctl enable --now cron

Python packages:

python3 -m pip install --user numpy pandas scikit-learn joblib

Create a working area:

sudo mkdir -p /usr/local/sbin /var/lib/ai-backup
sudo chown -R "$USER":"$USER" /var/lib/ai-backup

Project 1: Predictive scheduling — Run backups when your system is quiet

We’ll: 1) Collect light system telemetry. 2) Use a tiny ML script to pick the quietest hour in the next 24h. 3) Schedule the backup for that time using at.

1. Collect telemetry (load, disk util) every 5 minutes

Save as /usr/local/sbin/ai-backup-metrics.sh and make executable:

sudo tee /usr/local/sbin/ai-backup-metrics.sh >/dev/null <<'EOF'
#!/usr/bin/env bash
set -euo pipefail
DIR=/var/lib/ai-backup
CSV="$DIR/sys_metrics.csv"
mkdir -p "$DIR"
if [ ! -f "$CSV" ]; then
  echo "ts,load1,disk_util" > "$CSV"
fi

ts=$(date -Is)
load1=$(awk '{print $1}' /proc/loadavg)

# Average disk utilization across devices if iostat is present
disk_util=""
if command -v iostat >/dev/null 2>&1; then
  disk_util=$(iostat -dx 1 2 | awk '/^(sd|vd|xvd|nvme)/ {sum+=$NF; n++} END{if(n>0) printf "%.1f", sum/n; else print ""}')
fi

echo "$ts,$load1,${disk_util:-}" >> "$CSV"
EOF
sudo chmod +x /usr/local/sbin/ai-backup-metrics.sh

Add a root cron entry to sample every 5 minutes:

sudo tee /etc/cron.d/ai-backup-metrics >/dev/null <<'EOF'
*/5 * * * * root /usr/local/sbin/ai-backup-metrics.sh
EOF

2. ML picker: choose tomorrow’s quietest hour

Save as /usr/local/sbin/ai-backup-pick-window.py:

#!/usr/bin/env python3
import sys, os, math
from datetime import datetime, timedelta
import pandas as pd
import numpy as np
from sklearn.cluster import KMeans

CSV = "/var/lib/ai-backup/sys_metrics.csv"

def main():
    if not os.path.exists(CSV):
        # Default to 03:07 tomorrow if no data
        t = (datetime.now() + timedelta(days=1)).replace(hour=3, minute=7, second=0, microsecond=0)
        print(t.strftime("%Y%m%d%H%M"))
        return

    df = pd.read_csv(CSV, parse_dates=["ts"])
    # Keep last 30 days
    cutoff = datetime.now() - timedelta(days=30)
    df = df[df["ts"] >= cutoff].copy()
    if df.empty:
        t = (datetime.now() + timedelta(days=1)).replace(hour=3, minute=7, second=0, microsecond=0)
        print(t.strftime("%Y%m%d%H%M"))
        return

    # Fill missing disk_util with median
    if "disk_util" not in df.columns:
        df["disk_util"] = np.nan
    df["disk_util"] = pd.to_numeric(df["disk_util"], errors="coerce")
    df["disk_util"].fillna(df["disk_util"].median(skipna=True), inplace=True)

    df["hour"] = df["ts"].dt.hour
    hourly = df.groupby("hour", as_index=False).agg({"load1":"mean","disk_util":"mean"}).sort_values("hour")
    X = hourly[["load1","disk_util"]].values

    # If not enough data, simple fallback: pick the hour with min (load1 + disk)
    if len(hourly) < 3:
        best_hour = int(hourly.assign(score=hourly["load1"]+hourly["disk_util"]).sort_values("score").iloc[0]["hour"])
    else:
        k = min(3, len(hourly))
        km = KMeans(n_clusters=k, n_init="auto", random_state=42).fit(X)
        centers = km.cluster_centers_
        # Choose cluster with minimal combined center
        scores = centers.sum(axis=1)
        best_cluster = int(np.argmin(scores))
        # Hours in best cluster
        best_hours = hourly[km.labels_ == best_cluster]["hour"].tolist()
        # Pick the soonest occurrence within next 24h
        now = datetime.now()
        candidates = []
        for h in best_hours:
            # today or tomorrow depending on current time
            base = now.replace(minute=7, second=0, microsecond=0)
            candidate = base.replace(hour=h)
            if candidate <= now:
                candidate += timedelta(days=1)
            # restrict to next 24h
            if candidate <= now + timedelta(hours=24):
                candidates.append(candidate)
        if not candidates:
            # Fallback to the min-scoring single hour
            best_hour = int(hourly.assign(score=hourly["load1"]+hourly["disk_util"]).sort_values("score").iloc[0]["hour"])
            candidate = now.replace(hour=best_hour, minute=7, second=0, microsecond=0)
            if candidate <= now:
                candidate += timedelta(days=1)
        else:
            candidate = min(candidates)
        print(candidate.strftime("%Y%m%d%H%M"))
        return

    # Simple-mode candidate
    now = datetime.now()
    candidate = now.replace(hour=best_hour, minute=7, second=0, microsecond=0)
    if candidate <= now:
        candidate += timedelta(days=1)
    print(candidate.strftime("%Y%m%d%H%M"))

if __name__ == "__main__":
    main()

Make it executable:

sudo chmod +x /usr/local/sbin/ai-backup-pick-window.py

3. The backup job and scheduler

Example backup script (Restic). Save as /usr/local/sbin/ai-backup-run.sh:

#!/usr/bin/env bash
set -euo pipefail

# Configure Restic repo and credentials (adjust to your environment)
export RESTIC_REPOSITORY=/backup/restic
export RESTIC_PASSWORD_FILE=/etc/restic.pw

HIST=/var/lib/ai-backup/backup_history.csv
mkdir -p "$(dirname "$HIST")"
if [ ! -f "$HIST" ]; then
  echo "ts,size_bytes,duration_s,exit_code" > "$HIST"
fi

START=$(date +%s)
TS=$(date -Is)

set +e
restic backup /home /etc --exclude-file /etc/backup_excludes.txt --verbose
RC=$?
set -e

SIZE=0
if [ $RC -eq 0 ]; then
  # Get size of latest snapshot
  SIZE=$(restic stats latest --mode raw-data --json | awk -F'[,:}]' '/"total_size"/{gsub(/ /,""); print $3; exit}')
fi
DUR=$(( $(date +%s) - START ))

echo "$TS,${SIZE:-0},$DUR,$RC" >> "$HIST"

# Optional: rotate/prune policy example
# restic forget --keep-daily 7 --keep-weekly 4 --keep-monthly 6 --prune

exit $RC

Make executable:

sudo chmod +x /usr/local/sbin/ai-backup-run.sh

Scheduler wrapper: /usr/local/sbin/ai-backup-schedule.sh:

#!/usr/bin/env bash
set -euo pipefail
WHEN=$( /usr/local/sbin/ai-backup-pick-window.py )
echo "/usr/local/sbin/ai-backup-run.sh" | at -t "$WHEN"
echo "Scheduled backup at $WHEN"

sudo chmod +x /usr/local/sbin/ai-backup-schedule.sh

Run the scheduler once per day (e.g., 00:05):

sudo tee /etc/cron.d/ai-backup-scheduler >/dev/null <<'EOF'
5 0 * * * root /usr/local/sbin/ai-backup-schedule.sh
EOF

Result: Your backup is queued for the quietest hour in the next 24 hours, re-evaluated daily as conditions change.

Project 2: Anomaly detection — Catch bad backups before they bite

We’ll score the latest backup’s size and duration using Isolation Forest. If it’s anomalous, we’ll run restic check (or your tool’s verifier) and log a warning.

Script: /usr/local/sbin/ai-backup-anomaly.py

#!/usr/bin/env python3
import os, sys
import pandas as pd
import numpy as np
from sklearn.ensemble import IsolationForest
from datetime import datetime

HIST = "/var/lib/ai-backup/backup_history.csv"

def main():
    if not os.path.exists(HIST):
        print("no_history")
        sys.exit(0)

    df = pd.read_csv(HIST, parse_dates=["ts"])
    if len(df) < 10:
        print("insufficient_data")
        sys.exit(0)

    # Use numeric features
    X = df[["size_bytes","duration_s"]].fillna(0).astype(float)
    # Basic normalization (optional)
    Xn = (X - X.mean()) / (X.std().replace(0,1))

    iso = IsolationForest(contamination=0.1, random_state=42)
    iso.fit(Xn)

    last = Xn.tail(1).values
    pred = iso.predict(last)[0]  # -1 anomalous, 1 normal
    score = iso.decision_function(last)[0]

    status = "anomaly" if pred == -1 else "ok"
    print(f"{status},{score:.4f}")

if __name__ == "__main__":
    main()

Hook this into the end of your backup run by appending to /usr/local/sbin/ai-backup-run.sh just after writing backup_history.csv:

# Anomaly detection and auto-verification
AD=$(/usr/local/sbin/ai-backup-anomaly.py || true)
if echo "$AD" | grep -q '^anomaly'; then
  logger -t ai-backup "Anomalous backup detected ($AD). Running 'restic check'."
  restic check || logger -t ai-backup "Restic check reported issues."
fi

Swap restic check for borg check if you’re using Borg.

Result: Spikes, dips, or odd durations trigger verification immediately, not days later.

Project 3: AI-assisted tiering — Decide what goes local vs. cloud

Goal: Classify files into “Tier A” (fast local) and “Tier B” (cheap cloud) using simple features (size, age, extension). Then sync with rsync (local) and rclone (cloud).

1. Export file metadata

ROOT=/data
OUT=/var/lib/ai-backup/files.csv
find "$ROOT" -type f -printf '%p,%s,%T@,%f\n' > "$OUT"
# columns: path,size_bytes,mtime_epoch,filename

2. Train + predict tiers

Save /usr/local/sbin/ai-backup-tiering.py:

#!/usr/bin/env python3
import os, sys
import pandas as pd
import numpy as np
from sklearn.linear_model import LogisticRegression

SRC = "/var/lib/ai-backup/files.csv"
TIER_A_LIST = "/var/lib/ai-backup/tierA.list"
TIER_B_LIST = "/var/lib/ai-backup/tierB.list"

HIGH_VALUE_EXT = set(["doc","docx","xls","xlsx","ods","odt","ppt","pptx","pdf","txt","md","csv","sql","yml","yaml","json"])

def main():
    if not os.path.exists(SRC):
        print("files_csv_missing", file=sys.stderr)
        sys.exit(1)
    df = pd.read_csv(SRC, names=["path","size","mtime","fname"])
    df["ext"] = df["fname"].str.lower().str.rsplit(".", n=1).str[-1]
    df["age_days"] = (pd.Timestamp.now().timestamp() - df["mtime"]) / 86400.0

    # Heuristic labels for bootstrap training
    df["label"] = 0
    df.loc[(df["ext"].isin(HIGH_VALUE_EXT)) | (df["age_days"] < 14), "label"] = 1

    # Features
    X = pd.DataFrame({
        "log_size": np.log1p(df["size"].astype(float)),
        "age_days": df["age_days"].astype(float),
        "is_high_ext": df["ext"].isin(HIGH_VALUE_EXT).astype(int)
    })
    y = df["label"].astype(int)

    # Train quick model
    if len(df) >= 50 and y.nunique() > 1:
        model = LogisticRegression(max_iter=200)
        model.fit(X, y)
        proba = model.predict_proba(X)[:,1]
    else:
        # Not enough data: use heuristic probability
        proba = 0.7*X["is_high_ext"] + 0.3*(X["age_days"] < 14).astype(float)

    df["tierA_score"] = proba
    # Threshold: tuneable. Start at 0.5
    df["tier"] = np.where(df["tierA_score"] >= 0.5, "A", "B")

    df.loc[df["tier"]=="A","path"].to_csv(TIER_A_LIST, index=False, header=False)
    df.loc[df["tier"]=="B","path"].to_csv(TIER_B_LIST, index=False, header=False)

    print(f"tierA: { (df['tier']=='A').sum() } files")
    print(f"tierB: { (df['tier']=='B').sum() } files")

if __name__ == "__main__":
    main()

sudo chmod +x /usr/local/sbin/ai-backup-tiering.py

3. Sync tiers

Configure rclone once:

rclone config
# create remote name like: myremote:s3bucket/path

Sync commands:

# Local fast tier (e.g., RAID/NAS)
RSYNC_DEST=/backup/local-fast
mkdir -p "$RSYNC_DEST"
rsync -a --files-from=/var/lib/ai-backup/tierA.list / "$RSYNC_DEST"

# Cloud/cold tier (via rclone)
RCLONE_REMOTE="myremote:s3bucket/cold"
rclone copy --files-from /var/lib/ai-backup/tierB.list / "$RCLONE_REMOTE"

Automate tiering daily:

sudo tee /etc/cron.d/ai-backup-tiering >/dev/null <<'EOF'
30 1 * * * root ROOT=/data OUT=/var/lib/ai-backup/files.csv \
  bash -lc 'find "$ROOT" -type f -printf "%p,%s,%T@,%f\n" > "$OUT" && /usr/local/sbin/ai-backup-tiering.py && \
  rsync -a --files-from=/var/lib/ai-backup/tierA.list / /backup/local-fast && \
  rclone copy --files-from /var/lib/ai-backup/tierB.list / myremote:s3bucket/cold'
EOF

Result: High-value/recent files land on fast local storage; cold/large/old stuff streams to cloud—automatically.

Real-world notes and tips

  • Start simple, then tighten: Run the predictive scheduler alongside your existing schedule for a week and compare.

  • Visibility: Tail /var/log/syslog or journalctl -t ai-backup for anomaly alerts.

  • Trust but verify: Keep a weekly restic check or borg check --repair in a separate cron just in case.

  • Security: Store secrets in root-owned files with 600 permissions; avoid embedding credentials in scripts.

  • Dry runs: Use --dry-run flags for rsync and rclone while testing tiering.

Conclusion and next steps

You don’t need a GPU or a data science team to make your backups smarter. With a few small Python helpers and your usual Bash glue, you can:

  • Schedule backups when the system is actually idle.

  • Detect and respond to suspicious runs immediately.

  • Place data on the right storage tier to balance cost and recovery speed.

Your next step: 1) Install the prerequisites. 2) Enable Project 1 (predictive scheduling) and let it run for a few days. 3) Turn on anomaly detection, then add tiering once you trust the flow.

Have questions, improvements, or want a systemd-timer variant? Reach out—and if you ship this, share your tweaks so others can benefit.

Related posts