Repository: radiocosmology/alpenhorn

Issue:
Need a "node init" in client
We need some way to trigger "initialisation" of a node in the CLI (i.e. creation of the `ALPENHORN_NODE` file for a newly-created StorageNode.

Hints from issue comments:

Solution patch:
diff --git a/alpenhorn/cli/node/__init__.py b/alpenhorn/cli/node/__init__.py
index 3f46828cd..8b8651d56 100644
--- a/alpenhorn/cli/node/__init__.py
+++ b/alpenhorn/cli/node/__init__.py
@@ -1,32 +1,22 @@
 """Alpenhorn CLI interface for operations on `StorageNode`s."""
 
-import os
-import re
-import sys
-from collections import defaultdict
-
 import click
-import peewee as pw
-
-from ... import db
-from ...common import util
-from ...db import ArchiveAcq, ArchiveFile, ArchiveFileCopy, StorageGroup, StorageNode
 
 from .activate import activate
 from .autoclean import autoclean
 from .clean import clean
 from .create import create
 from .deactivate import deactivate
+from .init import init
 from .list import list_
 from .modify import modify
 from .rename import rename
+from .scan import scan
 from .show import show
 from .stats import stats
 from .sync import sync
 from .verify import verify
 
-RE_LOCK_FILE = re.compile(r"^\..*\.lock$")
-
 
 @click.group(context_settings={"help_option_names": ["-h", "--help"]})
 def cli():
@@ -38,275 +28,12 @@ def cli():
 cli.add_command(clean, "clean")
 cli.add_command(create, "create")
 cli.add_command(deactivate, "deactivate")
+cli.add_command(init, "init")
 cli.add_command(list_, "list")
 cli.add_command(modify, "modify")
 cli.add_command(rename, "rename")
+cli.add_command(scan, "scan")
 cli.add_command(show, "show")
 cli.add_command(stats, "stats")
 cli.add_command(sync, "sync")
 cli.add_command(verify, "verify")
-
-
-@cli.command()
-@click.argument("node_name", metavar="NODE")
-@click.option("-v", "--verbose", count=True)
-@click.option(
-    "--acq",
-    help="Limit import to specified acquisition directories.",
-    multiple=True,
-    default=None,
-)
-@click.option(
-    "--register-new", help="Register new files instead of ignoring them.", is_flag=True
-)
-@click.option("--dry", "-d", help="Dry run. Do not modify database.", is_flag=True)
-def scan(node_name, verbose, acq, register_new, dry):
-    """Scan the current directory for known acquisition files and add them into the database for NODE.
-
-    This command is useful for manually maintaining an archive where we cannot
-    run alpenhornd in the usual manner.
-    """
-    config_connect()
-
-    # Keep track of state as we process the files
-    added_files = []  # Files we have added to the database
-    corrupt_files = []  # Known files which are corrupt
-    registered_files = []  # Files already registered in the database
-    unknown_files = []  # Files not known in the database
-
-    known_acqs = []  # Directories which are known acquisitions
-    new_acqs = []  # Directories which were newly registered acquisitions
-    not_acqs = []  # Directories which were not known acquisitions
-
-    # Fetch a reference to the node
-    try:
-        this_node = StorageNode.select().where(StorageNode.name == node_name).get()
-    except pw.DoesNotExist:
-        click.echo("Unknown node:", node_name)
-        exit(1)
-
-    cwd = os.getcwd()
-    # Construct a dictionary of directories that might be acquisitions and the of
-    # list files that they contain
-    db_acqs = ArchiveAcq.select(ArchiveAcq.name)
-    acq_files = defaultdict(list)
-    if len(acq) == 0:
-        tops = [cwd]
-    else:
-        db_acqs = db_acqs.where(ArchiveAcq.name >> acq)
-        tops = []
-        for acq_name in acq:
-            acq_dir = os.path.join(this_node.root, acq_name)
-            if not os.path.isdir(acq_dir):
-                print(
-                    'Aquisition "%s" does not exist in this node. Ignoring.' % acq_name,
-                    file=sys.stderr,
-                )
-                continue
-            if acq_dir == cwd:
-                # the current directory is one of the limiting acquisitions, so
-                # we can ignore all others in the `--acq` list
-                tops = [acq_dir]
-                break
-            elif cwd.startswith(acq_dir):
-                # the current directory is inside one of the limiting
-                # acquisitions, so we can just walk its subtree
-                tops = [cwd]
-                break
-            elif acq_dir.startswith(cwd):
-                # the acquisition is inside the current directory, so we can
-                # just walk its subtree
-                tops.append(acq_dir)
-            else:
-                print(
-                    'Acquisition "%s" is outside the current directory and will be ignored.'
-                    % acq_name,
-                    file=sys.stderr,
-                )
-
-    for top in tops:
-        for d, ds, fs in os.walk(top):
-            d = os.path.relpath(d, this_node.root)
-            if d == ".":  # skip the node root directory
-                continue
-            acq_type_name = ac.AcqType.detect(d, this_node)
-            if acq_type_name:
-                _, acq_name = acq_type_name
-                if d == acq_name:
-                    # the directory is the acquisition
-                    acq_files[acq_name] += [
-                        f
-                        for f in fs
-                        if not RE_LOCK_FILE.match(f)
-                        and not os.path.isfile(os.path.join(d, ".{}.lock".format(f)))
-                    ]
-                if d.startswith(acq_name + "/"):
-                    # the directory is inside an acquisition
-                    acq_dirname = os.path.relpath(d, acq_name)
-                    acq_files[acq_name] += [
-                        (acq_dirname + "/" + f)
-                        for f in fs
-                        if not RE_LOCK_FILE.match(f)
-                        and not os.path.isfile(os.path.join(d, ".{}.lock".format(f)))
-                    ]
-            else:
-                not_acqs.append(d)
-
-    with click.progressbar(acq_files, label="Scanning acquisitions") as acq_iter:
-        for acq_name in acq_iter:
-            try:
-                acq = ArchiveAcq.select().where(ArchiveAcq.name == acq_name).get()
-                known_acqs.append(acq_name)
-
-                # Fetch lists of all files in this acquisition, and all
-                # files in this acq with local copies
-                file_names = [f.name for f in acq.files]
-                local_file_names = [
-                    f.name
-                    for f in acq.files.join(ArchiveFileCopy).where(
-                        ArchiveFileCopy.node == this_node
-                    )
-                ]
-            except pw.DoesNotExist:
-                if register_new:
-                    acq_type, _ = ac.AcqType.detect(acq_name, this_node)
-                    acq = ArchiveAcq(name=acq_name, type=acq_type)
-                    if not dry:
-                        # TODO: refactor duplication with auto_import.add_acq
-                        with db.database_proxy.atomic():
-                            # insert the archive record
-                            acq.save()
-                            # and generate the metadata table
-                            acq_type.acq_info.new(acq, this_node)
-
-                    new_acqs.append(acq_name)
-
-                    # Because it's a newly imported acquisition, all files within it are new also
-                    file_names = []
-                    local_file_names = []
-                else:
-                    not_acqs.append(acq_name)
-                    continue
-
-            for f_name in acq_files[acq_name]:
-                file_path = os.path.join(acq_name, f_name)
-
-                # Check if file exists in database
-                if not register_new and f_name not in file_names:
-                    unknown_files.append(file_path)
-                    continue
-
-                # Check if file is already registered on this node
-                if f_name in local_file_names:
-                    registered_files.append(file_path)
-                else:
-                    abs_path = os.path.join(this_node.root, file_path)
-                    if f_name in file_names:
-                        # it is a known file
-                        archive_file = (
-                            ArchiveFile.select()
-                            .where(ArchiveFile.name == f_name, ArchiveFile.acq == acq)
-                            .get()
-                        )
-
-                        # TODO: decide if, when the file is corrupted, we still
-                        # register the file as `has_file="X"` or just _continue_
-                        if os.path.getsize(abs_path) != archive_file.size_b:
-                            corrupt_files.append(file_path)
-                            continue
-                        else:
-                            if verbose > 2:
-                                print('Computing md5sum of "{}"'.format(f_name))
-                            md5sum = util.md5sum_file(abs_path, cmd_line=False)
-                            if md5sum != archive_file.md5sum:
-                                corrupt_files.append(file_path)
-                                continue
-                    else:
-                        # not a known file, register the new ArchiveFile instance
-                        file_type = ac.FileType.detect(f_name, acq, this_node)
-                        if not file_type:
-                            unknown_files.append(file_path)
-                            continue
-
-                        if verbose > 2:
-                            print('Computing md5sum of "{}"'.format(f_name))
-                        md5sum = util.md5sum_file(abs_path, cmd_line=False)
-                        size_b = os.path.getsize(abs_path)
-                        archive_file = ArchiveFile(
-                            name=f_name,
-                            acq=acq,
-                            type=file_type,
-                            size_b=size_b,
-                            md5sum=md5sum,
-                        )
-                        if not dry:
-                            archive_file.save()
-
-                    added_files.append(file_path)
-                    if not dry:
-                        copy_size_b = os.stat(abs_path).st_blocks * 512
-                        ArchiveFileCopy.create(
-                            file=archive_file,
-                            node=this_node,
-                            has_file="Y",
-                            wants_file="Y",
-                            size_b=copy_size_b,
-                        )
-
-    # now find the minimum unknown acqs paths that we can report
-    not_acqs_roots = []
-    last_acq_root = ""
-    for d in sorted(not_acqs):
-        common = os.path.commonprefix([last_acq_root, d])
-        if common == "":
-            for acq_name in known_acqs:
-                if acq_name.startswith(d):
-                    break
-            else:
-                for acq_name in new_acqs:
-                    if acq_name.startswith(d):
-                        break
-                else:
-                    not_acqs_roots.append(d)
-            last_acq_root = d
-
-    print("\n==== Summary ====")
-    print()
-    if register_new:
-        print("Registered %i new acquisitions" % len(new_acqs))
-    print("Added %i files" % len(added_files))
-    print()
-    print("%i corrupt files." % len(corrupt_files))
-    print("%i files already registered." % len(registered_files))
-    print("%i files not known" % len(unknown_files))
-    print("%i directories were not acquisitions." % len(not_acqs_roots))
-
-    if verbose > 0:
-        print()
-        if register_new:
-            print("New acquisitions:")
-            for an in sorted(new_acqs):
-                print(an)
-            print()
-
-        print("Added files:")
-        for fn in sorted(added_files):
-            print(fn)
-
-        print()
-
-    if verbose > 1:
-        print("Corrupt:")
-        for fn in sorted(corrupt_files):
-            print(fn)
-        print()
-
-        print("Unknown files:")
-        for fn in sorted(unknown_files):
-            print(fn)
-        print()
-
-        print("Unknown acquisitions:")
-        for fn in sorted(not_acqs_roots):
-            print(fn)
-        print()
diff --git a/alpenhorn/cli/node/create.py b/alpenhorn/cli/node/create.py
index d82e02940..d7ea0bdb5 100644
--- a/alpenhorn/cli/node/create.py
+++ b/alpenhorn/cli/node/create.py
@@ -4,7 +4,7 @@
 import json
 import peewee as pw
 
-from ...db import database_proxy, StorageGroup, StorageNode
+from ...db import database_proxy, StorageGroup, StorageNode, ArchiveFileImportRequest
 from ..options import cli_option, exactly_one, set_storage_type, set_io_config
 from ..cli import echo
 
@@ -50,6 +50,9 @@
     "Incompatible with --create-group",
 )
 @cli_option("host")
+@click.option(
+    "--init", is_flag=True, help='Initialise the new node (see "node init --help")'
+)
 @cli_option("io_config")
 @cli_option("io_var")
 @cli_option("max_total")
@@ -70,6 +73,7 @@ def create(
     field,
     group,
     host,
+    init,
     io_config,
     io_var,
     max_total,
@@ -131,7 +135,7 @@ def create(
             except pw.DoesNotExist:
                 raise click.ClickException("no such group: " + group)
 
-        StorageNode.create(
+        node = StorageNode.create(
             name=node_name,
             root=root,
             host=host,
@@ -149,6 +153,10 @@ def create(
             io_config=json.dumps(io_config) if io_config else None,
         )
 
-        if create_group:
-            echo(f'Created storage group "{node_name}".')
-        echo(f'Created storage node "{node_name}".')
+        # Create AFIR for node init
+        if init:
+            ArchiveFileImportRequest.create(node=node, path="ALPENHORN_NODE")
+
+    if create_group:
+        echo(f'Created storage group "{node_name}".')
+    echo(f'Created storage node "{node_name}".')
diff --git a/alpenhorn/cli/node/init.py b/alpenhorn/cli/node/init.py
new file mode 100644
index 000000000..54e5bc07e
--- /dev/null
+++ b/alpenhorn/cli/node/init.py
@@ -0,0 +1,36 @@
+"""alpenhorn node init command"""
+
+import click
+import peewee as pw
+
+from ...db import database_proxy, ArchiveFileImportRequest, StorageNode
+from ..cli import echo
+
+
+@click.command()
+@click.argument("name", metavar="NODE")
+def init(name):
+    """Initialise a known Storage Node.
+
+    This requests the daemon initialise the Node named NODE.  NODE must
+    already exist in the data index.  NODE must be active for the daemon
+    to act on this request.
+
+    What initialisation means can vary by node I/O class, but typically
+    it means creating the top-level "ALPENHORN_NODE" file.
+
+    Note: to create a new node, use "node create".  Node initialsation can
+    also happen at node creation time by using the "--init" flag with
+    "node create".
+    """
+
+    with database_proxy.atomic():
+        try:
+            node = StorageNode.get(name=name)
+        except pw.DoesNotExist:
+            raise click.ClickException("no such node: " + name)
+
+        # Add request
+        ArchiveFileImportRequest.create(node=node, path="ALPENHORN_NODE")
+
+    echo('Requested initialisation of Node "{name}".')
diff --git a/alpenhorn/cli/node/scan.py b/alpenhorn/cli/node/scan.py
new file mode 100644
index 000000000..6b4c6d170
--- /dev/null
+++ b/alpenhorn/cli/node/scan.py
@@ -0,0 +1,62 @@
+"""alpenhorn node scan command"""
+
+import click
+import pathlib
+import peewee as pw
+
+from ...db import ArchiveFileImportRequest, StorageNode, database_proxy
+from ..cli import echo
+
+
+@click.command()
+@click.argument("name", metavar="NODE")
+@click.argument("path", metavar="PATH", required=False, default=".")
+@click.option(
+    "--register-new",
+    "-n",
+    help="Register unknown files instead of ignoring them.",
+    is_flag=True,
+)
+def scan(name, path, register_new):
+    """Scan a node for files to import.
+
+    This command will submit a request to ask the daemon managing NODE to
+    scan PATH for new files to add to the Node.  if PATH is absolute, it
+    must start with the Node root path.  Otherwise, it's assumed PATH is
+    relative to Node root.  If not given, PATH defaults to "." (i.e. the scan
+    is performed on the entire Node).
+
+    By default, only files already registered in the data index will be imported
+    to NODE.  Unknown files will be skipped.  This can be changed, however, by
+    using the --register-new flag, which tells the daemon to create new records
+    for new files which are determined to be importable.
+    """
+
+    with database_proxy.atomic():
+        try:
+            node = StorageNode.get(name=name)
+        except pw.DoesNotExist:
+            raise click.ClickException("no such node: " + name)
+
+        # Strip node root, if absolute
+        path = pathlib.PurePath(path)
+
+        if path.is_absolute():
+            if node.root is None:
+                raise click.ClickException(
+                    f'PATH can\'t be absolute: Node "{node.name}" has no root.'
+                )
+
+            try:
+                path = path.relative_to(node.root)
+            except ValueError as e:
+                raise click.ClickException(
+                    f'absolute path "{path}" outside node root: ' + node.root
+                ) from e
+
+        # submit the request
+        ArchiveFileImportRequest.create(
+            node=node, path=str(path), recurse=True, register=register_new
+        )
+
+        echo(f'Added request for scan of "{path}" on Node "{node.name}".')
diff --git a/alpenhorn/daemon/auto_import.py b/alpenhorn/daemon/auto_import.py
index 96ade7099..a25852d1f 100644
--- a/alpenhorn/daemon/auto_import.py
+++ b/alpenhorn/daemon/auto_import.py
@@ -61,6 +61,8 @@ def import_file(
         If not None, req will be marked as complete if the import isn't skipped.
     """
 
+    path = pathlib.PurePath(path)
+
     # Occasionally the watchdog sends events on the node root directory itself. Skip these.
     if path == pathlib.PurePath(node.db.root):
         log.debug("Skipping import request of node root")
@@ -86,7 +88,9 @@ def import_file(
     # Ignore the node info file.  NB: this happens even on nodes which
     # don't use an ALPENHORN_NODE file, meaning even on those nodes,
     # this path is disallowed as a data file.
-    if str(path) == "ALPENHORN_NODE":
+    if pathlib.PurePath(node.db.root).joinpath(path) == pathlib.PurePath(
+        node.db.root
+    ).joinpath("ALPENHORN_NODE"):
         log.debug("ignoring ALPENHORN_NODE file during import")
         if req:
             log.info(f"Completed import request #{req.id}.")
diff --git a/alpenhorn/daemon/update.py b/alpenhorn/daemon/update.py
index d5dd6292d..1dcc8be55 100644
--- a/alpenhorn/daemon/update.py
+++ b/alpenhorn/daemon/update.py
@@ -271,24 +271,64 @@ def idle(self) -> bool:
         False otherwise."""
         return self._queue.fifo_size(self.io.fifo) == 0
 
-    def update_active(self) -> bool:
-        """Check if we are actually active on the host.
+    def check_init(self) -> bool:
+        """Check if the node is initialised.
 
         This is an I/O check, rather than a database check.
 
         Returns
         -------
-        active : bool
-            Whether or not the node is active.
+        initialised : bool
+            Whether or not the node is initialised.
         """
 
-        if self.db.active:
-            if self.io.check_active():
-                return True
-            else:
-                log.warning(f'Ignoring node "{self.name}": not ready for I/O.')
-        else:
+        def _async(
+            task: Task, node: UpdateableNode, req: ArchiveFileImportRequest
+        ) -> None:
+            """Task async to initialise `node`, if necessary.
+
+            Calls `req.complete()` if initialisation succeeds.
+            """
+
+            # recheck
+            if node.io.check_init():
+                log.info(f'Node "{node.name}" already initialised.')
+                req.complete()
+                return
+
+            # Run the init and check result
+            if node.io.init() and node.io.check_init():
+                log.info(f'Node "{node.name}" initialised.')
+                req.complete()
+                return
+
+            # Otherwise, fail, and don't complete req
+            log.warning(f'Initialisation failed for node "{node.name}".')
+
+        if not self.db.active:
             log.warning(f'Ignoring node "{self.name}": deactivated during update.')
+            return False
+
+        if self.io.check_init():
+            return True
+
+        # We're active but not initialised.  Is there a pending init request?
+        try:
+            req = ArchiveFileImportRequest.get(
+                node=self.db, path="ALPENHORN_NODE", completed=0
+            )
+            # Create a task to init the node
+            Task(
+                func=_async,
+                queue=self._queue,
+                key=self.io.fifo,
+                args=(self, req),
+                name=f'Init Node "{self.name}"',
+            )
+            log.info(f'Requesting init of node "{self.name}".')
+            # But then we still ignore it for now.
+        except pw.DoesNotExist:
+            log.warning(f'Ignoring node "{self.name}": not initialised.')
 
         return False
 
@@ -441,6 +481,15 @@ def update_import(self) -> None:
                 req.complete()
                 continue
 
+            if req.path == "ALPENHORN_NODE":
+                # This part of the update only runs if the node is already initialised,
+                # so this request can't be something we need to deal with here
+                log.info(
+                    f'Ignoring node init request for "{self.name}": already initialised.'
+                )
+                req.complete()
+                continue
+
             if req.recurse:
                 # If recursion was requested, run a scan on the path
                 Task(
@@ -848,7 +897,7 @@ def update_loop(queue: FairMultiFIFOQueue, pool: WorkerPool | EmptyPool) -> None
                 continue
 
             # Check if the node is actually active
-            if not node.update_active():
+            if not node.check_init():
                 del nodes[name]  # Not active
                 continue
 
diff --git a/alpenhorn/db/archive.py b/alpenhorn/db/archive.py
index a0209fda5..b8ff972e2 100644
--- a/alpenhorn/db/archive.py
+++ b/alpenhorn/db/archive.py
@@ -128,8 +128,8 @@ class ArchiveFileImportRequest(base_model):
     node : foreign key
         The StorageNode on which the import should happen
     path : string
-        The path to import.  If this is a directory and
-        recurse is True, it will be recursively scanned.
+        The path to import.  If this is a directory and recurse is True,
+        it will be recursively scanned.
     recurse : bool
         If True, recursively scan path for files, if path is a directory.
     register : bool
@@ -139,6 +139,15 @@ class ArchiveFileImportRequest(base_model):
         Set to true when the import request has completed.
     timestamp : datetime
         The UTC time when the request was made.
+
+    If `path` is the special value "ALPENHORN_NODE", then the request is
+    a node initialisation request, instead of a normal import request.
+    This only happens on active nodes which aren't already initialised:
+    an initialisation request on an already initialised node is ignored.
+
+    Note: How node initialisation works is dependent on the I/O class of
+    the node.  There is no requirement for initialisation to create an
+    ALPENHORN_NODE node file.
     """
 
     node = pw.ForeignKeyField(StorageNode, backref="import_requests")
diff --git a/alpenhorn/io/base.py b/alpenhorn/io/base.py
index 600bbc828..0d76bf611 100644
--- a/alpenhorn/io/base.py
+++ b/alpenhorn/io/base.py
@@ -16,11 +16,16 @@
 if TYPE_CHECKING:
     import os
     from collections.abc import Iterator
-    from ..acquisition import ArchiveFile
-    from ..archive import ArchiveFileCopy, ArchiveFileCopyRequest
-    from ..queue import FairMultiFIFOQueue
-    from ..storage import StorageNode, StorageGroup
-    from ..update import UpdateableNode
+    from ..db import (
+        ArchiveFile,
+        ArchiveFileCopy,
+        ArchiveFileCopyRequest,
+        ArchiveFileImportRequest,
+        StorageNode,
+        StorageGroup,
+    )
+    from ..scheduler import FairMultiFIFOQueue
+    from ..daemon.update import UpdateableNode
 
 log = logging.getLogger(__name__)
 
@@ -253,18 +258,17 @@ def check(self, copy: ArchiveFileCopy) -> None:
         """
         raise NotImplementedError("method must be re-implemented in subclass.")
 
-    def check_active(self) -> bool:
-        """Check that this is an active node.
+    def check_init(self) -> bool:
+        """Check that this node is initialised.
 
         This check should be done by inspecting the storage system, rather than
-        checking the database, because is meant to catch instances where the
-        "active" bit in the database is incorrect.
+        checking the database.
 
-        Returns True if the node is active, or False if inactive.
+        Returns True if the node is initialised, or False if not.
 
-        If this can't be determined, it should return self.node.active (i.e.
-        assume the database is correct), which is the default behaviour."""
-        return self.node.active
+        The default is to just return False (i.e. assume it's never initialised.)
+        """
+        return False
 
     def delete(self, copies: list[ArchiveFileCopy]) -> None:
         """Delete the ArchiveFileCopy list `copies` from the node.
@@ -325,6 +329,20 @@ def fits(self, size_b: int) -> bool:
         """
         raise NotImplementedError("method must be re-implemented in subclass.")
 
+    def init(self) -> bool:
+        """Initialise this node.
+
+        This method will only be called if `check_init` returns False.
+        If initialisation is successful, subsequent `check_init` calls should
+        return True.
+
+        Returns
+        -------
+        init_successful : bool
+            Did initialisation succeed?
+        """
+        raise NotImplementedError("method must be re-implemented in subclass.")
+
     def locked(self, path: os.PathLike) -> bool:
         """Is file `path` locked?
 
diff --git a/alpenhorn/io/default.py b/alpenhorn/io/default.py
index 4923c0218..d5fc88936 100644
--- a/alpenhorn/io/default.py
+++ b/alpenhorn/io/default.py
@@ -193,15 +193,15 @@ def check(self, copy: ArchiveFileCopy) -> None:
             name=f"Check file {copy.file.path} on {self.node.name}",
         )
 
-    def check_active(self) -> bool:
-        """Check that this is an active node.
+    def check_init(self) -> bool:
+        """Check that this node is initialised.
 
         Checks that the file `node.root`/ALPENHORN_NODE exists and
         contains the name of the node.
 
         Returns
         -------
-        active : bool
+        init : bool
             True if the `ALPENHORN_NODE` check succeeded;
             False otherwise.
         """
@@ -332,6 +332,29 @@ def fits(self, size_b: int) -> bool:
         """
         return self.reserve_bytes(size_b, check_only=True)
 
+    def init(self) -> bool:
+        """Initialise this node.
+
+        We do that by creating the ALPENHORN_NODE file, if it
+        doesn't already exit.
+        """
+
+        # Sanity check
+        if self.check_init():
+            return True
+
+        # Create file
+        try:
+            with open(
+                pathlib.Path(self.node.root).joinpath("ALPENHORN_NODE"), mode="wt"
+            ) as f:
+                f.write(self.node.name + "\n")
+        except IOError as e:
+            log.warning(f"Node initialistion failed: {e}")
+            return False
+
+        return True
+
     def locked(self, path: os.PathLike) -> bool:
         """Is file `path` locked?
 
diff --git a/alpenhorn/io/lustrehsm.py b/alpenhorn/io/lustrehsm.py
index 89f317ffc..79178b37d 100644
--- a/alpenhorn/io/lustrehsm.py
+++ b/alpenhorn/io/lustrehsm.py
@@ -440,13 +440,12 @@ def _async(task: Task, node_io: LustreHSMNodeIO, copy: ArchiveFileCopy) -> None:
         else:
             log.debug(f"Skipping check of {copy.path}: restore in progress.")
 
-    def check_active(self) -> bool:
-        """Check that this is an active node.
+    def check_init(self) -> bool:
+        """Check that this node is initialised.
 
-        There's no ALPENHORN_NODE file on HSM,
-        so this just returns `self.node.active`.
+        There's no ALPENHORN_NODE file on HSM, so just return True.
         """
-        return self.node.active
+        return True
 
     def exists(self, path: pathlib.PurePath) -> bool:
         """Does `path` exist?
@@ -488,6 +487,15 @@ def fits(self, size_b: int) -> bool:
         """
         return True
 
+    def init(self) -> bool:
+        """Initialise this node.
+
+        This should never be called, because check_init always returns
+        True, but in case it does, we implement it anyways.
+        """
+        # Initialisaiton is a no-op
+        return True
+
     def open(self, path: os.PathLike | str, binary: bool = True) -> IO:
         """Open the file specified by `path` for reading.
 


Test patch:
diff --git a/tests/cli/node/test_create.py b/tests/cli/node/test_create.py
index 791105a84..e528138a9 100644
--- a/tests/cli/node/test_create.py
+++ b/tests/cli/node/test_create.py
@@ -2,7 +2,7 @@
 
 import json
 import pytest
-from alpenhorn.db import StorageGroup, StorageNode
+from alpenhorn.db import StorageGroup, StorageNode, ArchiveFileImportRequest
 
 
 def test_no_group(clidb, cli):
@@ -265,3 +265,15 @@ def test_bad_min_avail(clidb, cli):
 
     cli(2, ["node", "create", "TEST", "--create-group", "--min-avail=-1"])
     cli(2, ["node", "create", "TEST", "--create-group", "--min-avail=none"])
+
+
+def test_init(clidb, cli):
+    """Test create with --init"""
+
+    cli(0, ["node", "create", "TEST", "--init", "--create-group"])
+
+    node = StorageNode.get(name="TEST")
+    afir = ArchiveFileImportRequest.get(id=1)
+
+    assert afir.node == node
+    assert afir.path == "ALPENHORN_NODE"
diff --git a/tests/cli/node/test_init.py b/tests/cli/node/test_init.py
new file mode 100644
index 000000000..2616b29f1
--- /dev/null
+++ b/tests/cli/node/test_init.py
@@ -0,0 +1,36 @@
+"""Test CLI: alpenhorn node init"""
+
+from alpenhorn.db import StorageGroup, StorageNode, ArchiveFileImportRequest, utcnow
+
+
+def test_no_node(clidb, cli):
+    """Test a missing NODE name"""
+
+    group = StorageGroup.create(name="Group")
+    StorageNode.create(name="Node", group=group)
+
+    cli(2, ["node", "init"])
+
+
+def test_bad_node(clidb, cli):
+    """Test a bad node name"""
+
+    group = StorageGroup.create(name="Group")
+    StorageNode.create(name="Node", group=group)
+
+    cli(1, ["node", "init", "MISSING"])
+
+
+def test_init(clidb, cli):
+    """Test the default invocation."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="Node", group=group, root="/root")
+
+    cli(0, ["node", "init", "Node"])
+
+    # Request created
+    afir = ArchiveFileImportRequest.get(id=1)
+
+    assert afir.node == node
+    assert afir.path == "ALPENHORN_NODE"
diff --git a/tests/cli/node/test_scan.py b/tests/cli/node/test_scan.py
new file mode 100644
index 000000000..bfa0189b2
--- /dev/null
+++ b/tests/cli/node/test_scan.py
@@ -0,0 +1,130 @@
+"""Test CLI: alpenhorn node scan"""
+
+from alpenhorn.db import StorageGroup, StorageNode, ArchiveFileImportRequest, utcnow
+
+
+def test_no_node(clidb, cli):
+    """Test a missing NODE name"""
+
+    group = StorageGroup.create(name="Group")
+    StorageNode.create(name="Node", group=group)
+
+    cli(2, ["node", "scan"])
+
+
+def test_bad_node(clidb, cli):
+    """Test a bad node name"""
+
+    group = StorageGroup.create(name="Group")
+    StorageNode.create(name="Node", group=group)
+
+    cli(1, ["node", "scan", "MISSING"])
+
+
+def test_default(clidb, cli):
+    """Test the default invocation."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="Node", group=group, root="/root")
+
+    cli(0, ["node", "scan", "Node"])
+
+    # Request created
+    afir = ArchiveFileImportRequest.get(id=1)
+
+    assert afir.node == node
+    assert afir.path == "."
+    assert afir.recurse == 1
+    assert afir.register == 0
+
+
+def test_relpath(clidb, cli):
+    """Test a relative PATH."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="Node", group=group, root="/root")
+
+    cli(0, ["node", "scan", "Node", "relative/path"])
+
+    # Request created
+    afir = ArchiveFileImportRequest.get(id=1)
+
+    assert afir.node == node
+    assert afir.path == "relative/path"
+    assert afir.recurse == 1
+    assert afir.register == 0
+
+
+def test_relpath(clidb, cli):
+    """Test a relative PATH."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="Node", group=group, root="/root")
+
+    cli(0, ["node", "scan", "Node", "relative/path"])
+
+    # Request created
+    afir = ArchiveFileImportRequest.get(id=1)
+
+    assert afir.node == node
+    assert afir.path == "relative/path"
+    assert afir.recurse == 1
+    assert afir.register == 0
+
+
+def test_abspath(clidb, cli):
+    """Test an absolute PATH."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="Node", group=group, root="/root")
+
+    cli(0, ["node", "scan", "Node", "/root/absolute/path"])
+
+    # Request created
+    afir = ArchiveFileImportRequest.get(id=1)
+
+    assert afir.node == node
+    assert afir.path == "absolute/path"
+    assert afir.recurse == 1
+    assert afir.register == 0
+
+
+def test_bad_abspath(clidb, cli):
+    """Test an invalid absolute PATH."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="Node", group=group, root="/root")
+
+    cli(1, ["node", "scan", "Node", "/absolute/path"])
+
+    # No request created
+    assert ArchiveFileImportRequest.select().count() == 0
+
+
+def test_abspath_no_root(clidb, cli):
+    """Test an absolute PATH but NODE has no root."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="Node", group=group, root=None)
+
+    cli(1, ["node", "scan", "Node", "/absolute/path"])
+
+    # No request created
+    assert ArchiveFileImportRequest.select().count() == 0
+
+
+def test_new(clidb, cli):
+    """Test --register-new."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="Node", group=group, root=None)
+
+    cli(0, ["node", "scan", "Node", "--register-new"])
+
+    # Request created
+    afir = ArchiveFileImportRequest.get(id=1)
+
+    assert afir.node == node
+    assert afir.path == "."
+    assert afir.recurse == 1
+    assert afir.register == 1
diff --git a/tests/daemon/test_update_node.py b/tests/daemon/test_update_node.py
index bae748940..7c0704d8a 100644
--- a/tests/daemon/test_update_node.py
+++ b/tests/daemon/test_update_node.py
@@ -121,27 +121,57 @@ def test_idle(unode, queue):
     assert unode.idle is True
 
 
-def test_update_active(unode):
-    """Test UpdateableNode.update_active."""
+def test_check_init(unode):
+    """Test UpdateableNode.check_init."""
 
     # Starts out active
     unode.db.active = True
     unode.db.save()
     assert unode.db.active
 
-    # Pretend node is actually active
-    with patch.object(unode.io, "check_active", lambda: True):
-        assert unode.update_active()
+    # Pretend node is actually initialised
+    with patch.object(unode.io, "check_init", lambda: True):
+        assert unode.check_init()
     assert unode.db.active
     assert StorageNode.select(StorageNode.active).limit(1).scalar()
 
-    # Pretend node is actually not active
-    with patch.object(unode.io, "check_active", lambda: False):
-        assert not unode.update_active()
+    # Pretend node is actually not initialised
+    with patch.object(unode.io, "check_init", lambda: False):
+        assert not unode.check_init()
     assert unode.db.active
     assert StorageNode.select(StorageNode.active).limit(1).scalar()
 
 
+def test_update_do_init(xfs, unode, queue, archivefileimportrequest):
+    """Test handling an init request"""
+
+    # the req
+    req = archivefileimportrequest(node=unode.db, path="ALPENHORN_NODE")
+
+    # Activate node
+    unode.db.active = True
+    unode.db.save()
+
+    xfs.create_dir("/node")
+
+    # Call check_init
+    unode.check_init()
+
+    # Task is queued
+    assert queue.qsize == 1
+
+    # Run task
+    task, fifo = queue.get()
+    task()
+    queue.task_done(fifo)
+
+    # Node is initialised
+    assert unode.io.check_init() is True
+
+    # Request is complete
+    assert ArchiveFileImportRequest.get(id=req.id).completed == 1
+
+
 def test_update_free_space(unode):
     """Test UpdateableNode.update_free_space."""
 
@@ -523,3 +553,19 @@ def test_update_import_scan(unode, queue, archivefileimportrequest):
 
     # Mocked scan was called
     mock.assert_called_with(task, unode, queue, pathlib.Path("import/path"), True, afir)
+
+
+def test_update_import_alpenhorn_node(unode, queue, archivefileimportrequest):
+    """Test update_import() doesn't try to import ALPENHORN_NODE.
+
+    Even if we try to trick it...
+    """
+
+    afir = archivefileimportrequest(
+        path="./ALPENHORN_NODE", node=unode.db, recurse=False, register=True
+    )
+
+    unode.update_import()
+
+    # nothing queued
+    assert queue.qsize == 0
diff --git a/tests/io/test_defaultnode.py b/tests/io/test_defaultnode.py
index 738e2f9ba..f50d0346e 100644
--- a/tests/io/test_defaultnode.py
+++ b/tests/io/test_defaultnode.py
@@ -14,17 +14,33 @@ def test_bytes_avail(xfs, unode):
     assert unode.io.bytes_avail() == 10000.0
 
 
-def test_check_active(xfs, unode):
-    """test DefaultNodeIO.check_active()"""
+def test_check_init(xfs, unode):
+    """test DefaultNodeIO.check_init()"""
 
-    assert unode.io.check_active() is False
+    assert unode.io.check_init() is False
 
     xfs.create_file("/node/ALPENHORN_NODE", contents="not-node")
-    assert unode.io.check_active() is False
+    assert unode.io.check_init() is False
 
     with open("/node/ALPENHORN_NODE", mode="w") as f:
         f.write("simplenode")
-    assert unode.io.check_active() is True
+    assert unode.io.check_init() is True
+
+
+def test_init(xfs, unode, queue):
+    """test DefaultNodeIO.init()"""
+
+    # Somewhere for I/O to happen
+    xfs.create_dir("/node")
+
+    # Not initialised
+    assert unode.io.check_init() is False
+
+    # Run init
+    assert unode.io.init() is True
+
+    # Now initialised
+    assert unode.io.check_init() is True
 
 
 def test_exists(unode, xfs):