Merge pull request #53 from flyingcircusio/DIR-770-status-in-directory

Call command on backup completion

changelog.d/20230908_002924_jb_DIR_770_status_in_directory.rst

@@ -0,0 +1 @@
+- Call configured script on (scheduler triggered) backup completion

doc/backy.conf.example

@@ -1,6 +1,7 @@
 global:
     base-dir: /my/backydir
     worker-limit: 3
+    backup-completed-callback: /path/to/script.sh
 schedules:
     default:
         daily:
@@ -37,8 +38,8 @@ jobs:
     test03:
         schedule: hourly
         source:
-            type: flyingcircus,
+            type: flyingcircus
             consul_acl_token: a211c244-846b-11e5-999b-081196cf15b4
-            vm: test03,
+            vm: test03
             pool: test
             image: test03.root

doc/man-backy.rst

@@ -281,6 +281,10 @@ config
         Maximum number of concurrent processes spawned by the scheduler.
         Defaults to 1 (no parallel backups).
 
+    backup-completed-callback
+        Command/Script to invoke after the scheduler successfully completed a backup.
+        The first argument is the job name. The output of `backy status --yaml` is available on stdin.
+
     status-file
         Path to a YAML status dump which is regularly updated by the scheduler
         and evaluated by **backy check**. Defaults to `{base-dir}/status`.

lib.nix

@@ -73,9 +73,12 @@ in
   checks = {
     pytest = runCommand "pytest" {
       nativeBuildInputs = [ poetryEnv ];
+      src = ./.;
     } ''
+      unpackPhase
+      cd *-source
       export BACKY_CMD=${poetryApplication}/bin/backy
-      cd ${./.}
+      patchShebangs src
       pytest -vv -p no:cacheprovider --no-cov
       touch $out
     '';

src/backy/daemon.py

@@ -28,6 +28,7 @@ class BackyDaemon(object):
     # config defaults, will be overriden from config file
     worker_limit: int = 1
     base_dir: str
+    backup_completed_callback: Optional[str]
     status_file: str
     status_interval: int = 30
     telnet_addrs: str = "::1, 127.0.0.1"
@@ -63,6 +64,7 @@ def _read_config(self):
         g = self.config.get("global", {})
         self.worker_limit = int(g.get("worker-limit", type(self).worker_limit))
         self.base_dir = g.get("base-dir")
+        self.backup_completed_callback = g.get("backup-completed-callback")
         self.status_file = g.get("status-file", p.join(self.base_dir, "status"))
         self.status_interval = int(
             g.get("status-interval", type(self).status_interval)

src/backy/main.py

@@ -9,6 +9,7 @@
 import humanize
 import structlog
 import tzlocal
+import yaml
 from prettytable import PrettyTable
 from structlog.stdlib import BoundLogger
 
@@ -39,17 +40,20 @@ def __init__(self, path, log):
         self.path = path
         self.log = log
 
-    def status(self):
+    def status(self, yaml_: bool):
         b = backy.backup.Backup(self.path, self.log)
+        if yaml_:
+            print(yaml.safe_dump([r.to_dict() for r in b.clean_history]))
+            return
         total_bytes = 0
 
         tz = tzlocal.get_localzone()
         t = PrettyTable(
             [f"Date ({tz})", "ID", "Size", "Duration", "Tags", "Trust"]
         )
         t.align = "l"
-        t.align["Size"] = "r"
-        t.align["Durat"] = "r"
+        t.align["Size"] = "r"  # type: ignore
+        t.align["Durat"] = "r"  # type: ignore
 
         for r in b.history:
             total_bytes += r.stats.get("bytes_written", 0)
@@ -209,6 +213,7 @@ def setup_argparser():
 Show backup status. Show inventory and summary information.
 """,
     )
+    p.add_argument("--yaml", dest="yaml_", action="store_true")
     p.set_defaults(func="status")
 
     # upgrade

src/backy/revision.py

@@ -94,7 +94,12 @@ def materialize(self):
 
     def write_info(self):
         self.log.debug("writing-info", tags=", ".join(self.tags))
-        metadata = {
+        with SafeFile(self.info_filename, encoding="utf-8") as f:
+            f.open_new("wb")
+            yaml.safe_dump(self.to_dict(), f)
+
+    def to_dict(self):
+        return {
             "uuid": self.uuid,
             "backend_type": self.backend_type,
             "timestamp": self.timestamp,
@@ -105,9 +110,6 @@ def write_info(self):
             "trust": self.trust.value,
             "tags": list(self.tags),
         }
-        with SafeFile(self.info_filename, encoding="utf-8") as f:
-            f.open_new("wb")
-            yaml.safe_dump(metadata, f)
 
     def distrust(self):
         self.log.info("distrusted")

src/backy/scheduler.py

@@ -184,6 +184,7 @@ async def run_forever(self):
                     await self.run_backup(next_tags)
                     await self.run_expiry()
                     await self.run_purge()
+                    await self.run_callback()
             except asyncio.CancelledError:
                 raise
             except Exception:
@@ -274,6 +275,62 @@ async def run_purge(self):
                 pass
             raise
 
+    async def run_callback(self):
+        if not self.daemon.backup_completed_callback:
+            self.log.debug("callback-not-configured")
+            return
+
+        self.log.info("callback-started")
+        read, write = os.pipe()
+        backy_proc = await asyncio.create_subprocess_exec(
+            BACKY_CMD,
+            "-b",
+            self.path,
+            "-l",
+            self.logfile,
+            "status",
+            "--yaml",
+            stdin=subprocess.DEVNULL,
+            stdout=write,
+            stderr=subprocess.DEVNULL,
+        )
+        os.close(write)
+        callback_proc = await asyncio.create_subprocess_exec(
+            self.daemon.backup_completed_callback,
+            self.name,
+            stdin=read,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        os.close(read)
+        try:
+            stdout, stderr = await callback_proc.communicate()
+            return_code1 = await backy_proc.wait()
+            self.log.info(
+                "callback-finished",
+                return_code1=return_code1,
+                return_code2=callback_proc.returncode,
+                subprocess_pid1=backy_proc.pid,
+                subprocess_pid2=callback_proc.pid,
+                stdout=stdout.decode() if stdout else None,
+                stderr=stderr.decode() if stderr else None,
+            )
+        except asyncio.CancelledError:
+            self.log.warning(
+                "callback-cancelled",
+                subprocess_pid1=backy_proc.pid,
+                subprocess_pid2=callback_proc.pid,
+            )
+            try:
+                backy_proc.terminate()
+            except ProcessLookupError:
+                pass
+            try:
+                callback_proc.terminate()
+            except ProcessLookupError:
+                pass
+            raise
+
     def start(self):
         assert self._task is None
         self._task = self.daemon.loop.create_task(

src/backy/tests/test_callback.sh

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+cat > "${1}.callback_stdin"

src/backy/tests/test_daemon.py

@@ -5,9 +5,11 @@
 import re
 import signal
 import time
+from pathlib import Path
 from unittest import mock
 
 import pytest
+import yaml
 
 from backy import utils
 from backy.backends.chunked import ChunkedFileBackend
@@ -31,6 +33,7 @@ async def daemon(tmpdir, event_loop, log):
     base-dir: {base_dir}
     status-interval: 1
     telnet_port: 1234
+    backup-completed-callback: {Path(__file__).parent / "test_callback.sh"}
 schedules:
     default:
         daily:
@@ -143,7 +146,23 @@ async def test_run_backup(daemon, log):
     with backend.open("r") as f:
         assert f.read() == b"I am your father, Luke!"
 
-    daemon.terminate()
+
+async def test_run_callback(daemon, log):
+    job = daemon.jobs["test01"]
+
+    await job.run_backup({"manual:asdf"})
+    await job.run_callback()
+
+    with open("test01.callback_stdin", "r") as f:
+        r = yaml.safe_load(f)[0]
+        # see directory api before changing this
+        assert isinstance(r["uuid"], str)
+        assert isinstance(r["trust"], str)
+        assert isinstance(r["timestamp"], datetime.datetime)
+        assert isinstance(r["tags"][0], str)
+        assert isinstance(r["stats"]["bytes_written"], int)
+        assert isinstance(r["stats"]["duration"], float)
+        # assert isinstance(r["location"], str)
 
 
 def test_spread(daemon):
@@ -295,6 +314,7 @@ async def failing_coroutine(*args, **kw):
     monkeypatch.setattr(job, "_wait_for_deadline", null_coroutine)
     monkeypatch.setattr(job, "run_expiry", null_coroutine)
     monkeypatch.setattr(job, "run_purge", null_coroutine)
+    monkeypatch.setattr(job, "run_callback", null_coroutine)
     monkeypatch.setattr(job, "run_backup", failing_coroutine)
 
     # This patch causes a single run through the generator loop.

src/backy/tests/test_main.py

@@ -88,7 +88,7 @@ def test_call_status(capsys, backup, argv, monkeypatch):
         Ellipsis(
             """\
 (<backy.main.Command object at 0x...>,)
-{}
+{'yaml_': False}
 """
         )
         == out
@@ -97,7 +97,7 @@ def test_call_status(capsys, backup, argv, monkeypatch):
         Ellipsis(
             """\
 ... D command/invoked                args='... -v -b ... status'
-... D command/parsed                 func='status' func_args={}
+... D command/parsed                 func='status' func_args={'yaml_': False}
 ... D command/successful             \n\
 """
         )
@@ -241,7 +241,7 @@ def do_raise(*args, **kw):
         Ellipsis(
             """\
 ... D command/invoked                args='... -l ... -b ... status'
-... D command/parsed                 func='status' func_args={}
+... D command/parsed                 func='status' func_args={'yaml_': False}
 ... E command/failed                 exception_class='builtins.RuntimeError' exception_msg='test'
 exception>\tTraceback (most recent call last):
 exception>\t  File ".../src/backy/main.py", line ..., in main
@@ -262,7 +262,7 @@ def test_commands_wrapper_status(backup, tmpdir, capsys, clock, tz_berlin, log):
     revision.timestamp = backy.utils.now()
     revision.materialize()
 
-    commands.status()
+    commands.status(yaml_=False)
     out, err = capsys.readouterr()
 
     assert err == ""
@@ -276,3 +276,37 @@ def test_commands_wrapper_status(backup, tmpdir, capsys, clock, tz_berlin, log):
 1 revisions containing 0 Bytes data (estimated)
 """
     )
+
+
+def test_commands_wrapper_status_yaml(
+    backup, tmpdir, capsys, clock, tz_berlin, log
+):
+    commands = backy.main.Command(str(tmpdir), log)
+
+    revision = Revision(backup, log, "1")
+    revision.timestamp = backy.utils.now()
+    revision.stats["duration"] = 3.5
+    revision.stats["bytes_written"] = 42
+    revision.materialize()
+    revision2 = Revision(backup, log, "2")  # ignored
+    revision2.materialize()
+
+    commands.status(yaml_=True)
+    out, err = capsys.readouterr()
+
+    assert err == ""
+    assert (
+        out
+        == """\
+- backend_type: chunked
+  parent: ''
+  stats:
+    bytes_written: 42
+    duration: 3.5
+  tags: []
+  timestamp: 2015-09-01 07:06:47+00:00
+  trust: trusted
+  uuid: '1'
+
+"""
+    )