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")
 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_speed",
 153                            value=scenario._bandwidth * 1024 * 1024)
 154
 155         resp = src.command("migrate_set_downtime",
 156                            value=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-cache-size",
 186                                value=(hardware._mem * 1024 * 1024 * 1024 / 100 *
 187                                       scenario._compression_xbzrle_cache))
 188
 189         resp = src.command("migrate", uri=connect_uri)
 190
 191         post_copy = False
 192         paused = False
 193
 194         progress_history = []
 195
 196         start = time.time()
 197         loop = 0
 198         while True:
 199             loop = loop + 1
 200             time.sleep(0.05)
 201
 202             progress = self._migrate_progress(src)
 203             if (loop % 20) == 0:
 204                 src_qemu_time.append(self._cpu_timing(src_pid))
 205                 src_vcpu_time.extend(self._vcpu_timing(src_pid, src_threads))
 206
 207             if (len(progress_history) == 0 or
 208                 (progress_history[-1]._ram._iterations <
 209                  progress._ram._iterations)):
 210                 progress_history.append(progress)
 211
 212             if progress._status in ("completed", "failed", "cancelled"):
 213                 if progress._status == "completed" and paused:
 214                     dst.command("cont")
 215                 if progress_history[-1] != progress:
 216                     progress_history.append(progress)
 217
 218                 if progress._status == "completed":
 219                     if self._verbose:
 220                         print("Sleeping %d seconds for final guest workload run" % self._sleep)
 221                     sleep_secs = self._sleep
 222                     while sleep_secs > 1:
 223                         time.sleep(1)
 224                         src_qemu_time.append(self._cpu_timing(src_pid))
 225                         src_vcpu_time.extend(self._vcpu_timing(src_pid, src_threads))
 226                         sleep_secs -= 1
 227
 228                 return [progress_history, src_qemu_time, src_vcpu_time]
 229
 230             if self._verbose and (loop % 20) == 0:
 231                 print("Iter %d: remain %5dMB of %5dMB (total %5dMB @ %5dMb/sec)" % (
 232                     progress._ram._iterations,
 233                     progress._ram._remaining_bytes / (1024 * 1024),
 234                     progress._ram._total_bytes / (1024 * 1024),
 235                     progress._ram._transferred_bytes / (1024 * 1024),
 236                     progress._ram._transfer_rate_mbs,
 237                 ))
 238
 239             if progress._ram._iterations > scenario._max_iters:
 240                 if self._verbose:
 241                     print("No completion after %d iterations over RAM" % scenario._max_iters)
 242                 src.command("migrate_cancel")
 243                 continue
 244
 245             if time.time() > (start + scenario._max_time):
 246                 if self._verbose:
 247                     print("No completion after %d seconds" % scenario._max_time)
 248                 src.command("migrate_cancel")
 249                 continue
 250
 251             if (scenario._post_copy and
 252                 progress._ram._iterations >= scenario._post_copy_iters and
 253                 not post_copy):
 254                 if self._verbose:
 255                     print("Switching to post-copy after %d iterations" % scenario._post_copy_iters)
 256                 resp = src.command("migrate-start-postcopy")
 257                 post_copy = True
 258
 259             if (scenario._pause and
 260                 progress._ram._iterations >= scenario._pause_iters and
 261                 not paused):
 262                 if self._verbose:
 263                     print("Pausing VM after %d iterations" % scenario._pause_iters)
 264                 resp = src.command("stop")
 265                 paused = True
 266
 267     def _get_common_args(self, hardware, tunnelled=False):
 268         args = [
 269             "noapic",
 270             "edd=off",
 271             "printk.time=1",
 272             "noreplace-smp",
 273             "cgroup_disable=memory",
 274             "pci=noearly",
 275             "console=ttyS0",
 276         ]
 277         if self._debug:
 278             args.append("debug")
 279         else:
 280             args.append("quiet")
 281
 282         args.append("ramsize=%s" % hardware._mem)
 283
 284         cmdline = " ".join(args)
 285         if tunnelled:
 286             cmdline = "'" + cmdline + "'"
 287
 288         argv = [
 289             "-accel", "kvm",
 290             "-cpu", "host",
 291             "-kernel", self._kernel,
 292             "-initrd", self._initrd,
 293             "-append", cmdline,
 294             "-chardev", "stdio,id=cdev0",
 295             "-device", "isa-serial,chardev=cdev0",
 296             "-m", str((hardware._mem * 1024) + 512),
 297             "-smp", str(hardware._cpus),
 298         ]
 299
 300         if self._debug:
 301             argv.extend(["-device", "sga"])
 302
 303         if hardware._prealloc_pages:
 304             argv_source += ["-mem-path", "/dev/shm",
 305                             "-mem-prealloc"]
 306         if hardware._locked_pages:
 307             argv_source += ["-overcommit", "mem-lock=on"]
 308         if hardware._huge_pages:
 309             pass
 310
 311         return argv
 312
 313     def _get_src_args(self, hardware):
 314         return self._get_common_args(hardware)
 315
 316     def _get_dst_args(self, hardware, uri):
 317         tunnelled = False
 318         if self._dst_host != "localhost":
 319             tunnelled = True
 320         argv = self._get_common_args(hardware, tunnelled)
 321         return argv + ["-incoming", uri]
 322
 323     @staticmethod
 324     def _get_common_wrapper(cpu_bind, mem_bind):
 325         wrapper = []
 326         if len(cpu_bind) > 0 or len(mem_bind) > 0:
 327             wrapper.append("numactl")
 328             if cpu_bind:
 329                 wrapper.append("--physcpubind=%s" % ",".join(cpu_bind))
 330             if mem_bind:
 331                 wrapper.append("--membind=%s" % ",".join(mem_bind))
 332
 333         return wrapper
 334
 335     def _get_src_wrapper(self, hardware):
 336         return self._get_common_wrapper(hardware._src_cpu_bind, hardware._src_mem_bind)
 337
 338     def _get_dst_wrapper(self, hardware):
 339         wrapper = self._get_common_wrapper(hardware._dst_cpu_bind, hardware._dst_mem_bind)
 340         if self._dst_host != "localhost":
 341             return ["ssh",
 342                     "-R", "9001:localhost:9001",
 343                     self._dst_host] + wrapper
 344         else:
 345             return wrapper
 346
 347     def _get_timings(self, vm):
 348         log = vm.get_log()
 349         if not log:
 350             return []
 351         if self._debug:
 352             print(log)
 353
 354         regex = r"[^\s]+\s\((\d+)\):\sINFO:\s(\d+)ms\scopied\s\d+\sGB\sin\s(\d+)ms"
 355         matcher = re.compile(regex)
 356         records = []
 357         for line in log.split("\n"):
 358             match = matcher.match(line)
 359             if match:
 360                 records.append(TimingRecord(int(match.group(1)),
 361                                             int(match.group(2)) / 1000.0,
 362                                             int(match.group(3))))
 363         return records
 364
 365     def run(self, hardware, scenario, result_dir=os.getcwd()):
 366         abs_result_dir = os.path.join(result_dir, scenario._name)
 367
 368         if self._transport == "tcp":
 369             uri = "tcp:%s:9000" % self._dst_host
 370         elif self._transport == "rdma":
 371             uri = "rdma:%s:9000" % self._dst_host
 372         elif self._transport == "unix":
 373             if self._dst_host != "localhost":
 374                 raise Exception("Running use unix migration transport for non-local host")
 375             uri = "unix:/var/tmp/qemu-migrate-%d.migrate" % os.getpid()
 376             try:
 377                 os.remove(uri[5:])
 378                 os.remove(monaddr)
 379             except:
 380                 pass
 381
 382         if self._dst_host != "localhost":
 383             dstmonaddr = ("localhost", 9001)
 384         else:
 385             dstmonaddr = "/var/tmp/qemu-dst-%d-monitor.sock" % os.getpid()
 386         srcmonaddr = "/var/tmp/qemu-src-%d-monitor.sock" % os.getpid()
 387
 388         src = QEMUMachine(self._binary,
 389                           args=self._get_src_args(hardware),
 390                           wrapper=self._get_src_wrapper(hardware),
 391                           name="qemu-src-%d" % os.getpid(),
 392                           monitor_address=srcmonaddr)
 393
 394         dst = QEMUMachine(self._binary,
 395                           args=self._get_dst_args(hardware, uri),
 396                           wrapper=self._get_dst_wrapper(hardware),
 397                           name="qemu-dst-%d" % os.getpid(),
 398                           monitor_address=dstmonaddr)
 399
 400         try:
 401             src.launch()
 402             dst.launch()
 403
 404             ret = self._migrate(hardware, scenario, src, dst, uri)
 405             progress_history = ret[0]
 406             qemu_timings = ret[1]
 407             vcpu_timings = ret[2]
 408             if uri[0:5] == "unix:":
 409                 os.remove(uri[5:])
 410             if self._verbose:
 411                 print("Finished migration")
 412
 413             src.shutdown()
 414             dst.shutdown()
 415
 416             return Report(hardware, scenario, progress_history,
 417                           Timings(self._get_timings(src) + self._get_timings(dst)),
 418                           Timings(qemu_timings),
 419                           Timings(vcpu_timings),
 420                           self._binary, self._dst_host, self._kernel,
 421                           self._initrd, self._transport, self._sleep)
 422         except Exception as e:
 423             if self._debug:
 424                 print("Failed: %s" % str(e))
 425             try:
 426                 src.shutdown()
 427             except:
 428                 pass
 429             try:
 430                 dst.shutdown()
 431             except:
 432                 pass
 433
 434             if self._debug:
 435                 print(src.get_log())
 436                 print(dst.get_log())
 437             raise
 438