Repository: radiocosmology/alpenhorn

Issue:
Prometheus client for server?
Deploying alpenhorn2 to `wind` will require either:
* rewriting `alpenmonitor`, a Perl script which scrapes the alpenhorn log file to create prometheus metrics; or else:
* add a prometheus client into the alpenhorn server, which is probably a better idea, and long-term easier to maintain.

Discussion:

Solution patch:
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index e92351a1b..2c4d3b023 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -56,33 +56,14 @@ jobs:
       with:
         python-version: ${{ matrix.python-version }}
 
-    - name: Install pip dependencies
+    - name: Install with dependencies
       run: pip install -e .[test]
 
-    - name: Run tests
+    - name: Run tests without prometheus client
       run: pytest -v .
 
-  # run-docker-tests:
+    - name: Install prometheus client
+      run: pip install prometheus_client
 
-  #   runs-on: ubuntu-latest
-  #   steps:
-  #   - uses: actions/checkout@v2
-
-  #   - name: Set up Python 3.9
-  #     uses: actions/setup-python@v1
-  #     with:
-  #       python-version: 3.9
-
-  #   - name: Install pip dependencies
-  #     run: |
-  #       pip install -r requirements.txt
-  #       pip install -r test-requirements.txt
-  #       python setup.py develop
-
-  #   - name: Build docker image
-  #     run: docker build -f tests/docker/Dockerfile.alpenhorn -t alpenhorn .
-
-  #   - name: Run docker tests
-  #     run: pytest tests/docker/test_docker.py
-  #     env:
-  #       RUN_DOCKER_TESTS: true
+    - name: Run tests with prometheus client
+      run: pytest -v .
diff --git a/alpenhorn/common/config.py b/alpenhorn/common/config.py
index a3bf17234..a790900cf 100644
--- a/alpenhorn/common/config.py
+++ b/alpenhorn/common/config.py
@@ -135,12 +135,20 @@
         # job will rum forever if it doesn't exit; not recommended).
         pull_timeout_base: 300
         pull_bytes_per_second: 20000000
+
+        # Prometheus client port.  Setting this to a positive value will
+        # cause the daemon to start the prometheus client HTTP server to
+        # serve GET requests on that port (but only when not running in
+        # --exit-after-update mode).  If not set, or set to a non-positive
+        # value, then the prometheus client is not started.
+        prom_client_port: 8080
 """
 
 from __future__ import annotations
 
 import logging
 import os
+from typing import Any
 
 import yaml
 from click import ClickException
@@ -155,6 +163,7 @@
         "auto_import_interval": 30,
         "auto_verify_min_days": 7,
         "num_workers": 0,
+        "prom_client_port": 0,
         "serial_io_timeout": 900,
         "update_interval": 60,
     },
@@ -215,7 +224,7 @@ def load_config(cli_conf: os.PathLike, cli: bool) -> None:
         )
 
 
-def merge_dict_tree(a: dict, b: dict) -> dict:
+def merge_dict_tree(a: Any, b: Any) -> Any:
     """Merge two dictionaries recursively.
 
     The following rules applied:
diff --git a/alpenhorn/common/metrics.py b/alpenhorn/common/metrics.py
new file mode 100644
index 000000000..7bd70718c
--- /dev/null
+++ b/alpenhorn/common/metrics.py
@@ -0,0 +1,358 @@
+"""Alpenhorn interface to prometheus_client
+
+This module provides an interface between alpenhorn
+and the prometheus client.
+
+Metrics should be created and accessed through the `Metric` class, which
+is a light-weight wrapper around the prometheus_client's Counter and
+Gauge metrics.
+
+The `Metric` class adds a restriction that all instances of a metric of a
+given name have the same set of labels.  The `Metric` class also provides a
+mechanism to bind particular labels, meaning repetition of common label
+values is not necessary.
+
+If the `prometheus_client` module cannot be imported, most of this module
+does nothing.
+"""
+
+from __future__ import annotations
+
+from ..common import config
+
+try:
+    import prometheus_client as prom
+except ModuleNotFoundError:
+    prom = None
+
+# A dict of all the metrics we're using.  Keys are names.
+# Values are two-tuples with elements:
+#  * the prometheus_client metric object
+#  * the set of all labelnames
+_metrics = {}
+
+
+class Metric:
+    """A wrapper for prometheus_client metrics."""
+
+    def __init__(
+        self,
+        name: str,
+        description: str,
+        counter: bool = False,
+        unbound: list | tuple | set = (),
+        bound: dict = {},
+    ) -> None:
+        """Create a new metric.
+
+        Labels in the metric can be bound (set to a fixed value) or unbound
+        (with no fixed value).  Names of unbound labels are listed in `unbound`.
+        Bound labels with their values are set by the `bound` dict.
+
+        All instances of a metric of a given `name` must have the same set of
+        label names (but different instances may change which of those are bound
+        and which are unbound), They must also have the same type (i.e. the same
+        value for `counter`).  All instances with the same `name` control the same
+        underlying metric.
+
+        Newly-created metrics are implicitly set to zero, but, unless they have
+        no labels at all, they won't actually be reported until their value is
+        explicitly set/updated.
+
+        Parameters
+        ----------
+        name:
+            Name of the metric, excluding the initial `alpenhorn_`.
+        description:
+            A human-readable description of the data in the metric.
+        counter:
+            If True, create a Counter metric.  Otherwise a Gauge metric is created.
+        unbound:
+            A set (or list or tuple) of names of unbound labels.
+        bound:
+            A dict of label key-value pairs for bound labels.
+
+        Raises
+        ------
+        KeyError:
+            The same label name was specified in both `unbound` and `bound`.
+        TypeError:
+            An attempt was made to access an existing metric using the wrong
+            value for `counter`.
+        ValueError:
+            An attempt was made to access an existing metric using the wrong
+            set of label names.
+        """
+        global _metrics
+
+        # keys in "bound" can't also appear in "unbound"
+        for key in bound.keys():
+            if key in unbound:
+                raise KeyError(f'label "{key}" is both bound and unbound')
+
+        self._name = name
+        self._desc = description
+        self._unbound_labels = set(unbound)
+        self._bound_labels = bound
+        self._counter = counter
+
+        # We don't save any metrics, if we have no prometheus_client
+        if prom is None:
+            self._metric = None
+            return
+
+        # Metric type
+        _type = prom.Counter if counter else prom.Gauge
+
+        # Get or create the prom metric
+        if name in _metrics:
+            existing_metric, existing_labels = _metrics[name]
+            # Validate access: must be using the correct type and list of labels
+            if not isinstance(existing_metric, _type):
+                raise TypeError(f"wrong metric type for metric: {name}")
+            # Make sure labelnames is the same
+            if existing_labels != set(self.labelnames):
+                raise ValueError(
+                    f"wrong labels for metric.  Expected: {existing_labels}"
+                )
+            self._metric = existing_metric
+        else:
+            self._metric = _type(
+                "alpenhorn_" + name, description, labelnames=self.labelnames
+            )
+            _metrics[name] = (self._metric, set(self.labelnames))
+
+    def bind(self, **labels: str) -> Metric:
+        """Bind values to labels.
+
+        This method returns a _copy_ of the Metric with the currently unbound
+        labels newly bound to the values specified by the keywords.  Not all
+        unbound labels need be specified.  Unbound labels not specified remain
+        unbound in the returned copy.
+
+        The original Metric is left unchanged.
+
+        Raises
+        ------
+        TypeError:
+            A keyword parameter did not correspond to one of the unbound labels.
+        """
+        unbound = set(self._unbound_labels)
+        bound = self._bound_labels.copy()
+
+        # Bind the new label values
+        for key in labels.keys():
+            if key not in unbound:
+                raise TypeError(f'"{key}" is not an unbound label')
+            unbound.remove(key)
+            bound[key] = labels[key]
+
+        # Return the child
+        return Metric(
+            name=self._name,
+            description=self._desc,
+            counter=self._counter,
+            unbound=unbound,
+            bound=bound,
+        )
+
+    def _check_unbound_covered(self, labels: dict) -> None:
+        """Check cover of given labels.
+
+        Checks that `labels` provides a value for all unbound labels
+        (and nothing more).
+
+        Raises
+        ------
+        TypeError:
+            An extra label was included.
+        ValueError:
+            A label was missing.
+        """
+
+        keys = set(labels.keys())
+
+        missing_keys = self._unbound_labels - keys
+        if missing_keys:
+            raise ValueError("not bound: " + ", ".join(missing_keys))
+
+        extra_keys = keys - self._unbound_labels
+        if extra_keys:
+            raise TypeError("not unbound: " + ", ".join(extra_keys))
+
+    @property
+    def labelnames(self) -> list[str]:
+        """An ordered list of label names."""
+
+        return sorted(self._unbound_labels | set(self._bound_labels.keys()))
+
+    def labelvalues(self, **labels: str) -> list[str]:
+        """Return the list of label values.
+
+        Values for all unbound labels must be specified by keyword.
+
+        The returned list is guaranteed to be in the same order as the list
+        of label names given by `labelnames`.
+        """
+        self._check_unbound_covered(labels)
+
+        # Merge the labels
+        labels |= self._bound_labels
+
+        # Return values in the correct order
+        return [labels[key] for key in sorted(labels.keys())]
+
+    def _labelled_metric(self, labels: dict) -> prom.metrics.MetricWrapperBase:
+        """Returns the labelled metric.
+
+        Returns a `prometheus_client` child metric for the labelset resulting
+        from merging the unbound and bound labels.
+
+        Values for all unbound labels must be specified in the supplied `labels`
+        dict.
+        """
+        # Unlabelled metrics have no children, so we must return the parent
+        if not self._unbound_labels and not self._bound_labels:
+            return self._metric
+
+        self._check_unbound_covered(labels)
+
+        if self._metric:
+            return self._metric.labels(**labels, **self._bound_labels)
+        return None
+
+    def add(self, value: float, /, **labels: str) -> None:
+        """Add `value` to the metric.
+
+        For Counter metrics, `value` must be positive.
+
+        Values for all unbound labels must be specified in the supplied `labels`
+        dict.
+        """
+        if self._metric:
+            self._labelled_metric(labels).inc(value)
+
+    def inc(self, /, **labels: str) -> None:
+        """Increment the metric by one.
+
+        Values for all unbound labels must be specified in the supplied `labels`
+        dict.
+        """
+        self.add(1, **labels)
+
+    def dec(self, /, **labels: str) -> None:
+        """Decrement the metric by one.
+
+        Calling this metric on a counter will result in an error.
+
+        Values for all unbound labels must be specified in the supplied `labels`
+        dict.
+        """
+        # We'll let prometheus_client throw the exception for counters.
+        self.add(-1, **labels)
+
+    def set(self, value: float, /, **labels: str) -> None:
+        """Set the metric to `value`.
+
+        For counter metrics, the only allowed value is zero.
+
+        Values for all unbound labels must be specified in the supplied `labels`
+        dict.
+        """
+        if self._metric is None:
+            return
+
+        if self._counter:
+            # For counters, Metric.set(0) is converted into a `.reset()` call
+            if value:
+                raise ValueError("attempt to set counter to non-zero value")
+            self._labelled_metric(labels).reset()
+        else:
+            # Gauges have a .set() method.
+            self._labelled_metric(labels).set(value)
+
+    def remove(self, /, **labels: str) -> None:
+        """Remove a labelset from the metric.
+
+        Returns the child metric with the labelset resulting from merging the
+        unbound and bound labels.
+
+        Values for all unbound labels must be specified in the supplied `labels`
+        dict.
+
+        If no such metric exists, this function does nothing.
+
+        Paramters:
+        ----------
+        labels:
+            The key-value pairs of the unbound labels for the labelset.
+        """
+        if self._metric:
+            self._metric.remove(self.labelvalues(**labels))
+
+    def clear(self) -> None:
+        """Remove all labelsets from the metric.
+
+        The metric, per se, is not deleted.
+        """
+        if self._metric:
+            self._metric.clear()
+
+
+def by_name(name: str) -> Metric:
+    """Retrieve the pre-made Metric called `name`.
+
+    This function returns several pre-made Metric instances for use in cases
+    where there is no other natural place to define the metric.
+
+    Metrics returned by this function have no bound labels.  You may call
+    `.bind()` on the returned metric if you wish to bind some of them.
+
+    Parameters
+    ----------
+    name:
+        The name of the Metric to return
+
+    Raises
+    ------
+    ValueError:
+        No such Metric was found with the requested name.
+    """
+
+    if name == "requests_completed":
+        return Metric(
+            name,
+            "Count of completed requests",
+            counter=True,
+            unbound=("type", "result", "node", "group"),
+        )
+    if name == "transfers":
+        return Metric(
+            name,
+            "Count of transfer attempts",
+            counter=True,
+            unbound=("result", "node_from", "group_to"),
+        )
+
+    raise ValueError(f"no such metric: {name}")
+
+
+def start_promclient() -> None:
+    """Start the prometheus client
+
+    The client is only started if `daemon.prom_client_port`
+    is set to a positive value in the alpenhorn config.
+    """
+
+    # Get the port number.  If not found, we just return here.
+    try:
+        port = int(config.config["daemon"]["prom_client_port"])
+        if port <= 0:
+            return
+    except KeyError:
+        return
+
+    # Okay, we're good to start the http server, if we can
+    if prom is not None:
+        prom.disable_created_metrics()
+        prom.start_http_server(port)
diff --git a/alpenhorn/common/util.py b/alpenhorn/common/util.py
index efcca07e3..35c07dff1 100644
--- a/alpenhorn/common/util.py
+++ b/alpenhorn/common/util.py
@@ -12,6 +12,7 @@
 import click
 
 from . import config, extensions, logger
+from .metrics import Metric
 
 if TYPE_CHECKING:
     from collections.abc import Callable
@@ -314,7 +315,13 @@ def md5sum_file(filename: str, hr: bool = True) -> str | None:
     --------
     http://stackoverflow.com/questions/1131220/get-md5-hash-of-big-files-in-python
     """
-    return asyncio.run(_md5sum_file(filename, hr))
+    metric = Metric("hash_running_count", "Count of in-progress MD5 hashing")
+
+    metric.inc()
+    result = asyncio.run(_md5sum_file(filename, hr))
+    metric.dec()
+
+    return result
 
 
 def get_hostname() -> str:
diff --git a/alpenhorn/daemon/auto_import.py b/alpenhorn/daemon/auto_import.py
index a52956a0a..caf62fc67 100644
--- a/alpenhorn/daemon/auto_import.py
+++ b/alpenhorn/daemon/auto_import.py
@@ -15,7 +15,7 @@
 from watchdog.events import FileSystemEventHandler
 
 from .. import db
-from ..common import config, extensions
+from ..common import config, extensions, metrics
 from ..common.util import invalid_import_path
 from ..db import (
     ArchiveAcq,
@@ -38,6 +38,41 @@
 log = logging.getLogger(__name__)
 
 
+def import_request_done(req: ArchiveFileImportRequest | None, result: str) -> None:
+    """Record a completed import request.
+
+    Including updating metrics.
+
+    Parameters
+    ----------
+    req:
+        Request that has been completed.  If None, this function does nothing.
+    result:
+        The result of the import request.  Used in the metric
+    """
+
+    if not req:
+        return
+
+    count = (
+        ArchiveFileImportRequest.update(completed=1)
+        .where(ArchiveFileImportRequest.id == req.id)
+        .execute()
+    )
+
+    # Only update metrics if this actually completed the request
+    if not count:
+        return
+
+    log.info(f"Completed import request #{req.id}.")
+    metrics.by_name("requests_completed").inc(
+        type="import",
+        result=result,
+        node=req.node.name,
+        group=req.node.group.name,
+    )
+
+
 def import_file(
     node: UpdateableNode,
     queue: FairMultiFIFOQueue,
@@ -70,9 +105,7 @@ def import_file(
     # Skip these.
     if path == pathlib.PurePath(node.db.root):
         log.debug("Skipping import request of node root")
-        if req:
-            log.info(f"Completed import request #{req.id}.")
-            req.complete()
+        import_request_done(req, "ignored")
         return
 
     # Strip node root, if path is absolute
@@ -84,9 +117,7 @@ def import_file(
             log.warning(
                 f"Ignoring import of {path}: not rooted under node {node.db.root}"
             )
-            if req:
-                log.info(f"Completed import request #{req.id}.")
-                req.complete()
+            import_request_done(req, "ignored")
             return
 
     # Ignore the node info file.  NB: this happens even on nodes which
@@ -96,9 +127,7 @@ def import_file(
         node.db.root
     ).joinpath("ALPENHORN_NODE"):
         log.debug("ignoring ALPENHORN_NODE file during import")
-        if req:
-            log.info(f"Completed import request #{req.id}.")
-            req.complete()
+        import_request_done(req, "ignored")
         return
 
     # New import task.
@@ -143,9 +172,7 @@ def _import_file(
     fullpath = pathlib.Path(node.db.root).joinpath(path)
     if fullpath.is_symlink() or not fullpath.is_file():
         log.debug(f'Not importing "{path}": not a file.')
-        if req:
-            log.info(f"Completed import request #{req.id}.")
-            req.complete()
+        import_request_done(req, "invalid")
         return
 
     log.debug(f'Considering "{path}" for import to node {node.name}.')
@@ -172,18 +199,14 @@ def _import_file(
     else:
         # Detection failed, so we're done.
         log.info(f"Not importing non-acquisition path: {path}")
-        if req:
-            log.info(f"Completed import request #{req.id}.")
-            req.complete()
+        import_request_done(req, "no_detection")
         return
 
     # Vet acq_name from extension
     rejection_reason = invalid_import_path(str(acq_name))
     if rejection_reason:
         log.warning(f'Rejecting invalid acq path "{acq_name}": {rejection_reason}')
-        if req:
-            log.info(f"Completed import request #{req.id}.")
-            req.complete()
+        import_request_done(req, "bad_acq")
         return
 
     file_name = path.relative_to(acq_name)
@@ -191,9 +214,7 @@ def _import_file(
     # If a copy already exists, we're done
     if node.db.named_copy_tracked(acq_name, file_name):
         log.debug(f"Not importing {path}: already known")
-        if req:
-            log.info(f"Completed import request #{req.id}.")
-            req.complete()
+        import_request_done(req, "duplicate")
         return
 
     # Begin a transaction
@@ -210,9 +231,7 @@ def _import_file(
                 log.info(f'Acquisition "{acq_name}" added to DB.')
             else:
                 log.info(f'Not importing unregistered acquistion: "{acq_name}".')
-                if req:
-                    log.info(f"Completed import request #{req.id}.")
-                    req.complete()
+                import_request_done(req, "unregistered")
                 return
 
         # Add the file, if necessary.
@@ -235,9 +254,7 @@ def _import_file(
                 log.info(f'File "{path}" added to DB.')
             else:
                 log.info(f'Not importing unregistered file: "{path}".')
-                if req:
-                    log.info(f"Completed import request #{req.id}.")
-                    req.complete()
+                import_request_done(req, "unregistered")
                 return
 
         try:
@@ -272,9 +289,7 @@ def _import_file(
             )
             log.info(f'Imported file copy "{path}" on node "{node.name}".')
 
-    if req:
-        log.info(f"Completed import request #{req.id}.")
-        req.complete()
+    import_request_done(req, "success")
 
     # Run post-add actions, if any
     ioutil.post_add(node.db, file_)
@@ -477,9 +492,9 @@ def scan(
         else:
             import_file(node, queue, file, register, None)
 
-    if req:
-        log.info(f"Completed import request #{req.id}.")
-        req.complete()
+    # This is successful because we've successfully scanned the
+    # tree, whether or not that resulted in any imports.
+    import_request_done(req, "success")
 
 
 def stop_observers() -> None:
diff --git a/alpenhorn/daemon/entry.py b/alpenhorn/daemon/entry.py
index e67ba58b5..2b5936062 100755
--- a/alpenhorn/daemon/entry.py
+++ b/alpenhorn/daemon/entry.py
@@ -6,7 +6,7 @@
 import click
 
 from .. import db
-from ..common import config
+from ..common import config, metrics
 from ..common.util import help_config_option, start_alpenhorn, version_option
 from ..scheduler import FairMultiFIFOQueue, pool
 from . import auto_import, update
@@ -60,6 +60,10 @@ def entry(conf, once):
     # Connect to the database
     db.connect()
 
+    # Start the prometheus client, if appropriate.
+    if not once:
+        metrics.start_promclient()
+
     # Set up the task queue
     queue = FairMultiFIFOQueue()
 
diff --git a/alpenhorn/daemon/update.py b/alpenhorn/daemon/update.py
index fb3d9ff12..aa4cb8792 100644
--- a/alpenhorn/daemon/update.py
+++ b/alpenhorn/daemon/update.py
@@ -10,8 +10,9 @@
 
 import peewee as pw
 
-from ..common import config, util
+from ..common import config, metrics, util
 from ..common.extensions import io_module
+from ..common.metrics import Metric
 from ..db import (
     ArchiveFileCopy,
     ArchiveFileCopyRequest,
@@ -241,6 +242,22 @@ def __init__(self, queue: FairMultiFIFOQueue, node: StorageNode) -> None:
         self.db = None
         self.reinit(node)
 
+        # Metrics that we want to delete when this node goes away
+        self._idle_metric = Metric(
+            "node_idle", "Node is idle", bound={"name": self.name}
+        )
+
+    def __del__(self):
+        # Delete metrics
+        try:
+            self._idle_metric.remove()
+        except (KeyError, AttributeError):
+            # AttributeError will be raised if this instance wasn't fully initialised
+            # KeyError will be raised if the prom metric was never created (because
+            #   we never set the metric to anything)
+            #   cf. https://github.com/prometheus/client_python/pull/1077
+            pass
+
     def reinit(self, node: StorageNode) -> bool:
         """Re-initialise the instance with a new database object.
 
@@ -289,19 +306,19 @@ def _async(
         ) -> None:
             """Task async to initialise `node`, if necessary.
 
-            Calls `req.complete()` if initialisation succeeds.
+            Completes `req` only if initialisation succeeds.
             """
 
             # recheck
             if node.io.check_init():
                 log.info(f'Node "{node.name}" already initialised.')
-                req.complete()
+                auto_import.import_request_done(req, "duplicate")
                 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()
+                auto_import.import_request_done(req, "success")
                 return
 
             # Otherwise, fail, and don't complete req
@@ -398,6 +415,8 @@ def update_idle(self) -> None:
         The idle updates are run if the regular update() ran but
         the node is currently idle.
         """
+
+        self._idle_metric.set(self.idle)
         if self._updated and self.idle:
             # Do any I/O class idle updates
             self.io.idle_update(self._io_happened)
@@ -501,7 +520,7 @@ def update_import(self) -> None:
             path = pathlib.Path(req.path)
             if path.is_absolute():
                 log.info(f'Not importing to "{self.name}" absolute path: {path}')
-                req.complete()
+                auto_import.import_request_done(req, "invalid")
                 continue
 
             if req.path == "ALPENHORN_NODE":
@@ -511,7 +530,7 @@ def update_import(self) -> None:
                     f'Ignoring node init request for "{self.name}": '
                     "already initialised."
                 )
-                req.complete()
+                auto_import.import_request_done(req, "duplicate")
                 continue
 
             if req.recurse:
@@ -524,7 +543,7 @@ def update_import(self) -> None:
                         "Ignoring import request of unresolvable scan path: "
                         f"{fullpath}: {e}"
                     )
-                    req.complete()
+                    auto_import.import_request_done(req, "invalid")
                     continue
 
                 # Recompute the relative path after resolution, or skip scan if we're
@@ -535,7 +554,7 @@ def update_import(self) -> None:
                     log.warning(
                         f"Ignoring import request of out-of-tree scan path: {fullpath}"
                     )
-                    req.complete()
+                    auto_import.import_request_done(req, "invalid")
                     continue
 
                 # Run scan
@@ -554,7 +573,7 @@ def update_import(self) -> None:
                         f'Ignoring request for import of invalid path "{req.path}": '
                         + rejection_reason
                     )
-                    req.complete()
+                    auto_import.import_request_done(req, "invalid")
                     continue
 
                 # Try to directly import the path
@@ -584,6 +603,12 @@ def update(self) -> None:
 
         if idle and do_update:
             log.info(f'Updating node "{self.name}".')
+            Metric(
+                "node_update",
+                "Count of updates on a node",
+                counter=True,
+                bound={"name": self.name},
+            ).inc()
 
             # Check the integrity of any questionable files (has_file=M)
             for copy in ArchiveFileCopy.select().where(
@@ -677,6 +702,21 @@ def __init__(
 
         self.reinit(group=group, nodes=nodes, idle=idle)
 
+        # Metrics that we want to delete when this node goes away
+        self._idle_metric = Metric(
+            "group_idle", "Group is idle", bound={"name": self.name}
+        )
+
+    def __del__(self):
+        # Delete metrics
+        try:
+            self._idle_metric.remove()
+        except (KeyError, AttributeError):
+            # AttributeError will be raised if this instance wasn't fully initialised
+            # KeyError will be raised if the prom metric was never created (because
+            #   we never set the metric to anything)
+            pass
+
     def reinit(
         self, *, group: StorageGroup, nodes: list[UpdateableNode], idle: bool
     ) -> None:
@@ -735,6 +775,14 @@ def update_pull(self, req: ArchiveFileCopyRequest) -> None:
         req : ArchiveFileCopyRequest
             The pull request to process.
         """
+
+        # The only label left unbound here is "result"
+        comp_metric = metrics.by_name("requests_completed").bind(
+            type="copy",
+            node=req.node_from.name,
+            group=req.group_to.name,
+        )
+
         # What's the current situation on the destination?
         copy_state = self.db.state_on_node(req.file)[0]
         if copy_state == "Y":
@@ -749,6 +797,7 @@ def update_pull(self, req: ArchiveFileCopyRequest) -> None:
             ArchiveFileCopyRequest.update(cancelled=True).where(
                 ArchiveFileCopyRequest.id == req.id
             ).execute()
+            comp_metric.inc(result="duplicate")
             return
         if copy_state == "M":
             log.warning(
@@ -792,6 +841,7 @@ def update_pull(self, req: ArchiveFileCopyRequest) -> None:
             ArchiveFileCopyRequest.update(cancelled=True).where(
                 ArchiveFileCopyRequest.id == req.id
             ).execute()
+            comp_metric.inc(result="missing")
             return
         if state == "M":
             log.info(
@@ -831,6 +881,12 @@ def update(self) -> None:
         # cancelled the update
         if self._init_idle and do_update:
             log.info(f'Updating group "{self.name}".')
+            Metric(
+                "group_update",
+                "Count of updates on a group",
+                counter=True,
+                bound={"name": self.name},
+            ).inc()
 
             # Remember ArchiveFiles that we're pulling, so we don't end up with
             # overlapping pulls (which would try to write to the same file).
@@ -860,6 +916,7 @@ def update_idle(self) -> None:
         The idle updates are run if the regular update() ran but
         the group was idle when it finished.
         """
+        self._idle_metric.set(1 if self._do_idle_updates else 0)
         if self._do_idle_updates:
             self.io.idle_update()
 
@@ -904,6 +961,23 @@ def update_loop(
     nodes = {}
     groups = {}
 
+    loop_time_metric = Metric(
+        "main_loop_time_seconds", description="Main loop execution time", counter=False
+    )
+    loop_count_metric = Metric(
+        "main_loops", description="Completed main loops", counter=True
+    )
+    node_avail_metric = Metric(
+        "node_available",
+        description="Node is available (active and initialized)",
+        unbound={"name"},
+    )
+    group_avail_metric = Metric(
+        "group_available",
+        description="Group is available (active and initialized)",
+        unbound={"name"},
+    )
+
     while not global_abort.is_set():
         loop_start = time.time()
 
@@ -935,6 +1009,7 @@ def update_loop(
                 auto_import.update_observer(node, queue, force_stop=True)
 
                 log.info(f'Node "{name}" no longer available.')
+                node_avail_metric.set(0, name=name)
             else:
                 vetted_nodes[name] = node
         nodes = vetted_nodes
@@ -952,6 +1027,7 @@ def update_loop(
             else:
                 # No existing node: create a new one.
                 log.info(f'Node "{name}" now available.')
+                node_avail_metric.set(1, name=name)
                 nodes[name] = UpdateableNode(queue, new_nodes[name])
 
             node = nodes[name]
@@ -987,6 +1063,7 @@ def update_loop(
         for name, group in groups.items():
             if name not in new_groups:
                 log.info(f'Group "{name}" no longer available.')
+                group_avail_metric.set(0, name=name)
             else:
                 vetted_groups[name] = group
         groups = vetted_groups
@@ -1001,6 +1078,7 @@ def update_loop(
             else:
                 # No existing group: create a new one.
                 log.info(f'Group "{name}" now available.')
+                group_avail_metric.set(1, name=name)
                 groups[name] = UpdateableGroup(queue=queue, **new_groups[name])
 
         # Node updates
@@ -1040,6 +1118,9 @@ def update_loop(
         loop_time = time.time() - loop_start
         log.info(f"Main loop execution was {util.pretty_deltat(loop_time)}.")
 
+        loop_time_metric.set(loop_time)
+        loop_count_metric.inc()
+
         # Pool and queue info
         log.info(
             f"Tasks: {queue.qsize} queued, {queue.deferred_size} deferred, "
@@ -1084,6 +1165,10 @@ def serial_io(queue: FairMultiFIFOQueue) -> None:
     """
     start_time = time.time()
 
+    task_metric = Metric(
+        "serialio_tasks", "Count of tasks run via Serial I/O", counter=True
+    )
+
     # Handle tasks for, say, 15 minutes at most
     while time.time() - start_time < config.config["daemon"]["serial_io_timeout"]:
         # Get a task
@@ -1093,6 +1178,8 @@ def serial_io(queue: FairMultiFIFOQueue) -> None:
         if item is None:
             break
 
+        task_metric.inc()
+
         # Run the task
         task, key = item
 
@@ -1100,3 +1187,5 @@ def serial_io(queue: FairMultiFIFOQueue) -> None:
         task()
         queue.task_done(key)
         log.info(f"Finished task {task}")
+
+    Metric("serialio_loops", "Count of Serial I/O loops", counter=True).inc()
diff --git a/alpenhorn/db/archive.py b/alpenhorn/db/archive.py
index 6dc82e665..f9164f529 100644
--- a/alpenhorn/db/archive.py
+++ b/alpenhorn/db/archive.py
@@ -162,11 +162,3 @@ class ArchiveFileImportRequest(base_model):
     register = pw.BooleanField(default=False)
     completed = pw.BooleanField(default=False)
     timestamp = pw.DateTimeField(default=pw.utcnow, null=True)
-
-    def complete(self) -> None:
-        """Set this request to complete in the database."""
-
-        # If we're already complete, this function does nothing
-        if not self.completed:
-            self.completed = True
-            self.save(only=[ArchiveFileImportRequest.completed])
diff --git a/alpenhorn/io/_default_asyncs.py b/alpenhorn/io/_default_asyncs.py
index 770901f4a..565a0467e 100644
--- a/alpenhorn/io/_default_asyncs.py
+++ b/alpenhorn/io/_default_asyncs.py
@@ -11,6 +11,8 @@
 
 import peewee as pw
 
+from ..common import metrics
+from ..common.metrics import Metric
 from ..common.util import timeout_call
 from ..daemon.update import RemoteNode
 from ..db import ArchiveFileCopy, ArchiveFileCopyRequest, utcnow
@@ -48,6 +50,18 @@ def pull_async(
         The request we're fulfilling.
     """
 
+    comp_metric = metrics.by_name("requests_completed").bind(
+        type="copy",
+        node=req.node_from.name,
+        group=req.group_to.name,
+    )
+    pullrun_metric = Metric(
+        "pull_running_count",
+        "Count of in-progress pulls",
+        unbound={"method", "remote"},
+        bound={"node": io.node.name},
+    )
+
     # Recheck the database to see if a file hasn't shown
     # up somehow
     if io.node.filecopy_state(req.file) == "Y":
@@ -58,6 +72,7 @@ def pull_async(
         ArchiveFileCopyRequest.update(cancelled=1).where(
             ArchiveFileCopyRequest.id == req.id
         ).execute()
+        comp_metric.inc(result="duplicate")
         return
 
     # Before we were queued, NodeIO reserved space for this file.
@@ -124,11 +139,15 @@ def pull_async(
             # calculate the md5 hash as it goes, so we'll do that to save doing
             # it at the end.
             log.info(f"Pulling remote file {req.file.path} using bbcp")
+            pullrun_metric.inc(method="bbcp", remote="1")
             ioresult = ioutil.bbcp(from_path, to_file, req.file.size_b)
+            pullrun_metric.dec(method="bbcp", remote="1")
         elif shutil.which("rsync") is not None:
             # Next try rsync over ssh.
             log.info(f"Pulling remote file {req.file.path} using rsync")
+            pullrun_metric.inc(method="rsync", remote="1")
             ioresult = ioutil.rsync(from_path, to_file, req.file.size_b, local)
+            pullrun_metric.dec(method="rsync", remote="1")
         else:
             # We have no idea how to transfer the file...
             log.error("No commands available to complete remote pull.")
@@ -143,7 +162,9 @@ def pull_async(
         # But don't do this if it creates a hardlink between an archive node and
         # a non-archive node
         if req.node_from.archive == io.node.archive:
+            pullrun_metric.inc(method="link", remote="0")
             ioresult = ioutil.hardlink(from_path, to_dir, req.file.name)
+            pullrun_metric.dec(method="link", remote="0")
             if ioresult is not None:
                 log.info(f"Hardlinked local file {req.file.path}")
         else:
@@ -153,13 +174,17 @@ def pull_async(
         if ioresult is None:
             if shutil.which("rsync") is not None:
                 log.info(f"Pulling local file {req.file.path} using rsync")
+                pullrun_metric.inc(method="rsync", remote="0")
                 ioresult = ioutil.rsync(from_path, to_file, req.file.size_b, local)
+                pullrun_metric.dec(method="rsync", remote="0")
             else:
                 # No rsync?  Just use shutil.copy, I guess
                 log.warning("Falling back on shutil.copy to complete local pull.")
+                pullrun_metric.inc(method="internal", remote="0")
                 ioresult = ioutil.local_copy(
                     from_path, to_dir, req.file.name, req.file.size_b
                 )
+                pullrun_metric.dec(method="internal", remote="0")
 
     # Delete the placeholder, if we created it
     placeholder.unlink(missing_ok=True)
@@ -211,6 +236,13 @@ def check_async(task: Task, io: BaseNodeIO, copy: ArchiveFileCopy) -> None:
     copyname = copy.file.path
     fullpath = copy.path
 
+    Metric(
+        "verification_checks",
+        "Count of verification checks",
+        counter=True,
+        bound={"node": io.node.name},
+    ).inc()
+
     # Does the copy exist?
     if fullpath.exists():
         # First check the size
@@ -296,6 +328,19 @@ def delete_async(
         fullpath = copy.path
         try:
             fullpath.unlink()  # Remove the actual file
+            Metric(
+                "deleted_files",
+                "Count of deleted files",
+                counter=True,
+                bound={"node": name},
+            ).inc()
+            if copy.file.size_b:
+                Metric(
+                    "deleted_bytes",
+                    "Size of deleted files",
+                    counter=True,
+                    bound={"node": name},
+                ).add(copy.file.size_b)
             log.info(f"Removed file copy {shortname} on {name}")
         except OSError as e:
             if e.errno == errno.ENOENT:
diff --git a/alpenhorn/io/ioutil.py b/alpenhorn/io/ioutil.py
index e4b2d46d2..f2eb42728 100644
--- a/alpenhorn/io/ioutil.py
+++ b/alpenhorn/io/ioutil.py
@@ -14,7 +14,8 @@
 import peewee as pw
 
 from .. import db
-from ..common import config, util
+from ..common import config, metrics, util
+from ..common.metrics import Metric
 from ..db import (
     ArchiveFileCopy,
     ArchiveFileCopyRequest,
@@ -505,10 +506,17 @@ def copy_request_done(
         or False if the transfer failed.
     """
 
+    # The only label left unbound here is "result"
+    transf_metric = metrics.by_name("transfers").bind(
+        node_from=req.node_from.name,
+        group_to=req.group_to.name,
+    )
+
     # Check the result
     if not success:
         if stderr is None:
             stderr = "Unspecified error."
+            transf_metric.inc(result="failure")
         if check_src:
             # If the copy didn't work, then the remote file may be corrupted.
             log.error("Copy failed.  Marking source file suspect.")
@@ -517,10 +525,12 @@ def copy_request_done(
                 ArchiveFileCopy.file == req.file,
                 ArchiveFileCopy.node == req.node_from,
             ).execute()
+            transf_metric.inc(result="check_src")
         else:
             # An error occurred that can't be due to the source being corrupt
             log.error("Copy failed")
             log.info(f"Output: {stderr}")
+            transf_metric.inc(result="failure")
         return False
 
     # Otherwise, transfer was completed, remember end time
@@ -538,6 +548,7 @@ def copy_request_done(
             ArchiveFileCopy.file == req.file,
             ArchiveFileCopy.node == req.node_from,
         ).execute()
+        transf_metric.inc(result="integrity")
         return False
 
     # Transfer successful
@@ -580,6 +591,23 @@ def copy_request_done(
             transfer_completed=utcfromtimestamp(end_time),
         ).where(ArchiveFileCopyRequest.id == req.id).execute()
 
+    # Update metrics
+    metrics.by_name("requests_completed").inc(
+        type="copy",
+        node=req.node_from.name,
+        group=req.group_to.name,
+        result="success",
+    )
+    transf_metric.inc(result="success")
+
+    # This can be used to measure throughput
+    Metric(
+        "pulled_bytes",
+        "Count of bytes pulled",
+        counter=True,
+        bound={"node_from": req.node_from.name, "group_to": req.group_to.name},
+    ).add(req.file.size_b)
+
     # Run post-add actions, if any
     post_add(io.node, req.file)
 
@@ -589,7 +617,7 @@ def copy_request_done(
 def remove_filedir(
     node: StorageNode, dirname: pathlib.Path, tree_lock: UpDownLock
 ) -> None:
-    """Try to delete an file's parenty directorie(s) from a node
+    """Try to delete a file's parent directory(s) from a node
 
     Will attempt to remove the enitre tree given as `dirname`
     while holding the `tree_lock` down until reaching `node.root`.
diff --git a/alpenhorn/scheduler/pool.py b/alpenhorn/scheduler/pool.py
index 4548126f5..239dbd2f9 100644
--- a/alpenhorn/scheduler/pool.py
+++ b/alpenhorn/scheduler/pool.py
@@ -7,6 +7,7 @@
 
 from peewee import OperationalError
 
+from ..common.metrics import Metric
 from .queue import FairMultiFIFOQueue
 
 log = logging.getLogger(__name__)
@@ -70,26 +71,44 @@ def run(self) -> None:
         # Put the worker id in `threadlocal`, so tasks can access it
         global threadlocal
         threadlocal.worker_id = self._worker_id
+        metric_running = Metric(
+            "worker_running",
+            "worker is running",
+            counter=False,
+            bound={"id": self._worker_id},
+        )
+        metric_running.set(1)
+        metric_idle = Metric(
+            "worker_idle",
+            "worker is idle (waiting for a task)",
+            counter=False,
+            bound={"id": self._worker_id},
+        )
 
         while True:
+            metric_idle.set(1)
             # Exit if told to stop
             if global_abort.is_set():
                 log.info("Stopped due to global abort.")
+                metric_running.set(0)
                 return None
             if self._worker_stop.is_set():
                 log.info("Stopped.")
+                metric_running.set(0)
                 return None
 
             # Wait for a task:
             item = self._queue.get(timeout=5)
 
             if item is not None:
+                metric_idle.set(0)
                 task, key = item
 
                 # Abandon working if alpenhornd is aborting
                 if global_abort.is_set():
                     log.info("Stopped due to global abort.")
                     self._queue.task_done(key)
+                    metric_running.set(0)
                     return None
 
                 # Otherwise, execute the task.
@@ -115,6 +134,7 @@ def run(self) -> None:
                         log.exception(
                             "Aborting due to uncaught exception in task cleanup"
                         )
+                        metric_running.set(0)
                         return 1
 
                     log.info(f"Finished task {task}")
@@ -124,10 +144,12 @@ def run(self) -> None:
                     task.requeue()
 
                     log.error(f"Exiting due to db error: {operr}")
+                    metric_running.set(0)
                     return 1  # Thread exits, will be respawned in the main loop
                 except Exception:
                     global_abort.set()
                     log.exception("Aborting due to uncaught exception in task")
+                    metric_running.set(0)
                     return 1
 
                 self._queue.task_done(key)
@@ -155,7 +177,7 @@ class WorkerPool:
         The task queue
     """
 
-    __slots__ = ["_all_workers", "_mutex", "_queue", "_workers"]
+    __slots__ = ["_all_workers", "_metric_worker_count", "_mutex", "_queue", "_workers"]
 
     def __init__(self, num_workers: int, queue: FairMultiFIFOQueue) -> None:
         self._queue = queue
@@ -171,6 +193,13 @@ def __init__(self, num_workers: int, queue: FairMultiFIFOQueue) -> None:
         # workers stopped by del_worker(), which may still be running.
         self._all_workers = []
 
+        self._metric_worker_count = Metric(
+            "worker_count",
+            "Number of worker threads",
+            counter=False,
+            bound={"pool_type": "WorkerPool"},
+        )
+
         # Start initial workers
         for _ in range(num_workers):
             self._new_worker()
@@ -211,6 +240,8 @@ def _new_worker(self, index: int | None = None) -> None:
         # Start working
         worker.start()
 
+        self._metric_worker_count.inc()
+
     def add_worker(self, blocking: bool = True) -> None:
         """Increment the number of workers in the pool.
 
@@ -254,6 +285,8 @@ def del_worker(self, blocking: bool = True) -> None:
                 # Fire the stop event
                 worker.stop_working()
 
+                self._metric_worker_count.dec()
+
             # Release the lock
             self._mutex.release()
         else:
@@ -308,6 +341,7 @@ def shutdown(self) -> None:
             # Probably we're about to exit, but just so everything stays copacetic:
             self._workers = []
             self._all_workers = []
+            self._metric_worker_count.set(0)
 
 
 class EmptyPool:
@@ -316,6 +350,15 @@ class EmptyPool:
     It has the same methods as WorkerPool, but does nothing and is always
     empty."""
 
+    def __init__(self) -> None:
+        # This is never updated
+        Metric(
+            "worker_count",
+            "Number of worker threads",
+            counter=False,
+            bound={"pool_type": "EmptyPool"},
+        )
+
     def __len__(self) -> None:
         return 0
 
diff --git a/alpenhorn/scheduler/queue.py b/alpenhorn/scheduler/queue.py
index 2832193c4..39d391e91 100644
--- a/alpenhorn/scheduler/queue.py
+++ b/alpenhorn/scheduler/queue.py
@@ -20,6 +20,8 @@
 from time import monotonic, sleep
 from typing import Any
 
+from ..common.metrics import Metric
+
 
 class FairMultiFIFOQueue:
     """Create a new Fair Multi-FIFO Queue"""
@@ -35,6 +37,10 @@ class FairMultiFIFOQueue:
         "_keys_by_inprogress",
         "_lock",
         "_not_empty",
+        "_qcount",
+        "_qcount_all",
+        "_qcount_any",
+        "_qlock",
         "_total_inprogress",
         "_total_queued",
     ]
@@ -71,6 +77,56 @@ def __init__(self) -> None:
         # of the primary _lock.
         self._dlock = threading.Lock()
 
+        # Metrics
+
+        # Count of tasks
+        self._qcount = Metric(
+            "queue_count", "Count of queued tasks", unbound=["fifo", "status"]
+        )
+
+        # Marginalised over status
+        self._qcount_any = Metric(
+            "queue_count",
+            "Count of queued tasks",
+            unbound=["fifo"],
+            bound={"status": "any"},
+        )
+
+        # Marignalised over fifo
+        self._qcount_all = Metric(
+            "queue_count",
+            "Count of queued tasks",
+            unbound=["status"],
+            bound={"fifo": "_ALL_"},
+        )
+
+        # Track locked fifos
+        self._qlock = Metric(
+            "queue_locked", "The queue fifo is locked", unbound=["fifo"]
+        )
+
+    def _inc_metrics(self, fifo: str, status: str) -> None:
+        """Increment the queue_count metric.
+
+        Adds one to the metric with fifo=fifo and status=status.  Also adds
+        one to the marginalised ..._any and ..._all metrics.
+        """
+
+        self._qcount.inc(fifo=fifo, status=status)
+        self._qcount_any.inc(fifo=fifo)
+        self._qcount_all.inc(status=status)
+
+    def _dec_metrics(self, fifo: str, status: str) -> None:
+        """Decrement the queue_count metric.
+
+        Subtracts one from the metric with fifo=fifo and status=status.  Also
+        subtracts one from the marginalised ..._any and ..._all metrics.
+        """
+
+        self._qcount.dec(fifo=fifo, status=status)
+        self._qcount_any.dec(fifo=fifo)
+        self._qcount_all.dec(status=status)
+
     def task_done(self, key: Hashable) -> None:
         """Report that a task from the FIFO named `key` is done.
 
@@ -108,11 +164,13 @@ def task_done(self, key: Hashable) -> None:
             # because there can't be anything else in-progress from this FIFO.  So,
             # it's always reasonable to try to unlock a FIFO here.
             self._fifo_locks.discard(key)
+            self._qlock.set(0, fifo=key)
 
             # Decrement counts
             count -= 1
             self._inprogress_counts[key] = count
             self._total_inprogress -= 1
+            self._dec_metrics(fifo=key, status="in-progress")
 
             # Put the fifo back in the ordered list.
             #
@@ -162,6 +220,9 @@ def join(self) -> None:
         with self._dlock:
             self._joining = False
 
+        # Clear the metric.  This clears the marginalised versions, too
+        self._qcount.clear()
+
     @property
     def qsize(self) -> int:
         """Total number of queued tasks.
@@ -240,6 +301,8 @@ def _put(self, item: Any, key: Hashable, exclusive: bool) -> None:
         fifo.append((item, exclusive))
         self._total_queued += 1
 
+        self._inc_metrics(fifo=key, status="queued")
+
     def put(
         self, item: Any, key: Hashable, exclusive: bool = False, wait: float = 0
     ) -> bool:
@@ -295,6 +358,7 @@ def put(
                 heapq.heappush(
                     self._deferrals, (monotonic() + wait, item, key, exclusive)
                 )
+                self._inc_metrics(fifo=key, status="deferred")
         else:
             # Immediate put
             with self._not_empty:
@@ -350,6 +414,7 @@ def _get(self, t: float) -> tuple[Any, Hashable] | None:
             while len(self._deferrals) > 0 and self._deferrals[0][0] <= monotonic():
                 # heappop removes and returns self._deferrals[0]
                 _, item, key, exclusive = heapq.heappop(self._deferrals)
+                self._dec_metrics(fifo=key, status="deferred")
                 self._put(item, key, exclusive)
 
         # If the queue is still empty, time out
@@ -415,9 +480,14 @@ def _get(self, t: float) -> tuple[Any, Hashable] | None:
         self._total_queued -= 1
         self._total_inprogress += 1
 
+        # Update metrics
+        self._inc_metrics(fifo=key, status="in-progress")
+        self._dec_metrics(fifo=key, status="queued")
+
         # Lock this FIFO, if item is exclusive
         if exclusive:
             self._fifo_locks.add(key)
+            self._qlock.set(1, fifo=key)
 
         # Increment the in-progress count and file the key in the right
         # place in _keys_by_inprogress
diff --git a/pyproject.toml b/pyproject.toml
index 35f719606..e8cf831cd 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -24,6 +24,9 @@ dependencies = [
 ]
 
 [project.optional-dependencies]
+prometheus = [
+  "prometheus_client",
+]
 test = [
   "chimedb @ git+https://github.com/chime-experiment/chimedb.git",
   "docker >= 3.0",


Test patch:
diff --git a/tests/common/test_metrics.py b/tests/common/test_metrics.py
new file mode 100644
index 000000000..713916170
--- /dev/null
+++ b/tests/common/test_metrics.py
@@ -0,0 +1,373 @@
+"""Test common.metrics."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from alpenhorn.common import metrics
+
+prometheus_client = pytest.importorskip(
+    "prometheus_client", exc_type=ModuleNotFoundError
+)
+Counter = prometheus_client.metrics.Counter
+Gauge = prometheus_client.metrics.Gauge
+REGISTRY = prometheus_client.registry.REGISTRY
+
+
+@pytest.fixture
+def cleanup():
+    """Clean up after metric tests."""
+    yield
+
+    # Delete all metrics from the prometheus client registry and our dict
+    while metrics._metrics:
+        key, value = metrics._metrics.popitem()
+        REGISTRY.unregister(value[0])
+
+
+def test_nothing(cleanup):
+    """Test nothing.
+
+    The only point of this test is to trigger the cleanup fixture so
+    everything following starts with an empty _metric cache.
+    """
+    assert True
+
+
+def test_label_clash(cleanup):
+    """Can't both bind a label and leave it unbound."""
+    with pytest.raises(KeyError):
+        metrics.Metric("name", "desc", unbound={"a", "b", "c"}, bound={"c": "d"})
+
+
+def test_label_change(cleanup):
+    """Accessing the same metric with different labels shouldn't work."""
+
+    metrics.Metric("name", "desc", unbound={"a", "b", "c"})
+
+    with pytest.raises(ValueError):
+        # Missing a label
+        metrics.Metric("name", "desc", unbound={"a", "b"})
+
+    with pytest.raises(ValueError):
+        # Extra label
+        metrics.Metric("name", "desc", unbound={"a", "b", "c", "d"})
+
+    with pytest.raises(ValueError):
+        # Missing and extra label
+        metrics.Metric("name", "desc", unbound={"a", "b", "d"})
+
+
+def test_labelnames(cleanup):
+    """Test Metric.labelnames"""
+    metric = metrics.Metric("unbound", "desc", unbound={"h", "b", "c", "d", "f"})
+    assert metric.labelnames == ["b", "c", "d", "f", "h"]
+
+    metric = metrics.Metric(
+        "mixed", "desc", unbound={"a", "b", "c"}, bound={"d": "e", "f": "g"}
+    )
+    assert metric.labelnames == ["a", "b", "c", "d", "f"]
+
+    metric = metrics.Metric(
+        "bound", "desc", bound={"i": "a", "b": "b", "c": "c", "d": "e", "f": "g"}
+    )
+    assert metric.labelnames == ["b", "c", "d", "f", "i"]
+
+
+def test_labelvalues(cleanup):
+    """Test Metric.labelvalues"""
+    metric = metrics.Metric("unbound", "desc", unbound={"h", "b", "c", "d", "f"})
+    assert metric.labelvalues(h=1, b=2, c=3, d=4, f=5) == [2, 3, 4, 5, 1]
+
+    metric = metrics.Metric(
+        "mixed", "desc", unbound={"a", "b", "c"}, bound={"d": "e", "f": "g"}
+    )
+    assert metric.labelvalues(a=6, b=7, c=8) == [6, 7, 8, "e", "g"]
+
+    metric = metrics.Metric(
+        "bound", "desc", bound={"i": "a", "b": "b", "c": "c", "d": "e", "f": "g"}
+    )
+    assert metric.labelvalues() == ["b", "c", "e", "g", "a"]
+
+
+def test_labelvalues_bad(cleanup):
+    """Test invalid labels to labelvalues()"""
+    metric = metrics.Metric(
+        "name", "desc", unbound={"a", "b", "c"}, bound={"d": "e", "f": "g"}
+    )
+
+    # must bind all labels
+    with pytest.raises(ValueError):
+        metric.labelvalues(a=1)
+
+    # Can't add extra labels
+    with pytest.raises(TypeError):
+        metric.labelvalues(a=1, b=2, c=3, h=4)
+
+    # Can't re-bind bound labels
+    with pytest.raises(TypeError):
+        metric.labelvalues(a=1, b=2, c=3, d=4)
+
+
+def test_bind(cleanup):
+    """Test Metric.bind"""
+    base = metrics.Metric("name", "desc", unbound=["a", "b", "c"])
+
+    # All labels are unbound
+    assert base.labelvalues(a=1, b=2, c=3) == [1, 2, 3]
+
+    # Can't bind non-existent labels
+    with pytest.raises(TypeError):
+        base.bind(d=0)
+
+    # This is fine.  It just makes a copy.
+    copy = base.bind()
+
+    # All labels are still unbound
+    assert copy.labelvalues(a=1, b=2, c=3) == [1, 2, 3]
+
+    # Not the same instance as base
+    assert copy is not base
+
+    # Bind some labels
+    copy = base.bind(a=4)
+
+    # Only two unbound labels, now
+    assert copy.labelvalues(b=2, c=3) == [4, 2, 3]
+
+    # Fails because a is now bound
+    with pytest.raises(TypeError):
+        assert copy.labelvalues(a=1, b=2, c=3) == [1, 2, 3]
+
+    # But the base hasn't changed
+    assert base.labelvalues(a=1, b=2, c=3) == [1, 2, 3]
+
+    # Bind the remaining labels
+    copy2 = copy.bind(b=5, c=6)
+
+    # Now no unbound labels
+    assert copy2.labelvalues() == [4, 5, 6]
+
+
+def test_labelled_metric(cleanup):
+    """Test that _labelled_metric does return a prometheus metric."""
+    gauge = metrics.Metric(
+        "gauge", "desc", unbound={"a", "b", "c"}, bound={"d": "e", "f": "g"}
+    )
+
+    submetric = gauge._labelled_metric({"a": "a", "b": "b", "c": "c"})
+    assert isinstance(submetric, Gauge)
+
+    counter = metrics.Metric(
+        "counter",
+        "desc",
+        counter=True,
+        unbound={"a", "b", "c"},
+        bound={"d": "e", "f": "g"},
+    )
+    submetric = counter._labelled_metric({"a": "a", "b": "b", "c": "c"})
+    assert isinstance(submetric, Counter)
+
+
+def test_unlabelled_metric(cleanup):
+    """Test using an unlabelled metric."""
+
+    # In this case, Metric._labelled_metric _can't_ call the
+    # prom client metric's .label().  So, it just returns the parent
+    # metric, which is then used for the .set call
+
+    metric = metrics.Metric("name", "desc")
+
+    # _labelled_metric just returns _metric now
+    assert metric._labelled_metric(labels={}) is metric._metric
+
+
+def test_add(cleanup):
+    """Test Metric.add"""
+
+    gauge = metrics.Metric(
+        "gauge", "desc", unbound=["a", "b", "c"], bound={"d": "e", "f": "g"}
+    )
+
+    # Replace the internal prom metric with some mocks (childmock plays
+    # the role of the labelled child metric)
+    mock = MagicMock()
+    childmock = MagicMock()
+    mock.labels.return_value = childmock
+    gauge._metric = mock
+
+    gauge.add(1, a=1, b=2, c=3)
+
+    mock.labels.assert_called_with(a=1, b=2, c=3, d="e", f="g")
+    # .add is implemented via .inc on the prom metric
+    childmock.inc.assert_called_with(1)
+
+    # Not fully bound is an error
+    with pytest.raises(ValueError):
+        gauge.add(0, b=5, c=6)
+
+    gauge.add(-3, a=4, b=5, c=6)
+
+    mock.labels.assert_called_with(a=4, b=5, c=6, d="e", f="g")
+    childmock.inc.assert_called_with(-3)
+
+
+def test_inc(cleanup):
+    """Test Metric.inc"""
+
+    metric = metrics.Metric(
+        "name", "desc", unbound=["a", "b", "c"], bound={"d": "e", "f": "g"}
+    )
+
+    # Replace the internal prom metric with some mocks (childmock plays
+    # the role of the labelled child metric)
+    mock = MagicMock()
+    childmock = MagicMock()
+    mock.labels.return_value = childmock
+    metric._metric = mock
+
+    metric.inc(a=1, b=2, c=3)
+
+    mock.labels.assert_called_with(a=1, b=2, c=3, d="e", f="g")
+    # The passed .inc value is always explicit
+    childmock.inc.assert_called_with(1)
+
+    # Not fully bound is an error
+    with pytest.raises(ValueError):
+        metric.inc(b=5, c=6)
+
+
+def test_dec(cleanup):
+    """Test Metric.dec"""
+
+    metric = metrics.Metric(
+        "name", "desc", unbound=["a", "b", "c"], bound={"d": "e", "f": "g"}
+    )
+
+    # Replace the internal prom metric with some mocks (childmock plays
+    # the role of the labelled child metric)
+    mock = MagicMock()
+    childmock = MagicMock()
+    mock.labels.return_value = childmock
+    metric._metric = mock
+
+    metric.dec(a=1, b=2, c=3)
+
+    mock.labels.assert_called_with(a=1, b=2, c=3, d="e", f="g")
+    childmock.inc.assert_called_with(-1)
+
+    # Not fully bound is an error
+    with pytest.raises(ValueError):
+        metric.dec(b=5, c=6)
+
+
+def test_set_gauge(cleanup):
+    """Test Metric.set on a gauge"""
+
+    gauge = metrics.Metric(
+        "name", "desc", unbound=["a", "b", "c"], bound={"d": "e", "f": "g"}
+    )
+
+    # Replace the internal prom metric with some mocks (childmock plays
+    # the role of the labelled child metric)
+    mock = MagicMock()
+    childmock = MagicMock()
+    mock.labels.return_value = childmock
+    gauge._metric = mock
+
+    # Not fully bound is an error
+    with pytest.raises(ValueError):
+        gauge.set(-1, b=5, c=6)
+
+    gauge.set(-2, a=4, b=5, c=6)
+
+    # We always use the keyword calling convention with .labels
+    mock.labels.assert_called_with(a=4, b=5, c=6, d="e", f="g")
+    childmock.set.assert_called_with(-2)
+
+
+def test_set_counter(cleanup):
+    """Test Metric.set on a counter"""
+
+    counter = metrics.Metric(
+        "name",
+        "desc",
+        counter=True,
+        unbound=["a", "b", "c"],
+        bound={"d": "e", "f": "g"},
+    )
+
+    # Replace the internal prom metric with some mocks (childmock plays
+    # the role of the labelled child metric)
+    mock = MagicMock()
+    childmock = MagicMock()
+    mock.labels.return_value = childmock
+    counter._metric = mock
+
+    # Not fully bound is an error
+    with pytest.raises(ValueError):
+        counter.set(0, b=5, c=6)
+
+    # Can't use a non-zero value
+    with pytest.raises(ValueError):
+        counter.set(2, a=4, b=5, c=6)
+
+    # This resets the counter
+    counter.set(0, a=4, b=5, c=6)
+
+    mock.labels.assert_called_with(a=4, b=5, c=6, d="e", f="g")
+    childmock.reset.assert_called()
+    childmock.set.assert_not_called()
+
+
+def test_remove(cleanup):
+    """Test Metric.remove"""
+
+    metric = metrics.Metric(
+        "name", "desc", unbound=["a", "b", "c"], bound={"d": "e", "f": "g"}
+    )
+
+    # Replace the internal prom metric with a mock
+    mock = MagicMock()
+    metric._metric = mock
+
+    # Not fully bound is an error
+    with pytest.raises(ValueError):
+        metric.remove(b=5, c=6)
+
+    metric.remove(a=4, b=5, c=6)
+
+    # No keywords allowed here; must pass the ordered labelvalue list
+    mock.remove.assert_called_with([4, 5, 6, "e", "g"])
+
+
+def test_by_name(cleanup):
+    """Test by_name."""
+
+    metric = metrics.by_name("transfers")
+    assert isinstance(metric, metrics.Metric)
+
+    with pytest.raises(ValueError):
+        metrics.by_name("no_such_metric")
+
+
+@pytest.mark.alpenhorn_config({"daemon": {"prom_client_port": "0"}})
+def test_start_promclient_off(set_config):
+    """Test start_promclient with no port."""
+
+    mock = MagicMock()
+    with patch("prometheus_client.start_http_server", mock):
+        metrics.start_promclient()
+
+    mock.assert_not_called()
+
+
+@pytest.mark.alpenhorn_config({"daemon": {"prom_client_port": "1234"}})
+def test_start_promclient_on(set_config):
+    """Test start_promclient with a port."""
+
+    mock = MagicMock()
+    with patch("prometheus_client.start_http_server", mock):
+        metrics.start_promclient()
+
+    mock.assert_called_with(1234)
diff --git a/tests/daemon/test_auto_import.py b/tests/daemon/test_auto_import.py
index c8d23ea82..329a49a8a 100644
--- a/tests/daemon/test_auto_import.py
+++ b/tests/daemon/test_auto_import.py
@@ -11,10 +11,25 @@
 from alpenhorn.daemon import auto_import
 from alpenhorn.daemon.update import UpdateableNode
 from alpenhorn.db.acquisition import ArchiveAcq, ArchiveFile
-from alpenhorn.db.archive import ArchiveFileCopy
+from alpenhorn.db.archive import ArchiveFileCopy, ArchiveFileImportRequest
 from alpenhorn.io.lfs import HSMState
 
 
+def test_import_request_done(simpleimportrequest):
+    """Test import_request_done()."""
+
+    assert ArchiveFileImportRequest.get(id=simpleimportrequest.id).completed == 0
+
+    auto_import.import_request_done(simpleimportrequest, True)
+
+    assert ArchiveFileImportRequest.get(id=simpleimportrequest.id).completed == 1
+
+    # Can be called multiple times
+    auto_import.import_request_done(simpleimportrequest, True)
+
+    assert ArchiveFileImportRequest.get(id=simpleimportrequest.id).completed == 1
+
+
 def test_import_file_bad_paths(queue, unode):
     """Test that bad paths are rejected by import_file."""
 
diff --git a/tests/db/test_archive.py b/tests/db/test_archive.py
index d15c12c67..9a83d9edd 100644
--- a/tests/db/test_archive.py
+++ b/tests/db/test_archive.py
@@ -198,21 +198,3 @@ def test_archivefileimportrequest_model(
 
     # Not unique
     archivefileimportrequest(path="min_path", node=minnode)
-
-
-def test_importreqeust_complete(simpleimportrequest):
-    """Test ArchiveFileImportRequest.complete()."""
-
-    assert simpleimportrequest.completed == 0
-    assert ArchiveFileImportRequest.get(id=simpleimportrequest.id).completed == 0
-
-    simpleimportrequest.complete()
-
-    assert simpleimportrequest.completed == 1
-    assert ArchiveFileImportRequest.get(id=simpleimportrequest.id).completed == 1
-
-    # Can be called multiple times
-    simpleimportrequest.complete()
-
-    assert simpleimportrequest.completed == 1
-    assert ArchiveFileImportRequest.get(id=simpleimportrequest.id).completed == 1