Repository: radiocosmology/alpenhorn

Issue:
Limited run-then-exit mode for the daemon
> Actually, maybe we should have an option for the daemon which tells it to run once, wait for I/O to finish and then exit.  Then people could run the daemon as an on-demand foreground process, in these sorts of cases, rather than forcing it to run as a long-running background process.

_Originally posted by @ketiltrout in https://github.com/radiocosmology/alpenhorn/issues/229#issuecomment-2477711216_
            

Discussion:

Patch:
diff --git a/alpenhorn/daemon/entry.py b/alpenhorn/daemon/entry.py
index db96b7c..e67ba58 100755
--- a/alpenhorn/daemon/entry.py
+++ b/alpenhorn/daemon/entry.py
@@ -33,12 +33,25 @@ sys.excepthook = log_exception
     default=None,
     metavar="FILE",
 )
+@click.option(
+    "once",
+    "--exit-after-update",
+    "-o",
+    "--once",
+    is_flag=True,
+    help="Run the update loop once, wait for updates to complete, and then exit.",
+)
 @version_option
 @help_config_option
-def entry(conf):
-    """Alpenhorn data management daemon.
+def entry(conf, once):
+    """Alpenhornd: data management daemon.
+
+    The alpenhorn daemon can be used to manage Storage Nodes.  See the alpenhorn
+    documentation for details on how to run the daemon.
 
-    This daemon can be used to manage Storage Nodes.
+    By default, the daemon will keep running until killed, but you can instead tell
+    it to run only a single update pass and then exit after updates have completed
+    by using the "--exit-after-update" flag.
     """
 
     # Initialise alpenhorn
@@ -65,7 +78,7 @@ def entry(conf):
 
     # Enter main loop
     try:
-        update.update_loop(queue, wpool)
+        update.update_loop(queue, wpool, once)
     # Catch keyboard interrupt
     except KeyboardInterrupt:
         log.info("Exiting due to SIGINT")
diff --git a/alpenhorn/daemon/update.py b/alpenhorn/daemon/update.py
index 6600926..48c4f15 100644
--- a/alpenhorn/daemon/update.py
+++ b/alpenhorn/daemon/update.py
@@ -846,7 +846,9 @@ class UpdateableGroup(updateable_base):
             self.io.idle_update()
 
 
-def update_loop(queue: FairMultiFIFOQueue, pool: WorkerPool | EmptyPool) -> None:
+def update_loop(
+    queue: FairMultiFIFOQueue, pool: WorkerPool | EmptyPool, once: bool
+) -> None:
     """Main loop of alepnhornd.
 
     This is the main update loop for the alpenhorn daemon.
@@ -857,9 +859,12 @@ def update_loop(queue: FairMultiFIFOQueue, pool: WorkerPool | EmptyPool) -> None
         the task manager
     pool : WorkerPool
         the pool of worker threads (may be empty)
+    once : bool
+        If True, only run the loop once, wait for the queue to empty,
+        and then exit.  If False, loop forever.
 
-    The daemon cycles through the update loop until it is terminated in
-    one of three ways:
+    If `once` is false, the daemon cycles through the update loop until it is
+    terminated in one of three ways:
 
     - receiving SIGINT (AKA KeyboardInterrupt).  This causes a clean exit.
     - a global abort caused by an uncaught exception in a worker thread.  This
@@ -1017,12 +1022,33 @@ def update_loop(queue: FairMultiFIFOQueue, pool: WorkerPool | EmptyPool) -> None
             f"{queue.inprogress_size} in-progress on {len(pool)} workers"
         )
 
-        # Avoid looping too fast.
-        remaining = config.config["daemon"]["update_interval"] - loop_time
-        if remaining > 0:
-            global_abort.wait(remaining)  # Stops waiting if a global abort is triggered
+        if once:
+            # If we're in Exit-after-update mode, wait for updates to complete
+            # and then return
+            first_time = True
+            while True:
+                if queue.qsize + queue.inprogress_size + queue.deferred_size == 0:
+                    log.info("Update complete.  Exiting.")
+                    return
+
+                if first_time:
+                    first_time = False
+                    log.info("Waiting for updates to complete.")
+
+                # Wait a bit
+                global_abort.wait(config.config["daemon"]["update_interval"])
+                log.info(
+                    f"Tasks: {queue.qsize} queued, {queue.deferred_size} deferred, "
+                    f"{queue.inprogress_size} in-progress on {len(pool)} workers"
+                )
+        else:
+            # Not in EAU mode.  Avoid looping too fast.
+            remaining = config.config["daemon"]["update_interval"] - loop_time
+            if remaining > 0:
+                # Stops waiting if a global abort is triggered
+                global_abort.wait(remaining)
 
-    # Warn on exit
+    # Warn on abnormal exit
     log.warning("Exiting due to global abort")
 
 


Test patch:
diff --git a/tests/daemon/test_entry.py b/tests/daemon/test_entry.py
index ada9b75..4bd77b6 100644
--- a/tests/daemon/test_entry.py
+++ b/tests/daemon/test_entry.py
@@ -114,8 +114,7 @@ def e2e_db(xfs, clidb_noinit, hostname):
         host=hostname,
         active=True,
         avail_gb=2000.0 / 2**30,
-        # This is mostly ignored
-        io_config='{"quota_group": "qgroup", "headroom": 10}',
+        io_config='{"quota_group": "qgroup", "headroom": 10, "restore_wait": 1}',
     )
     sf1 = StorageNode.create(
         name="sf1", group=nlgrp, root="/sf1", host=hostname, active=True
@@ -304,7 +303,7 @@ def e2e_config(xfs, hostname, clidb_uri):
         ],
         "database": {"url": "sqlite:///?database=" + urlquote(clidb_uri) + "&uri=true"},
         "logging": {"level": "debug"},
-        "daemon": {"num_workers": 0},
+        "daemon": {"num_workers": 0, "update_interval": 1},
     }
 
     # Put it in a file
@@ -318,17 +317,18 @@ def e2e_config(xfs, hostname, clidb_uri):
         "/nl1/acq1/release.me": "restored",
     }
 )
-def test_cli(e2e_db, e2e_config, mock_lfs, mock_rsync, loop_once):
+@pytest.mark.lfs_hsm_restore_result("restore")
+def test_cli(e2e_db, e2e_config, mock_lfs, mock_rsync):
     runner = CliRunner()
 
-    result = runner.invoke(entry, catch_exceptions=False)
+    result = runner.invoke(entry, ["--once"], catch_exceptions=False)
 
     assert result.exit_code == 0
 
     # Check HSM
     lfs = mock_lfs(quota_group="qgroup")
     assert lfs.hsm_state("/nl2/acq1/correct.me") == lfs.HSM_RESTORED
-    assert lfs.hsm_state("/nl1/acq1/restore.me") == lfs.HSM_RESTORING
+    assert lfs.hsm_state("/nl1/acq1/restore.me") == lfs.HSM_RESTORED
     assert lfs.hsm_state("/nl1/acq1/release.me") == lfs.HSM_RELEASED
 
     # Check results
@@ -347,9 +347,9 @@ def test_cli(e2e_db, e2e_config, mock_lfs, mock_rsync, loop_once):
     correctme = ArchiveFile.get(name="correct.me")
     assert ArchiveFileCopy.get(file=correctme).ready
 
-    # restore.me is not ready (because the main loop only ran once).
+    # restore.me is ready
     restoreme = ArchiveFile.get(name="restore.me")
-    assert not ArchiveFileCopy.get(file=restoreme).ready
+    assert ArchiveFileCopy.get(file=restoreme).ready
 
     # pull.me has been pulled
     tp1 = StorageNode.get(name="tp1")
diff --git a/tests/daemon/test_update.py b/tests/daemon/test_update.py
index 9f99264..69de7fb 100644
--- a/tests/daemon/test_update.py
+++ b/tests/daemon/test_update.py
@@ -53,7 +53,7 @@ def test_update_abort():
 
     # This should do nothing except exit, so passing
     # a couple of Nones shouldn't be a problem
-    update.update_loop(None, None)
+    update.update_loop(None, None, False)
 
     # Reset
     pool.global_abort.clear()
@@ -64,7 +64,7 @@ def test_update_no_nodes(
 ):
     """Test update_loop with no active nodes."""
 
-    update.update_loop(queue, emptypool)
+    update.update_loop(queue, emptypool, False)
 
     mock_serial_io.assert_called_once_with(queue)
 
@@ -83,7 +83,7 @@ def test_update_node_not_idle(
 
     xfs.create_file("/mocknode/ALPENHORN_NODE", contents="mocknode")
 
-    update.update_loop(queue, emptypool)
+    update.update_loop(queue, emptypool, False)
 
     # Node update started
     mockio.node.before_update.assert_called_once()
@@ -107,7 +107,7 @@ def test_update_node_idle(
 
     xfs.create_file("/mocknode/ALPENHORN_NODE", contents="mocknode")
 
-    update.update_loop(queue, emptypool)
+    update.update_loop(queue, emptypool, False)
 
     # Node update started
     mockio.node.before_update.assert_called_once()
@@ -130,7 +130,7 @@ def test_update_node_cancelled(
 
     xfs.create_file("/mocknode/ALPENHORN_NODE", contents="mocknode")
 
-    update.update_loop(queue, emptypool)
+    update.update_loop(queue, emptypool, False)
 
     # Node update started
     mockio.node.before_update.assert_called_once()
@@ -221,7 +221,7 @@ def test_update_group_not_idle_node(
 
     xfs.create_file("/mocknode/ALPENHORN_NODE", contents="mocknode")
 
-    update.update_loop(queue, emptypool)
+    update.update_loop(queue, emptypool, False)
 
     # Node update started
     mockio.node.before_update.assert_called_once()
@@ -258,7 +258,7 @@ def test_update_group_not_idle_group(
 
     xfs.create_file("/mocknode/ALPENHORN_NODE", contents="mocknode")
 
-    update.update_loop(queue, emptypool)
+    update.update_loop(queue, emptypool, False)
 
     # Idle update didn't happen
     mockio.group.idle_update.assert_not_called()
@@ -280,7 +280,7 @@ def test_update_group_idle(
 
     xfs.create_file("/mocknode/ALPENHORN_NODE", contents="mocknode")
 
-    update.update_loop(queue, emptypool)
+    update.update_loop(queue, emptypool, False)
 
     # Group update started
     mockio.group.before_update.assert_called_once()
@@ -305,7 +305,7 @@ def test_update_group_cancelled(
 
     xfs.create_file("/mocknode/ALPENHORN_NODE", contents="mocknode")
 
-    update.update_loop(queue, emptypool)
+    update.update_loop(queue, emptypool, False)
 
     # Group update started
     mockio.group.before_update.assert_called_once()