Merge deployed native Alloy and backup recovery changes [skip ci]

.gitignore

@@ -16,3 +16,5 @@ __pycache__/
 *.env
 .env
 *.bak
+._*
+.DS_Store

actions/backup/run.py

@@ -7,12 +7,44 @@ non-`backup` subcommand passes straight through to restic, so init/snapshots/for
 behave exactly as before."""
 import json
 import os
+from pathlib import Path
+import time
 from engine.descriptor import Meta, Arg
 
 METADATA = Meta(summary="Run the resticontainer label-driven backup (init-if-needed, backup, forget --prune).",
                 args=[Arg("telegram_bot_url", "Telegram Bot URL for notifications")])
 TITLE = "Restic Host Backup"
 
+def write_backup_status(path, state, timestamp):
+    path = Path(path)
+    previous = json.loads(path.read_text()) if path.exists() else {}
+    status = {"state": state, "updated_at": timestamp,
+              "last_success": timestamp if state == "success" else previous.get("last_success", 0)}
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_name(path.name + f".{os.getpid()}.tmp")
+    temporary.write_text(json.dumps(status) + "\n")
+    temporary.chmod(0o644)
+    temporary.replace(path)
+
+def restore_containers(ctx, container_ids):
+    """Recover previously running backup targets, retrying namespace dependents."""
+    pending = list(container_ids)
+    if not pending:
+        return
+    for _ in range(len(pending) + 1):
+        cp = ctx.sys.run(["docker", "inspect", *pending], capture=True)
+        stopped = [c["Id"] for c in json.loads(cp.stdout) if not c["State"]["Running"]]
+        if not stopped:
+            return
+        progress = False
+        for container_id in stopped:
+            cp = ctx.sys.run(["docker", "start", container_id], check=False, capture=True)
+            progress = progress or cp.returncode == 0
+        if not progress:
+            raise RuntimeError("Backup left containers stopped: " + ", ".join(stopped))
+        pending = stopped
+    raise RuntimeError("Backup containers did not stay running: " + ", ".join(pending))
+
 def run(ctx, args):
     log, node, url = ctx.log, ctx.node.name, args.telegram_bot_url
     core = ctx.paths.core
@@ -42,6 +74,8 @@ def run(ctx, args):
         except (ValueError, TypeError):
             return None
 
+    status_path = Path("/var/lib/apex/backup-status.json")
+    write_backup_status(status_path, "running", time.time())
     try:
         log.info("Checking restic repository availability...")
         if restic("snapshots", check=False).returncode == 0:
@@ -50,8 +84,16 @@ def run(ctx, args):
             ctx.notify.error(TITLE, url, "restic init failed"); raise SystemExit(1)
         before = host_snapshot_ids()
         log.info("Backing up `restic.*`-labelled services (resticontainer discovers paths)...")
-        if restic("backup", "--host", host, "--tag", tag, "--compression", "max",
+        targets = ctx.sys.run(["docker", "ps", "-q", "--no-trunc",
-                  check=False).returncode != 0:
+                               "--filter", "label=restic.enable=true",
+                               "--filter", "label=restic.backup.stop=true"], capture=True).stdout.split()
+        try:
+            result = restic("backup", "--host", host, "--tag", tag, "--compression", "max",
+                            check=False)
+        finally:
+            # Resticontainer may start a network-namespace dependent before its owner.
+            restore_containers(ctx, targets)
+        if result.returncode != 0:
             ctx.notify.error(TITLE, url, "restic backup failed"); raise SystemExit(1)
         # Guard: the label-driven backup is a silent no-op if no service carries the
         # `restic.enable` label (resticontainer finds zero paths and skips restic).
@@ -70,9 +112,12 @@ def run(ctx, args):
         log.success("Restic backup completed.")
         if not ctx.notify.success(TITLE, url, f"Restic backup + retention succeeded for {node}"):
             raise SystemExit(1)   # bash exited 1 when the success telegram failed to send
+        write_backup_status(status_path, "success", time.time())
     except SystemExit:
+        write_backup_status(status_path, "failed", time.time())
         raise
     except Exception as e:
+        write_backup_status(status_path, "failed", time.time())
         log.error(f"{type(e).__name__}: {e}")
         ctx.notify.error(TITLE, url, f"backup failed: {e}")
         raise SystemExit(1) from e

actions/configure/alloy.py

@@ -0,0 +1,77 @@
+"""Deploy the opt-in native Alloy service using shared host/guest configs."""
+import json
+from pathlib import Path
+import re
+import tempfile
+
+from engine.descriptor import Flag, Meta
+
+METADATA = Meta(summary="Configure native Alloy from configs/alloy/config.json.",
+                args=[Flag("--dry-run", "Validate the deployment plan without changes.")])
+
+
+def deployment(ctx):
+    settings = Path(ctx.paths.configs) / "alloy/config.json"
+    if not settings.exists():
+        return None
+    values = json.loads(settings.read_text())
+    role = values.get("role")
+    if role not in ("host", "guest"):
+        raise ValueError("Alloy role must be host or guest")
+    env_file = values.get("environment_file", "")
+    if not re.fullmatch(r"/[A-Za-z0-9_./-]+", env_file):
+        raise ValueError("Alloy environment_file must be an absolute path without whitespace")
+    source = Path(ctx.paths.commons) / "etc/alloy" / (
+        "native-host.alloy" if role == "host" else "native.alloy")
+    config = source.read_text()
+    extra = Path(ctx.paths.configs) / "alloy/extra.alloy"
+    if extra.is_file():
+        config += "\n" + extra.read_text()
+    unit = ("[Service]\nEnvironmentFile=" + env_file + "\n"
+            'ExecStart=\nExecStart=/usr/bin/alloy run --storage.path=/var/lib/alloy --server.http.listen-addr=127.0.0.1:12346 /etc/alloy/config.alloy\n'
+            + ("User=root\nGroup=root\n" if role == "guest" else "") +
+            "SupplementaryGroups=systemd-journal" + (" docker" if role == "guest" else "") + "\n")
+    return config, unit, env_file
+
+
+def run(ctx, args):
+    plan = deployment(ctx)
+    if plan is None:
+        ctx.log.info("Native Alloy is not configured for this node; skipping.")
+        return
+    config, unit, env_file = plan
+    if args.dry_run:
+        ctx.log.info("Native Alloy deployment plan validated; no changes made.")
+        return
+    ctx.sys.sudo(["test", "-f", env_file])
+    # Validate with the actual service environment without printing credentials.
+    validator = '''import os, stat, subprocess, sys
+from pathlib import Path
+# Environment files deliberately use plain KEY=value, with no shell evaluation.
+secret = Path(sys.argv[1])
+info = secret.stat()
+if info.st_uid != 0 or stat.S_IMODE(info.st_mode) & 0o077:
+    raise RuntimeError('Alloy environment file must be root-owned with mode 0600 or stricter')
+for line in secret.read_text().splitlines():
+    if not line.strip() or line.lstrip().startswith('#'): continue
+    key, value = line.split('=', 1)
+    os.environ[key] = value
+subprocess.run(['alloy', 'validate', sys.argv[2]], check=True)
+'''
+    with tempfile.TemporaryDirectory(prefix="apex-alloy-") as directory:
+        config_path = Path(directory) / "config.alloy"
+        unit_path = Path(directory) / "apex.conf"
+        config_path.write_text(config)
+        unit_path.write_text(unit)
+        ctx.sys.sudo(["python3", "-c", validator, env_file, str(config_path)])
+        ctx.sys.sudo(["install", "-d", "-m", "0755", "/etc/alloy",
+                      "/etc/systemd/system/alloy.service.d"])
+        ctx.sys.sudo(["install", "-m", "0644", str(config_path), "/etc/alloy/config.alloy"])
+        ctx.sys.sudo(["install", "-m", "0644", str(unit_path),
+                      "/etc/systemd/system/alloy.service.d/apex.conf"])
+    ctx.sys.sudo(["systemctl", "unmask", "alloy.service"])
+    ctx.sys.daemon_reload()
+    ctx.sys.sudo(["systemctl", "enable", "alloy.service"])
+    ctx.sys.sudo(["systemctl", "restart", "alloy.service"])
+    ctx.sys.sudo(["systemctl", "is-active", "--quiet", "alloy.service"])
+    ctx.log.success("Configured native Alloy.")

ci/packer/apex-debian-13.pkr.hcl

@@ -79,6 +79,7 @@ build {
     scripts = [
       "ci/packer/scripts/01-base-system.sh",
       "ci/packer/scripts/02-storage-docker.sh",
+      "ci/packer/scripts/02-alloy-package.sh",
       "ci/packer/scripts/03-bootstrap-helper.sh",
       "ci/packer/scripts/04-cleanup.sh"
     ]

ci/packer/scripts/02-alloy-package.sh

@@ -0,0 +1,18 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Service configuration and credentials are deployed later by apex configure/alloy.
+alloy_version=1.21.3
+alloy_sha256=dbbd705de8ab37dec0cbd6361a3fc7e754d6b2854b57ce672a83e6366199ba73
+[[ "$(dpkg --print-architecture)" == amd64 ]]
+alloy_deb=$(mktemp --suffix=.deb)
+trap 'rm -f "$alloy_deb"' EXIT
+curl --fail --location --retry 3 --output "$alloy_deb" \
+  "https://github.com/grafana/alloy/releases/download/v${alloy_version}/alloy-${alloy_version}-1.amd64.deb"
+printf '%s  %s\n' "$alloy_sha256" "$alloy_deb" | sha256sum --check --status
+systemctl mask alloy.service
+DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends "$alloy_deb"
+systemctl disable --now alloy.service
+systemctl unmask alloy.service
+! systemctl is-active --quiet alloy.service
+! systemctl is-enabled --quiet alloy.service

etc/alloy/native-host.alloy

@@ -0,0 +1,102 @@
+// Shared native Grafana Alloy configuration for the physical host.
+// Metrics and logs use the endpoints and credentials supplied through
+// APEX_PROM_PUSH_URL, APEX_LOKI_PUSH_URL, and APEX_OBS_BASIC_AUTH_USER/PASS.
+
+// ----------------------------------------------------------------------------
+// METRICS
+// ----------------------------------------------------------------------------
+
+// Native filesystem views; physical capacity is collected on the Proxmox host.
+prometheus.exporter.unix "node" {
+  enable_collectors = ["textfile"]
+  textfile {
+    directory = "/var/lib/alloy/textfile"
+  }
+  procfs_path = "/proc"
+  sysfs_path  = "/sys"
+  rootfs_path = "/"
+}
+
+// Container metrics (cadvisor-equivalent) via the docker socket.
+
+
+// Stamp a static `node` label onto each exporter's targets.
+discovery.relabel "node_metrics" {
+  targets = prometheus.exporter.unix.node.targets
+  rule {
+    action       = "replace"
+    target_label = "node"
+    replacement  = sys.env("APEX_NODE_HOST")
+  }
+  // Stamp a stable `instance` (the node's public host) instead of the default,
+  // which is the Alloy container id/address.
+  rule {
+    action       = "replace"
+    target_label = "instance"
+    replacement  = sys.env("APEX_NODE_HOST")
+  }
+}
+
+
+prometheus.scrape "node" {
+  targets         = discovery.relabel.node_metrics.output
+  forward_to      = [prometheus.remote_write.default.receiver]
+  scrape_interval = "30s"
+}
+
+
+// The configured Prometheus-compatible endpoint must accept remote writes.
+prometheus.remote_write "default" {
+  endpoint {
+    url = sys.env("APEX_PROM_PUSH_URL")
+    basic_auth {
+      username = sys.env("APEX_OBS_BASIC_AUTH_USER")
+      password = sys.env("APEX_OBS_BASIC_AUTH_PASS")
+    }
+    queue_config {
+      capacity             = 5000
+      max_shards           = 10
+      min_shards           = 1
+      max_samples_per_send = 1000
+      batch_send_deadline  = "5s"
+      min_backoff          = "100ms"
+      max_backoff          = "10s"
+    }
+  }
+  wal {
+    truncate_frequency = "30m"
+    min_keepalive_time = "5m"
+    max_keepalive_time = "30m"
+  }
+}
+
+// ----------------------------------------------------------------------------
+// LOGS
+// ----------------------------------------------------------------------------
+
+// systemd journal (host services, ufw, sshd, crowdsec, etc.).
+loki.source.journal "systemd" {
+  path    = "/var/log/journal"
+  max_age = "12h"
+  labels  = {
+    node = sys.env("APEX_NODE_HOST"),
+    job  = "systemd-journal",
+  }
+  forward_to = [loki.write.default.receiver]
+}
+
+// All docker containers' stdout/stderr.
+
+
+
+
+// Push logs to the central Loki.
+loki.write "default" {
+  endpoint {
+    url = sys.env("APEX_LOKI_PUSH_URL")
+    basic_auth {
+      username = sys.env("APEX_OBS_BASIC_AUTH_USER")
+      password = sys.env("APEX_OBS_BASIC_AUTH_PASS")
+    }
+  }
+}

etc/alloy/native.alloy

@@ -0,0 +1,128 @@
+// Shared native Grafana Alloy configuration for Docker guests.
+// Metrics and logs use the endpoints and credentials supplied through
+// APEX_PROM_PUSH_URL, APEX_LOKI_PUSH_URL, and APEX_OBS_BASIC_AUTH_USER/PASS.
+
+// ----------------------------------------------------------------------------
+// METRICS
+// ----------------------------------------------------------------------------
+
+// Native filesystem views; physical capacity is collected on the Proxmox host.
+
+
+// Container metrics (cadvisor-equivalent) via the docker socket.
+prometheus.exporter.cadvisor "containers" {
+  docker_host                  = "unix:///var/run/docker.sock"
+  store_container_labels       = false
+  allowlisted_container_labels = [
+    "com.docker.compose.project",
+    "com.docker.compose.service",
+  ]
+  docker_only                  = true
+  enabled_metrics              = [
+    "cpu",
+    "cpuLoad",
+    "diskIO",
+    "memory",
+    "network",
+    "oom_event",
+  ]
+}
+
+// Stamp a static `node` label onto each exporter's targets.
+
+discovery.relabel "container_metrics" {
+  targets = prometheus.exporter.cadvisor.containers.targets
+  rule {
+    action       = "replace"
+    target_label = "node"
+    replacement  = sys.env("APEX_NODE_HOST")
+  }
+  rule {
+    action       = "replace"
+    target_label = "instance"
+    replacement  = sys.env("APEX_NODE_HOST")
+  }
+}
+
+
+prometheus.scrape "containers" {
+  targets         = discovery.relabel.container_metrics.output
+  forward_to      = [prometheus.remote_write.default.receiver]
+  scrape_interval = "30s"
+}
+
+// The configured Prometheus-compatible endpoint must accept remote writes.
+prometheus.remote_write "default" {
+  endpoint {
+    url = sys.env("APEX_PROM_PUSH_URL")
+    basic_auth {
+      username = sys.env("APEX_OBS_BASIC_AUTH_USER")
+      password = sys.env("APEX_OBS_BASIC_AUTH_PASS")
+    }
+    queue_config {
+      capacity             = 5000
+      max_shards           = 10
+      min_shards           = 1
+      max_samples_per_send = 1000
+      batch_send_deadline  = "5s"
+      min_backoff          = "100ms"
+      max_backoff          = "10s"
+    }
+  }
+  wal {
+    truncate_frequency = "30m"
+    min_keepalive_time = "5m"
+    max_keepalive_time = "30m"
+  }
+}
+
+// ----------------------------------------------------------------------------
+// LOGS
+// ----------------------------------------------------------------------------
+
+// systemd journal (host services, ufw, sshd, crowdsec, etc.).
+loki.source.journal "systemd" {
+  path    = "/var/log/journal"
+  max_age = "12h"
+  labels  = {
+    node = sys.env("APEX_NODE_HOST"),
+    job  = "systemd-journal",
+  }
+  forward_to = [loki.write.default.receiver]
+}
+
+// All docker containers' stdout/stderr.
+discovery.docker "containers" {
+  host = "unix:///var/run/docker.sock"
+}
+discovery.relabel "container_logs" {
+  targets = discovery.docker.containers.targets
+  rule {
+    action        = "replace"
+    source_labels = ["__meta_docker_container_name"]
+    regex         = "/?(.*)"
+    target_label  = "container"
+  }
+  rule {
+    action       = "replace"
+    target_label = "node"
+    replacement  = sys.env("APEX_NODE_HOST")
+  }
+}
+loki.source.docker "containers" {
+  host          = "unix:///var/run/docker.sock"
+  targets       = discovery.relabel.container_logs.output
+  relabel_rules = discovery.relabel.container_logs.rules
+  forward_to    = [loki.write.default.receiver]
+}
+
+// Push logs to the central Loki.
+loki.write "default" {
+  endpoint {
+    url = sys.env("APEX_LOKI_PUSH_URL")
+    basic_auth {
+      username = sys.env("APEX_OBS_BASIC_AUTH_USER")
+      password = sys.env("APEX_OBS_BASIC_AUTH_PASS")
+    }
+  }
+}

tests/test_backup_run.py

@@ -0,0 +1,49 @@
+import json
+from pathlib import Path
+from types import SimpleNamespace
+import unittest
+import tempfile
+from unittest.mock import Mock
+
+from engine import overlay
+
+backup = overlay.load_module(str(Path(__file__).resolve().parents[1] / 'actions/backup/run.py'),
+                             'test_backup_action')
+
+
+class BackupRecovery(unittest.TestCase):
+    def test_failed_attempt_preserves_last_success_and_records_new_failure(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'status.json'
+            backup.write_backup_status(path, 'success', 100)
+            backup.write_backup_status(path, 'running', 200)
+            backup.write_backup_status(path, 'failed', 210)
+            status = json.loads(path.read_text())
+            self.assertEqual(status['last_success'], 100)
+            self.assertEqual(status['state'], 'failed')
+            self.assertEqual(status['updated_at'], 210)
+
+    def test_recovers_namespace_owner_before_dependent_without_starting_other_containers(self):
+        running = {'komf': False, 'komga': False, 'intentionally-stopped': False}
+        def docker(command, **kwargs):
+            if command[1] == 'inspect':
+                return SimpleNamespace(returncode=0, stdout=json.dumps([
+                    {'Id': name, 'State': {'Running': running[name]}} for name in command[2:]]))
+            name = command[2]
+            if name == 'komf' and not running['komga']:
+                return SimpleNamespace(returncode=1)
+            running[name] = True
+            return SimpleNamespace(returncode=0)
+        ctx = SimpleNamespace(sys=SimpleNamespace(run=docker), log=Mock())
+        backup.restore_containers(ctx, ['komf', 'komga'])
+        self.assertTrue(running['komf'])
+        self.assertTrue(running['komga'])
+        self.assertFalse(running['intentionally-stopped'])
+
+    def test_unrecoverable_container_prevents_success(self):
+        def docker(command, **kwargs):
+            return SimpleNamespace(returncode=0 if command[1] == 'inspect' else 1,
+                                   stdout='[{"Id":"broken","State":{"Running":false}}]')
+        ctx = SimpleNamespace(sys=SimpleNamespace(run=docker), log=Mock())
+        with self.assertRaisesRegex(RuntimeError, 'broken'):
+            backup.restore_containers(ctx, ['broken'])

tests/test_configure_alloy.py

@@ -0,0 +1,52 @@
+"""Native Alloy opt-in and validation boundaries."""
+import json
+from pathlib import Path
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import Mock
+
+from engine import overlay
+
+ROOT = Path(__file__).resolve().parents[1]
+alloy = overlay.load_module(str(ROOT / 'actions/configure/alloy.py'), 'test_alloy_action')
+
+
+class NativeAlloy(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.configs = Path(self.temp.name)
+        self.ctx = SimpleNamespace(paths=SimpleNamespace(configs=str(self.configs), commons=str(ROOT)),
+                                   sys=Mock(), log=Mock())
+
+    def configure(self, **settings):
+        (self.configs / 'alloy').mkdir(exist_ok=True)
+        (self.configs / 'alloy/config.json').write_text(json.dumps(settings))
+
+    def test_unconfigured_node_is_untouched(self):
+        alloy.run(self.ctx, SimpleNamespace(dry_run=False))
+        self.ctx.sys.sudo.assert_not_called()
+
+    def test_invalid_role_or_environment_path_cannot_change_service(self):
+        for role, path in [('unknown', '/etc/alloy/secrets'), ('host', '/tmp/file\nUser=root')]:
+            self.configure(role=role, environment_file=path)
+            with self.assertRaises(ValueError):
+                alloy.run(self.ctx, SimpleNamespace(dry_run=False))
+            self.ctx.sys.sudo.assert_not_called()
+
+    def test_dry_run_never_installs_or_starts(self):
+        self.configure(role='guest', environment_file='/etc/alloy/secrets')
+        alloy.run(self.ctx, SimpleNamespace(dry_run=True))
+        self.ctx.sys.sudo.assert_not_called()
+
+    def test_failed_config_validation_leaves_existing_service_untouched(self):
+        self.configure(role='host', environment_file='/etc/alloy/secrets')
+        def execute(command):
+            if command[0] == 'python3':
+                raise RuntimeError('invalid Alloy configuration')
+            self.assertEqual(command, ['test', '-f', '/etc/alloy/secrets'])
+        self.ctx.sys.sudo.side_effect = execute
+        with self.assertRaisesRegex(RuntimeError, 'invalid Alloy'):
+            alloy.run(self.ctx, SimpleNamespace(dry_run=False))
+        self.ctx.sys.daemon_reload.assert_not_called()