Libguestfs - Nov 2019 - [PATCH 1/2] rhv-upload: Extract cancel_transfer() function

Fix NameError introduced by inocomplete change to extract create_transfer() by
extracting cancel_transfer() function.

Finally extract finallize_transfer() function, dealing with the poorly
documented and over complicated oVirt SDK.

Tested with:
libguestfs-1.40.2-4.fc30.x86_64
nbdkit-1.12.8-1.fc30.x86_64

Tested flows:
- Successful import
- Error in close() - by injecting long sleep in vdsm verifaction step
- Error in close() - by injecting error in vdsm verification step
- Error in open() - by injecting error in imageio OPTIONS handler
- Error in zero() - by injecting error in imageio PATCH handler
- Error in flush() - by injecting error in imageio PATCH handler

Nir Soffer (2):
  rhv-upload: Extract cancel_transfer() function
  rhv-upload: Extract finalize_transfer() function

 v2v/rhv-upload-plugin.py | 118 ++++++++++++++++++++++++++-------------
 1 file changed, 78 insertions(+), 40 deletions(-)

-- 
2.21.0

Extract a cancel_transfer() function, so we can cancel a transfer in
plugin handlers without keeping the transfer_service around.

Fixes a NameError when starting a transfer fails:

    During handling of the above exception, another exception occurred:

    Traceback (most recent call last):
      File "/var/tmp/rhvupload.Vw0CIU/rhv-upload-plugin.py", line 97,
in open
        transfer_service.cancel()
    NameError: name 'transfer_service' is not defined

Thanks: Martin Kletzander
---
 v2v/rhv-upload-plugin.py | 12 +++++++++++-
 1 file changed, 11 insertions(+), 1 deletion(-)

diff --git a/v2v/rhv-upload-plugin.py b/v2v/rhv-upload-plugin.py
index a2d09458..b9b9e967 100644
--- a/v2v/rhv-upload-plugin.py
+++ b/v2v/rhv-upload-plugin.py
@@ -93,7 +93,7 @@ def open(readonly):
         options = get_options(http, destination_url)
         http = optimize_http(http, host, options)
     except:
-        transfer_service.cancel()
+        cancel_transfer(connection, transfer)
         raise
 
     debug("imageio features: flush=%(can_flush)r trim=%(can_trim)r "
@@ -531,6 +531,16 @@ def create_transfer(connection, disk, host):
 
     return transfer
 
+def cancel_transfer(connection, transfer):
+    """
+    Cancel a transfer, removing the transfer disk.
+    """
+    debug("canceling transfer %s" % transfer.id)
+    transfer_service = (connection.system_service()
+                            .image_transfers_service()
+                            .image_transfer_service(transfer.id))
+    transfer_service.cancel()
+
 # oVirt imageio operations
 
 def parse_transfer_url(transfer):
-- 
2.21.0

Extract finalize_transfer() function hiding the complexity of finalizing
and waiting for a transfer, and separating plugin logic like writing the
disk_id file from oVirt logic.

Since canceling a transfer is not part of finalizing, but the current
plugin policy, move cancellation after finalize failure to close().

When waiting for the disk status change, handle the ILLEGAL status, so
we can fail quickly if oVirt failed to finalized and paused the
transfer, instead of waiting for timeout.
---
 v2v/rhv-upload-plugin.py | 106 +++++++++++++++++++++++++--------------
 1 file changed, 67 insertions(+), 39 deletions(-)

diff --git a/v2v/rhv-upload-plugin.py b/v2v/rhv-upload-plugin.py
index b9b9e967..85dec7a6 100644
--- a/v2v/rhv-upload-plugin.py
+++ b/v2v/rhv-upload-plugin.py
@@ -306,10 +306,7 @@ def close(h):
     http = h['http']
     connection = h['connection']
     transfer = h['transfer']
-
-    transfer_service = (connection.system_service()
-                            .image_transfers_service()
-                            .image_transfer_service(transfer.id))
+    disk_id = h['disk_id']
 
     # This is sometimes necessary because python doesn't set up
     # sys.stderr to be line buffered and so debug, errors or
@@ -323,49 +320,22 @@ def close(h):
     # the transfer will delete the disk.
     if h['failed']:
         try:
-            transfer_service.cancel()
+            cancel_transfer(connection, transfer)
         finally:
             connection.close()
         return
 
+    # Try to finalize the transfer. On errors the transfer may be paused by the
+    # system, and we need to cancel the transfer to remove the disk.
     try:
-        disk_id = h['disk_id']
-
-        transfer_service.finalize()
-
-        # Wait until the transfer disk job is completed since
-        # only then we can be sure the disk is unlocked.  As this
-        # code is not very clear, what's happening is that we are
-        # waiting for the transfer object to cease to exist, which
-        # falls through to the exception case and then we can
-        # continue.
-        disk_service = (
-            connection.system_service().disks_service().disk_service(disk_id))
-        start = time.time()
-        try:
-            while True:
-                time.sleep(1)
-                disk = disk_service.get()
-                if disk.status == types.DiskStatus.LOCKED:
-                    if time.time() > start + timeout:
-                        raise RuntimeError("timed out waiting for transfer
"
-                                           "to finalize")
-                    continue
-                if disk.status == types.DiskStatus.OK:
-                    debug("finalized after %s seconds" % (time.time()
- start))
-                    break
-        except sdk.NotFoundError:
-            raise RuntimeError("transfer failed: disk %s not found" %
disk_id)
-
+        finalize_transfer(connection, transfer, disk_id)
+    except:
+        cancel_transfer(connection, transfer)
+        raise
+    else:
         # Write the disk ID file.  Only do this on successful completion.
         with builtins.open(params['diskid_file'], 'w') as fp:
             fp.write(disk_id)
-
-    except:
-        # If oVirt engine fails to finalize the transfer, it will pause the
-        # transfer and keep the disk.
-        transfer_service.cancel()
-        raise
     finally:
         connection.close()
 
@@ -541,6 +511,64 @@ def cancel_transfer(connection, transfer):
                             .image_transfer_service(transfer.id))
     transfer_service.cancel()
 
+def finalize_transfer(connection, transfer, disk_id):
+    """
+    Finalize a transfer, making the transfer disk available.
+
+    If finalizing succeeds, transfer's phase will change to
FINISHED_SUCCESS
+    and the transer's disk status will change to OK.  On errors, the
transfer's
+    phase will changed to FINISHED_FAILURE and the disk status will change to
+    ILLEGAL and it will be removed. In both cases the transfer entity will be
+    removed shortly after.
+
+    If oVirt fails to finalize the transfer, transfer's phase will changed
to
+    PAUSED_SYSTEM. In this case the disk's status will change to ILLEGAL
and it
+    will not be removed.
+
+    For simplicity, we track only disk's status changes.
+
+    For more info see:
+    -
http://ovirt.github.io/ovirt-engine-api-model/4.4/#services/image_transfer
+    -
http://ovirt.github.io/ovirt-engine-sdk/master/types.m.html#ovirtsdk4.types.ImageTransfer
+    """
+    debug("finalizing transfer %s" % transfer.id)
+    transfer_service = (connection.system_service()
+                            .image_transfers_service()
+                            .image_transfer_service(transfer.id))
+
+    start = time.time()
+
+    transfer_service.finalize()
+
+    disk_service = (connection.system_service()
+                        .disks_service()
+                        .disk_service(disk_id))
+
+    while True:
+        time.sleep(1)
+        try:
+            disk = disk_service.get()
+        except sdk.NotFoundError:
+            # Disk verification failed and the system removed the disk.
+            raise RuntimeError(
+                "transfer %s failed: disk %s was removed"
+                % (transfer.id, disk_id))
+
+        if disk.status == types.DiskStatus.ILLEGAL:
+            # Disk verification failed or transfer was paused by the system.
+            raise RuntimeError(
+                "transfer %s failed: disk is ILLEGAL" % transfer.id)
+
+        if disk.status == types.DiskStatus.OK:
+            debug("transfer %s finalized in %.3f seconds"
+                  % (transfer.id, time.time() - start))
+            break
+
+        if time.time() > start + timeout:
+            raise RuntimeError(
+                "timed out waiting for transfer %s to finalize"
+                % transfer.id)
+
 # oVirt imageio operations
 
 def parse_transfer_url(transfer):
-- 
2.21.0

On Wed, Nov 20, 2019 at 1:19 AM Nir Soffer <nirsof@gmail.com>
wrote:
>
> Fix NameError introduced by inocomplete change to extract create_transfer()
by
> extracting cancel_transfer() function.
>
> Finally extract finallize_transfer() function, dealing with the poorly
> documented and over complicated oVirt SDK.
>
> Tested with:
> libguestfs-1.40.2-4.fc30.x86_64
> nbdkit-1.12.8-1.fc30.x86_64
>
> Tested flows:
> - Successful import
> - Error in close() - by injecting long sleep in vdsm verifaction step
> - Error in close() - by injecting error in vdsm verification step
> - Error in open() - by injecting error in imageio OPTIONS handler
> - Error in zero() - by injecting error in imageio PATCH handler
> - Error in flush() - by injecting error in imageio PATCH handler
Here is output from the tests runs:

## Simulate verification error by injecting fake error in vdsm.
...
nbdkit: python[1]: debug: client sent NBD_CMD_DISC, closing connection
nbdkit: python[1]: debug: close
finalizing transfer 01aecf9a-2629-40e4-8528-f310a4ad6c82
virtual copying rate: 3368.6 M bits/sec
canceling transfer 01aecf9a-2629-40e4-8528-f310a4ad6c82
nbdkit: python[1]: error:
/home/nsoffer/src/virt-v2v/tmp/rhvupload.SZBNp8/rhv-upload-plugin.py:
close: error: ['Traceback (most recent call last):\n', '  File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.SZBNp8/rhv-upload-plugin.py",
line 332, in close\n    finalize_transfer(connection, transfer,
disk_id)\n', '  File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.SZBNp8/rhv-upload-plugin.py",
line 561, in finalize_transfer\n    "transfer %s failed: disk is
ILLEGAL" % transfer.id)\n', 'RuntimeError: transfer
01aecf9a-2629-40e4-8528-f310a4ad6c82 failed: disk is ILLEGAL\n']


## Simulate inaccessible storage by adding long sleep in vdsm.
...
nbdkit: python[1]: debug: client sent NBD_CMD_DISC, closing connection
nbdkit: python[1]: debug: close
finalizing transfer c05c578e-df99-4fb5-80f0-6944051dbee0
virtual copying rate: 2281.9 M bits/sec
canceling transfer c05c578e-df99-4fb5-80f0-6944051dbee0
nbdkit: python[1]: error:
/home/nsoffer/src/virt-v2v/tmp/rhvupload.YKpW3U/rhv-upload-plugin.py:
close: error: ['Traceback (most recent call last):\n', '  File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.YKpW3U/rhv-upload-plugin.py",
line 332, in close\n    finalize_transfer(connection, transfer,
disk_id)\n', '  File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.YKpW3U/rhv-upload-plugin.py",
line 561, in finalize_transfer\n    "transfer %s failed: disk is
ILLEGAL" % transfer.id)\n', 'RuntimeError: transfer
c05c578e-df99-4fb5-80f0-6944051dbee0 failed: disk is ILLEGAL\n']


## Simulate error after transfer was started by returning invalid json
in imageio OPTIONS handler
...
disk.id = 'a2bad783-0736-42a5-92d1-d76d133453bf'
transfer.id = 'fd297a0d-1c9b-49ac-9104-829358127f23'
canceling transfer fd297a0d-1c9b-49ac-9104-829358127f23
nbdkit: python[1]: error:
/home/nsoffer/src/virt-v2v/tmp/rhvupload.nejhOD/rhv-upload-plugin.py:
open: error: ['Traceback (most recent call last):\n', '  File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.nejhOD/rhv-upload-plugin.py",
line 94, in open\n    options = get_options(http, destination_url)\n',
'  File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.nejhOD/rhv-upload-plugin.py",
line 618, in get_options\n    j = json.loads(data)\n', '  File
"/usr/lib64/python3.7/json/__init__.py", line 348, in loads\n
return _default_decoder.decode(s)\n', '  File
"/usr/lib64/python3.7/json/decoder.py", line 337, in decode\n    obj,
end = self.raw_decode(s, idx=_w(s, 0).end())\n', '  File
"/usr/lib64/python3.7/json/decoder.py", line 355, in raw_decode\n
raise JSONDecodeError("Expecting value", s, err.value) from
None\n',
'json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char
0)\n']
qemu-img: Could not open 'json:{ "file.driver": "nbd",
"file.path":
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.nejhOD/nbdkit0.sock",
"file.export": "/" }': Failed to read option reply:
Unexpected
end-of-file before all bytes were read


## Simulate error in zero by injecting error in imageio PATCH/zero
...
nbdkit: python[1]: debug: zero count=458752 offset=6441926656 may_trim=1 fua=0
unexpected response from imageio server:
could not zero sector offset 6441926656 size 458752
500: Internal Server Error
b'Server failed to perform the request, check logs'
nbdkit: python[1]: error:
/home/nsoffer/src/virt-v2v/tmp/rhvupload.3rv9Nk/rhv-upload-plugin.py:
zero: error: ['Traceback (most recent call last):\n', '  File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.3rv9Nk/rhv-upload-plugin.py",
line 229, in zero\n    (offset, count))\n', '  File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.3rv9Nk/rhv-upload-plugin.py",
line 149, in request_failed\n    raise RuntimeError("%s: %d %s: %r" %
(msg, status, reason, body[:200]))\n', "RuntimeError: could not zero
sector offset 6441926656 size 458752: 500 Internal Server Error:
b'Server failed to perform the request, check logs'\n"]
nbdkit: python[1]: debug: sending error reply: Input/output error
nbdkit: python[1]: debug: pwrite count=458752 offset=6441926656 fua=0
nbdkit: python[1]: debug: pwrite count=65536 offset=6442385408 fua=0
    (100.00/100%)
nbdkit: python[1]: debug: flush
nbdkit: python[1]: debug: flush
nbdkit: python[1]: debug: client sent NBD_CMD_DISC, closing connection
nbdkit: python[1]: debug: close
canceling transfer 86af4ae7-e1dc-4514-ad95-d7587678f946
virtual copying rate: 1213.8 M bits/sec


## Simulate error in flush by injecting error in imageio PATCH/flush
...
nbdkit: python[1]: debug: pwrite count=65536 offset=6442385408 fua=0
    (100.00/100%)
nbdkit: python[1]: debug: flush
unexpected response from imageio server:
could not flush
500: Internal Server Error
b'Server failed to perform the request, check logs'
nbdkit: python[1]: error:
/home/nsoffer/src/virt-v2v/tmp/rhvupload.vPoylv/rhv-upload-plugin.py:
flush: error: ['Traceback (most recent call last):\n', '  File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.vPoylv/rhv-upload-plugin.py",
line 302, in flush\n    request_failed(h, r, "could not
flush")\n', '
File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.vPoylv/rhv-upload-plugin.py",
line 149, in request_failed\n    raise RuntimeError("%s: %d %s: %r" %
(msg, status, reason, body[:200]))\n', "RuntimeError: could not flush:
500 Internal Server Error: b'Server failed to perform the request,
check logs'\n"]
nbdkit: python[1]: debug: sending error reply: Input/output error
nbdkit: python[1]: debug: flush
unexpected response from imageio server:
could not flush
500: Internal Server Error
b'Server failed to perform the request, check logs'
nbdkit: python[1]: error:
/home/nsoffer/src/virt-v2v/tmp/rhvupload.vPoylv/rhv-upload-plugin.py:
flush: error: ['Traceback (most recent call last):\n', '  File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.vPoylv/rhv-upload-plugin.py",
line 302, in flush\n    request_failed(h, r, "could not
flush")\n', '
File
"/home/nsoffer/src/virt-v2v/tmp/rhvupload.vPoylv/rhv-upload-plugin.py",
line 149, in request_failed\n    raise RuntimeError("%s: %d %s: %r" %
(msg, status, reason, body[:200]))\n', "RuntimeError: could not flush:
500 Internal Server Error: b'Server failed to perform the request,
check logs'\n"]
nbdkit: python[1]: debug: sending error reply: Input/output error
nbdkit: python[1]: debug: client sent NBD_CMD_DISC, closing connection
nbdkit: python[1]virtual copying rate: 2227.0 M bits/sec
: debug: close
canceling transfer 6508e39c-6956-4333-8f27-1f2b55377986
virt-v2v: error: transfer of disk 1/1 failed, see earlier error messages

Nir

On Wed, Nov 20, 2019 at 01:19:31AM +0200, Nir Soffer
wrote:
>Fix NameError introduced by inocomplete change to extract create_transfer()
by
>extracting cancel_transfer() function.
>
>Finally extract finallize_transfer() function, dealing with the poorly
>documented and over complicated oVirt SDK.
>
>Tested with:
>libguestfs-1.40.2-4.fc30.x86_64
>nbdkit-1.12.8-1.fc30.x86_64
>
>Tested flows:
>- Successful import
>- Error in close() - by injecting long sleep in vdsm verifaction step
>- Error in close() - by injecting error in vdsm verification step
>- Error in open() - by injecting error in imageio OPTIONS handler
>- Error in zero() - by injecting error in imageio PATCH handler
>- Error in flush() - by injecting error in imageio PATCH handler
>
>Nir Soffer (2):
>  rhv-upload: Extract cancel_transfer() function
>  rhv-upload: Extract finalize_transfer() function
>
Both look OK to me, although I have not tested the second one yet.  There are
just a few typos in the second patch that could get some s/changed/change/
treatment, but it's so minor I don't want send another mail just for
that =)
> v2v/rhv-upload-plugin.py | 118 ++++++++++++++++++++++++++-------------
> 1 file changed, 78 insertions(+), 40 deletions(-)
>
>-- 
>2.21.0
>

On Wed, Nov 20, 2019 at 1:19 AM Nir Soffer <nirsof@gmail.com>
wrote:
>
> Extract a cancel_transfer() function, so we can cancel a transfer in
> plugin handlers without keeping the transfer_service around.
>
> Fixes a NameError when starting a transfer fails:
>
>     During handling of the above exception, another exception occurred:
>
>     Traceback (most recent call last):
>       File "/var/tmp/rhvupload.Vw0CIU/rhv-upload-plugin.py", line
97, in open
>         transfer_service.cancel()
>     NameError: name 'transfer_service' is not defined
Richard, this should be merged before you release 1.48. Without this we can
leave leftover disks if get_options() raises.
> Thanks: Martin Kletzander
> ---
>  v2v/rhv-upload-plugin.py | 12 +++++++++++-
>  1 file changed, 11 insertions(+), 1 deletion(-)
>
> diff --git a/v2v/rhv-upload-plugin.py b/v2v/rhv-upload-plugin.py
> index a2d09458..b9b9e967 100644
> --- a/v2v/rhv-upload-plugin.py
> +++ b/v2v/rhv-upload-plugin.py
> @@ -93,7 +93,7 @@ def open(readonly):
>          options = get_options(http, destination_url)
>          http = optimize_http(http, host, options)
>      except:
> -        transfer_service.cancel()
> +        cancel_transfer(connection, transfer)
>          raise
>
>      debug("imageio features: flush=%(can_flush)r trim=%(can_trim)r
"
> @@ -531,6 +531,16 @@ def create_transfer(connection, disk, host):
>
>      return transfer
>
> +def cancel_transfer(connection, transfer):
> +    """
> +    Cancel a transfer, removing the transfer disk.
> +    """
> +    debug("canceling transfer %s" % transfer.id)
> +    transfer_service = (connection.system_service()
> +                            .image_transfers_service()
> +                            .image_transfer_service(transfer.id))
> +    transfer_service.cancel()
> +
>  # oVirt imageio operations
>
>  def parse_transfer_url(transfer):
> --
> 2.21.0
>