monitoring: publish reports for delete, undelete and transfer

borg delete, borg undelete and borg transfer now publish a monitoring report as
the last action while the store is open (skipped on --dry-run), with host/user
defaulting to the local machine like prune. Stats record the number of archives
deleted / undeleted / transferred (plus considered/skipped and transferred
bytes). These appear as their own units in borg monitor (command=delete /
undelete / transfer).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: ThomasWaldmann

src/borg/archiver/delete_cmd.py

@@ -1,6 +1,7 @@
 import logging
 
 from ._common import with_repository
+from .. import monitoring
 from ..constants import *  # NOQA
 from ..helpers import format_archive, CommandError, bin_to_hex, archivename_validator
 from ..helpers.argparsing import ArgumentParser
@@ -32,7 +33,7 @@ def do_delete(self, args, repository):
                 "or just delete the whole repository (might be much faster)."
             )
 
-        deleted = False
+        deleted_count = 0
         logger_list = logging.getLogger("borg.output.list")
         for i, archive_info in enumerate(archive_infos, 1):
             name, id, hex_id = archive_info.name, archive_info.id, bin_to_hex(archive_info.id)
@@ -46,17 +47,24 @@ def do_delete(self, args, repository):
             except KeyError:
                 self.print_warning(f"Archive {name} {hex_id} not found ({i}/{count}).")
             else:
-                deleted = True
+                deleted_count += 1
                 if self.output_list:
                     msg = "Would delete: {} ({}/{})" if dry_run else "Deleted archive: {} ({}/{})"
                     logger_list.info(msg.format(archive_formatted, i, count))
         if dry_run:
             logger.info("Finished dry-run.")
-        elif deleted:
+        elif deleted_count:
             manifest.write()
             self.print_warning('Done. Run "borg compact" to free space.', wc=None)
         else:
             self.print_warning("Aborted.", wc=None)
+        if not dry_run:
+            monitoring.publish_command_report(
+                repository,
+                manifest.key,
+                "delete",
+                stats={"archives_deleted": deleted_count, "archives_considered": count},
+            )
         return
 
     def build_parser_delete(self, subparsers, common_parser, mid_common_parser):

src/borg/archiver/transfer_cmd.py

@@ -1,4 +1,5 @@
 from ._common import with_repository, with_other_repository, Highlander
+from .. import monitoring
 from ..archive import Archive, cached_hash, DownloadPipeline
 from ..chunkers import get_chunker
 from ..constants import *  # NOQA
@@ -186,6 +187,9 @@ def do_transfer(self, args, *, repository, manifest, cache, other_repository=Non
 
         upgrader = UpgraderCls(cache=cache, args=args)
 
+        transferred_count = 0
+        skipped_count = 0
+        transferred_size = 0
         for archive_info in archive_infos:
             name, id, ts = archive_info.name, archive_info.id, archive_info.ts
             id_hex, ts_str = bin_to_hex(id), ts.isoformat()
@@ -201,9 +205,11 @@ def do_transfer(self, args, *, repository, manifest, cache, other_repository=Non
                 # Useful for Borg 1.x -> 2 transfers; we have unique names in Borg 1.x.
                 # Also useful for Borg 2 -> 2 transfers with metadata changes (ID changes).
                 print(f"{name} {ts_str}: archive is already present in destination repo, skipping.")
+                skipped_count += 1
             elif not dry_run and manifest.archives.exists_name_and_id(name, id):
                 # Useful for Borg 2 -> 2 transfers without changes (ID stays the same)
                 print(f"{name} {id_hex}: archive is already present in destination repo, skipping.")
+                skipped_count += 1
             else:
                 if not dry_run:
                     print(f"{name} {ts_str} {id_hex}: copying archive to destination repo...")
@@ -253,6 +259,8 @@ def do_transfer(self, args, *, repository, manifest, cache, other_repository=Non
                         archive.stats.show_progress(final=True)
                     additional_metadata = upgrader.upgrade_archive_metadata(metadata=other_archive.metadata)
                     archive.save(additional_metadata=additional_metadata)
+                    transferred_count += 1
+                    transferred_size += transfer_size
                     print(
                         f"{name} {ts_str} {id_hex}: finished. "
                         f"transfer_size: {format_file_size(transfer_size)} "
@@ -267,6 +275,19 @@ def do_transfer(self, args, *, repository, manifest, cache, other_repository=Non
                         f"present_size: {format_file_size(present_size)}"
                     )
 
+        if not dry_run:
+            monitoring.publish_command_report(
+                repository,
+                manifest.key,
+                "transfer",
+                stats={
+                    "archives_transferred": transferred_count,
+                    "archives_skipped": skipped_count,
+                    "archives_considered": count,
+                    "transferred_size": transferred_size,
+                },
+            )
+
     def build_parser_transfer(self, subparsers, common_parser, mid_common_parser):
         from ._common import process_epilog
         from ._common import define_archive_filters_group

src/borg/archiver/undelete_cmd.py

@@ -1,6 +1,7 @@
 import logging
 
 from ._common import with_repository
+from .. import monitoring
 from ..constants import *  # NOQA
 from ..helpers import format_archive, CommandError, bin_to_hex, archivename_validator
 from ..helpers.argparsing import ArgumentParser
@@ -29,7 +30,7 @@ def do_undelete(self, args, repository):
         if not args.name and not args.match_archives and args.first == 0 and args.last == 0:
             raise CommandError("Aborting: if you really want to undelete all archives, please use -a 'sh:*'.")
 
-        undeleted = False
+        undeleted_count = 0
         logger_list = logging.getLogger("borg.output.list")
         for i, archive_info in enumerate(archive_infos, 1):
             name, id, hex_id = archive_info.name, archive_info.id, bin_to_hex(archive_info.id)
@@ -39,17 +40,24 @@ def do_undelete(self, args, repository):
             except KeyError:
                 self.print_warning(f"Archive {name} {hex_id} not found ({i}/{count}).")
             else:
-                undeleted = True
+                undeleted_count += 1
                 if self.output_list:
                     msg = "Would undelete: {} ({}/{})" if dry_run else "Undeleted archive: {} ({}/{})"
                     logger_list.info(msg.format(format_archive(archive_info), i, count))
         if dry_run:
             logger.info("Finished dry-run.")
-        elif undeleted:
+        elif undeleted_count:
             manifest.write()
             self.print_warning("Done.", wc=None)
         else:
             self.print_warning("Aborted.", wc=None)
+        if not dry_run:
+            monitoring.publish_command_report(
+                repository,
+                manifest.key,
+                "undelete",
+                stats={"archives_undeleted": undeleted_count, "archives_considered": count},
+            )
         return
 
     def build_parser_undelete(self, subparsers, common_parser, mid_common_parser):

src/borg/testsuite/archiver/monitor_cmd_test.py

@@ -108,6 +108,40 @@ def test_prune_publishes_its_own_report(archivers, request, monkeypatch):
     assert prune["stats"]["archives_kept"] == 1
 
 
+def test_delete_and_undelete_publish_reports(archivers, request, monkeypatch):
+    archiver = request.getfixturevalue(archivers)
+    create_regular_file(archiver.input_path, "file1", contents=b"some data")
+    cmd(archiver, "repo-create", RK_ENCRYPTION)
+    cmd(archiver, "create", "a1", "input")
+    cmd(archiver, "create", "a2", "input")
+    cmd(archiver, "delete", "a2")
+    cmd(archiver, "undelete", "a2")
+    monkeypatch.setenv("BORG_MONITORING_KEY", _monitoring_key(archiver))
+
+    entries = _entries(archiver)
+    assert {"delete", "undelete"} <= set(entries)
+    assert entries["delete"]["report"]["stats"]["archives_deleted"] == 1
+    assert entries["undelete"]["report"]["stats"]["archives_undeleted"] == 1
+
+
+def test_transfer_publishes_report(archivers, request, monkeypatch):
+    from .transfer_cmd_test import setup_repos
+
+    archiver = request.getfixturevalue(archivers)
+    with setup_repos(archiver, monkeypatch) as other_repo1:
+        create_regular_file(archiver.input_path, "file1", contents=b"some data")
+        cmd(archiver, "create", "arch1", "input")
+        cmd(archiver, "create", "arch2", "input")
+    cmd(archiver, "transfer", other_repo1)
+    monkeypatch.setenv("BORG_MONITORING_KEY", _monitoring_key(archiver))
+
+    entries = _entries(archiver)
+    assert "transfer" in entries
+    stats = entries["transfer"]["report"]["stats"]
+    assert stats["archives_transferred"] == 2
+    assert stats["archives_considered"] == 2
+
+
 def test_keep_evicts_old_objects(archivers, request, monkeypatch):
     archiver = request.getfixturevalue(archivers)
     create_regular_file(archiver.input_path, "file1", contents=b"some data")