diff --git a/AGENTS.md b/AGENTS.md index bce46cb..167b6a5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -34,8 +34,10 @@ uv run ruff check . && uv run ruff format . - **A snapshot only exists the day it was restored.** Any change to the snapshot, send or restore paths must be proven by a test that restores. - **Destruction stays deliberate.** Nothing deletes a volume or a snapshot on - its own initiative; only an explicit command or the purge pattern the user - asked for. + its own initiative; only an explicit command, the purge pattern the user + asked for, or the mount of a volume whose replication the user scheduled, + which keeps what the volume held as a snapshot before replacing it. The one + thing deleted without a copy is an empty volume. - **Fix the root cause, at the shared function.** Before editing a handler, check its siblings: one guard in a helper beats a guard in every caller. - **Pure inside, effects at the edges.** Pattern parsing, name validation and @@ -53,8 +55,8 @@ uv run ruff check . && uv run ruff format . ## Commits and pull requests -- Every change goes through a pull request, on a branch starting from `master`. - Never commit to `master` directly, and never open a pull request from a +- Every change goes through a pull request, on a branch starting from `main`. + Never commit to `main` directly, and never open a pull request from a long-lived work branch that carries unrelated commits. - One pull request, one subject. If the description needs the word "and", it is two pull requests. diff --git a/CHANGES.rst b/CHANGES.rst index 40270f5..5a934ec 100644 --- a/CHANGES.rst +++ b/CHANGES.rst @@ -12,6 +12,87 @@ CHANGELOG subvolume where a directory already stood. Removing it meant deleting the directory by hand. +- A volume can ask for its scheduled jobs as it is created. An option named + after an action, ``-o replicate:node2=1``, writes that line in the + schedule of the host the volume is created on, when no line says the same + thing already; a line already there is left alone, paused or not, and none + is removed on its own. This is how a Docker Swarm service, with + ``volume-opt=replicate:node2=1``, says once what happens to its volume on + whatever host it lands on, including a host that kept the volume from an + earlier deployment. An option that is neither ``copyonwrite``, + ``compression`` nor an action is now refused, where it was ignored: an + option nobody read would leave a replication unscheduled and nothing said. + +- A replicated volume now follows its container from one host to the other. + On a volume with a ``replicate:`` line scheduled, the first mount + asks that host for the last snapshot of the volume to appear there, + receives it, and restores it when it is the last to have appeared here and + came from another host; the last unmount snapshots the volume and sends + that snapshot. What the volume held is kept as a snapshot first. A host + that does not answer at mount refuses the mount, and pausing the line is + how to mount without asking. The README section "Move an application + between hosts" says the rest: what a crash keeps aside, why the clocks of + the hosts do not decide, and the thirty seconds Docker gives a mount. + + **Every** ``replicate:`` **line already in** ``schedule.csv`` **changes + meaning with this version**: the volumes it names start being brought to + what that host holds when a container starts, and sent to it when the + last one stops, and a replication host that is down keeps their + containers from starting. Pause the line, or delete it, for a volume that + must not. + + While no container uses a volume here, each scheduled round of its + replication also fetches from the other host what appeared there since, + without restoring it, so that a mount receives a difference and not a + whole volume. + + The scheduler's replication lock moves into the plugin, where the unmount + needs it too: a round finding a replication under way skips its turn as + before, where an unmount waits for it and then sends the final state. + +- ``buttervolume replicate `` snapshots a volume and sends that + snapshot in one step, which is what a scheduled replication does at each + round, and what moving an application by hand needed as two commands. The + scheduler now asks the plugin for that one step, through the new + ``/VolumeDriver.Replicate`` endpoint, instead of a snapshot, a send and a + cleanup of its own. Which replications are under way is the plugin's + business from now on, and a round that finds one under way is skipped as + before. + +- A send now refuses to bury a history it never saw. Before sending, the + remote host is asked what the last snapshot of the volume to appear there + is; when this host neither holds it nor holds the trace of exchanging it, + another host has sent its work there since, or somebody restored an older + snapshot there, and a send on top of that would pass this host's copy off + as the most recent one everywhere. The send is refused, the error names + the snapshot to receive first, and a scheduled replication that is refused + takes back the snapshot it took for the occasion and says so at every + round. A volume at rest sends nothing, so it asks nothing. + +- A restore no longer leaves a snapshot behind when the volume held nothing + new. What the volume held is kept as a snapshot before it is replaced, and + that snapshot is now the previous one when nothing changed since it, none + at all when the volume was empty; and a restore of the snapshot the volume + already holds does nothing. The answer of ``/VolumeDriver.Snapshot.Restore`` + always carries ``VolumeBackup``, naming the snapshot that holds what the + volume held or empty when it held nothing, and a new ``Restored`` field + saying whether the volume was replaced. + +- A snapshot is now compared with the last snapshot taken of the volume on + this host, in the order they were taken, and no longer with a snapshot + received from another host that happens to carry a later date. Such a + snapshot was never taken of this volume, and comparing with it made a host + at rest snapshot its stale volume again the minute after a fresher copy + arrived, then send that. + +- A receive now fetches the last snapshot that appeared on the other host, + taken there or received there, instead of the one carrying the latest date + in its name. That date is written by the clock of whichever host took the + snapshot, so a host whose clock ran ahead passed its copy off as the most + recent one, and the README could only advise keeping the clocks in + agreement. BTRFS numbers subvolumes in the order it creates them, and that + order is what the other host is now asked for. + - The plugin starts again, and the ``buttervolume`` command can be run inside it. A docker plugin is not started from the image configuration, so the ``PATH`` and ``PYTHONPATH`` the Dockerfile sets were not there and the diff --git a/README.rst b/README.rst index e933d5a..3d1c209 100644 --- a/README.rst +++ b/README.rst @@ -411,6 +411,32 @@ When you delete the volume with ``docker rm -v `` or ``docker volume rm ``, the BTRFS subvolume is deleted. If you snapshotted the volume elsewhere in the meantime, the snapshots won't be deleted. +A volume can ask for its scheduled jobs as it is created: an option named +after an action, with a number of minutes as its value, writes that line in +the schedule of the host the volume is created on:: + + docker volume create -d ccomb/buttervolume -o replicate:node2=1 -o purge:4h:1d:1w=60 db + +The line is written when no line says the same thing, and left alone when +one does, paused or not, so creating the volume again changes nothing. It is +never removed on its own. That is how a Docker Swarm service says, once, what +happens to its volume on whatever host it lands on:: + + docker service create --mount type=volume,source=db,target=/data,volume-driver=ccomb/buttervolume,volume-opt=replicate:node2=1 ... + +In a compose file, the same goes under ``driver_opts``, with the value +written as a string:: + + volumes: + db: + driver: ccomb/buttervolume + driver_opts: + "replicate:node2": "1" + +An option that is neither ``copyonwrite``, ``compression`` nor an action is +refused, and the volume is not created: an option nobody read would leave +the replication unscheduled and nothing said. + Managing volumes and snapshots ------------------------------ @@ -427,6 +453,7 @@ When buttervolume is installed, it provides a command line tool restore Restore a snapshot (optionally to a different volume) clone Clone a volume as new volume send Send a snapshot to another host + replicate Snapshot a volume and send the snapshot to another host receive Receive from another host its last snapshot of a volume sync Synchronise a volume from a remote host volume rm Delete a snapshot @@ -468,14 +495,20 @@ to live in ``/var/lib/buttervolume/volumes``. Restore a snapshot ------------------ -You can restore a snapshot as a volume. The current volume will first -be snapshotted, deleted, then replaced with the snapshot. If you provide a -volume name instead of a snapshot, the **latest snapshot** is restored. So no -data is lost if you do something wrong. Please take care of stopping the -container before restoring a snapshot:: +You can restore a snapshot as a volume. What the volume holds is kept as a +snapshot first, then the volume is deleted and replaced with the snapshot. If +you provide a volume name instead of a snapshot, the **latest snapshot** is +restored. So no data is lost if you do something wrong. Please take care of +stopping the container before restoring a snapshot:: buttervolume restore +The command prints the name of the snapshot that holds what the volume held. +That is a new one only when the volume had changed since its last snapshot: +when it had not, that last snapshot is the one, and nothing is written. An +empty volume has nothing to keep, and a volume that already holds exactly the +snapshot asked for is left alone. + ```` is the name of the snapshot, not the full path. It is expected to live in ``/var/lib/buttervolume/snapshots``. @@ -539,8 +572,11 @@ to the volume. **One host writes at a time: replicate.** This is the failover case, and the "move this application to another node" case. It works with any data, including a database, because a BTRFS snapshot is a coherent image of the -whole volume at one instant. What replication cannot do is merge: two hosts -replicating to each other and restoring would each throw away the work of the +whole volume at one instant. A replicated volume follows its container: the +host a container starts on takes what the other hosts hold, and the host it +stops on sends what it wrote (see `Move an application between hosts`_). What +replication cannot do is merge: two hosts writing to the same volume at the +same time and replicating to each other would each throw away the work of the other, and the last one to speak would win. **Several hosts write at the same time: synchronize.** Each host pulls from @@ -579,18 +615,35 @@ consuming a lot of bandwith or disk space:: to live in ``/var/lib/buttervolume/snapshots`` and is replicated to the same path on the remote host. -What the remote host already holds is read from the trace kept locally, -named ``@@``, and nothing is asked of the remote host. -That trace is written after each send, and after each receive from that host, -since a snapshot that has just arrived from a host is a snapshot that host -holds. So a snapshot whose trace is there is not sent a second time, and a -copy deleted on the remote host behind Buttervolume's back goes unnoticed: -delete the trace as well, and the next send carries the whole volume again. +To snapshot a volume and send that snapshot in one step, which is what a +scheduled replication does at each round, name the volume instead:: + + buttervolume replicate + +It prints the name of the snapshot the remote host now holds. A volume +unchanged since its last snapshot sends that one, and sends nothing when the +remote host already has it. -A replication scheduled on a volume at rest therefore costs nothing at all: the -snapshot it would take is not taken, its trace says the remote host already has -it, and nothing crosses the network. Replicating the same volume to two hosts -every minute is a reasonable thing to schedule. +What the remote host already holds is read from the trace kept locally, +named ``@@``. That trace is written after each send, +and after each receive from that host, since a snapshot that has just arrived +from a host is a snapshot that host holds. So a snapshot whose trace is there +is not sent a second time, and a copy deleted on the remote host behind +Buttervolume's back goes unnoticed: delete the trace as well, and the next +send carries the whole volume again. + +Before a snapshot is sent, the remote host is asked what the last snapshot of +that volume to appear there is. When that one is unknown here, neither held +nor traced, the volume over there has moved on without this host: another +host sent its work there, or somebody restored an older snapshot there. A send +over that would make this host's copy pass for the most recent one on every +host and bury the other history under it, so it is refused, and the error +says which snapshot to receive first, or to delete over there. + +A replication scheduled on a volume at rest costs nothing at all: the snapshot +it would take is not taken, its trace says the remote host already has it, and +nothing crosses the network, not even that question. Replicating the same +volume to two hosts every minute is a reasonable thing to schedule. ```` is the hostname or IP address of the remote host. The snapshot is @@ -619,6 +672,88 @@ The default SSH_PORT of the ssh server included in the plugin is **1122**. You c change it with `docker plugin set ccomb/buttervolume SSH_PORT=` before enabling the plugin. +Move an application between hosts +--------------------------------- + +Say the volume ``db`` of an application is replicated to ``node2`` from every +host the application can run on, with the same line scheduled on each of +them:: + + buttervolume schedule replicate:node2 1 db + +or, with Docker Swarm, asked for by the volume itself, so that the line is +written on whatever host the service lands on (see `Creating and deleting +volumes`_):: + + docker service create --mount type=volume,source=db,target=/data,volume-driver=ccomb/buttervolume,volume-opt=replicate:node2=1 ... + +The application runs on ``host1``, which sends a snapshot to ``node2`` every +minute the volume changed. Stop it there and start it on ``host2``, by hand or +because Docker Swarm moved it, and the volume follows: + +- when the last container using ``db`` stops on ``host1``, the volume is + snapshotted and that snapshot sent to ``node2``, right then, before Docker + goes on. A replication under way from the scheduled round is waited for, so + what leaves is the final state; +- when the first container using ``db`` starts on ``host2``, ``node2`` is asked + for the last snapshot of ``db`` that appeared there, it is received when it + is not here already, and it becomes the volume. What the volume held on + ``host2`` is kept as a snapshot first, unless a snapshot already holds + exactly that, or the volume was empty, which is what a volume Docker just + created is. The volume being no longer empty, Docker does not copy the + content of the image into it. + +The volume is brought to the last snapshot that appeared on this host **when +that one came from another host**. A snapshot taken here since, by a +scheduled snapshot or by the stop of a container, means the volume's own +history goes on, and nothing is restored: a container that restarts on the +same host, and a host that reboots without stopping its containers, keep +what they wrote. "Last" is read from the order BTRFS created the snapshots +in, never from the dates in their names, so the clocks of the hosts do not +decide which copy is the most recent one. + +A host that does not answer when a container starts **refuses the mount**, +and Docker reports the error. Mounting on "there is nothing over there" when +nobody said so would start the application on whatever this host holds, +possibly the state of a week ago, and its writes would then bury the work +done elsewhere. To start the application anyway, pause the replication on +that host; the mount then asks nobody:: + + buttervolume schedule replicate:node2 pause db + +A host that does not answer when a container stops leaves the snapshot here, +and the scheduled replication sends it when the host is back. Docker reports +the error and stops the container anyway. + +The other side of the same rule is the send. A host that crashed with writes +it never sent comes back with another history of the volume, older than what +``node2`` received from the host the application moved to. Its scheduled +replication is refused, and says so at every round, until a container starts +there again: the mount then brings the work done elsewhere in, and keeps the +unsent writes as a snapshot, named in the log. A rounds' send is also refused +after somebody restored an older snapshot on ``node2`` by hand; the error +names the snapshot to receive first, or to delete over there. + +While no container uses the volume on a host, each scheduled round fetches +from ``node2`` what appeared there since, without restoring it, so that a +mount has a difference to receive and not a whole volume. That matters +because Docker gives a plugin thirty seconds to answer a mount, and the +first container of a volume this host has never seen receives the whole +volume. Either give Docker more patience:: + + docker plugin install --disable ccomb/buttervolume + docker plugin enable --timeout 600 ccomb/buttervolume + +or create the volume with its replication scheduled ahead of the container, +and let a round go by. + +Docker calls the plugin once per container, so a volume shared by two +containers on the same host changes hands at the first start and at the last +stop only. The plugin counts them in memory: after it restarts, the next +container to stop replicates the volume even when another one still runs, +which sends a coherent state a minute early and harms nothing. + + Receive a snapshot from another host ------------------------------------ @@ -628,10 +763,10 @@ The other direction, for a host that wants back what another one holds:: It names a **volume**, where ``send`` names a snapshot: whoever receives does not know what the other host has, which is precisely the question this asks. -The most recent snapshot that host keeps of that volume is fetched into -``/var/lib/buttervolume/snapshots``, and its name is printed. Only the -difference crosses the network when the two hosts still share an older -snapshot to build on. +The last snapshot of that volume that appeared on that host, taken there or +received there, is fetched into ``/var/lib/buttervolume/snapshots``, and its +name is printed. Only the difference crosses the network when the two hosts +still share an older snapshot to build on. The command **does not restore anything**. It brings a snapshot over, and which snapshot becomes the volume stays a separate, explicit decision:: @@ -639,6 +774,10 @@ which snapshot becomes the volume stays a separate, explicit decision:: buttervolume receive node2 www buttervolume restore www +On a host where the volume is replicated to ``node2``, that decision is taken +by the next mount, which restores the last snapshot to have appeared here when +it came from another host (see `Move an application between hosts`_). + A host that keeps no snapshot of that volume and a host that could not answer are two different answers, and never the same one. An unreachable host, a refused connection, an ssh that takes too long: each is reported as the error @@ -646,10 +785,12 @@ it is. Only a host that answered and holds nothing is reported as holding nothing. Reading silence as "there is nothing over there" is how the good copy of a volume gets replaced by an older one. -A snapshot carries the moment it was taken on the machine that took it, and -"the most recent" is read from that name. A host whose clock runs ahead -therefore passes its copy off as the most recent one, so the hosts replicating -to each other should agree on the time. +A snapshot carries in its name the moment it was taken, by the clock of the +machine that took it, and that date is not what decides which one is the last. +BTRFS numbers subvolumes in the order it creates them, and a received snapshot +is created on arrival, so "the last one" is read from that order on the host +that answers. A host whose clock runs ahead does not pass its copy off as the +most recent one. Synchronize a volume from another host volume diff --git a/buttervolume/api.py b/buttervolume/api.py index ea98104..3d1cd23 100644 --- a/buttervolume/api.py +++ b/buttervolume/api.py @@ -155,6 +155,14 @@ def send(snapshot, host, test=False): return get_from(_post("/VolumeDriver.Snapshot.Send", payload, test), "") is not False +def replicate(volume, host, test=False): + """Snapshot a volume and send that snapshot to a host, and answer its name.""" + payload = {"Name": volume, "Host": host} + if test: + payload["Test"] = True + return get_from(_post("/VolumeDriver.Replicate", payload, test), "Snapshot") + + def receive(volume, host, test=False): """Fetch the last snapshot another host has of a volume, and answer its name.""" payload = {"Name": volume, "Host": host} diff --git a/buttervolume/btrfs.py b/buttervolume/btrfs.py index 48d4e49..d26ab82 100644 --- a/buttervolume/btrfs.py +++ b/buttervolume/btrfs.py @@ -8,12 +8,19 @@ Nothing here knows what a volume is, which is why ``BtrfsError`` lives here rather than with the errors the API answers with. + +What ``btrfs subvolume show`` prints is read here too, for one subvolume or +for a whole directory of them, and read into ``Listed``: the name, the rank a +subvolume was created at, and whether it arrived through ``btrfs receive``. +That rank is how the plugin orders snapshots in time, because the date in a +snapshot's name is the clock of whichever host took it. """ import contextlib import os import tempfile import time +from dataclasses import dataclass from subprocess import PIPE, CalledProcessError, Popen, TimeoutExpired from subprocess import run as _run @@ -61,6 +68,122 @@ def run_safe(cmd, timeout, error=BtrfsError): raise error(f"Command could not be run: {' '.join(cmd)}\n{e}") from e +# The fields ``btrfs subvolume show`` prints, one per line, before the list of +# snapshots. Reading them by name rather than by line number keeps a version +# that prints one more of them from shifting the others. +SHOW_FIELDS = frozenset( + [ + "Name", + "UUID", + "Parent UUID", + "Received UUID", + "Creation time", + "Subvolume ID", + "Generation", + "Gen at creation", + "Parent ID", + "Top level ID", + "Flags", + "Send transid", + "Send time", + "Receive transid", + "Receive time", + "Quota group", + ] +) + + +def parse_shows(text): + """The subvolumes described by this ``btrfs subvolume show`` output, one dict each. + + Pure: the output of one call, or of several concatenated, which is how a + remote host describes a whole directory in a single command. Each + description opens with the path of the subvolume, unindented, and goes on + with indented ``Field: value`` lines. What follows ``Snapshot(s):`` is a + list of paths, not fields, and is left out. + """ + shown = [] + for line in text.splitlines(): + if not line.strip(): + continue + if not line[0].isspace(): + shown.append({}) + continue + field, _, value = line.strip().partition(":") + if shown and field in SHOW_FIELDS: + shown[-1][field] = value.strip() + return shown + + +@dataclass(frozen=True) +class Listed: + """A subvolume of a directory: its name, its rank, and whether it was received. + + The rank is the subvolume id, which a filesystem hands out in creation + order, so it says in what order the subvolumes appeared there, whatever + their names say. A subvolume that arrived through ``btrfs receive`` carries + the id of the one it was sent from, and ``received`` says so; a writable + snapshot made from it does not inherit that mark, nor do the snapshots + taken from that writable one. + """ + + name: str + rank: int + received: bool + + +def parse_listing(text): + """The subvolumes of a directory in creation order, read from their descriptions. + + Pure: ``text`` is what ``listing_command`` prints, the ``show`` of every + subvolume of the directory. A description missing a field it should have + raises rather than being skipped, because a listing that is read wrong is + answered as a host holding less than it does. + """ + listed = [] + for shown in parse_shows(text): + try: + listed.append( + Listed(shown["Name"], int(shown["Subvolume ID"]), shown["Received UUID"] != "-") + ) + except (KeyError, ValueError) as e: + raise BtrfsError(f"Unreadable subvolume description {shown}: {e}") from e + return sorted(listed, key=lambda s: s.rank) + + +def listing_command(directory): + """The shell command that describes every subvolume of this directory. + + Meant to run on another host over ssh, which is why it is a shell string: + ``cd`` fails when the directory is not there and the version check when + ``btrfs`` is not, and either failure reaches the caller as a non-zero + status, where an empty output only ever means a directory holding no + subvolume. A directory that is not a subvolume prints nothing, which is + what it is worth. The names never enter the command: the shell lists them + itself. + """ + return ( + f"cd {directory} && btrfs --version > /dev/null && " + '{ for s in */; do btrfs subvolume show "${s%/}" 2> /dev/null; done; true; }' + ) + + +def subvolumes_in(directory): + """The subvolumes of this directory, in the order they appeared there. + + A directory that is not a subvolume is left out, the way ``list_volumes`` + leaves it out of the volumes: it is not something ``btrfs`` made. + """ + listed = [] + for name in os.listdir(directory): + subvolume = Subvolume(os.path.join(directory, name)) + if not subvolume.exists(): + continue + shown = subvolume.show() + listed.append(Listed(name, int(shown["Subvolume ID"]), shown["Received UUID"] != "-")) + return sorted(listed, key=lambda s: s.rank) + + class Subvolume: """basic wrapper around the CLI""" @@ -69,33 +192,18 @@ def __init__(self, path): self.path = os.path.abspath(path) def show(self): - """Parse btrfs subvolume show output""" + """What ``btrfs subvolume show`` says of this subvolume, as a dict.""" raw = run_safe( ["btrfs", "subvolume", "show", self.path], timeout=SHOW_TIMEOUT, error=BtrfsSubvolumeError, ) - lines = raw.split("\n") - - if len(lines) < 13: + shown = parse_shows(raw) + if len(shown) != 1: raise BtrfsSubvolumeError( f"Unexpected output format from 'btrfs subvolume show {self.path}'" ) - - # Parse key-value pairs from lines 1-12 - output = {} - for line in lines[1:12]: - if ":" in line: - k, v = line.split(":", 1) - output[k.strip()] = v.strip() - - # Check for snapshots section - if len(lines) > 12 and "Snapshot(s):" in lines[12]: - output["Snapshot(s)"] = [s.strip() for s in lines[13:] if s.strip()] - else: - output["Snapshot(s)"] = [] - - return output + return shown[0] def exists(self): """Check if this path is a valid BTRFS subvolume""" diff --git a/buttervolume/cli.py b/buttervolume/cli.py index b077a9d..cfca93d 100644 --- a/buttervolume/cli.py +++ b/buttervolume/cli.py @@ -186,6 +186,13 @@ def send(args, test=False): return api.send(args.snapshot[0], args.host[0], test=test) +def replicate(args, test=False): + res = api.replicate(args.volume[0], args.host[0], test=test) + if res: + print(res) + return res + + def receive(args, test=False): res = api.receive(args.volume[0], args.host[0], test=test) if res: @@ -363,6 +370,16 @@ def main(): parser_send.add_argument("host", metavar="host", nargs=1, help="Host to send the snapshot to") parser_send.add_argument("snapshot", metavar="snapshot", nargs=1, help="Snapshot to send") + parser_replicate = subparsers.add_parser( + "replicate", help="Snapshot a volume and send the snapshot to another host" + ) + parser_replicate.add_argument( + "host", metavar="host", nargs=1, help="Host to send the snapshot to" + ) + parser_replicate.add_argument( + "volume", metavar="volume", nargs=1, help="Volume to snapshot and send" + ) + parser_receive = subparsers.add_parser( "receive", help="Receive the last snapshot another host has of a volume" ) @@ -440,6 +457,7 @@ def main(): parser_restore.set_defaults(func=restore) parser_clone.set_defaults(func=clone) parser_send.set_defaults(func=send) + parser_replicate.set_defaults(func=replicate) parser_receive.set_defaults(func=receive) parser_sync.set_defaults(func=sync) parser_remove.set_defaults(func=remove) diff --git a/buttervolume/names.py b/buttervolume/names.py index fa4a307..a97f102 100644 --- a/buttervolume/names.py +++ b/buttervolume/names.py @@ -175,33 +175,67 @@ def sent_snapshots(volume, host, names): ) -def _taken_snapshots(volume, names): - """The snapshots taken of this volume among these names, the traces left out. +def taken_snapshots(volume, names): + """The snapshots of this volume among these names, in that order, traces left out. A trace of a send is named after the snapshot it was made from, plus the - target, so `www@2026-08-26T10:00:00.000000@node3` sorts after the snapshot - itself and would pass for the most recent one. + target, so `www@2026-08-26T10:00:00.000000@node3` would pass for a snapshot + of the volume. The order is the one the names come in, which is the order + they appeared on the host that listed them: the date in a name is what + that host's clock said, and two hosts need not agree on the time. Every name of this volume is read, and one that cannot be read raises rather than being dropped. These names come from another machine, and a listing we could not read must never be answered as a host with nothing. """ - return [s for s in map(Snapshot.parse, snapshots_of(volume, names)) if not s.host] + return [ + s + for s in map(Snapshot.parse, (n for n in names if n.startswith(volume + "@"))) + if not s.host + ] + + +def snapshot_to_restore(empty, events, candidates): + """Which snapshot a replicated volume is brought to before its first container starts. + + Pure. ``events`` are the snapshots of the volume as they appeared on this + host, oldest first, each as ``(name, came_from_another_host)``; + ``candidates`` are the names the hosts the volume is replicated to hold as + their last, already fetched. The answer is a name, or None when the volume + is what it should be. + + An empty volume, the one Docker just created or recreated, takes the last + candidate to have appeared here, and the last snapshot when the hosts hold + nothing. Any other volume is brought to the last snapshot to have appeared + here when that one came from another host: another host wrote since this + one did. When the last one was taken here, the volume's own history goes + on, whatever the dates in the names say, and whatever was received before. + """ + if not events: + return None + if empty: + arrived = [name for name, _ in events if name in set(candidates)] + return arrived[-1] if arrived else events[-1][0] + name, from_elsewhere = events[-1] + return name if from_elsewhere else None def snapshot_to_fetch(volume, remote_names, local_names): """What to ask that host for, and the parent both sides already hold. The pair `(snapshot, parent)`, or None when the host holds no snapshot of - this volume. The parent is the most recent older snapshot the two sides - have in common, which is what an incremental transfer is built on; None - when they have none, and then the whole volume has to come over. + this volume. ``remote_names`` come in the order they appeared over there, + so the one to fetch is the last of them: the last that host took or + received, whatever the dates in the names say. The parent is the last + earlier one the two sides have in common, which is what an incremental + transfer is built on; None when they have none, and then the whole volume + has to come over. The parent is read from the two listings rather than from the trace of a send, which says what we once sent there. The question here is what that host has now, and it has just been asked. """ - remote = _taken_snapshots(volume, remote_names) + remote = taken_snapshots(volume, remote_names) if not remote: return None # what we hold is read as plain names, none of which has to make sense: a diff --git a/buttervolume/plugin.py b/buttervolume/plugin.py index 4227174..f75b153 100644 --- a/buttervolume/plugin.py +++ b/buttervolume/plugin.py @@ -54,13 +54,15 @@ parsed, sent_snapshots, snapshot_to_fetch, + snapshot_to_restore, snapshots_of, + taken_snapshots, validate_hostname, validate_snapshot_name, validate_volume_name, ) from buttervolume.purge import Pattern, compute_purges -from buttervolume.schedule import Entry, Job, read_schedule, write_schedule +from buttervolume.schedule import Entry, Job, Replicate, read_schedule, write_schedule log = logging.getLogger() @@ -282,17 +284,17 @@ def run_btrfs_receive(remote_host, remote_snapshot_path, remote_parent_path=None def snapshots_on_remote(remote_host, remote_path, port): - """The names this host keeps in that directory. + """The names this host keeps in that directory, in the order they appeared there. An empty answer means it answered and keeps nothing there. A host that could not answer raises instead, and is never read as a host with nothing: acting on "there is nothing over there" when nobody said so is how the good copy of a volume gets replaced by an older one. - The whole directory is listed rather than a `volume@*` pattern. The - pattern would carry a name into a shell on the other machine, and `ls` - leaves with the same non-zero status when nothing matches as when it - failed, which is exactly the difference this function exists to keep. + The whole directory is described rather than a `volume@*` pattern, which + would carry a name into a shell on the other machine. The order is the one + the filesystem over there hands out, so the last name is the last snapshot + that host took or received, whatever its clock wrote in the name. """ validate_hostname(remote_host) ssh_cmd = [ @@ -302,7 +304,7 @@ def snapshots_on_remote(remote_host, remote_path, port): "-o", "StrictHostKeyChecking=no", remote_host, - f"ls -1 {remote_path}", + btrfs.listing_command(remote_path), ] try: listing = subprocess.run(ssh_cmd, capture_output=True, timeout=REMOTE_TIMEOUT) @@ -318,7 +320,7 @@ def snapshots_on_remote(remote_host, remote_path, port): # nothing is decoded strictly: a name we cannot read is one this host # should not have written, and it is refused later by name rather than # here by an error nobody expected - return listing.stdout.decode(errors="replace").splitlines() + return [s.name for s in btrfs.parse_listing(listing.stdout.decode(errors="replace"))] @route("/Plugin.Activate") @@ -326,15 +328,89 @@ def plugin_activate(_): return {"Implements": ["VolumeDriver"]} +VOLUME_OPTIONS = ("copyonwrite", "compression") + + +def is_a_timer(text): + """Whether this names a number of minutes a job waits between two runs. + + isdecimal, not isnumeric: "²".isnumeric() is True and int("²") raises, + and the scheduler reads this timer back with int(). + """ + return text.isdecimal() and int(text) > 0 + + +def jobs_in_options(opts): + """The jobs the options of a volume ask to schedule, as (action, timer) pairs. + + Pure. An option named after an action, ``replicate:node2`` with a number + of minutes as its value, is a line to write in the schedule. An option + that is neither that nor one of the volume's own is refused: one nobody + reads would leave a replication unscheduled, and nothing said. + """ + jobs = [] + for key, value in opts.items(): + if key in VOLUME_OPTIONS: + continue + try: + Job.parse(key) + except ValidationError as e: + raise ValidationError( + f"Unknown option '{key}'. Options are {', '.join(VOLUME_OPTIONS)}, or an action " + f"to schedule on the volume with a number of minutes as its value: {e}" + ) from e + timer = str(value) + if not is_a_timer(timer): + raise ValidationError( + f"Invalid value '{value}' for option '{key}': it must be a number of minutes" + ) + jobs.append((key, timer)) + return jobs + + +def refuse_if_paused(name, jobs): + """A paused schedule takes no line, and says so before a volume is created.""" + if jobs and os.path.exists(SCHEDULE_DISABLED): + raise ValidationError( + f"Schedule is globally paused, so nothing can be scheduled on {name}: resume it first" + ) + + +def schedule_lines(name, jobs): + """Write these jobs of this volume in the schedule, leaving alone the lines already there. + + A line already written keeps its timer and stays paused if it is: the + options of a volume say what to schedule when nothing is, and pausing a + line is somebody's decision. + """ + if not jobs: + return + refuse_if_paused(name, jobs) + if not os.path.exists(SCHEDULE): + os.makedirs(dirname(SCHEDULE), exist_ok=True) + with open(SCHEDULE, "w") as f: + f.write("") + schedule = {(entry.name, entry.action): entry for entry in read_schedule(SCHEDULE)} + for action, timer in jobs: + if (name, action) not in schedule: + schedule[(name, action)] = Entry(name, action, timer, "True") + log.info("Scheduled %s of %s every %s minutes, as its options ask", action, name, timer) + write_schedule(SCHEDULE, schedule.values()) + + @route("/VolumeDriver.Create") def volume_create(req): + """Create a volume, and schedule what its options ask for. + + The options are read before anything is created, so a bad one leaves + nothing behind, and the jobs they ask for are scheduled whether the + volume is created or already there: a service that Docker Swarm deploys + again onto a host that kept the volume has to find its replication + scheduled all the same. + """ name = req["Name"] opts = req.get("Opts", {}) or {} - volpath = volumepath(name) - # volume already exists? - if name in [v["Name"] for v in list_volumes()["Volumes"]]: - return {"Err": ""} cow = opts.get("copyonwrite", "true").lower() if cow not in ["true", "false"]: @@ -346,6 +422,13 @@ def volume_create(req): return { "Err": f"Invalid option for compression: {compression}. Valid options: {', '.join(valid_compression[1:])}" } + jobs = jobs_in_options(opts) + refuse_if_paused(name, jobs) + + # volume already exists? + if name in [v["Name"] for v in list_volumes()["Volumes"]]: + schedule_lines(name, jobs) + return {"Err": ""} btrfs.Subvolume(volpath).create(cow=cow == "true") @@ -358,12 +441,25 @@ def volume_create(req): log.warning(f"Could not enable compression for volume {name}: {e}") # Don't fail volume creation if compression setting fails + schedule_lines(name, jobs) return {"Err": ""} @route("/VolumeDriver.Mount") def volume_mount(req): - return {"Mountpoint": existing_volume(req["Name"]).path, "Err": ""} + """Give the path of the volume, brought first to what the other hosts hold. + + Only the first mount does that, and only for a volume with a replication + scheduled. A mount that fails leaves the count where it was: Docker will + not call Unmount for a container it did not start. + """ + name = req["Name"] + volume = existing_volume(name) + with mount_lock(name): + if not mounted.get(name): + take_over(name, req.get("Test", False)) + mounted[name] = mounted.get(name, 0) + 1 + return {"Mountpoint": volume.path, "Err": ""} @route("/VolumeDriver.Path") @@ -372,7 +468,13 @@ def volume_path(req): @route("/VolumeDriver.Unmount") -def volume_unmount(_): +def volume_unmount(req): + """Send the final state of the volume to the other hosts, once its last container stopped.""" + name = req["Name"] + with mount_lock(name): + mounted[name] = max(0, mounted.get(name, 0) - 1) + if not mounted[name]: + hand_over(name, req.get("Test", False)) return {"Err": ""} @@ -384,7 +486,13 @@ def volume_get(req): @route("/VolumeDriver.Remove") def volume_remove(req): - existing_volume(req["Name"]).delete() + name = req["Name"] + existing_volume(name).delete() + # Docker removes no volume a container uses, so the count is what a + # restart left behind, and the volume recreated under this name starts + # from nothing + with mount_lock(name): + mounted.pop(name, None) return {"Err": ""} @@ -500,13 +608,39 @@ def keep_trace(snapshot, remote_host): log.warning("Failed to delete old snapshot %s: %s", str(old), str(e)) +def continues(volume_name, snapshot, remote_host): + """Whether a send from this volume continues the history that snapshot belongs to. + + It does when the snapshot was taken from this volume, when the volume was + made from it by a restore, or when this host is the one that sent it there: + the trace of that send is a snapshot taken here, where the trace left by + a receive carries the mark of the subvolume it was received from. What + the volume holds is then the continuation of what that host holds. A + snapshot fetched from that host and never restored is none of these: the + volume here is another history, and sending it would bury that one. + """ + volume = existing_volume(volume_name).show() + path = snapshotpath(str(snapshot)) + if os.path.exists(path): + shown = btrfs.Subvolume(path).show() + if shown["Parent UUID"] == volume["UUID"] or volume["Parent UUID"] == shown["UUID"]: + return True + trace = snapshotpath(str(snapshot.sent_to(remote_host))) + return os.path.exists(trace) and btrfs.Subvolume(trace).show()["Received UUID"] == "-" + + @route("/VolumeDriver.Snapshot.Send") def snapshot_send(req): - """The last sent snapshot is remembered by adding a suffix with the target""" - test = req.get("Test", False) - snapshot_name = req["Name"] - remote_host = req["Host"] + """Send a snapshot to another host.""" + send_snapshot(req["Name"], req["Host"], req.get("Test", False)) + return {"Err": ""} + +def send_snapshot(snapshot_name, remote_host, test=False): + """Send this snapshot to that host, incrementally when a previous send allows it. + + The last sent snapshot is remembered by adding a suffix with the target. + """ snapshot = Snapshot.parse(snapshot_name) if snapshot.host: # sending a trace would name its own trace, and btrfs would refuse @@ -536,6 +670,24 @@ def snapshot_send(req): parent_path = snapshotpath(str(latest)) if latest else None port = os.getenv("SSH_PORT", "1122") + # the last snapshot of this volume that appeared over there has to be one + # this volume continues, or the volume over there has moved on without + # this host: another host sent it, or somebody restored it there. Sending + # over that would make this host's copy pass for the most recent one + # everywhere and bury the other history under it, so it is refused, and + # said. A host that holds nothing of the volume lets the first send + # through, and a volume at rest never gets here, so nothing is asked of a + # host at rest. + arrived = taken_snapshots( + snapshot.volume, snapshots_on_remote(remote_host, remote_snapshots, port) + ) + if arrived and not continues(snapshot.volume, arrived[-1], remote_host): + raise ReplicationError( + f"{remote_host} holds {arrived[-1]}, which this host never saw: the volume " + f"there has moved on. Receive it first with `buttervolume receive {remote_host} " + f"{snapshot.volume}`, or delete it there" + ) + try: log.info("Sending snapshot %s to %s", snapshot_path, remote_host) run_btrfs_send_receive(snapshot_path, remote_host, remote_snapshots, parent_path, port) @@ -567,7 +719,172 @@ def snapshot_send(req): keep_trace(snapshot, remote_host) - return {"Err": ""} + +# One replication of a volume at a time. A replication is a snapshot and a +# send, and two of them interleaved on the same volume would send the same +# snapshot twice, or a newer one before an older. The scheduled round refuses +# to wait, since a send longer than its timer would otherwise pile rounds up +# behind it; the unmount of a volume waits, because what it has to send is the +# final state, after whatever was under way. The locks live in memory alone, +# and that is right: a daemon that stops has no replication under way either. +replications = {} +# Docker calls Mount once per container that uses a volume and Unmount once per +# container that stops, so a volume shared by two containers is mounted twice. +# This counts them: the first mount and the last unmount are the two moments a +# replicated volume changes hands, and nothing in between is. A restarted +# plugin counts from zero again, and that costs, at worst, a replication at +# the stop of a container that was not the last one, which sends a coherent +# state a minute early, and a mount that asks the other hosts while a +# container runs here, which finds nothing newer there. +mounted = {} +mounts = {} +locks_lock = threading.Lock() + + +def lock_of(table, name): + """The lock of this volume in this table, made on first use.""" + with locks_lock: + return table.setdefault(name, threading.Lock()) + + +def replication_lock(name): + return lock_of(replications, name) + + +def mount_lock(name): + return lock_of(mounts, name) + + +def replicate(name, remote_host, test=False, wait=False): + """Snapshot this volume and send that snapshot to this host, as one step. + + Answers the name of the snapshot that host now holds. A send that fails + takes back the snapshot this call took for it, and only that one: a + volume nobody wrote to gives back the snapshot of an earlier call, which a + failure here is no reason to delete. ``wait`` says what to do when a + replication of this volume is under way: wait for it, or refuse. + """ + lock = replication_lock(name) + if not lock.acquire(blocking=wait): + raise ReplicationError(f"Replication of {name} already in progress") + try: + snapshot, created = take_snapshot(name) + try: + send_snapshot(snapshot, remote_host, test) + except Exception: + if created: + try: + btrfs.Subvolume(snapshotpath(snapshot)).delete() + log.info("Removed snapshot %s of the failed replication", snapshot) + except BtrfsError as e: + log.warning( + "Could not remove snapshot %s of the failed replication: %s", snapshot, e + ) + raise + return snapshot + finally: + lock.release() + + +@route("/VolumeDriver.Replicate") +def volume_replicate(req): + """Snapshot a volume and send the snapshot to another host, as one step. + + On a host where no container uses the volume, what the other host holds + is fetched first. It is worth having before a container asks for the + volume here: the mount then has a difference to receive rather than a + whole volume, within the time Docker gives it. Nothing is restored here; + which snapshot becomes the volume is decided at the mount, and only then. + """ + name, remote_host, test = req["Name"], req["Host"], req.get("Test", False) + with mount_lock(name): + idle = not mounted.get(name) + if idle: + fetch(name, remote_host, test) + return {"Err": "", "Snapshot": replicate(name, remote_host, test)} + + +def replication_hosts(name): + """The hosts this volume is replicated to, read from the lines of the schedule that run. + + A paused line, and a paused schedule, name no host. Pausing is how a + volume is mounted without asking a host that is down, and unmounted + without sending to it. + """ + if not os.path.exists(SCHEDULE): + return [] + hosts = [] + for entry in read_schedule(SCHEDULE): + if entry.name != name or not entry.enabled: + continue + try: + job = Job.parse(entry.action) + except ValidationError: + continue + if isinstance(job, Replicate): + hosts.append(job.host) + return hosts + + +def take_over(name, test=False): + """Bring a replicated volume to what the other hosts hold, before its first container starts. + + Each host the volume is replicated to is asked for the last snapshot of + it that appeared there, and that one is received when it is not here. The + volume is then brought to the last snapshot that came from another host, + when it is the last to have appeared here: a snapshot taken here since is + the volume's own history going on. An empty volume takes what the hosts + hold. A host that does not answer stops the mount, and the error says how + to mount without asking: acting on "there is nothing over there" when + nobody said so is how the good copy of a volume gets replaced by an older + one, and here the application would then write on top of it. + """ + hosts = replication_hosts(name) + if not hosts: + return + candidates = [] + for host in hosts: + try: + fetched = fetch(name, host, test) + except ReplicationError as e: + raise ReplicationError( + f"Not mounting {name} without knowing what {host} holds: {e}. To mount it " + f"anyway, pause its replication: buttervolume schedule replicate:{host} pause {name}" + ) from e + if fetched: + candidates.append(fetched) + # read with stat rather than listed: listing a directory updates its + # access time, which the restore would take for a change. BTRFS keeps the + # size of a directory as twice the length of the names in it, so zero is + # empty. + empty = os.stat(volumepath(name)).st_size == 0 + target = snapshot_to_restore(empty, snapshots_here(name), candidates) + if target: + log.info("Bringing %s to %s before it is mounted", name, target) + restore(target, name) + + +def hand_over(name, test=False): + """Send the final state of a replicated volume to the other hosts, once its last container stopped. + + A replication under way, from a scheduled round, is waited for: what has + to go is the state after it. A host that cannot be reached is reported and + the others are still sent to; the scheduled replication sends the state + to it at its next round, when it is back. + """ + errors = [] + for host in replication_hosts(name): + try: + snapshot = replicate(name, host, test, wait=True) + log.info("Sent the final state of %s to %s as %s", name, host, snapshot) + except Exception as e: + log.error("Could not send the final state of %s to %s: %s", name, host, e) + errors.append(f"{host}: {e}") + if errors: + raise ReplicationError( + f"The final state of {name} could not be sent to {'; '.join(errors)}. " + "The scheduled replication will send it at its next round" + ) # One receive at a time. A `btrfs receive` that stops early leaves behind what @@ -610,11 +927,19 @@ def receive_or_clean(remote_host, remote_snapshot_path, path, remote_parent_path @route("/VolumeDriver.Snapshot.Receive") def snapshot_receive(req): - """Fetch from another host the most recent snapshot it has of a volume""" - test = req.get("Test", False) - volume = req["Name"] - remote_host = req["Host"] + """Fetch from another host the last snapshot of a volume that appeared there""" + snapshot = fetch(req["Name"], req["Host"], req.get("Test", False)) + if not snapshot: + raise SnapshotNotFoundError(f"{req['Host']} keeps no snapshot of volume '{req['Name']}'") + return {"Err": "", "Snapshot": snapshot} + +def fetch(volume, remote_host, test=False): + """Fetch the last snapshot of this volume that appeared on that host, and name it. + + None when that host answered that it keeps no snapshot of the volume. A + host that does not answer raises: it is never read as a host with nothing. + """ validate_volume_name(volume) validate_hostname(remote_host) remote_snapshots = SNAPSHOTS_PATH if not test else TEST_REMOTE_PATH @@ -629,7 +954,7 @@ def snapshot_receive(req): os.listdir(SNAPSHOTS_PATH), ) if not found: - raise SnapshotNotFoundError(f"{remote_host} keeps no snapshot of volume '{volume}'") + return None snapshot, parent = found path = snapshotpath(str(snapshot)) @@ -646,7 +971,7 @@ def snapshot_receive(req): ) log.info("Snapshot %s is already here, nothing to receive", snapshot) keep_trace(snapshot, remote_host) - return {"Err": "", "Snapshot": str(snapshot)} + return str(snapshot) remote_path = join(remote_snapshots, str(snapshot)) parent_path = join(remote_snapshots, str(parent)) if parent else None @@ -668,7 +993,7 @@ def snapshot_receive(req): # that host holds this snapshot, we have just read it there. Without # this trace the first send back would carry the whole volume again. keep_trace(snapshot, remote_host) - return {"Err": "", "Snapshot": str(snapshot)} + return str(snapshot) # One snapshot at a time. The endpoint below takes a copy, compares it with the @@ -681,21 +1006,46 @@ def snapshot_receive(req): snapshotting = threading.Lock() -@route("/VolumeDriver.Snapshot") -def volume_snapshot(req): - """Snapshot a volume in the SNAPSHOTS dir, unless it holds nothing new. +def snapshots_here(volume_name): + """The snapshots of this volume on this host, in the order they appeared, with their origin. - The answer names the snapshot that holds the state of the volume, and says - in "Created" whether this call is the one that took it. A volume nobody - wrote to gives the name of the previous snapshot and takes none, which is - what keeps a replication scheduled every minute from filling the disk with + Pairs of the name and whether it came from another host. The traces of + the sends are left out: their content is that of their own snapshot, but + the next send would refuse one if named. The order is the one BTRFS + created them in, not the one the dates in their names spell. + """ + return [ + (listed.name, listed.received) + for listed in btrfs.subvolumes_in(SNAPSHOTS_PATH) + for s in parsed([listed.name]) + if s.volume == volume_name and not s.host + ] + + +def snapshots_taken_here(volume_name): + """The snapshots taken of this volume on this host, in the order they were taken. + + The snapshots received from another host are left out: they were never + taken of this volume, so nothing says how the volume compares to them. + """ + return [name for name, received in snapshots_here(volume_name) if not received] + + +def take_snapshot(name): + """Snapshot this volume, unless it holds nothing new since the last snapshot. + + Answers the name of the snapshot that holds the state of the volume, and + whether this call is the one that took it. A volume nobody wrote to gives + the name of the last snapshot taken of it and takes none, which is what + keeps a replication scheduled every minute from filling the disk with identical copies. A caller that deletes what it created has to read that flag: the name alone no longer says whose snapshot it is. """ - name = req["Name"] volume = existing_volume(name) with snapshotting: + taken = snapshots_taken_here(name) + previous = taken[-1] if taken else None timestamped = new_snapshot_name(name) path = snapshotpath(timestamped) volume.snapshot(path, readonly=True) @@ -703,24 +1053,13 @@ def volume_snapshot(req): # BTRFS cannot compare a live volume to a snapshot, since a send needs # a readonly subvolume, so the comparison happens after the fact: the # copy just taken is deleted when it holds nothing the previous one - # does not. The traces of the sends are left out: their content is that - # of their own snapshot, but naming one here would have the next send - # refuse it. - previous = max( - ( - s - for s in parsed(os.listdir(SNAPSHOTS_PATH)) - if s.volume == name and not s.host and str(s) != timestamped - ), - key=str, - default=None, - ) + # does not. if previous: try: - if btrfs.Subvolume(path).is_same_as(snapshotpath(str(previous))): + if btrfs.Subvolume(path).is_same_as(snapshotpath(previous)): btrfs.Subvolume(path).delete() log.info("%s has not changed since %s", name, previous) - return {"Err": "", "Snapshot": str(previous), "Created": False} + return previous, False except BtrfsError as e: # neither a comparison we could not make nor a deletion that # failed is a reason to lose a snapshot: the copy is kept, and @@ -731,7 +1070,18 @@ def volume_snapshot(req): previous, e, ) - return {"Err": "", "Snapshot": timestamped, "Created": True} + return timestamped, True + + +@route("/VolumeDriver.Snapshot") +def volume_snapshot(req): + """Snapshot a volume in the SNAPSHOTS dir, unless it holds nothing new. + + The answer names the snapshot that holds the state of the volume, and says + in "Created" whether this call is the one that took it. + """ + snapshot, created = take_snapshot(req["Name"]) + return {"Err": "", "Snapshot": snapshot, "Created": created} @route("/VolumeDriver.Snapshot.List", "GET") @@ -785,9 +1135,7 @@ def schedule(req): schedule[(name, action)] = replace(schedule[(name, action)], active="True") else: del schedule[(name, action)] - elif timer.isdecimal() and int(timer) > 0: - # isdecimal, not isnumeric: "²".isnumeric() is True and int("²") - # raises, and the scheduler reads this timer back with int() + elif is_a_timer(timer): validate_volume_name(name) Job.parse(action) schedule[(name, action)] = Entry(name, action, timer, "True") @@ -830,12 +1178,52 @@ def schedule_enable(_): @route("/VolumeDriver.Snapshot.Restore") def snapshot_restore(req): + """Replace a volume with a snapshot, keeping what the volume held.""" + return restore(req["Name"], req.get("Target")) + + +def keep_what_it_holds(volume, target_name, snapshot_path): + """Snapshot this volume before it is replaced, and name where its content went. + + The copy taken goes through the rule a snapshot goes through: when the + volume holds exactly what the last snapshot taken of it holds, that + snapshot is the answer and the copy is deleted. And when it holds exactly + what the snapshot about to be restored holds, there is nothing to keep and + nothing to restore: the answer is None, and the copy is deleted too. An + empty volume, which is what Docker hands over and what a `docker volume + rm` leaves behind once recreated, has nothing to keep either, and the + answer is the empty string. + + Whether the volume is empty is read on the copy, not on the volume: + reading a directory updates its access time, the comparison below would + see that as a change, and the volume would never be found unchanged. """ - Snapshot a volume and overwrite it with the specified snapshot. + with snapshotting: + taken = snapshots_taken_here(target_name) + previous = taken[-1] if taken else None + copy = new_snapshot_name(target_name) + copy_path = snapshotpath(copy) + volume.snapshot(copy_path, readonly=True) + if not os.listdir(copy_path): + btrfs.Subvolume(copy_path).delete() + return "" + if btrfs.Subvolume(copy_path).is_same_as(snapshot_path): + btrfs.Subvolume(copy_path).delete() + return None + if previous and btrfs.Subvolume(copy_path).is_same_as(snapshotpath(previous)): + btrfs.Subvolume(copy_path).delete() + return previous + return copy + + +def restore(snapshot_name, target_name=None): + """Make this snapshot the volume, and answer where what the volume held went. + + ``VolumeBackup`` names the snapshot that holds what the volume held before, + and is empty when it held nothing. ``Restored`` says whether the volume + was replaced: it is not when it already holds exactly this snapshot, so + asking twice does the same as asking once. """ - snapshot_name = req["Name"] - target_name = req.get("Target") - if "@" not in snapshot_name: # we're passing the name of the volume. Use the latest snapshot. volume_name = validate_volume_name(snapshot_name) @@ -852,20 +1240,26 @@ def snapshot_restore(req): target_name = target_name or Snapshot.parse(snapshot_name).volume target_path = volumepath(target_name) volume = btrfs.Subvolume(target_path) - res = {"Err": ""} if not snapshot.exists(): raise SnapshotNotFoundError(f"Snapshot '{snapshot_name}' is not a valid BTRFS subvolume") + backup = "" if volume.exists(): - # backup and delete - stamped_name = new_snapshot_name(target_name) - volume.snapshot(snapshotpath(stamped_name), readonly=True) - res["VolumeBackup"] = stamped_name + backup = keep_what_it_holds(volume, target_name, snapshot_path) + if backup is None: + log.info("%s already holds %s, nothing to restore", target_name, snapshot_name) + return {"Err": "", "VolumeBackup": "", "Restored": False} volume.delete() snapshot.snapshot(target_path) - return res + log.info( + "Restored %s as %s, what it held is kept as %s", + snapshot_name, + target_name, + backup or "nothing", + ) + return {"Err": "", "VolumeBackup": backup, "Restored": True} @route("/VolumeDriver.Clone") diff --git a/buttervolume/scheduler.py b/buttervolume/scheduler.py index 0425375..e6af2e4 100644 --- a/buttervolume/scheduler.py +++ b/buttervolume/scheduler.py @@ -11,8 +11,9 @@ ``schedule.py``: ``schedule.csv``, which says what to run, and ``lastruns.csv``, where the scheduler writes down the date of every job that succeeded. It reads that second file at each round rather than remembering anything, so a daemon -that stops picks its jobs up where it left them. Only the replications under -way live in memory, and a daemon that stops has none. +that stops picks its jobs up where it left them. Nothing lives in memory +here: which replications are under way is the plugin's business, since the +unmount of a volume replicates too and has to wait for the same ones. The thread itself is started and stopped by ``cli.py``, which owns the daemon and its signals. What is here only knows how to run one round. @@ -25,7 +26,7 @@ from os.path import exists from subprocess import CalledProcessError -from buttervolume import ReplicationError, ValidationError, api +from buttervolume import ValidationError, api from buttervolume.config import LAST_RUNS, SCHEDULE, TIMER from buttervolume.schedule import ( Entry, @@ -41,10 +42,6 @@ log = logging.getLogger() -# The volumes a replication is under way for. It lives in memory alone, and -# that is right: a daemon that stops has no replication under way either. -ReplicationInProgress = set() - def is_due(entry, last, now): """Has this line waited its timer since the last time it ran? @@ -99,41 +96,20 @@ def run_snapshot(job: Snapshot, name, test=False): @run_job.register def run_replicate(job: Replicate, name, test=False): - if name in ReplicationInProgress: - log.warning(f"Replication of {name} already in progress, skipping.") - return False - log.info("Starting scheduled replication of %s", name) - snap = None - created = False - try: - ReplicationInProgress.add(name) - snap, created = api.snapshot(name, test=test) - if not snap: - log.info("Could not snapshot %s", name) - return False - log.info("Replicating %s", snap) - if not api.send(snap, job.host, test=test): - # the same road as an exception: the error is already logged, and - # a snapshot taken for this replication has no reason to stay - raise ReplicationError(f"Could not send {snap} to {job.host}") - log.info("Successfully replicated %s to %s", name, snap) - return True - except Exception as e: - log.warning("Replication failed: %s", e) - # remove the snapshot this round created for the replication, and - # only that one: an unchanged volume gives back the snapshot of an - # earlier round, which a failure here is no reason to delete - if snap and created: - if api.remove(snap, test=test): - log.info("Removed snapshot %s for failed replication", snap) - else: - log.warning( - "Could not remove snapshot %s of the failed replication", - snap, - ) + """One call: the plugin snapshots, sends, and takes back its snapshot if the send fails. + + A replication of the same volume still under way, from a round whose send + outlasted the timer, is answered as an error by the plugin, so this round + is not spent and comes back at the next one. + """ + log.info("Starting scheduled replication of %s to %s", name, job.host) + snap = api.replicate(name, job.host, test=test) + if not snap: + # the error is already logged by the client + log.warning("Replication failed: %s to %s", name, job.host) return False - finally: - ReplicationInProgress.remove(name) + log.info("Successfully replicated %s to %s as %s", name, job.host, snap) + return True @run_job.register diff --git a/test.py b/test.py index 924021c..82f4883 100755 --- a/test.py +++ b/test.py @@ -27,7 +27,7 @@ from webtest import TestApp -from buttervolume import ValidationError, btrfs, cli, plugin, schedule, scheduler +from buttervolume import ValidationError, btrfs, cli, plugin, schedule from buttervolume.config import ( DTFORMAT, SNAPSHOTS_PATH, @@ -40,6 +40,7 @@ new_snapshot, sent_snapshots, snapshot_to_fetch, + snapshot_to_restore, snapshots_of, ) from buttervolume.purge import Pattern, compute_purges @@ -177,7 +178,7 @@ def _cleanup_stale_loop_devices(self): pass # Don't fail the test if cleanup fails def test_replication_lock(self): - """Check that the replication lock prevents concurrent replications""" + """A replication under way makes the next round skip, and a final send wait""" # create a volume with a file name = PREFIX_TEST_VOLUME + uuid.uuid4().hex self.create_a_volume_with_a_file(name) @@ -190,9 +191,10 @@ def test_replication_lock(self): # simulate a long-running replication def slow_send(*args, **kwargs): time.sleep(2) - return True - with patch("buttervolume.api.send") as mock_send: + with patch("buttervolume.plugin.snapshots_on_remote", return_value=[]), patch( + "buttervolume.plugin.send_snapshot" + ) as mock_send: mock_send.side_effect = slow_send # run the scheduler in a separate thread t = threading.Thread( @@ -202,15 +204,23 @@ def slow_send(*args, **kwargs): # wait for the replication to start time.sleep(1) # check that the replication is in progress - self.assertIn(name, scheduler.ReplicationInProgress) + self.assertTrue(plugin.replication_lock(name).locked()) # run the scheduler again runjobs(SCHEDULE, True, last_runs=LAST_RUNS) - # check that the second replication was skipped + # check that the second replication was skipped, not queued mock_send.assert_called_once() - # wait for the replication to finish + # a caller that has the final state to send waits its turn instead + waiting = threading.Thread( + target=plugin.replicate, + args=(name, "localhost"), + kwargs={"test": True, "wait": True}, + ) + waiting.start() t.join() + waiting.join() + self.assertEqual(mock_send.call_count, 2) # check that the lock is released - self.assertNotIn(name, scheduler.ReplicationInProgress) + self.assertFalse(plugin.replication_lock(name).locked()) # unschedule self.app.post( "/VolumeDriver.Schedule", @@ -274,7 +284,9 @@ def test_replication_cleanup_on_failure(self): "/VolumeDriver.Schedule", json.dumps({"Name": name, "Action": "replicate:localhost", "Timer": 1}), ) - with patch("buttervolume.api.send") as mock_send: + with patch("buttervolume.plugin.snapshots_on_remote", return_value=[]), patch( + "buttervolume.plugin.send_snapshot" + ) as mock_send: mock_send.side_effect = Exception("replication failed") runjobs(SCHEDULE, True, last_runs=LAST_RUNS) mock_send.assert_called_once() @@ -282,7 +294,7 @@ def test_replication_cleanup_on_failure(self): snapshots = [s for s in os.listdir(SNAPSHOTS_PATH) if s.startswith(name + "@")] self.assertEqual(snapshots, []) # the lock was released - self.assertNotIn(name, scheduler.ReplicationInProgress) + self.assertFalse(plugin.replication_lock(name).locked()) # unschedule self.app.post( "/VolumeDriver.Schedule", @@ -297,9 +309,11 @@ def test_a_replication_that_failed_is_not_reported_as_a_success(self): "/VolumeDriver.Schedule", json.dumps({"Name": name, "Action": "replicate:localhost", "Timer": 1}), ) - with patch("buttervolume.api.send") as mock_send: - # what the client answers when the endpoint fills the Err field - mock_send.return_value = False + with patch("buttervolume.plugin.snapshots_on_remote", return_value=[]), patch( + "buttervolume.plugin.send_snapshot" + ) as mock_send: + # what fills the Err field of the answer + mock_send.side_effect = plugin.ReplicationError("the remote host refused") with self.assertLogs(level=logging.INFO) as log_capture: runjobs(SCHEDULE, True, last_runs=LAST_RUNS) @@ -325,7 +339,10 @@ def test_send_timeout_is_not_retried(self): snap = jsonloads(self.app.post("/VolumeDriver.Snapshot", json.dumps({"Name": name})).body)[ "Snapshot" ] - with patch("buttervolume.plugin.run_btrfs_send_receive") as mock_send: + # a host that answered and holds nothing, so the send goes ahead + with patch("buttervolume.plugin.snapshots_on_remote", return_value=[]), patch( + "buttervolume.plugin.run_btrfs_send_receive" + ) as mock_send: mock_send.side_effect = plugin.ReplicationTimeoutError("waited too long") resp = jsonloads( self.app.post( @@ -424,6 +441,82 @@ def test(self): resp = jsonloads(self.app.post("/VolumeDriver.List", "{}").body) self.assertEqual(resp["Volumes"], []) + def schedule_of(self, name): + return [ + line + for line in jsonloads(self.app.get("/VolumeDriver.Schedule.List").body)["Schedule"] + if line["Name"] == name + ] + + def test_a_volume_created_with_an_action_as_option_has_it_scheduled(self): + """What Docker Swarm can say about a volume, on whatever host it lands""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + opts = {"Opts": {"replicate:localhost": "1", "purge:2h:1d": 60}} + resp = jsonloads( + self.app.post("/VolumeDriver.Create", json.dumps({"Name": name, **opts})).body + ) + self.assertEqual(resp, {"Err": ""}) + self.assertEqual( + self.schedule_of(name), + [ + {"Name": name, "Action": "replicate:localhost", "Timer": "1", "Active": "True"}, + {"Name": name, "Action": "purge:2h:1d", "Timer": "60", "Active": "True"}, + ], + ) + # somebody pauses the line, and the service is deployed again onto + # this host, which still has the volume: the line is left as it is + self.app.post( + "/VolumeDriver.Schedule", + json.dumps({"Name": name, "Action": "replicate:localhost", "Timer": "pause"}), + ) + resp = jsonloads( + self.app.post("/VolumeDriver.Create", json.dumps({"Name": name, **opts})).body + ) + self.assertEqual(resp, {"Err": ""}) + self.assertEqual(self.schedule_of(name)[0]["Active"], "False") + self.assertEqual(len(self.schedule_of(name)), 2) + # and one deleted comes back, since the options still ask for it + self.app.post( + "/VolumeDriver.Schedule", + json.dumps({"Name": name, "Action": "purge:2h:1d", "Timer": 0}), + ) + self.app.post("/VolumeDriver.Create", json.dumps({"Name": name, **opts})) + self.assertEqual( + [line["Action"] for line in self.schedule_of(name)], + ["replicate:localhost", "purge:2h:1d"], + ) + + def test_an_option_nobody_reads_is_refused_before_anything_is_created(self): + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + for opts in ( + {"replicat:localhost": "1"}, + {"replicate:localhost": "soon"}, + {"nocopy": "true"}, + ): + resp = jsonloads( + self.app.post("/VolumeDriver.Create", json.dumps({"Name": name, "Opts": opts})).body + ) + self.assertTrue(resp["Err"], opts) + self.assertFalse(os.path.exists(join(VOLUMES_PATH, name)), opts) + self.assertEqual(self.schedule_of(name), []) + + def test_a_paused_schedule_refuses_a_volume_that_asks_for_a_line(self): + """Creating it with nothing scheduled and nothing said is the one thing not to do""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + with patch("buttervolume.plugin.SCHEDULE_DISABLED", SCHEDULE + ".disabled"): + self.app.post("/VolumeDriver.Schedule.Pause") + try: + resp = jsonloads( + self.app.post( + "/VolumeDriver.Create", + json.dumps({"Name": name, "Opts": {"replicate:localhost": "1"}}), + ).body + ) + finally: + self.app.post("/VolumeDriver.Schedule.Resume") + self.assertIn("globally paused", resp["Err"]) + self.assertFalse(os.path.exists(join(VOLUMES_PATH, name))) + def test_enabled_cow(self): """Check that cow is enabled by default""" # create a volume with a file @@ -592,6 +685,84 @@ def test_sending_the_same_snapshot_twice_leaves_the_remote_copy_whole(self): with open(join(VOLUMES_PATH, target, "foobar")) as f: self.assertEqual(f.read(), "foobar") + @unittest.skipIf( + os.environ.get("BUTTERVOLUME_LOCAL_TEST"), "SSH not available in local test mode" + ) + def test_a_send_refuses_to_bury_a_history_it_never_saw(self): + """The last snapshot that appeared over there has to be one this host knows""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(name) + first = jsonloads(self.app.post("/VolumeDriver.Snapshot", json.dumps({"Name": name})).body)[ + "Snapshot" + ] + self.app.post( + "/VolumeDriver.Snapshot.Send", + json.dumps({"Name": first, "Host": "localhost", "Test": True}), + ) + # a trace that host keeps of its own sends is not a snapshot of the volume + btrfs.Subvolume(join(TEST_REMOTE_PATH, first)).snapshot( + join(TEST_REMOTE_PATH, f"{first}@node3"), readonly=True + ) + self.write_a_byte(name) + second = jsonloads( + self.app.post("/VolumeDriver.Snapshot", json.dumps({"Name": name})).body + )["Snapshot"] + resp = jsonloads( + self.app.post( + "/VolumeDriver.Snapshot.Send", + json.dumps({"Name": second, "Host": "localhost", "Test": True}), + ).body + ) + self.assertEqual(resp, {"Err": ""}) + + # another host ran the application meanwhile and sent its work over there + other = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(other) + with open(join(VOLUMES_PATH, other, "foobar"), "w") as f: + f.write("the work of the other host") + foreign = f"{name}@2000-01-01T00:00:00.000000" + btrfs.Subvolume(join(VOLUMES_PATH, other)).snapshot( + join(TEST_REMOTE_PATH, foreign), readonly=True + ) + self.write_a_byte(name) + third = jsonloads(self.app.post("/VolumeDriver.Snapshot", json.dumps({"Name": name})).body)[ + "Snapshot" + ] + resp = jsonloads( + self.app.post( + "/VolumeDriver.Snapshot.Send", + json.dumps({"Name": third, "Host": "localhost", "Test": True}), + ).body + ) + self.assertIn(foreign, resp["Err"]) + self.assertIn("buttervolume receive localhost", resp["Err"]) + self.assertNotIn(third, os.listdir(TEST_REMOTE_PATH)) + + # a scheduled replication is refused the same way, and takes back the + # snapshot it took for the occasion, round after round + self.write_a_byte(name) + self.app.post( + "/VolumeDriver.Schedule", + json.dumps({"Name": name, "Action": "replicate:localhost", "Timer": 1}), + ) + runjobs(SCHEDULE, True, last_runs=LAST_RUNS) + # no container uses the volume here, so the round fetched the other + # host's work first, whose trace replaced the trace of the second send + self.assertEqual( + sorted(s for s in os.listdir(SNAPSHOTS_PATH) if s.startswith(name + "@")), + sorted([first, second, third, foreign, f"{foreign}@localhost"]), + ) + self.app.post( + "/VolumeDriver.Schedule", + json.dumps({"Name": name, "Action": "replicate:localhost", "Timer": 0}), + ) + + # the other host's work is whole over there: restoring it is the proof + target = PREFIX_TEST_VOLUME + uuid.uuid4().hex + btrfs.Subvolume(join(TEST_REMOTE_PATH, foreign)).snapshot(join(VOLUMES_PATH, target)) + with open(join(VOLUMES_PATH, target, "foobar")) as f: + self.assertEqual(f.read(), "the work of the other host") + def test_a_host_that_cannot_answer_is_not_read_as_an_empty_host(self): """A listing nobody could read is an error, never a host with nothing""" name = PREFIX_TEST_VOLUME + uuid.uuid4().hex @@ -901,7 +1072,9 @@ def test_a_failed_replication_keeps_a_snapshot_it_did_not_create(self): "/VolumeDriver.Schedule", json.dumps({"Name": name, "Action": "replicate:localhost", "Timer": 1}), ) - with patch("buttervolume.api.send") as mock_send: + with patch("buttervolume.plugin.snapshots_on_remote", return_value=[]), patch( + "buttervolume.plugin.send_snapshot" + ) as mock_send: mock_send.side_effect = Exception("replication failed") runjobs(SCHEDULE, True, last_runs=LAST_RUNS) mock_send.assert_called_once() @@ -1179,6 +1352,408 @@ def test_restore(self): with open(join(path2, "foobar")) as f: self.assertEqual(f.read(), "foobar") + @contextmanager + def remote_next_door(self): + """Have the other host be the received directory, without any ssh. + + What ``Test: True`` does for the send, done for the listing and the + receive as well, so that a whole move of a volume can be played here. + The two hosts share the filesystem, and that is all they share. + """ + + def listing(remote_host, remote_path, port): + return [s.name for s in btrfs.subvolumes_in(remote_path)] + + def transfer(source, parent, into): + cmd = ["btrfs", "send"] + (["-p", parent] if parent else []) + [source] + run( + f"{' '.join(cmd)} | btrfs receive {into}", + shell=True, + check=True, + capture_output=True, + ) + + def send(snapshot_path, remote_host, remote_snapshots, parent_path=None, port="1122"): + transfer(snapshot_path, parent_path, remote_snapshots) + + def receive(remote_host, remote_snapshot_path, remote_parent_path=None, port="1122"): + transfer(remote_snapshot_path, remote_parent_path, SNAPSHOTS_PATH) + + with patch("buttervolume.plugin.snapshots_on_remote", side_effect=listing), patch( + "buttervolume.plugin.run_btrfs_send_receive", side_effect=send + ), patch("buttervolume.plugin.run_btrfs_receive", side_effect=receive) as received: + yield received + + def another_host_sends(self, name, content, stamp="2000-01-01T00:00:00.000000"): + """What another host running the application leaves on the remote host""" + other = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(other) + with open(join(VOLUMES_PATH, other, "foobar"), "w") as f: + f.write(content) + sent = f"{name}@{stamp}" + btrfs.Subvolume(join(VOLUMES_PATH, other)).snapshot( + join(TEST_REMOTE_PATH, sent), readonly=True + ) + return sent + + def replicated_volume(self, name, content="foobar"): + self.create_a_volume_with_a_file(name) + with open(join(VOLUMES_PATH, name, "foobar"), "w") as f: + f.write(content) + self.app.post( + "/VolumeDriver.Schedule", + json.dumps({"Name": name, "Action": "replicate:localhost", "Timer": 1}), + ) + + def mount(self, name): + return jsonloads( + self.app.post("/VolumeDriver.Mount", json.dumps({"Name": name, "Test": True})).body + ) + + def unmount(self, name): + return jsonloads( + self.app.post("/VolumeDriver.Unmount", json.dumps({"Name": name, "Test": True})).body + ) + + def content_of(self, name): + with open(join(VOLUMES_PATH, name, "foobar")) as f: + return f.read() + + def snapshots_of_volume(self, name, where=SNAPSHOTS_PATH): + return sorted(s for s in os.listdir(where) if s.startswith(name + "@")) + + def test_a_volume_follows_its_container_from_one_host_to_the_other(self): + """Mount takes what the other host holds, unmount sends the final state""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.replicated_volume(name, "written here") + with self.remote_next_door() as received: + # the first mount finds nothing over there and keeps the volume + self.assertEqual(self.mount(name)["Err"], "") + self.assertEqual(self.content_of(name), "written here") + self.assertEqual(self.snapshots_of_volume(name), []) + # the last unmount sends the state the container left + self.assertEqual(self.unmount(name), {"Err": ""}) + [first] = self.snapshots_of_volume(name, TEST_REMOTE_PATH) + + # the application runs on the other host meanwhile, and comes back + later = self.another_host_sends(name, "written elsewhere") + self.assertEqual(self.mount(name)["Err"], "") + self.assertEqual(self.content_of(name), "written elsewhere") + # what the volume held was already the first snapshot: no new one. + # The trace of the first send went, the receive left a newer one + self.assertEqual( + self.snapshots_of_volume(name), sorted([first, later, f"{later}@localhost"]) + ) + + # a second container on the same volume mounts and unmounts + # without the volume changing hands + received.reset_mock() + self.assertEqual(self.mount(name)["Err"], "") + self.assertEqual(self.unmount(name), {"Err": ""}) + received.assert_not_called() + self.assertEqual( + self.snapshots_of_volume(name, TEST_REMOTE_PATH), sorted([first, later]) + ) + self.assertEqual(self.content_of(name), "written elsewhere") + + # the last container writes and stops: its state goes out, on top + # of the history that came in + with open(join(VOLUMES_PATH, name, "foobar"), "w") as f: + f.write("written here again") + self.assertEqual(self.unmount(name), {"Err": ""}) + over_there = self.snapshots_of_volume(name, TEST_REMOTE_PATH) + self.assertEqual(len(over_there), 3) + [final] = [s for s in over_there if s not in (first, later)] + with open(join(TEST_REMOTE_PATH, final, "foobar")) as f: + self.assertEqual(f.read(), "written here again") + # and restoring it there is the proof it is whole + target = PREFIX_TEST_VOLUME + uuid.uuid4().hex + btrfs.Subvolume(join(TEST_REMOTE_PATH, final)).snapshot(join(VOLUMES_PATH, target)) + self.assertEqual(self.content_of(target), "written here again") + + def test_a_host_that_crashed_does_not_bury_the_work_done_elsewhere(self): + """Its unsent writes are kept aside, and the work done elsewhere wins""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.replicated_volume(name, "before the crash") + with self.remote_next_door(): + self.mount(name) + self.unmount(name) + [first] = self.snapshots_of_volume(name, TEST_REMOTE_PATH) + # the container runs again, writes, and the host crashes: no unmount + self.mount(name) + with open(join(VOLUMES_PATH, name, "foobar"), "w") as f: + f.write("never sent") + later = self.another_host_sends(name, "done elsewhere") + + # the scheduled round, with the plugin still counting a container: + # its send is refused, its snapshot taken back, round after round + for _ in range(2): + with self.assertLogs(level=logging.WARNING) as log_capture: + runjobs(SCHEDULE, True, last_runs=LAST_RUNS) + self.assertTrue(any(later in msg for msg in log_capture.output)) + self.assertEqual(self.snapshots_of_volume(name), [first, f"{first}@localhost"]) + self.assertEqual( + self.snapshots_of_volume(name, TEST_REMOTE_PATH), sorted([first, later]) + ) + + # the plugin restarts with the host and counts from zero: the next + # mount brings the work done elsewhere in, and keeps the rest aside + plugin.mounted.clear() + self.assertEqual(self.mount(name)["Err"], "") + self.assertEqual(self.content_of(name), "done elsewhere") + kept = [ + s + for s in self.snapshots_of_volume(name) + if s not in (first, f"{first}@localhost", later, f"{later}@localhost") + ] + self.assertEqual(len(kept), 1) + with open(join(SNAPSHOTS_PATH, kept[0], "foobar")) as f: + self.assertEqual(f.read(), "never sent") + + # the container crashes at once and the plugin restarts again: the + # snapshot kept aside is the last to have appeared here, and it + # was taken here, so nothing is restored over the work done elsewhere + plugin.mounted.clear() + self.assertEqual(self.mount(name)["Err"], "") + self.assertEqual(self.content_of(name), "done elsewhere") + self.assertEqual( + self.snapshots_of_volume(name), + sorted([first, later, f"{later}@localhost", kept[0]]), + ) + + def test_an_empty_volume_takes_what_the_other_hosts_hold(self): + """A volume Docker just created on a host the application moved to""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + with self.remote_next_door(): + sent = self.another_host_sends(name, "done elsewhere") + self.app.post("/VolumeDriver.Create", json.dumps({"Name": name})) + self.app.post( + "/VolumeDriver.Schedule", + json.dumps({"Name": name, "Action": "replicate:localhost", "Timer": 1}), + ) + self.assertEqual(self.mount(name)["Err"], "") + self.assertEqual(self.content_of(name), "done elsewhere") + # nothing was kept of an empty volume + self.assertEqual(self.snapshots_of_volume(name), [sent, f"{sent}@localhost"]) + + def test_a_volume_recreated_after_its_removal_takes_its_last_snapshot_back(self): + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.replicated_volume(name, "before the removal") + with self.remote_next_door(): + self.mount(name) + self.unmount(name) + self.app.post("/VolumeDriver.Remove", json.dumps({"Name": name})) + self.app.post("/VolumeDriver.Create", json.dumps({"Name": name})) + self.assertEqual(self.mount(name)["Err"], "") + self.assertEqual(self.content_of(name), "before the removal") + + def test_a_snapshot_received_by_hand_is_restored_at_the_next_mount(self): + """And a host at rest sends nothing meanwhile, whatever the dates say""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.replicated_volume(name, "written here") + with self.remote_next_door(): + self.mount(name) + self.unmount(name) + later = self.another_host_sends(name, "done elsewhere", "2999-01-01T00:00:00.000000") + resp = jsonloads( + self.app.post( + "/VolumeDriver.Snapshot.Receive", + json.dumps({"Name": name, "Host": "localhost", "Test": True}), + ).body + ) + self.assertEqual(resp["Snapshot"], later) + self.assertEqual(self.content_of(name), "written here") + # the volume is unchanged since its own last snapshot: nothing to send + runjobs(SCHEDULE, True, last_runs=LAST_RUNS) + self.assertEqual(len(self.snapshots_of_volume(name, TEST_REMOTE_PATH)), 2) + self.assertEqual(self.mount(name)["Err"], "") + self.assertEqual(self.content_of(name), "done elsewhere") + + def test_a_host_where_nothing_runs_fetches_what_the_other_holds_at_each_round(self): + """So that the mount has a difference to receive, not a whole volume""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.replicated_volume(name, "written here") + with self.remote_next_door() as received: + self.mount(name) + self.unmount(name) + later = self.another_host_sends(name, "done elsewhere") + runjobs(SCHEDULE, True, last_runs=LAST_RUNS) + self.assertIn(later, self.snapshots_of_volume(name)) + # fetched, not restored: which snapshot becomes the volume is the + # mount's decision + self.assertEqual(self.content_of(name), "written here") + received.reset_mock() + self.assertEqual(self.mount(name)["Err"], "") + self.assertEqual(self.content_of(name), "done elsewhere") + received.assert_not_called() + + def test_a_host_that_does_not_answer_refuses_the_mount(self): + """Mounting on "there is nothing over there" when nobody said so is the one thing not to do""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.replicated_volume(name) + with patch( + "buttervolume.plugin.snapshots_on_remote", + side_effect=plugin.ReplicationError("Could not read the snapshots of localhost"), + ): + resp = self.mount(name) + self.assertIn("Could not read the snapshots of localhost", resp["Err"]) + self.assertIn("replicate:localhost pause", resp["Err"]) + # pausing the line is the way out, and the mount asks nobody + self.app.post( + "/VolumeDriver.Schedule", + json.dumps({"Name": name, "Action": "replicate:localhost", "Timer": "pause"}), + ) + self.assertEqual(self.mount(name)["Err"], "") + self.assertEqual(self.unmount(name), {"Err": ""}) + + def test_a_volume_nobody_replicates_is_mounted_as_it_is(self): + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(name) + with patch("buttervolume.plugin.snapshots_on_remote") as listing: + self.assertEqual(self.mount(name)["Err"], "") + self.assertEqual(self.unmount(name), {"Err": ""}) + listing.assert_not_called() + self.assertEqual(self.snapshots_of_volume(name), []) + + def test_the_final_state_of_a_rolled_back_volume_still_goes_out(self): + """A restore by hand on the host that runs the application is its own history""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.replicated_volume(name, "first") + with self.remote_next_door(): + self.mount(name) + self.unmount(name) + [first] = self.snapshots_of_volume(name, TEST_REMOTE_PATH) + self.mount(name) + with open(join(VOLUMES_PATH, name, "foobar"), "w") as f: + f.write("second") + self.unmount(name) + self.mount(name) + # back to the first state, then on from there + self.restore_of(first) + with open(join(VOLUMES_PATH, name, "foobar"), "w") as f: + f.write("third") + self.assertEqual(self.unmount(name), {"Err": ""}) + over_there = self.snapshots_of_volume(name, TEST_REMOTE_PATH) + self.assertEqual(len(over_there), 3) + with open(join(TEST_REMOTE_PATH, over_there[-1], "foobar")) as f: + self.assertEqual(f.read(), "third") + + def test_a_mount_that_fails_is_not_counted(self): + """Docker calls no Unmount for a container it did not start""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.replicated_volume(name) + with patch( + "buttervolume.plugin.snapshots_on_remote", side_effect=plugin.ReplicationError("down") + ): + self.assertTrue(self.mount(name)["Err"]) + self.assertFalse(plugin.mounted.get(name)) + + def receive_locally(self, name, source): + """Put a copy of this subvolume in the snapshots directory, as a receive would + + The copy carries the mark of a subvolume that came from another host, + which is what these tests need, and nothing crosses a network. + """ + staged = join(TEST_REMOTE_PATH, name) + btrfs.Subvolume(source).snapshot(staged, readonly=True) + run( + f"btrfs send {staged} | btrfs receive {SNAPSHOTS_PATH}", + shell=True, + check=True, + capture_output=True, + ) + return join(SNAPSHOTS_PATH, name) + + def restore_of(self, snapshot, target=None): + return jsonloads( + self.app.post( + "/VolumeDriver.Snapshot.Restore", json.dumps({"Name": snapshot, "Target": target}) + ).body + ) + + def test_a_volume_unchanged_since_its_last_snapshot_is_not_backed_up_again(self): + """The backup of a restore goes through the rule a snapshot goes through""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(name) + first = jsonloads(self.app.post("/VolumeDriver.Snapshot", json.dumps({"Name": name})).body) + self.write_a_byte(name) + second = jsonloads(self.app.post("/VolumeDriver.Snapshot", json.dumps({"Name": name})).body) + + resp = self.restore_of(first["Snapshot"]) + self.assertEqual(resp, {"Err": "", "VolumeBackup": second["Snapshot"], "Restored": True}) + with open(join(VOLUMES_PATH, name, "foobar")) as f: + self.assertEqual(f.read(), "foobar") + # what the volume held is in the second snapshot, and nowhere else + self.assertEqual( + sorted(s for s in os.listdir(SNAPSHOTS_PATH) if s.startswith(name + "@")), + sorted([first["Snapshot"], second["Snapshot"]]), + ) + + def test_a_volume_that_already_holds_the_snapshot_is_left_alone(self): + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(name) + snapshot = jsonloads( + self.app.post("/VolumeDriver.Snapshot", json.dumps({"Name": name})).body + )["Snapshot"] + before = btrfs.Subvolume(join(VOLUMES_PATH, name)).show()["UUID"] + + resp = self.restore_of(snapshot) + self.assertEqual(resp, {"Err": "", "VolumeBackup": "", "Restored": False}) + # the very same subvolume, not a copy of it + self.assertEqual(btrfs.Subvolume(join(VOLUMES_PATH, name)).show()["UUID"], before) + self.assertEqual( + [s for s in os.listdir(SNAPSHOTS_PATH) if s.startswith(name + "@")], [snapshot] + ) + + def test_an_empty_volume_has_nothing_to_keep(self): + """What Docker hands over, and what a docker volume rm leaves once recreated""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(name) + snapshot = jsonloads( + self.app.post("/VolumeDriver.Snapshot", json.dumps({"Name": name})).body + )["Snapshot"] + empty = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.app.post("/VolumeDriver.Create", json.dumps({"Name": empty})) + + resp = self.restore_of(snapshot, target=empty) + self.assertEqual(resp, {"Err": "", "VolumeBackup": "", "Restored": True}) + with open(join(VOLUMES_PATH, empty, "foobar")) as f: + self.assertEqual(f.read(), "foobar") + self.assertEqual([s for s in os.listdir(SNAPSHOTS_PATH) if s.startswith(empty + "@")], []) + + def test_a_snapshot_is_compared_with_the_last_one_taken_here(self): + """A snapshot received from another host was never taken of this volume""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(name) + first = jsonloads(self.app.post("/VolumeDriver.Snapshot", json.dumps({"Name": name})).body) + # a fresher copy arrives from another host, carrying other content and + # a date that sorts after everything taken here + other = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(other) + self.write_a_byte(other) + self.receive_locally(f"{name}@2999-01-01T00:00:00.000000", join(VOLUMES_PATH, other)) + + # the volume itself did not change: nothing to snapshot, nothing to send + again = jsonloads(self.app.post("/VolumeDriver.Snapshot", json.dumps({"Name": name})).body) + self.assertEqual(again, {"Err": "", "Snapshot": first["Snapshot"], "Created": False}) + + def test_the_same_content_from_another_lineage_is_not_taken_for_the_same_snapshot(self): + """Two subvolumes that share no history are compared as different, and never raise""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(name) + twin = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(twin) + received = f"{name}@2026-01-01T00:00:00.000000" + self.receive_locally(received, join(VOLUMES_PATH, twin)) + + resp = self.restore_of(received) + self.assertEqual(resp["Err"], "") + self.assertTrue(resp["Restored"]) + # what the volume held is kept, since nothing could prove it identical + self.assertTrue(resp["VolumeBackup"].startswith(name + "@")) + with open(join(VOLUMES_PATH, name, "foobar")) as f: + self.assertEqual(f.read(), "foobar") + def test_clone(self): """Check we can clone as a new volume""" # create a volume with a file @@ -1409,6 +1984,38 @@ def test_run_safe_missing_command(self): with self.assertRaises(btrfs.BtrfsError): btrfs.run_safe(["there_is_no_such_command"], timeout=btrfs.SHOW_TIMEOUT) + def test_the_subvolumes_of_a_directory_come_in_the_order_they_appeared(self): + """Their names say what a clock said, their rank says what happened first""" + name = PREFIX_TEST_VOLUME + uuid.uuid4().hex + self.create_a_volume_with_a_file(name) + volume = btrfs.Subvolume(join(VOLUMES_PATH, name)) + later, earlier = f"{name}@2026-09-02T10:00:00.000000", f"{name}@2026-09-02T09:00:00.000000" + volume.snapshot(join(SNAPSHOTS_PATH, later), readonly=True) + volume.snapshot(join(SNAPSHOTS_PATH, earlier), readonly=True) + # a received copy is marked as such, a plain file is not a subvolume + run( + f"btrfs send {join(SNAPSHOTS_PATH, later)} | btrfs receive {TEST_REMOTE_PATH}", + shell=True, + check=True, + capture_output=True, + ) + with open(join(SNAPSHOTS_PATH, PREFIX_TEST_VOLUME + "stray"), "w") as f: + f.write("not a subvolume") + + listed = [s for s in btrfs.subvolumes_in(SNAPSHOTS_PATH) if s.name.startswith(name)] + self.assertEqual([s.name for s in listed], [later, earlier]) + self.assertEqual([s.received for s in listed], [False, False]) + received = [s for s in btrfs.subvolumes_in(TEST_REMOTE_PATH) if s.name.startswith(name)] + self.assertEqual([(s.name, s.received) for s in received], [(later, True)]) + + # the command a remote host runs describes the same directory the same way + described = check_output(["sh", "-c", btrfs.listing_command(SNAPSHOTS_PATH)]).decode() + self.assertEqual( + [s for s in btrfs.parse_listing(described) if s.name.startswith(name)], listed + ) + # and fails, rather than describing nothing, when the directory is not there + self.assertNotEqual(run(["sh", "-c", btrfs.listing_command("/no/such/dir")]).returncode, 0) + def test_compute_purge(self): now = datetime.now() snapshots = [ @@ -2004,15 +2611,24 @@ def test_the_parent_of_a_send_is_the_last_one_sent_to_that_host(self): class TestWhatToFetch(unittest.TestCase): """What to ask a host for, read from two listings and nothing else""" - def test_the_most_recent_snapshot_over_there_is_the_one_to_fetch(self): + def test_the_last_snapshot_that_appeared_over_there_is_the_one_to_fetch(self): remote = ["www@t1", "www@t2", "www@t3", "other@t9"] local = ["www@t1", "www@t2"] snapshot, parent = snapshot_to_fetch("www", remote, local) self.assertEqual(str(snapshot), "www@t3") - # the most recent one both sides already hold, which is what an - # incremental transfer is built on + # the last one both sides already hold, which is what an incremental + # transfer is built on self.assertEqual(str(parent), "www@t2") + def test_the_order_is_the_one_the_host_listed_and_not_the_dates_in_the_names(self): + # the host whose clock runs ahead wrote t9 first; what arrived last is + # the most recent, and the parent is the last one in common, not the + # one whose name says so + remote = ["www@t9", "www@t1", "www@t8", "www@t2"] + snapshot, parent = snapshot_to_fetch("www", remote, ["www@t9", "www@t1", "www@t8"]) + self.assertEqual(str(snapshot), "www@t2") + self.assertEqual(str(parent), "www@t8") + def test_a_trace_is_neither_fetched_nor_taken_for_a_parent(self): # that host keeps the traces of its own sends next to its snapshots, # and www@t3@node3 sorts after www@t3 @@ -2045,6 +2661,31 @@ def test_a_stray_file_of_our_own_stops_nothing(self): self.assertEqual(str(parent), "www@t1") +class TestWhatToRestore(unittest.TestCase): + """Which snapshot a mounted volume is brought to, decided from two lists and a flag""" + + def test_the_last_snapshot_to_appear_is_restored_when_it_came_from_elsewhere(self): + events = [("www@t1", False), ("www@t2", True)] + self.assertEqual(snapshot_to_restore(False, events, ["www@t2"]), "www@t2") + + def test_a_snapshot_taken_here_since_is_the_volume_going_on(self): + # what was received before, and what the dates say, do not matter + events = [("www@t9", True), ("www@t1", False)] + self.assertIsNone(snapshot_to_restore(False, events, ["www@t9"])) + + def test_an_empty_volume_takes_what_the_hosts_hold(self): + events = [("www@t1", False), ("www@t2", True), ("www@t3", False)] + self.assertEqual(snapshot_to_restore(True, events, ["www@t2"]), "www@t2") + # the last of the candidates to have appeared here, when there are two + self.assertEqual(snapshot_to_restore(True, events, ["www@t3", "www@t2"]), "www@t3") + # and its last snapshot when the hosts hold nothing of it + self.assertEqual(snapshot_to_restore(True, events, []), "www@t3") + + def test_a_volume_without_a_snapshot_is_left_as_it_is(self): + self.assertIsNone(snapshot_to_restore(True, [], [])) + self.assertIsNone(snapshot_to_restore(False, [], ["www@t1"])) + + class TestPurgePattern(unittest.TestCase): """The retention pattern, read without touching a filesystem""" @@ -2385,6 +3026,77 @@ def rows(): self.assertTrue(left[0].startswith("lastruns.csv."), left) +SHOWN = """\ +@home/var/lib/buttervolume/snapshots/www@2026-09-02T10:00:00.000000 +\tName: \t\t\twww@2026-09-02T10:00:00.000000 +\tUUID: \t\t\t63c084af-fc1e-0942-9307-2146d11d43a2 +\tParent UUID: \t\t- +\tReceived UUID: \t\t7cc405e8-abc2-d849-acd4-5f22acc780e0 +\tCreation time: \t\t2026-09-02 11:02:00 +0200 +\tSubvolume ID: \t\t69929 +\tGeneration: \t\t1923607 +\tGen at creation: \t1923607 +\tParent ID: \t\t961 +\tTop level ID: \t\t961 +\tFlags: \t\t\treadonly +\tSend transid: \t\t1923605 +\tSend time: \t\t2026-09-02 11:02:00 +0200 +\tReceive transid: \t1923607 +\tReceive time: \t\t2026-09-02 11:02:00 +0200 +\tSnapshot(s): +\t\t\t\tsnapshots/www@2026-09-02T10:00:00.000000@node2 +\tQuota group:\t\tn/a +@home/var/lib/buttervolume/snapshots/www@2026-09-02T09:00:00.000000 +\tName: \t\t\twww@2026-09-02T09:00:00.000000 +\tUUID: \t\t\t11111111-fc1e-0942-9307-2146d11d43a2 +\tParent UUID: \t\t- +\tReceived UUID: \t\t- +\tCreation time: \t\t2026-09-02 11:03:00 +0200 +\tSubvolume ID: \t\t69931 +\tGeneration: \t\t1923609 +\tGen at creation: \t1923609 +\tParent ID: \t\t961 +\tTop level ID: \t\t961 +\tFlags: \t\t\treadonly +\tSnapshot(s): +""" + + +class TestBtrfsOutput(unittest.TestCase): + """What btrfs prints, read without touching a filesystem""" + + def test_a_description_is_read_field_by_field(self): + shown = btrfs.parse_shows(SHOWN) + self.assertEqual(len(shown), 2) + self.assertEqual(shown[0]["Name"], "www@2026-09-02T10:00:00.000000") + self.assertEqual(shown[0]["Received UUID"], "7cc405e8-abc2-d849-acd4-5f22acc780e0") + self.assertEqual(shown[0]["Flags"], "readonly") + self.assertEqual(shown[0]["Quota group"], "n/a") + # the snapshots listed under a description are paths, not fields, + # even when a colon in their name makes them look like one + self.assertNotIn("snapshots/www@2026-09-02T10", shown[0]) + self.assertEqual(shown[1]["Received UUID"], "-") + + def test_a_listing_is_in_creation_order_and_says_what_was_received(self): + # the second one was created later, though its name says earlier + listed = btrfs.parse_listing(SHOWN) + self.assertEqual( + listed, + [ + btrfs.Listed("www@2026-09-02T10:00:00.000000", 69929, True), + btrfs.Listed("www@2026-09-02T09:00:00.000000", 69931, False), + ], + ) + + def test_a_directory_holding_no_subvolume_is_an_empty_listing(self): + self.assertEqual(btrfs.parse_listing(""), []) + + def test_a_description_missing_a_field_stops_everything(self): + # answering the other ones would be answering "less than there is" + with self.assertRaises(btrfs.BtrfsError): + btrfs.parse_listing("some/path\n\tName: \t\twww@t1\n\tFlags: \t\treadonly\n") + + @contextmanager def temporary_directory(path): """Give an empty directory at the path of your choosing.