From 5142a1fea843da8325cb6b2a16c4a4eec8ec798f Mon Sep 17 00:00:00 2001 From: lkleinax Date: Tue, 27 Oct 2026 15:09:41 +0100 Subject: [PATCH] fix: fix reading copying status by rsync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes #20 Signed-off-by: Łukasz Kleina --- mfd_kvm/hypervisor.py | 141 ++++++++++++--------- tests/unit/test_mfd_kvm/test_hypervisor.py | 74 +++++++++-- 2 files changed, 148 insertions(+), 67 deletions(-) diff --git a/mfd_kvm/hypervisor.py b/mfd_kvm/hypervisor.py index df579ca..d2f0cba 100644 --- a/mfd_kvm/hypervisor.py +++ b/mfd_kvm/hypervisor.py @@ -900,53 +900,82 @@ def clone_vm_hdd_image( msg=f"Cloning disk image {path_to_source_image} into {path_to_destination_image}, " f"timeout - {timeout} secs.", ) - if path_to_source_image.is_file(): - if path_to_destination_image.exists(): - # if source image contains suffix that matches checksum pattern use it for comparison - source_checksum = path_to_source_image.name.rsplit(".", maxsplit=1)[-1] - if re.fullmatch(r"[a-fA-F0-9]{32}", source_checksum): - logger.log( - log_levels.MODULE_DEBUG, - msg=f"Pre-calculated checksum found in the source filename: {source_checksum}. " - "It will be used for comparison with the destination image.", - ) - else: - logger.log( - log_levels.MODULE_DEBUG, - msg="Pre-calculated checksum not found in the source filename. " - "It will be calculated directly for comparison.", - ) - source_checksum = self._conn.execute_command(f"md5sum {path_to_source_image}").stdout.split()[0] - destination_checksum = self._conn.execute_command(f"md5sum {path_to_destination_image}").stdout.split()[ - 0 - ] - if source_checksum.lower() == destination_checksum.lower(): - logger.log( - log_levels.MODULE_DEBUG, - msg=f"Destination image {path_to_destination_image} already matches the source; " - "skipping cloning.", - ) - return path_to_destination_image + if not path_to_source_image.is_file(): + raise FileNotFoundError(f"Not found {path_to_source_image} in system.") + + if path_to_destination_image.exists(): + # if source image contains suffix that matches checksum pattern use it for comparison + source_checksum = path_to_source_image.name.rsplit(".", maxsplit=1)[-1] + if re.fullmatch(r"[a-fA-F0-9]{32}", source_checksum): + logger.log( + log_levels.MODULE_DEBUG, + msg=f"Pre-calculated checksum found in the source filename: {source_checksum}. " + "It will be used for comparison with the destination image.", + ) + else: logger.log( log_levels.MODULE_DEBUG, - msg=f"Destination image {path_to_destination_image} differs from the source; removing it.", + msg="Pre-calculated checksum not found in the source filename. " + "It will be calculated directly for comparison.", ) - path_to_destination_image.unlink() - - timeout_counter = TimeoutCounter(timeout) - copy_command = "rsync -aq" if self._rsync_available else "scp" - # show 5 column of ls, size of file - target_size = self._conn.execute_command( - f"ls {path_to_source_image} -l | awk '{{print $5}}'", shell=True - ).stdout - copy_process = self._conn.start_process( - f"{copy_command} {path_to_source_image} {path_to_destination_image}" + source_checksum = self._conn.execute_command(f"md5sum {path_to_source_image}").stdout.split()[0] + destination_checksum = self._conn.execute_command(f"md5sum {path_to_destination_image}").stdout.split()[0] + if source_checksum.lower() == destination_checksum.lower(): + logger.log( + log_levels.MODULE_DEBUG, + msg=f"Destination image {path_to_destination_image} already matches the source; " + "skipping cloning.", + ) + return path_to_destination_image + logger.log( + log_levels.MODULE_DEBUG, + msg=f"Destination image {path_to_destination_image} differs from the source; removing it.", ) - logger.log(log_levels.MODULE_DEBUG, msg="Checking if cloning is completed..") - while not timeout_counter: - if not copy_process.running: - logger.log(log_levels.MODULE_DEBUG, msg="Cloning completed.") - return path_to_destination_image + path_to_destination_image.unlink() + + timeout_counter = TimeoutCounter(timeout) + copy_command = "rsync -av" if self._rsync_available else "scp" + params = f" --info=progress2 --outbuf=N " if self._rsync_available else " " + # show 5 column of ls, size of file + target_size = self._conn.execute_command( + f"ls {path_to_source_image} -l | awk '{{print $5}}'", + shell=True, + ).stdout + copy_process = self._conn.start_process( + f"{copy_command}{params}{path_to_source_image} {path_to_destination_image}", + log_file=self._rsync_available, + ) + logger.log(log_levels.MODULE_DEBUG, msg="Checking if cloning is completed..") + while not timeout_counter: + if not copy_process.running: + logger.log(log_levels.MODULE_DEBUG, msg="Cloning completed.") + return path_to_destination_image + if self._rsync_available: + out = self._conn.execute_command( + f"cat -A {copy_process.log_path}", + shell=True, + expected_return_codes={}, + skip_logging=True, + ) + if out.return_code != 0: + logger.log( + log_levels.MODULE_DEBUG, + msg=f"Failed to read rsync log file {copy_process.log_path}. " + f"Return code: {out.return_code}, output: {out.stderr}", + ) + current_size = 0 + else: + last_line = str(out.stdout).split("^M")[-1] + logger.log(log_levels.MODULE_DEBUG, msg=f"Last line of rsync process:\n{last_line}") + if last_line.strip() == "": + current_size = 0 + elif "sent" in last_line or "total size" in last_line: + current_size = target_size + elif path_to_destination_image.name in last_line or "sending incremental file list" in last_line: + current_size = 0 + else: + current_size = last_line.strip().split()[0].replace(",", "") + else: current_size = ( self._conn.execute_command( f"ls {path_to_destination_image} -l | awk '{{print $5}}'", @@ -955,22 +984,20 @@ def clone_vm_hdd_image( if path_to_destination_image.exists() else 0 ) - try: - current_size = int(current_size) - except (TypeError, ValueError): - current_size = 0 - if current_size > 0: - logging.log( - log_levels.MODULE_DEBUG, - msg=f"still cloning... {current_size / int(target_size) * 100:.0f} %, next check in 30secs.", - ) - sleep(30) - else: - raise KVMHypervisorException( - f"Cloning image {path_to_source_image} not finished in given timeout: {timeout}" + try: + current_size = int(current_size) + except (TypeError, ValueError): + current_size = 0 + if current_size > 0: + logging.log( + log_levels.MODULE_DEBUG, + msg=f"still cloning... {current_size / int(target_size) * 100:.0f} %, next check in 30secs.", ) + sleep(30) else: - raise FileNotFoundError(f"Not found {path_to_source_image} in system.") + raise KVMHypervisorException( + f"Cloning image {path_to_source_image} not finished in given timeout: {timeout}" + ) def create_mdev( self, diff --git a/tests/unit/test_mfd_kvm/test_hypervisor.py b/tests/unit/test_mfd_kvm/test_hypervisor.py index fa0f517..413bbfc 100644 --- a/tests/unit/test_mfd_kvm/test_hypervisor.py +++ b/tests/unit/test_mfd_kvm/test_hypervisor.py @@ -905,7 +905,10 @@ def test_clone_vm_hdd_image_checks_destination_on_connection(self, hv, mocker, t "ls /remote/source.img -l | awk '{print $5}'", shell=True, ) - hv._conn.start_process.assert_called_once_with("rsync -aq /remote/source.img /remote/destination.img") + hv._conn.start_process.assert_called_once_with( + "rsync -av --info=progress2 --outbuf=N /remote/source.img /remote/destination.img", + log_file=True, + ) @pytest.mark.parametrize( "source_name", @@ -988,8 +991,12 @@ def test_clone_vm_hdd_image_filename_checksum_differs(self, hv, mocker, tmp_path assert result == destination_path assert not destination_path.exists() - assert f"Destination image {destination_path} differs from the source; removing it." in caplog.text - hv._conn.start_process.assert_called_once_with(f"rsync -aq {source_path} {destination_path}") + messages = [r.getMessage() for r in caplog.records] + assert f"Destination image {destination_path} differs from the source; removing it." in messages + hv._conn.start_process.assert_called_once_with( + f"rsync -av --info=progress2 --outbuf=N {source_path} {destination_path}", + log_file=True, + ) def test_clone_vm_hdd_image_timeout_exceeded(self, hv, mocker): timeout_mocker = mocker.patch("mfd_kvm.hypervisor.TimeoutCounter") @@ -1012,7 +1019,7 @@ def test_clone_vm_hdd_image_timeout_exceeded(self, hv, mocker): @pytest.mark.parametrize( ("rsync_return_code", "copy_command"), - [(0, "rsync -aq"), (1, "scp")], + [(0, "rsync -av --info=progress2 --outbuf=N"), (1, "scp")], ) def test_clone_vm_hdd_image_source_file_exists_and_succeeded(self, hv, mocker, rsync_return_code, copy_command): timeout_mocker = mocker.patch("mfd_kvm.hypervisor.TimeoutCounter") @@ -1039,38 +1046,86 @@ def test_clone_vm_hdd_image_source_file_exists_and_succeeded(self, hv, mocker, r mocker.call(f"ls {path_mocker} -l | awk '{{print $5}}'", shell=True), mocker.call(f"ls {path_mocker} -l | awk '{{print $5}}'", shell=True), ] + log_file = copy_command != "scp" assert hv._conn.start_process.call_args_list == [ - mocker.call(f"{copy_command} {path_mocker} {dest_path}"), - mocker.call(f"{copy_command} {path_mocker} {dest_path}"), + mocker.call(f"{copy_command} {path_mocker} {dest_path}", log_file=log_file), + mocker.call(f"{copy_command} {path_mocker} {dest_path}", log_file=log_file), ] assert new_names == [dest_path, dest_path] - def test_clone_vm_hdd_image_still_cloning(self, hv, mocker, caplog): + def test_clone_vm_hdd_image_still_cloning_scp(self, hv, mocker, caplog): caplog.set_level(log_levels.MODULE_DEBUG) timeout_mocker = mocker.patch("mfd_kvm.hypervisor.TimeoutCounter") timeout_mocker.return_value.__bool__.return_value = False process = mocker.create_autospec(RPyCProcess) type(process).running = mocker.PropertyMock(side_effect=[True, False]) + process.log_path = "/tmp/logfile.log" + hv.__dict__["_rsync_available"] = False hv._conn.start_process.return_value = process path_mocker = mocker.patch("mfd_kvm.hypervisor.Path") dest_path = mocker.patch("mfd_kvm.hypervisor.Path") mocker.patch("mfd_kvm.hypervisor.sleep") path_mocker.is_file.return_value = True dest_path.exists.side_effect = [False, True] + dest_path.name = "destination.img" hv._conn.path.side_effect = [path_mocker, dest_path] hv._conn.execute_command.side_effect = [ - ConnectionCompletedProcess(args="", return_code=0), ConnectionCompletedProcess(args="", stdout="100", return_code=0), ConnectionCompletedProcess(args="", stdout="10", return_code=0), ] hv.clone_vm_hdd_image(path_to_source_image=path_mocker, path_to_destination_image=dest_path) - assert "still cloning... 10 %, next check in 30secs." in caplog.text + messages = [r.getMessage() for r in caplog.records] + assert "still cloning... 10 %, next check in 30secs." in messages + + def test_clone_vm_hdd_image_still_cloning_rsync(self, hv, mocker, caplog): + caplog.set_level(log_levels.MODULE_DEBUG) + timeout_mocker = mocker.patch("mfd_kvm.hypervisor.TimeoutCounter") + timeout_mocker.return_value.__bool__.return_value = False + process = mocker.create_autospec(RPyCProcess) + type(process).running = mocker.PropertyMock(side_effect=[True, True, False]) + process.log_path = "/tmp/logfile.log" + hv.__dict__["_rsync_available"] = True + hv._conn.start_process.return_value = process + path_mocker = mocker.patch("mfd_kvm.hypervisor.Path") + dest_path = mocker.patch("mfd_kvm.hypervisor.Path") + mocker.patch("mfd_kvm.hypervisor.sleep") + path_mocker.is_file.return_value = True + dest_path.exists.side_effect = [False, True] + dest_path.name = "destination.img" + hv._conn.path.side_effect = [path_mocker, dest_path] + rsync_still_cloning_output = dedent( + """\ + sending incremental file list + source.img + ^M 32,768 0% 0.00kB/s 0:00:00 ^M 22,577,152 0% 21.31MB/s 0:11:32 ^M + 1,512,793,702 10% 19.56MB/s 0:12:33 + """ + ) + rsync_finished_cloning_output = dedent( + """\ + sending incremental file list + source.img + ^M 32,768 0% 0.00kB/s 0:00:00 ^M 22,577,152 0% 21.31MB/s 0:11:32 ^M + 1,512,793,702 10% 19.56MB/s 0:12:33 ^M + total size is 15,127,937,020 speedup is 1.00 + """ + ) + hv._conn.execute_command.side_effect = [ + ConnectionCompletedProcess(args="", stdout="15127937020", return_code=0), + ConnectionCompletedProcess(args="", stdout=rsync_still_cloning_output, return_code=0), + ConnectionCompletedProcess(args="", stdout=rsync_finished_cloning_output, return_code=0), + ] + hv.clone_vm_hdd_image(path_to_source_image=path_mocker, path_to_destination_image=dest_path) + messages = [r.getMessage() for r in caplog.records] + assert "still cloning... 10 %, next check in 30secs." in messages + assert "still cloning... 100 %, next check in 30secs." in messages @pytest.mark.parametrize("current_size", ["", "invalid", "0", "-1"]) def test_clone_vm_hdd_image_invalid_current_size(self, hv, mocker, caplog, current_size): caplog.set_level(log_levels.MODULE_DEBUG) timeout_mocker = mocker.patch("mfd_kvm.hypervisor.TimeoutCounter") timeout_mocker.return_value.__bool__.return_value = False + hv.__dict__["_rsync_available"] = False process = mocker.create_autospec(RPyCProcess) type(process).running = mocker.PropertyMock(side_effect=[True, False]) hv._conn.start_process.return_value = process @@ -1081,7 +1136,6 @@ def test_clone_vm_hdd_image_invalid_current_size(self, hv, mocker, caplog, curre dest_path.exists.side_effect = [False, True] hv._conn.path.side_effect = [path_mocker, dest_path] hv._conn.execute_command.side_effect = [ - ConnectionCompletedProcess(args="", return_code=0), ConnectionCompletedProcess(args="", stdout="100", return_code=0), ConnectionCompletedProcess(args="", stdout=current_size, return_code=0), ]