#!/usr/bin/env python3 # group: auto backup # # Copyright (c) 2022 Virtuozzo International GmbH # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . # import os import re from qemu.machine import QEMUMachine import iotests from iotests import qemu_img_create, qemu_io temp_img = os.path.join(iotests.test_dir, 'temp') source_img = os.path.join(iotests.test_dir, 'source') size = '1M' class TestCbwError(iotests.QMPTestCase): def tearDown(self): self.vm.shutdown() os.remove(temp_img) os.remove(source_img) def setUp(self): qemu_img_create('-f', iotests.imgfmt, source_img, size) qemu_img_create('-f', iotests.imgfmt, temp_img, size) qemu_io('-c', 'write 0 1M', source_img) opts = ['-nodefaults', '-display', 'none', '-machine', 'none'] self.vm = QEMUMachine(iotests.qemu_prog, opts, base_temp_dir=iotests.test_dir) self.vm.launch() def do_cbw_error(self, on_cbw_error): self.vm.cmd('blockdev-add', { 'node-name': 'cbw', 'driver': 'copy-before-write', 'on-cbw-error': on_cbw_error, 'file': { 'driver': iotests.imgfmt, 'file': { 'driver': 'file', 'filename': source_img, } }, 'target': { 'driver': iotests.imgfmt, 'file': { 'driver': 'blkdebug', 'image': { 'driver': 'file', 'filename': temp_img }, 'inject-error': [ { 'event': 'write_aio', 'errno': 5, 'immediately': False, 'once': True } ] } } }) self.vm.cmd('blockdev-add', { 'node-name': 'access', 'driver': 'snapshot-access', 'file': 'cbw' }) result = self.vm.qmp('human-monitor-command', command_line='qemu-io cbw "write 0 1M"') self.assert_qmp(result, 'return', '') result = self.vm.qmp('human-monitor-command', command_line='qemu-io access "read 0 1M"') self.assert_qmp(result, 'return', '') self.vm.shutdown() log = self.vm.get_log() log = re.sub(r'^\[I \d+\.\d+\] OPENED\n', '', log) log = re.sub(r'\[I \+\d+\.\d+\] CLOSED\n?$', '', log) log = iotests.filter_qemu_io(log) return log def test_break_snapshot_on_cbw_error(self): """break-snapshot behavior: Guest write succeed, but further snapshot-read fails, as snapshot is broken. """ log = self.do_cbw_error('break-snapshot') self.assertEqual(log, """\ wrote 1048576/1048576 bytes at offset 0 1 MiB, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec) read failed: Permission denied """) def test_break_guest_write_on_cbw_error(self): """break-guest-write behavior: Guest write fails, but snapshot-access continues working and further snapshot-read succeeds. """ log = self.do_cbw_error('break-guest-write') self.assertEqual(log, """\ write failed: Input/output error read 1048576/1048576 bytes at offset 0 1 MiB, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec) """) def do_cbw_timeout(self, on_cbw_error): self.vm.cmd('object-add', { 'qom-type': 'throttle-group', 'id': 'group0', 'limits': {'bps-write': 300 * 1024} }) self.vm.cmd('blockdev-add', { 'node-name': 'cbw', 'driver': 'copy-before-write', 'on-cbw-error': on_cbw_error, 'cbw-timeout': 1, 'file': { 'driver': iotests.imgfmt, 'file': { 'driver': 'file', 'filename': source_img, } }, 'target': { 'driver': 'throttle', 'throttle-group': 'group0', 'file': { 'driver': 'qcow2', 'file': { 'driver': 'file', 'filename': temp_img } } } }) self.vm.cmd('blockdev-add', { 'node-name': 'access', 'driver': 'snapshot-access', 'file': 'cbw' }) result = self.vm.qmp('human-monitor-command', command_line='qemu-io cbw "write 0 512K"') self.assert_qmp(result, 'return', '') # We need second write to trigger throttling result = self.vm.qmp('human-monitor-command', command_line='qemu-io cbw "write 512K 512K"') self.assert_qmp(result, 'return', '') result = self.vm.qmp('human-monitor-command', command_line='qemu-io access "read 0 1M"') self.assert_qmp(result, 'return', '') self.vm.shutdown() log = self.vm.get_log() log = re.sub(r'^\[I \d+\.\d+\] OPENED\n', '', log) log = re.sub(r'\[I \+\d+\.\d+\] CLOSED\n?$', '', log) log = iotests.filter_qemu_io(log) return log def test_timeout_break_guest(self): log = self.do_cbw_timeout('break-guest-write') # macOS and FreeBSD tend to represent ETIMEDOUT as # "Operation timed out", when Linux prefer # "Connection timed out" log = log.replace('Operation timed out', 'Connection timed out') self.assertEqual(log, """\ wrote 524288/524288 bytes at offset 0 512 KiB, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec) write failed: Connection timed out read 1048576/1048576 bytes at offset 0 1 MiB, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec) """) def test_timeout_break_snapshot(self): log = self.do_cbw_timeout('break-snapshot') self.assertEqual(log, """\ wrote 524288/524288 bytes at offset 0 512 KiB, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec) wrote 524288/524288 bytes at offset 524288 512 KiB, X ops; XX:XX:XX.X (XXX YYY/sec and XXX ops/sec) read failed: Permission denied """) if __name__ == '__main__': iotests.main(supported_fmts=['qcow2'], supported_protocols=['file'], required_fmts=['copy-before-write'])