Skip to content

fix: fix reading copying status by rsync #21

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lkleinax wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: fix reading copying status by rsync #21

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 84 additions & 57 deletions mfd_kvm/hypervisor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)
Comment on lines +937 to +947
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,
)
Comment on lines +954 to +959
Comment on lines +954 to +959
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(",", "")
Comment on lines +974 to +977
else:
current_size = (
self._conn.execute_command(
f"ls {path_to_destination_image} -l | awk '{{print $5}}'",
Expand All @@ -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,
Expand Down
74 changes: 64 additions & 10 deletions tests/unit/test_mfd_kvm/test_hypervisor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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")
Expand All @@ -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")
Expand All @@ -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
Expand All @@ -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),
]
Expand Down