tests/migration/guestperf/engine.py

   1 #
   2 # Migration test main engine
   3 #
   4 # Copyright (c) 2016 Red Hat, Inc.
   5 #
   6 # This library is free software; you can redistribute it and/or
   7 # modify it under the terms of the GNU Lesser General Public
   8 # License as published by the Free Software Foundation; either
   9 # version 2.1 of the License, or (at your option) any later version.
  10 #
  11 # This library is distributed in the hope that it will be useful,
  12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # Lesser General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU Lesser General Public
  17 # License along with this library; if not, see <http://www.gnu.org/licenses/>.
  18 #
  19
  20
  21 import os
  22 import re
  23 import sys
  24 import time
  25
  26 from guestperf.progress import Progress, ProgressStats
  27 from guestperf.report import Report
  28 from guestperf.timings import TimingRecord, Timings
  29
  30 sys.path.append(os.path.join(os.path.dirname(__file__),
  31                              '..', '..', '..', 'python'))
  32 from qemu.machine import QEMUMachine
  33
  34
  35 class Engine(object):
  36
  37     def __init__(self, binary, dst_host, kernel, initrd, transport="tcp",
  38                  sleep=15, verbose=False, debug=False):
  39
  40         self._binary = binary # Path to QEMU binary
  41         self._dst_host = dst_host # Hostname of target host
  42         self._kernel = kernel # Path to kernel image
  43         self._initrd = initrd # Path to stress initrd
  44         self._transport = transport # 'unix' or 'tcp' or 'rdma'
  45         self._sleep = sleep
  46         self._verbose = verbose
  47         self._debug = debug
  48
  49         if debug:
  50             self._verbose = debug
  51
  52     def _vcpu_timing(self, pid, tid_list):
  53         records = []
  54         now = time.time()
  55
  56         jiffies_per_sec = os.sysconf(os.sysconf_names['SC_CLK_TCK'])
  57         for tid in tid_list:
  58             statfile = "/proc/%d/task/%d/stat" % (pid, tid)
  59             with open(statfile, "r") as fh:
  60                 stat = fh.readline()
  61                 fields = stat.split(" ")
  62                 stime = int(fields[13])
  63                 utime = int(fields[14])
  64                 records.append(TimingRecord(tid, now, 1000 * (stime + utime) / jiffies_per_sec))
  65         return records
  66
  67     def _cpu_timing(self, pid):
  68         records = []
  69         now = time.time()
  70
  71         jiffies_per_sec = os.sysconf(os.sysconf_names['SC_CLK_TCK'])
  72         statfile = "/proc/%d/stat" % pid
  73         with open(statfile, "r") as fh:
  74             stat = fh.readline()
  75             fields = stat.split(" ")
  76             stime = int(fields[13])
  77             utime = int(fields[14])
  78             return TimingRecord(pid, now, 1000 * (stime + utime) / jiffies_per_sec)
  79
  80     def _migrate_progress(self, vm):
  81         info = vm.command("query-migrate")
  82
  83         if "ram" not in info:
  84             info["ram"] = {}
  85
  86         return Progress(
  87             info.get("status", "active"),
  88             ProgressStats(
  89                 info["ram"].get("transferred", 0),
  90                 info["ram"].get("remaining", 0),
  91                 info["ram"].get("total", 0),
  92                 info["ram"].get("duplicate", 0),
  93                 info["ram"].get("skipped", 0),
  94                 info["ram"].get("normal", 0),
  95                 info["ram"].get("normal-bytes", 0),
  96                 info["ram"].get("dirty-pages-rate", 0),
  97                 info["ram"].get("mbps", 0),
  98                 info["ram"].get("dirty-sync-count", 0)
  99             ),
 100             time.time(),
 101             info.get("total-time", 0),
 102             info.get("downtime", 0),
 103             info.get("expected-downtime", 0),
 104             info.get("setup-time", 0),
 105             info.get("x-cpu-throttle-percentage", 0),
 106         )
 107
 108     def _migrate(self, hardware, scenario, src, dst, connect_uri):
 109         src_qemu_time = []
 110         src_vcpu_time = []
 111         src_pid = src.get_pid()
 112
 113         vcpus = src.command("query-cpus-fast")
 114         src_threads = []
 115         for vcpu in vcpus:
 116             src_threads.append(vcpu["thread_id"])
 117
 118         # XXX how to get dst timings on remote host ?
 119
 120         if self._verbose:
 121             print("Sleeping %d seconds for initial guest workload run" % self._sleep)
 122         sleep_secs = self._sleep
 123         while sleep_secs > 1:
 124             src_qemu_time.append(self._cpu_timing(src_pid))
 125             src_vcpu_time.extend(self._vcpu_timing(src_pid, src_threads))
 126             time.sleep(1)
 127             sleep_secs -= 1
 128
 129         if self._verbose:
 130             print("Starting migration")
 131         if scenario._auto_converge:
 132             resp = src.command("migrate-set-capabilities",
 133                                capabilities = [
 134                                    { "capability": "auto-converge",
 135                                      "state": True }
 136                                ])
 137             resp = src.command("migrate-set-parameters",
 138                                x_cpu_throttle_increment=scenario._auto_converge_step)
 139
 140         if scenario._post_copy:
 141             resp = src.command("migrate-set-capabilities",
 142                                capabilities = [
 143                                    { "capability": "postcopy-ram",
 144                                      "state": True }
 145                                ])
 146             resp = dst.command("migrate-set-capabilities",
 147                                capabilities = [
 148                                    { "capability": "postcopy-ram",
 149                                      "state": True }
 150                                ])
 151
 152         resp = src.command("migrate-set-parameters",
 153                            max_bandwidth=scenario._bandwidth * 1024 * 1024)
 154
 155         resp = src.command("migrate-set-parameters",
 156                            downtime_limit=scenario._downtime / 1024.0)
 157
 158         if scenario._compression_mt:
 159             resp = src.command("migrate-set-capabilities",
 160                                capabilities = [
 161                                    { "capability": "compress",
 162                                      "state": True }
 163                                ])
 164             resp = src.command("migrate-set-parameters",
 165                                compress_threads=scenario._compression_mt_threads)
 166             resp = dst.command("migrate-set-capabilities",
 167                                capabilities = [
 168                                    { "capability": "compress",
 169                                      "state": True }
 170                                ])
 171             resp = dst.command("migrate-set-parameters",
 172                                decompress_threads=scenario._compression_mt_threads)
 173
 174         if scenario._compression_xbzrle:
 175             resp = src.command("migrate-set-capabilities",
 176                                capabilities = [
 177                                    { "capability": "xbzrle",
 178                                      "state": True }
 179                                ])
 180             resp = dst.command("migrate-set-capabilities",
 181                                capabilities = [
 182                                    { "capability": "xbzrle",
 183                                      "state": True }
 184                                ])
 185             resp = src.command("migrate-set-parameters",
 186                                xbzrle_cache_size=(
 187                                    hardware._mem *
 188                                    1024 * 1024 * 1024 / 100 *
 189                                    scenario._compression_xbzrle_cache))
 190
 191         resp = src.command("migrate", uri=connect_uri)
 192
 193         post_copy = False
 194         paused = False
 195
 196         progress_history = []
 197
 198         start = time.time()
 199         loop = 0
 200         while True:
 201             loop = loop + 1
 202             time.sleep(0.05)
 203
 204             progress = self._migrate_progress(src)
 205             if (loop % 20) == 0:
 206                 src_qemu_time.append(self._cpu_timing(src_pid))
 207                 src_vcpu_time.extend(self._vcpu_timing(src_pid, src_threads))
 208
 209             if (len(progress_history) == 0 or
 210                 (progress_history[-1]._ram._iterations <
 211                  progress._ram._iterations)):
 212                 progress_history.append(progress)
 213
 214             if progress._status in ("completed", "failed", "cancelled"):
 215                 if progress._status == "completed" and paused:
 216                     dst.command("cont")
 217                 if progress_history[-1] != progress:
 218                     progress_history.append(progress)
 219
 220                 if progress._status == "completed":
 221                     if self._verbose:
 222                         print("Sleeping %d seconds for final guest workload run" % self._sleep)
 223                     sleep_secs = self._sleep
 224                     while sleep_secs > 1:
 225                         time.sleep(1)
 226                         src_qemu_time.append(self._cpu_timing(src_pid))
 227                         src_vcpu_time.extend(self._vcpu_timing(src_pid, src_threads))
 228                         sleep_secs -= 1
 229
 230                 return [progress_history, src_qemu_time, src_vcpu_time]
 231
 232             if self._verbose and (loop % 20) == 0:
 233                 print("Iter %d: remain %5dMB of %5dMB (total %5dMB @ %5dMb/sec)" % (
 234                     progress._ram._iterations,
 235                     progress._ram._remaining_bytes / (1024 * 1024),
 236                     progress._ram._total_bytes / (1024 * 1024),
 237                     progress._ram._transferred_bytes / (1024 * 1024),
 238                     progress._ram._transfer_rate_mbs,
 239                 ))
 240
 241             if progress._ram._iterations > scenario._max_iters:
 242                 if self._verbose:
 243                     print("No completion after %d iterations over RAM" % scenario._max_iters)
 244                 src.command("migrate_cancel")
 245                 continue
 246
 247             if time.time() > (start + scenario._max_time):
 248                 if self._verbose:
 249                     print("No completion after %d seconds" % scenario._max_time)
 250                 src.command("migrate_cancel")
 251                 continue
 252
 253             if (scenario._post_copy and
 254                 progress._ram._iterations >= scenario._post_copy_iters and
 255                 not post_copy):
 256                 if self._verbose:
 257                     print("Switching to post-copy after %d iterations" % scenario._post_copy_iters)
 258                 resp = src.command("migrate-start-postcopy")
 259                 post_copy = True
 260
 261             if (scenario._pause and
 262                 progress._ram._iterations >= scenario._pause_iters and
 263                 not paused):
 264                 if self._verbose:
 265                     print("Pausing VM after %d iterations" % scenario._pause_iters)
 266                 resp = src.command("stop")
 267                 paused = True
 268
 269     def _get_common_args(self, hardware, tunnelled=False):
 270         args = [
 271             "noapic",
 272             "edd=off",
 273             "printk.time=1",
 274             "noreplace-smp",
 275             "cgroup_disable=memory",
 276             "pci=noearly",
 277             "console=ttyS0",
 278         ]
 279         if self._debug:
 280             args.append("debug")
 281         else:
 282             args.append("quiet")
 283
 284         args.append("ramsize=%s" % hardware._mem)
 285
 286         cmdline = " ".join(args)
 287         if tunnelled:
 288             cmdline = "'" + cmdline + "'"
 289
 290         argv = [
 291             "-accel", "kvm",
 292             "-cpu", "host",
 293             "-kernel", self._kernel,
 294             "-initrd", self._initrd,
 295             "-append", cmdline,
 296             "-chardev", "stdio,id=cdev0",
 297             "-device", "isa-serial,chardev=cdev0",
 298             "-m", str((hardware._mem * 1024) + 512),
 299             "-smp", str(hardware._cpus),
 300         ]
 301
 302         if self._debug:
 303             argv.extend(["-device", "sga"])
 304
 305         if hardware._prealloc_pages:
 306             argv_source += ["-mem-path", "/dev/shm",
 307                             "-mem-prealloc"]
 308         if hardware._locked_pages:
 309             argv_source += ["-overcommit", "mem-lock=on"]
 310         if hardware._huge_pages:
 311             pass
 312
 313         return argv
 314
 315     def _get_src_args(self, hardware):
 316         return self._get_common_args(hardware)
 317
 318     def _get_dst_args(self, hardware, uri):
 319         tunnelled = False
 320         if self._dst_host != "localhost":
 321             tunnelled = True
 322         argv = self._get_common_args(hardware, tunnelled)
 323         return argv + ["-incoming", uri]
 324
 325     @staticmethod
 326     def _get_common_wrapper(cpu_bind, mem_bind):
 327         wrapper = []
 328         if len(cpu_bind) > 0 or len(mem_bind) > 0:
 329             wrapper.append("numactl")
 330             if cpu_bind:
 331                 wrapper.append("--physcpubind=%s" % ",".join(cpu_bind))
 332             if mem_bind:
 333                 wrapper.append("--membind=%s" % ",".join(mem_bind))
 334
 335         return wrapper
 336
 337     def _get_src_wrapper(self, hardware):
 338         return self._get_common_wrapper(hardware._src_cpu_bind, hardware._src_mem_bind)
 339
 340     def _get_dst_wrapper(self, hardware):
 341         wrapper = self._get_common_wrapper(hardware._dst_cpu_bind, hardware._dst_mem_bind)
 342         if self._dst_host != "localhost":
 343             return ["ssh",
 344                     "-R", "9001:localhost:9001",
 345                     self._dst_host] + wrapper
 346         else:
 347             return wrapper
 348
 349     def _get_timings(self, vm):
 350         log = vm.get_log()
 351         if not log:
 352             return []
 353         if self._debug:
 354             print(log)
 355
 356         regex = r"[^\s]+\s\((\d+)\):\sINFO:\s(\d+)ms\scopied\s\d+\sGB\sin\s(\d+)ms"
 357         matcher = re.compile(regex)
 358         records = []
 359         for line in log.split("\n"):
 360             match = matcher.match(line)
 361             if match:
 362                 records.append(TimingRecord(int(match.group(1)),
 363                                             int(match.group(2)) / 1000.0,
 364                                             int(match.group(3))))
 365         return records
 366
 367     def run(self, hardware, scenario, result_dir=os.getcwd()):
 368         abs_result_dir = os.path.join(result_dir, scenario._name)
 369
 370         if self._transport == "tcp":
 371             uri = "tcp:%s:9000" % self._dst_host
 372         elif self._transport == "rdma":
 373             uri = "rdma:%s:9000" % self._dst_host
 374         elif self._transport == "unix":
 375             if self._dst_host != "localhost":
 376                 raise Exception("Running use unix migration transport for non-local host")
 377             uri = "unix:/var/tmp/qemu-migrate-%d.migrate" % os.getpid()
 378             try:
 379                 os.remove(uri[5:])
 380                 os.remove(monaddr)
 381             except:
 382                 pass
 383
 384         if self._dst_host != "localhost":
 385             dstmonaddr = ("localhost", 9001)
 386         else:
 387             dstmonaddr = "/var/tmp/qemu-dst-%d-monitor.sock" % os.getpid()
 388         srcmonaddr = "/var/tmp/qemu-src-%d-monitor.sock" % os.getpid()
 389
 390         src = QEMUMachine(self._binary,
 391                           args=self._get_src_args(hardware),
 392                           wrapper=self._get_src_wrapper(hardware),
 393                           name="qemu-src-%d" % os.getpid(),
 394                           monitor_address=srcmonaddr)
 395
 396         dst = QEMUMachine(self._binary,
 397                           args=self._get_dst_args(hardware, uri),
 398                           wrapper=self._get_dst_wrapper(hardware),
 399                           name="qemu-dst-%d" % os.getpid(),
 400                           monitor_address=dstmonaddr)
 401
 402         try:
 403             src.launch()
 404             dst.launch()
 405
 406             ret = self._migrate(hardware, scenario, src, dst, uri)
 407             progress_history = ret[0]
 408             qemu_timings = ret[1]
 409             vcpu_timings = ret[2]
 410             if uri[0:5] == "unix:":
 411                 os.remove(uri[5:])
 412
 413             if os.path.exists(srcmonaddr):
 414                 os.remove(srcmonaddr)
 415
 416             if self._dst_host == "localhost" and os.path.exists(dstmonaddr):
 417                 os.remove(dstmonaddr)
 418
 419             if self._verbose:
 420                 print("Finished migration")
 421
 422             src.shutdown()
 423             dst.shutdown()
 424
 425             return Report(hardware, scenario, progress_history,
 426                           Timings(self._get_timings(src) + self._get_timings(dst)),
 427                           Timings(qemu_timings),
 428                           Timings(vcpu_timings),
 429                           self._binary, self._dst_host, self._kernel,
 430                           self._initrd, self._transport, self._sleep)
 431         except Exception as e:
 432             if self._debug:
 433                 print("Failed: %s" % str(e))
 434             try:
 435                 src.shutdown()
 436             except:
 437                 pass
 438             try:
 439                 dst.shutdown()
 440             except:
 441                 pass
 442
 443             if self._debug:
 444                 print(src.get_log())
 445                 print(dst.get_log())
 446             raise
 447