Repository: radiocosmology/alpenhorn

Issue:
Ability to manipulate archivefilecopyrequests
In CHIME we occasionally get situations where copy requests can't be completed because the source file doesn't exist.  (Usually due to something happening while copying it to transport disk).

So, we end up with a bunch of messages like this:

```
Feb 20 13:46:08 ERROR >> Skipping request for 20191220T204152Z_chime26m_corr/03700704_0000.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=157964]
Feb 20 13:46:08 ERROR >> Skipping request for 20191220T204152Z_chime26m_corr/03703248_0000.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=157966]
Feb 20 13:46:08 ERROR >> Skipping request for 20191211T132558Z_chime_rawadc/002303.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=157968]
Feb 20 13:46:09 ERROR >> Skipping request for 20191220T204152Z_chimetiming_corr/03700132_0000.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=157969]
Feb 20 13:46:09 ERROR >> Skipping request for 20191211T132558Z_chime_rawadc/002305.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=157970]
Feb 20 13:46:09 ERROR >> Skipping request for 20191211T132558Z_chime_rawadc/002304.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=157971]
Feb 20 13:46:09 ERROR >> Skipping request for 20191211T132558Z_chime_rawadc/002315.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=158000]
Feb 20 13:46:09 ERROR >> Skipping request for 20191211T132558Z_chime_rawadc/002316.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=158001]
Feb 20 13:46:09 ERROR >> Skipping request for 20191220T204152Z_chimetiming_corr/03725580_0000.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=158003]
Feb 20 13:46:09 ERROR >> Skipping request for 20191211T132558Z_chime_rawadc/002317.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=158004]
Feb 20 13:46:09 ERROR >> Skipping request for 20191220T204152Z_chimetiming_corr/03730669_0000.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=158006]
Feb 20 13:46:09 ERROR >> Skipping request for 20191211T132558Z_chime_rawadc/002319.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=158008]
Feb 20 13:46:09 ERROR >> Skipping request for 20191211T132558Z_chime_rawadc/002318.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=158009]
Feb 20 13:46:09 ERROR >> Skipping request for 20191220T204152Z_chimetiming_corr/03735759_0000.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=158011]
Feb 20 13:46:09 ERROR >> Skipping request for 20191211T132558Z_chime_rawadc/002320.h5 since it is not available on node "CH-WX71DA4A068E". [file_id=158013]
```

Currently, to get rid of these messages, I manually cancel these requests by executing:
```
UPDATE archivefilecopyrequest
     LEFT JOIN archivefilecopy
          ON archivefilecopy.file_id = archivefilecopyrequest.file_id
          AND archivefilecopy.node_id = archivefilecopyrequest.node_from_id
     SET cancelled=1
     WHERE completed=0
          AND cancelled=0
          AND has_file='N';
```
but it would be nicer if this could be done through the CLI.

Hints from issue comments:
@cubranic: Proposed UI (based on Slack discussion):

- `request list [--file=ID] [--acq=ID] [--source=NODE] [--dest=GROUP]`
- `request copy `NODE GROUP [--file=ID] [--acq=ID]` (a synonym of `alpenhorn sync`)
- `request cancel ??`@jrs65: Maybe `create` instead of `copy` as the verb should act on the preceeding noun. `copy` makes it sound like it's copying a request not requesting the copy of a file.@jrs65: Otherwise I really like having better support for managing requests. It's not great to have to edit the DB by hand as we do at the moment.@cubranic: > Maybe `create` instead of `copy` as the verb should act on the preceeding noun. `copy` makes it sound like it's copying a request not requesting the copy of a file.

I'm OK with `create`, sure.

Solution patch:
diff --git a/alpenhorn/cli/file/__init__.py b/alpenhorn/cli/file/__init__.py
index 03d9827ae..b30a1189a 100644
--- a/alpenhorn/cli/file/__init__.py
+++ b/alpenhorn/cli/file/__init__.py
@@ -6,6 +6,7 @@
 from .create import create
 from .import_ import import_
 from .show import show
+from .sync import sync
 
 
 @click.group(context_settings={"help_option_names": ["-h", "--help"]})
@@ -16,3 +17,4 @@ def cli():
 cli.add_command(create, "create")
 cli.add_command(import_, "import")
 cli.add_command(show, "show")
+cli.add_command(sync, "sync")
diff --git a/alpenhorn/cli/file/sync.py b/alpenhorn/cli/file/sync.py
new file mode 100644
index 000000000..0e69823b8
--- /dev/null
+++ b/alpenhorn/cli/file/sync.py
@@ -0,0 +1,124 @@
+"""alpenhorn file sync command."""
+
+import click
+import peewee as pw
+
+from ...db import ArchiveFileCopyRequest, database_proxy
+from ..cli import echo
+from ..options import cli_option, file_from_path, resolve_group, resolve_node
+
+
+@click.command()
+@click.argument("path", metavar="FILE")
+@click.option("--cancel", is_flag=True, help="Cancel existing transfer requests.")
+@click.option(
+    "--force",
+    is_flag=True,
+    help="Create the transfer request even if the source Node does "
+    "not have a copy of the file.  Ignored if used with --cancel.",
+)
+@cli_option(
+    "from_", help="Source Node for the transfer request.  Required if not cancelling."
+)
+@cli_option(
+    "to",
+    help="Destination Group for the transfer request.  Required if not cancelling.",
+)
+@click.pass_context
+def sync(ctx, path, cancel, force, from_, to):
+    """Create or cancel a File transfer.
+
+    This command allows you to either create a new request to transfer
+    a file from a Node to a Group, or else lets you cancel existing
+    pending transfer requests for FILE.  The FILE should be specified as
+    "<acq_name>/<filename>".
+
+    \b
+    New Requests
+    ------------
+
+    Wihtout "--cancel", this command creates a request for the daemon to
+    transfer the file named FILE from the Node SORUCE_NODE into the
+    destination Group DEST_GROUP.  In this case, both "--from" and "--to"
+    are required.  The File must not already be present in the destination
+    Group.
+
+    By default, the command will only create the request if the file exists
+    on the source.  To skip this check, and make the request anyways use the
+    "--force" flag.  In this case, the daemon may be unable to complete such
+    a request, and may cancel it, if it can't find the source file when it
+    tries to handle the request.
+
+    \b
+    Cancelling Requests
+    -------------------
+
+    To cancel pending transfer requests for FILE use the "--cancel" flag.
+    Which requests are cancelled may be limited with the "--from" and "--to"
+    flags.  By default, all exisingt pending requests for FILE are cancelled.
+    """
+
+    # Usage checks: must use --cancel or else both --from AND --to
+    if not cancel:
+        if not from_:
+            raise click.UsageError("missing --from")
+        if not to:
+            raise click.UsageError("missing --to")
+
+    if cancel:
+        # Cancel mode.  Find matching requests and cancel them.
+        with database_proxy.atomic():
+            file_ = file_from_path(path)
+            query = ArchiveFileCopyRequest.update(cancelled=True).where(
+                ArchiveFileCopyRequest.file == file_,
+                ArchiveFileCopyRequest.cancelled == 0,
+                ArchiveFileCopyRequest.completed == 0,
+            )
+
+            if from_:
+                query = query.where(
+                    ArchiveFileCopyRequest.node_from == resolve_node(from_)
+                )
+            if to:
+                query = query.where(
+                    ArchiveFileCopyRequest.group_to == resolve_group(to)
+                )
+
+            count = query.execute()
+
+        if count == 1:
+            echo("Cancelled 1 request.")
+        elif count:
+            echo(f"Cancelled {count} requests.")
+        else:
+            echo("No requests to cancel.")
+        ctx.exit()
+
+    # Create mode.
+    with database_proxy.atomic():
+        file_ = file_from_path(path)
+        src = resolve_node(from_)
+        dest = resolve_group(to)
+
+        # Is the file already in the dest?
+        state, node = dest.state_on_node(file_)
+        if state == "Y":
+            echo(
+                f'Not done: {path} already present on Node "{node.name}" '
+                f'in destination Group "{to}".'
+            )
+            ctx.exit()
+
+        # Does the source file exist?
+        if src.filecopy_state(file_) != "Y":
+            if force:
+                echo("File missing from source: forcing request.")
+            else:
+                echo("Not done: File missing from source.")
+                ctx.exit()
+
+        # Create request
+        ArchiveFileCopyRequest.create(
+            file=file_, node_from=src, group_to=dest, completed=0, cancelled=0
+        )
+        echo("Request submitted.")
diff --git a/alpenhorn/cli/options.py b/alpenhorn/cli/options.py
index f51261fd0..e5ea15b1a 100644
--- a/alpenhorn/cli/options.py
+++ b/alpenhorn/cli/options.py
@@ -72,6 +72,9 @@ def cli_option(option: str, **extra_kwargs):
             "help": "Make node a field node (i.e. neither an archive node "
             "nor a transport node).  Incompatible with --archive or --transport.",
         }
+    elif option == "from_":
+        args = ("from_", "--from")
+        kwargs = {"metavar": "SOURCE_NODE", "help": "Source Node for the transfer."}
     elif option == "group":
         args = ("--group",)
         kwargs = {
@@ -148,6 +151,12 @@ def cli_option(option: str, **extra_kwargs):
             "help": "May be specified multiple times.  Restrict operation to "
             "files which exist in all specified target groups GROUP.",
         }
+    elif option == "to":
+        args = ("--to",)
+        kwargs = {
+            "metavar": "DEST_GROUP",
+            "help": "Destination Group for the transfer.",
+        }
     elif option == "transport":
         args = ("--transport",)
         kwargs = {


Test patch:
diff --git a/tests/cli/file/test_sync.py b/tests/cli/file/test_sync.py
new file mode 100644
index 000000000..4ddc1b6a8
--- /dev/null
+++ b/tests/cli/file/test_sync.py
@@ -0,0 +1,295 @@
+"""Test CLI: alpenhorn file sync"""
+
+from alpenhorn.db import (
+    ArchiveAcq,
+    ArchiveFile,
+    ArchiveFileCopy,
+    ArchiveFileCopyRequest,
+    StorageGroup,
+    StorageNode,
+    utcnow,
+)
+
+
+def test_bad_file(clidb, cli):
+    """Test a bad file."""
+
+    dest = StorageGroup.create(name="Dest")
+    StorageNode.create(name="DestNode", group=dest)
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    cli(1, ["file", "sync", "MIS/SING", "--from=Src", "--to=Dest"])
+
+
+def test_no_source(clidb, cli):
+    """Test a missing source node."""
+
+    dest = StorageGroup.create(name="Dest")
+    StorageNode.create(name="DestNode", group=dest)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(2, ["file", "sync", "Acq/File", "--to=Dest"])
+
+
+def test_bad_source(clidb, cli):
+    """Test a bad source node."""
+
+    dest = StorageGroup.create(name="Dest")
+    StorageNode.create(name="DestNode", group=dest)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(1, ["file", "sync", "Acq/File", "--from=MISSING", "--to=Dest"])
+
+
+def test_no_dest(clidb, cli):
+    """Test missing dest group."""
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(2, ["file", "sync", "Acq/File", "--from=Src"])
+
+
+def test_bad_dest(clidb, cli):
+    """Test a bad dest group."""
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(1, ["file", "sync", "Acq/File", "--from=Src", "--to=MISSING"])
+
+
+def test_no_src(clidb, cli):
+    """Test file missing from src."""
+
+    dest = StorageGroup.create(name="Dest")
+    node = StorageNode.create(name="DestNode", group=dest)
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(0, ["file", "sync", "Acq/File", "--from=Src", "--to=Dest"])
+
+    # No request
+    assert ArchiveFileCopyRequest.select().count() == 0
+
+
+def test_sync(clidb, cli):
+    """Test a good sync."""
+
+    dest = StorageGroup.create(name="Dest")
+    node = StorageNode.create(name="DestNode", group=dest)
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+    ArchiveFileCopy.create(file=file_, node=src, has_file="Y", wants_file="Y")
+
+    cli(0, ["file", "sync", "Acq/File", "--from=Src", "--to=Dest", "--force"])
+
+    # Request created
+    afcr = ArchiveFileCopyRequest.get(id=1)
+    assert afcr.file == file_
+    assert afcr.node_from == src
+    assert afcr.group_to == dest
+    assert afcr.completed == 0
+    assert afcr.cancelled == 0
+
+
+def test_force(clidb, cli):
+    """Test force with no source file."""
+
+    dest = StorageGroup.create(name="Dest")
+    node = StorageNode.create(name="DestNode", group=dest)
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+
+    cli(0, ["file", "sync", "Acq/File", "--from=Src", "--to=Dest", "--force"])
+
+    # Request created
+    afcr = ArchiveFileCopyRequest.get(id=1)
+    assert afcr.file == file_
+    assert afcr.node_from == src
+    assert afcr.group_to == dest
+    assert afcr.completed == 0
+    assert afcr.cancelled == 0
+
+
+def test_dest_has(clidb, cli):
+    """Test file already on dest."""
+
+    dest = StorageGroup.create(name="Dest")
+    node = StorageNode.create(name="DestNode", group=dest)
+
+    group = StorageGroup.create(name="SrcGroup")
+    src = StorageNode.create(name="Src", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file_ = ArchiveFile.create(name="File", acq=acq)
+    ArchiveFileCopy.create(file=file_, node=src, has_file="Y", wants_file="Y")
+    ArchiveFileCopy.create(file=file_, node=node, has_file="Y", wants_file="Y")
+
+    cli(0, ["file", "sync", "Acq/File", "--from=Src", "--to=Dest"])
+
+    # No request
+    assert ArchiveFileCopyRequest.select().count() == 0
+
+
+def test_cancel_bad_file(clidb, cli):
+    """Test a bad file with --cancel."""
+
+    cli(1, ["file", "sync", "MIS/SING", "--cancel"])
+
+
+def test_cancel_all(clidb, cli):
+    """Test --cancel without limit."""
+
+    group = StorageGroup.create(name="Group")
+    node = StorageNode.create(name="Node", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file1 = ArchiveFile.create(name="File1", acq=acq)
+    file2 = ArchiveFile.create(name="File2", acq=acq)
+
+    req0 = ArchiveFileCopyRequest.create(
+        file=file1, node_from=node, group_to=group, cancelled=0, completed=0
+    )
+    req1 = ArchiveFileCopyRequest.create(
+        file=file1, node_from=node, group_to=group, cancelled=0, completed=1
+    )
+    req2 = ArchiveFileCopyRequest.create(
+        file=file1, node_from=node, group_to=group, cancelled=1, completed=0
+    )
+    req3 = ArchiveFileCopyRequest.create(
+        file=file1, node_from=node, group_to=group, cancelled=1, completed=1
+    )
+    req4 = ArchiveFileCopyRequest.create(
+        file=file2, node_from=node, group_to=group, cancelled=0, completed=0
+    )
+
+    cli(0, ["file", "sync", "Acq/File1", "--cancel"])
+
+    # req0 newly cancelled.   req2, 3 previously cancelled.  Others unchanged.
+    assert ArchiveFileCopyRequest.get(id=req0.id).cancelled
+    assert not ArchiveFileCopyRequest.get(id=req1.id).cancelled
+    assert ArchiveFileCopyRequest.get(id=req2.id).cancelled
+    assert ArchiveFileCopyRequest.get(id=req3.id).cancelled
+    assert not ArchiveFileCopyRequest.get(id=req4.id).cancelled
+
+
+def test_cancel_bad_node(clidb, cli):
+    """Test --cancel with a bad --from."""
+
+    acq = ArchiveAcq.create(name="Acq")
+    file = ArchiveFile.create(name="File", acq=acq)
+
+    cli(1, ["file", "sync", "Acq/File", "--cancel", "--from=Missing"])
+
+
+def test_cancel_node(clidb, cli):
+    """Test --cancel with --from."""
+
+    group = StorageGroup.create(name="Group")
+    node1 = StorageNode.create(name="Node1", group=group)
+    node2 = StorageNode.create(name="Node2", group=group)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file = ArchiveFile.create(name="File", acq=acq)
+
+    ArchiveFileCopyRequest.create(
+        file=file, node_from=node1, group_to=group, cancelled=0, completed=0
+    )
+    ArchiveFileCopyRequest.create(
+        file=file, node_from=node2, group_to=group, cancelled=0, completed=0
+    )
+
+    cli(0, ["file", "sync", "Acq/File", "--cancel", "--from=Node1"])
+
+    assert ArchiveFileCopyRequest.get(node_from=node1).cancelled
+    assert not ArchiveFileCopyRequest.get(node_from=node2).cancelled
+
+
+def test_cancel_bad_group(clidb, cli):
+    """Test --cancel with a bad --to."""
+
+    acq = ArchiveAcq.create(name="Acq")
+    file = ArchiveFile.create(name="File", acq=acq)
+
+    cli(1, ["file", "sync", "Acq/File", "--cancel", "--to=Missing"])
+
+
+def test_cancel_group(clidb, cli):
+    """Test --cancel with --to."""
+
+    group1 = StorageGroup.create(name="Group1")
+    group2 = StorageGroup.create(name="Group2")
+    node = StorageNode.create(name="Node", group=group1)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file = ArchiveFile.create(name="File", acq=acq)
+
+    ArchiveFileCopyRequest.create(
+        file=file, node_from=node, group_to=group1, cancelled=0, completed=0
+    )
+    ArchiveFileCopyRequest.create(
+        file=file, node_from=node, group_to=group2, cancelled=0, completed=0
+    )
+
+    cli(0, ["file", "sync", "Acq/File", "--cancel", "--to=Group1"])
+
+    assert ArchiveFileCopyRequest.get(group_to=group1).cancelled
+    assert not ArchiveFileCopyRequest.get(group_to=group2).cancelled
+
+
+def test_cancel_full(clidb, cli):
+    """Test --cancel with both --from and --to."""
+
+    group1 = StorageGroup.create(name="Group1")
+    node1 = StorageNode.create(name="Node1", group=group1)
+
+    group2 = StorageGroup.create(name="Group2")
+    node2 = StorageNode.create(name="Node2", group=group2)
+
+    acq = ArchiveAcq.create(name="Acq")
+    file = ArchiveFile.create(name="File", acq=acq)
+
+    req11 = ArchiveFileCopyRequest.create(
+        file=file, node_from=node1, group_to=group1, cancelled=0, completed=0
+    )
+    req12 = ArchiveFileCopyRequest.create(
+        file=file, node_from=node1, group_to=group2, cancelled=0, completed=0
+    )
+    req21 = ArchiveFileCopyRequest.create(
+        file=file, node_from=node2, group_to=group1, cancelled=0, completed=0
+    )
+    req22 = ArchiveFileCopyRequest.create(
+        file=file, node_from=node2, group_to=group2, cancelled=0, completed=0
+    )
+
+    cli(0, ["file", "sync", "Acq/File", "--cancel", "--from=Node1", "--to=Group2"])
+
+    assert not ArchiveFileCopyRequest.get(id=req11.id).cancelled
+    assert ArchiveFileCopyRequest.get(id=req12.id).cancelled
+    assert not ArchiveFileCopyRequest.get(id=req21.id).cancelled
+    assert not ArchiveFileCopyRequest.get(id=req22.id).cancelled