some improvement

slapos/qemuqmpclient/__init__.py

@@ -31,6 +31,8 @@ import os
 import pprint
 import socket
 import time
+import errno
+import psutil
 from operator import itemgetter
 
 def parseArgument():
@@ -59,6 +61,37 @@ def parseArgument():
   parser.add_argument('remainding_argument_list', nargs=argparse.REMAINDER)
   return  parser.parse_args()
 
+def getInitialQemuResourceDict(pid_file):
+  """
+    Return CPU ad RAM initial values used to start Qemu process
+  """
+  if not os.path.exists(pid_file):
+    return None
+
+  with open(pid_file) as f:
+    pid = int(f.read())
+
+  try:
+    process = psutil.Process(pid)
+  except psutil.NoSuchProcess:
+    print 'No process with pid %s' % pid
+    return None
+  resource_dict = {'cpu': None, 'ram': None}
+  cmd_list = process.cmdline()
+  cpu_index = cmd_list.index('-smp')
+  ram_index = cmd_list.index('-m')
+  if cpu_index >= 0:
+    resource_dict['cpu'] = cmd_list[cpu_index + 1].split(',')[0]
+  if ram_index >= 0:
+    resource_dict['ram'] = cmd_list[ram_index + 1].split(',')[0]
+
+  return resource_dict
+
+class QmpCommandError(Exception):
+  pass
+
+class QmpDeviceRemoveError(Exception):
+  pass
 
 class QemuQMPWrapper(object):
   """
@@ -67,8 +100,12 @@ class QemuQMPWrapper(object):
   QMP API definition.
   """
   def __init__(self, unix_socket_location, auto_connect=True):
+    self._event_list = []
+    self.socket = None
+    self.sockf = None
     if auto_connect:
       self.socket = self.connectToQemu(unix_socket_location)
+      self.sockf = self.socket.makefile()
       self.capabilities()
 
   @staticmethod
@@ -94,21 +131,26 @@ class QemuQMPWrapper(object):
 
     return so
 
-
-  def _send(self, message, check_result=False):
-    self.socket.send(json.dumps(message))
-    data = self.socket.recv(65535)
-    try:
-      result = json.loads(data)
-      if check_result and result.get('return', None) != {} and 'error' in data:
-        raise Exception('ERROR: %s' % data)
-      return result
-    except ValueError:
-      # if error the raise
-      if "error" in data:
-        raise Exception('ERROR: %s' % data)
-      else:
-        print 'Wrong data: %s' % data
+  def _readResponse(self, only_event=False):
+    response = None
+    while True:
+      data = self.sockf.readline()
+      if not data:
+        return
+      response = json.loads(data)
+      if 'error' in response:
+        raise QmpCommandError(response["error"]["desc"])
+      if 'event' in response:
+        self._event_list.append(response)
+        print response
+        if not only_event:
+          continue
+
+      return response
+
+  def _send(self, message):
+    self.socket.sendall(json.dumps(message))
+    return self._readResponse()
 
   def _sendRetry(self, message):
     """
@@ -143,6 +185,43 @@ class QemuQMPWrapper(object):
     print 'Asking for capabilities...'
     self._send({'execute': 'qmp_capabilities'})
 
+  def getEventList(self, timeout=0, cleanup=False):
+    """
+    Get a list of available QMP events.
+    """
+
+    if self.socket is None:
+      return []
+    if cleanup:
+      self.cleanupEventList()
+    if not self._event_list and timeout > 0:
+      self.socket.settimeout(timeout)
+      try:
+        # Read then wait a bit
+        self._readResponse(only_event=True)
+      except socket.timeout:
+        pass
+      finally:
+        self.socket.settimeout(None)
+    else:
+      self.socket.setblocking(0)
+      try:
+        self._readResponse(only_event=True)
+      except socket.error, err:
+        if err[0] == errno.EAGAIN:
+          # No data available
+          pass
+      finally:
+        self.socket.setblocking(1)
+
+    return self._event_list
+
+  def cleanupEventList(self):
+    """
+    Clear current list of pending events.
+    """
+    self.__events = []
+
   def setVNCPassword(self, password):
     # Set VNC password
     print 'Setting VNC password...'
@@ -267,29 +346,67 @@ class QemuQMPWrapper(object):
 
     return mem_info_dict
 
-  def _removeDevice(self, dev_id, command_dict):
-    max_retry = 3
+  def _removeDevice(self, dev_id, command_dict, auto_reboot=False):
+    max_retry = 5
     result = None
-    while max_retry > 0 and result is None:
-      result = self._send(command_dict)
+    resend = True
+    stop_retry = False
+    while max_retry > 0:
       max_retry -= 1
-      if (not result or result.get('return', None) != {}) and \
-          max_retry > 0:
+      try:
+        if resend:
+          result = self._send(command_dict)
+      except QmpCommandError, e:
+        print "ERROR: ", str(e)
         print "%s\nRetry remove %r in few seconds..." % (result, dev_id)
-        time.sleep(3)
-        result = None
+        resend = True
+      else:
+        for event in self.getEventList(timeout=2, cleanup=True):
+          if 'ACPI_DEVICE_OST' == event['event']:
+            # request was received
+            resend = False
+          if 'DEVICE_DELETED' == event['event']:
+            # device was deleted
+            stop_retry = True
+            break
+        if stop_retry:
+          break
+        elif result is None and max_retry > 0:
+          print "Retry remove %r in few seconds..." % dev_id
+      time.sleep(2)
 
     if result is not None:
-      if result.get('return', None) != {}:
-        if result.get('error') and \
-            result['error'].get('class', '') == 'DeviceNotFound':
-          print 'Device %s was removed.' % dev_id
-        else:
-          raise ValueError("Error: Could not remove device %s... %s" % (
-                 dev_id, result))
-    else:
+      if result.get('return', None) == {} or ('error' in result and \
+          result['error'].get('class', '') == 'DeviceNotFound'):
+        print 'Device %s was removed.' % dev_id
+        return
+
+    # device was not remove after retries
+    if not auto_reboot:
       raise ValueError("Cannot remove device %s" % dev_id)
 
+    # try soft reboot of the VM
+    self.powerdown()
+    system_exited = False
+    # wait for ~10 seconds if the system exit, else quit Qemu
+    for i in range(0, 5):
+      for event in self.getEventList(timeout=2, cleanup=True):
+        if event['event'] == 'SHUTDOWN':
+          # qemu is about to exit
+          system_exited = True
+      if system_exited:
+        break
+      else:
+        time.sleep(2)
+
+    if not system_exited:
+      # hard reset the VM
+      print "Trying hard shutdown of the VM..."
+      self._send({"execute": "quit"})
+
+    raise QmpDeviceRemoveError("Stopped Qemu in order to remove the device %r" % dev_id)
+
+
   def _updateCPU(self, amount, cpu_model):
     """
       Add or remove CPU according current value
@@ -300,8 +417,8 @@ class QemuQMPWrapper(object):
     used_socket_id_list = []
     unremovable_cpu = 0
     cpu_hotplugable_list = self._send({
-        'execute': 'query-hotpluggable-cpus'
-      })['return']
+      'execute': 'query-hotpluggable-cpus'
+    })['return']
     cpu_hotplugable_list.reverse()
     for cpu in cpu_hotplugable_list:
       if cpu.get('qom-path', '') == '':
@@ -350,7 +467,7 @@ class QemuQMPWrapper(object):
         self._send({
             'execute': 'device_add',
             'arguments': empty_socket_list[i]
-        }, check_result=True)
+        })
 
     # check that hotplugged memery amount is consistent
     cpu_info = self.getCPUInfo()
@@ -360,19 +477,19 @@ class QemuQMPWrapper(object):
         " current CPU amount is %s" % (hotplug_amount, final_cpu_count))
     print "Done."
 
-  def _removeMemory(self, id_dict):
+  def _removeMemory(self, id_dict, auto_reboot=False):
     print "Trying to remove devices %s, %s..." % (id_dict['id'], id_dict['memdev'])
     self._removeDevice(id_dict['id'] ,{
       'execute': 'device_del',
       'arguments': {'id': id_dict['id']}
-    })
+    }, auto_reboot=auto_reboot)
     # when dimm is removed, remove memdev object
     self._removeDevice(id_dict['memdev'], {
       'execute': 'object-del',
       'arguments': {
         'id': id_dict['memdev']
       }
-    })
+    }, auto_reboot=auto_reboot)
 
   def _updateMemory(self, mem_size, slot_size, slot_amount, allow_reboot=False):
     """
@@ -417,16 +534,9 @@ class QemuQMPWrapper(object):
         'arguments': {
           'id': memdev
         }
-      })
+      }, auto_reboot=allow_reboot)
 
     num_slot_used = len(memory_id_list)
-    if num_slot_used > 0 and slot_size != memory_id_list[0]['size']:
-      # XXX - we won't change the defined size of RAM on slots on live,
-      # restart qemu will allow to change the value
-      if allow_reboot:
-        self.powerdown()
-        raise ValueError("The Size of RAM Slot changed. Rebooting...")
-
     if (mem_size % slot_size) != 0:
       raise ValueError("Memory size %r is not a multiple of %r" % (mem_size,
         slot_size))
@@ -442,22 +552,28 @@ class QemuQMPWrapper(object):
       raise ValueError("Memory size is not valid: %s" % option_dict)
     elif current_size > mem_size:
       # Request to remove memory
-      if allow_reboot:
-        # reboot so new memory size will be configured safely
-        self.powerdown()
-        raise ValueError("Rebooting because memory size was reduced...")
-
-      slot_remove = (current_size - mem_size) / slot_size
-      print "Removing %s memory slots of %s MB..." % (slot_remove, slot_size)
-
-      for i in range(num_slot_used, (num_slot_used - slot_remove), -1):
-        # remove all slot that won't be used
-        self._removeMemory(memory_id_list[i - 1])
+      to_remove_size = current_size - mem_size
+      print "Removing %s MB of memory..." % to_remove_size
+
+      for i in range(num_slot_used, 0, -1):
+        # remove all slots that won't be used
+        index = i - 1
+        to_remove_size -= memory_id_list[index]['size']
+        if to_remove_size >= 0:
+          self._removeMemory(memory_id_list[index], auto_reboot=allow_reboot)
+          if to_remove_size == 0:
+            break
+        else:
+          raise ValueError("Cannot remove the requested size of memory. " \
+            "Remaining to remove: %s, RAM slot size: %s" % (
+              to_remove_size + memory_id_list[index]['size'],
+              memory_id_list[index]['size'])
+            )
     elif current_size < mem_size:
       # ask for increase memory
       slot_add = (mem_size - current_size) / slot_size
 
-      print "Adding %s memory slots of %s MB..." % (slot_add, slot_size)
+      print "Adding %s memory slot(s) of %s MB..." % (slot_add, slot_size)
       for i in range(0, slot_add):
         index = num_slot_used + i + 1
         self._send({
@@ -475,7 +591,7 @@ class QemuQMPWrapper(object):
               'id': 'dimm%s' % index,
               'memdev': 'mem%s' % index
             }
-          }, check_result=True)
+          })
 
     # check that hotplugged memery amount is consistent
     mem_info = self.getMemoryInfo()

slapos/test/test_qemuqmpclient.py

@@ -31,7 +31,7 @@ import os
 import tempfile
 import shutil
 
-from slapos.qemuqmpclient import QemuQMPWrapper
+from slapos.qemuqmpclient import QemuQMPWrapper, QmpDeviceRemoveError
 
 class TestQemuQMPWrapper(unittest.TestCase):
 
@@ -43,6 +43,8 @@ class TestQemuQMPWrapper(unittest.TestCase):
     self.hotplugged_memory_amount = 0
     # slot of 1G
     self.memory_slot_size = 1024
+    self.event_list = []
+    self.fail = False
 
   def tearDown(self):
     if os.path.exists(self.base_dir):
@@ -73,8 +75,16 @@ class TestQemuQMPWrapper(unittest.TestCase):
         self.setChange('dimm', -1 * self.memory_slot_size)
       if message['arguments']['id'].startswith('cpu'):
         self.setChange('cpu', -1)
+    if self.fail:
+      return {"error": {"class": "CommandFailed", "message": ""}}
     return {"return": {}}
 
+  def fake_getEventList(self, timeout=0, cleanup=False):
+    if self.event_list:
+      return self.event_list
+    else:
+      return []
+
   def returnQueryResult(self, message):
     if message['execute'] == 'query-hotpluggable-cpus':
       # return 4 hotpluggable cpu slots
@@ -263,6 +273,8 @@ class TestQemuQMPWrapper(unittest.TestCase):
     self.free_cpu_slot_amount = 2
     qmpwrapper = QemuQMPWrapper(self.socket_file, auto_connect=False)
     qmpwrapper._send = self.fake_send
+    qmpwrapper.getEventList = self.fake_getEventList
+    self.event_list = [{"event": "DEVICE_DELETED"}]
     # add 2 more cpu
     cpu_option = {
       'device': 'cpu',
@@ -397,6 +409,8 @@ class TestQemuQMPWrapper(unittest.TestCase):
   def test_updateDevice_memory_delete(self):
     qmpwrapper = QemuQMPWrapper(self.socket_file, auto_connect=False)
     qmpwrapper._send = self.fake_send
+    qmpwrapper.getEventList = self.fake_getEventList
+    self.event_list = [{"event": "DEVICE_DELETED"}]
     self.hotplugged_memory_amount = 3072
     # slot of 1G
     self.memory_slot_size = 1024
@@ -436,6 +450,8 @@ class TestQemuQMPWrapper(unittest.TestCase):
   def test_updateDevice_memory_delete_all(self):
     qmpwrapper = QemuQMPWrapper(self.socket_file, auto_connect=False)
     qmpwrapper._send = self.fake_send
+    qmpwrapper.getEventList = self.fake_getEventList
+    self.event_list = [{"event": "DEVICE_DELETED"}]
     self.hotplugged_memory_amount = 3072
     # slot of 1G
     self.memory_slot_size = 1024
@@ -505,6 +521,8 @@ class TestQemuQMPWrapper(unittest.TestCase):
   def test_updateDevice_memory_will_reboot(self):
     qmpwrapper = QemuQMPWrapper(self.socket_file, auto_connect=False)
     qmpwrapper._send = self.fake_send
+    qmpwrapper.getEventList = self.fake_getEventList
+    self.fail = True
     self.hotplugged_memory_amount = 3072
     # slot of 1G
     self.memory_slot_size = 1024
@@ -516,15 +534,36 @@ class TestQemuQMPWrapper(unittest.TestCase):
       'slot': self.memory_slot_size,
       'canreboot': True
     }
-    with self.assertRaises(ValueError):
+    with self.assertRaises(QmpDeviceRemoveError):
       qmpwrapper.updateDevice(cpu_option)
     expected_result = [
       {'execute': 'query-memory-devices'},
       {'execute': 'query-memdev'},
-      {'execute': 'system_powerdown'}
+      {
+        'execute': 'device_del',
+        'arguments': {'id': u'dimm3'}
+      },
+      {
+        'execute': 'device_del',
+        'arguments': {'id': u'dimm3'}
+      },
+      {
+        'execute': 'device_del',
+        'arguments': {'id': u'dimm3'}
+      },
+      {
+        'execute': 'device_del',
+        'arguments': {'id': u'dimm3'}
+      },
+      {
+        'execute': 'device_del',
+        'arguments': {'id': u'dimm3'}
+      },
+      {'execute': 'system_powerdown'},
+      {'execute': 'quit'}
     ]
 
-    self.assertEquals(len(self.call_stack_list), 3)
+    self.assertEquals(len(self.call_stack_list), 9)
     self.assertEquals(self.call_stack_list, expected_result)
 
 if __name__ == '__main__':