[NEW PATCH] A new JSON-based protocol with the guest's agent. (via gerrit-bot)

Gal Hammer ghammer at redhat.com
Sun Aug 28 10:03:08 UTC 2011


New patch submitted by Gal Hammer (ghammer at redhat.com)

You can review this change at: http://gerrit.usersys.redhat.com/807

commit ceb0e7060b23593166fd084792d1a00159b68fe9
Author: Gal Hammer <ghammer at redhat.com>
Date:   Wed Aug 10 13:26:31 2011 +0300

    A new JSON-based protocol with the guest's agent.
    
    Change-Id: Ie6096b326cc58879c8f63b72e32586657fffeca0

diff --git a/vdsm/clientIF.py b/vdsm/clientIF.py
index c5169a4..27eb973 100644
--- a/vdsm/clientIF.py
+++ b/vdsm/clientIF.py
@@ -269,7 +269,6 @@ class clientIF:
                 (self.desktopLogin, 'desktopLogin'),
                 (self.desktopLogoff, 'desktopLogoff'),
                 (self.desktopLock, 'desktopLock'),
-                (self.sendHcCmdToDesktop, 'sendHcCmdToDesktop'),
                 (self.hibernate, 'hibernate'),
                 (self.monitorCommand, 'monitorCommand'),
                 (self.addNetwork, 'addNetwork'),
@@ -815,20 +814,6 @@ class clientIF:
         else:
             return errCode['nonresp']
 
-    def sendHcCmdToDesktop (self, vmId, message):
-        """
-        Send a command to the guest agent (depricated).
-        """
-        try:
-            vm = self.vmContainer[vmId]
-        except KeyError:
-            return errCode['noVM']
-        vm.guestAgent.sendHcCmdToDesktop(message)
-        if vm.guestAgent.isResponsive():
-            return {'status': doneCode}
-        else:
-            return errCode['nonresp']
-
     # take a rough estimate on how much free mem is available for new vm
     # memTotal = memFree + memCached + mem_used_by_non_qemu + resident  .
     # simply returning (memFree + memCached) is not good enough, as the
diff --git a/vdsm/guestIF.py b/vdsm/guestIF.py
index a648dec..a176e3e 100644
--- a/vdsm/guestIF.py
+++ b/vdsm/guestIF.py
@@ -1,7 +1,7 @@
 import traceback, logging, threading
 import time
 import socket
-import struct
+import json
 from config import config
 import utils
 import constants
@@ -23,36 +23,6 @@ def _filterXmlChars(u):
 
     return ''.join([maskRestricted(c) for c in u])
 
-
-class guestMType:
-    powerup=1
-    powerdown=2
-    heartbeat=3
-    guestName=4
-    guestOs=5
-    IPAddresses=6
-    lastSessionMessage=7
-    userInfo=8
-    newApp=9
-    flushApps=10
-    sessionLock=12
-    sessionUnlock=13
-    sessionLogoff=14
-    sessionLogon=15
-    agentCmd = 16 # obsolete
-    agentUninstalled = 17
-    sessionStartup = 18
-    sessionShutdown = 19
-
-class protocolMtype:
-    register, unregister, forward = range(1, 4)
-    error = 0x80000001
-
-
-headerLength = 3
-wordSize = 4
-headerLengthBytes = headerLength * wordSize
-
 class GuestAgent (threading.Thread):
     def __init__(self, socketName, log, user='Unknown', ips='', connect=True):
         self.log = log
@@ -61,15 +31,10 @@ class GuestAgent (threading.Thread):
         self._stopped = True
         self.guestStatus = None
         self.guestInfo = {'username': user, 'memUsage': 0, 'guestIPs': ips,
-                          'session': 'Unknown', 'appsList': []}
+                          'session': 'Unknown', 'appsList': [], 'disksUsage': [],
+                          'netIfaces': []}
         self._agentTimestamp = 0
 
-        # A temporary storage to hold the guest's application list during an
-        # update from the guest. Will be obselete if a new virtual channel
-        # (that can handle large messages) will be implemented and used.
-        self._tempAppsList = []
-        self._firstAppsList = True
-
         threading.Thread.__init__(self)
         if connect:
             self.start()
@@ -88,109 +53,72 @@ class GuestAgent (threading.Thread):
                 time.sleep(1)
         return False
 
-    #The protocol envelope:
-    #DWORD(32Bit) - ChannelID
-    #DWORD(32Bit) - Action/Cmd
-    #   1 - Register (to receive ChannelID events)
-    #   2 - UnRegister (from ChannelID events)
-    #   3 - Forward Message to channel
-    #   0x80000001 - Error UnknownChannel
-    #DWORD(32Bit) MessageLength(In Bytes)
-    #message payload
-    # Simple protocol at this point. Just a single byte
-    # with enumeration of the event
-    #   1 - PowerUp
-    #   2 - PowerDown
-    #   3 - HeartBeat
+    def _forward(self, cmd, args = {}):
+        args['__name__'] = cmd
+        message = (json.dumps(args) + '\n').encode('utf8')
+        self._sock.send(message)
+        self.log.log(logging.TRACE, 'sent %s', message)
 
-    CHANNEL = 1
-    def _forward(self, s):
-        lens = len(s)
-        t = struct.pack('>III%ds' % lens, self.CHANNEL,
-                    protocolMtype.forward, headerLengthBytes + lens, s)
-        self._sock.send(t)
-        self.log.log(logging.TRACE, 'sent %s', repr(t))
-
-    def _parseHeader(self, header):
-        channel, messageType, length = struct.unpack('>III', header)
-        if channel != self.CHANNEL:
-            self.log.error("Illegal channel id %d" % (channel))
-            return 0
-        if messageType != protocolMtype.forward:
-            self.log.error("Unexpected message type " + str((channel, messageType, length)))
-            return 0
-        return length - headerLengthBytes
-
-    def _parseBody(self, body):
-        guestMessage, = struct.unpack('>I', body[:4])
-        body = body[4:]
-        self.log.log(logging.TRACE, 'guest message %s body %s',
-                     guestMessage, body)
+    def _handleMessage(self, message, args):
+        self.log.log(logging.TRACE, "Guest's message %s: %s", message, args)
         if self.guestStatus == None:
             self.guestStatus = 'Running'
-        if guestMessage == guestMType.heartbeat:
-            self.guestStatus = 'Running'
-            self.guestInfo['memUsage'] = int(body.strip())
-        elif guestMessage == guestMType.powerup:
+        if message == 'heartbeat':
             self.guestStatus = 'Running'
-        elif guestMessage == guestMType.powerdown:
-            self.guestStatus = 'Powered down'
-            if self.guestInfo['username'] not in 'None': #in case powerdown event hit before logoff
-                self.guestInfo['lastUser'] = '' + self.guestInfo['username']
-                self.guestInfo['username'] = 'None'
-                self.guestInfo['lastLogout'] = time.time()
-        elif guestMessage == guestMType.guestName:
-            self.guestInfo['guestName'] = _filterXmlChars(unicode(body, 'utf8'))
-        elif guestMessage == guestMType.guestOs:
-            self.guestInfo['guestOs'] = _filterXmlChars(unicode(body, 'utf8'))
-        elif guestMessage == guestMType.IPAddresses:
-            guestIPs = body.strip().split()
-            self.log.debug(str(guestIPs))
-            self.guestInfo['guestIPs'] = _filterXmlChars(' '.join(guestIPs))
-        elif guestMessage == guestMType.lastSessionMessage:
-            lastSessionMessage = body
-            self.log.debug(lastSessionMessage)
-            if 'Logoff' in lastSessionMessage:
-                self.guestInfo['lastUser'] = '' + self.guestInfo['username']
-                self.guestInfo['username'] = 'None'
-                self.guestInfo['lastLogout'] = time.time()
-        elif guestMessage == guestMType.flushApps:
-            if self._tempAppsList == self.guestInfo['appsList'] == []:
-                self._firstAppsList = True
-            else:
-                self._firstAppsList = False
-            self.guestInfo['appsList'] = self._tempAppsList
-            self._tempAppsList = []
-        elif guestMessage == guestMType.newApp:
-            app = _filterXmlChars(unicode(body, 'utf8').strip())
-            if app not in self._tempAppsList:
-                self._tempAppsList.append(app)
-            if self._firstAppsList:
-                self.guestInfo['appsList'] = self._tempAppsList
-        elif guestMessage == guestMType.userInfo:
-            self.log.debug(body)
-            currentUser = _filterXmlChars(unicode(body, 'utf8'))
+            self.guestInfo['memUsage'] = int(args['free-ram'])
+        elif message == 'host-name':
+            self.guestInfo['guestName'] = _filterXmlChars(args['name'])
+        elif message == 'os-version':
+            self.guestInfo['guestOs'] = _filterXmlChars(args['version'])
+        elif message == 'network-interfaces':
+            interfaces = []
+            old_ips = ''
+            for iface in args['interfaces']:
+                iface['name'] = _filterXmlChars(iface['name'])
+                iface['hw'] = _filterXmlChars(iface['hw'])
+                iface['inet'] = map(_filterXmlChars, iface['inet'])
+                iface['inet6'] = map(_filterXmlChars, iface['inet6'])
+                interfaces.append(iface)
+                # Provide the old information which includes only the IP addresses.
+                old_ips += ' '.join(iface['inet']) + ' '
+            self.guestInfo['netIfaces'] = interfaces
+            self.guestInfo['guestIPs'] = old_ips
+        elif message == 'applications':
+            self.guestInfo['appsList'] = map(_filterXmlChars, args['applications'])
+        elif message == 'active-user':
+            currentUser = _filterXmlChars(args['name'])
             if (currentUser != self.guestInfo['username']) and not (currentUser=='Unknown' and self.guestInfo['username']=='None'):
                 self.guestInfo['username'] = currentUser
                 self.guestInfo['lastLogin'] = time.time()
             self.log.debug(repr(self.guestInfo['username']))
-        elif guestMessage == guestMType.sessionLogon:
+        elif message == 'session-logon':
             self.guestInfo['session'] = "UserLoggedOn"
-        elif guestMessage == guestMType.sessionLock:
+        elif message == 'session-locked':
             self.guestInfo['session'] = "Locked"
-        elif guestMessage == guestMType.sessionUnlock:
+        elif message == 'session-unlock':
             self.guestInfo['session'] = "Active"
-        elif guestMessage == guestMType.sessionLogoff:
+        elif message == 'session-logoff':
             self.guestInfo['session'] = "LoggedOff"
-        elif guestMessage == guestMType.agentUninstalled:
+        elif message == 'uninstalled':
             self.log.debug("RHEV agent was uninstalled.")
             self.guestInfo['appsList'] = []
-        elif guestMessage == guestMType.sessionStartup:
+        elif message == 'session-startup':
             self.log.debug("Guest system is started or restarted.")
-        elif guestMessage == guestMType.sessionShutdown:
+        elif message == 'session-shutdown':
             self.log.debug("Guest system shuts down.")
+        elif message == 'disks-usage':
+            disks = []
+            for disk in args['disks']:
+                disk['path'] = _filterXmlChars(disk['path'])
+                disk['fs'] = _filterXmlChars(disk['fs'])
+                # Converting to string because XML-RPC doesn't support 64-bit
+                # integers.
+                disk['total'] = _filterXmlChars(str(disk['total']))
+                disk['used'] = _filterXmlChars(str(disk['used']))
+                disks.append(disk)
+            self.guestInfo['disksUsage'] = disks
         else:
-            self.log.error('Unknown message type %s', guestMessage)
+            self.log.error('Unknown message type %s', message)
 
     def stop (self):
         self._stopped = True
@@ -220,7 +148,7 @@ class GuestAgent (threading.Thread):
     def desktopLock(self):
         try:
             self.log.debug("desktopLock called")
-            self._forward("lock screen")
+            self._forward("lock-screen")
         except:
             self.log.error(traceback.format_exc())
 
@@ -231,70 +159,82 @@ class GuestAgent (threading.Thread):
                 username = user + '@' + domain
             else:
                 username = user
-            username = username.encode('utf-8')
-            password = password.encode('utf-8')
-            s = struct.pack('>6sI%ds%ds' % (len(username), len(password) + 1),
-                        'login', len(username), username, password)
-            self._forward(s)
+            self._forward('login', { 'username' : username, "password" : password })
         except:
             self.log.error(traceback.format_exc())
 
     def desktopLogoff (self, force):
         try:
             self.log.debug("desktopLogoff called")
-            self._forward('log off')
+            self._forward('log-off')
         except:
             self.log.error(traceback.format_exc())
 
-    def sendHcCmdToDesktop (self, cmd):
+    def desktopShutdown (self, timeout, msg):
         try:
-            self.log.debug("sendHcCmdToDesktop('%s')"%(cmd))
-            self._forward(str(cmd))
+            self.log.debug("desktopShutdown called")
+            self._forward('shutdown', { 'timeout' : timeout, 'message' : msg })
         except:
             self.log.error(traceback.format_exc())
 
+    def _onChannelTimeout(self):
+        self.guestInfo['memUsage'] = 0
+        if self.guestStatus not in ("Powered down", "RebootInProgress"):
+            self.log.log(logging.TRACE, "Guest connection timed out")
+            self.guestStatus = None
+
     READSIZE = 2**16
     def _readBuffer(self):
-        while not self._stopped:
-            try:
-                s = self._sock.recv(self.READSIZE)
-                if s:
-                    self._buffer += s
-                    self._agentTimestamp = time.time()
-                    break
+        try:
+            s = self._sock.recv(self.READSIZE)
+            if s:
+                self._buffer += s
+            else:
+                # s is None if recv() exit with an error. The sleep() is
+                # called in order to prevent a 100% CPU utilization while
+                # trying to read from the socket again (and again...).
                 time.sleep(1)
-            except socket.timeout:
-                # TODO move these specific bits out of here
-                self.guestInfo['memUsage'] = 0
-                if self.guestStatus not in ("Powered down", "RebootInProgress"):
-                    self.log.log(logging.TRACE, "Guest connection timed out")
-                    self.guestStatus = None
-
-    def _readMessage(self):
-        msg = None
-        if len(self._buffer) >= headerLengthBytes:
-            msglen = self._parseHeader(self._buffer[:headerLengthBytes])
-            if len(self._buffer) >= headerLengthBytes + msglen:
-                msg = self._buffer[headerLengthBytes:headerLengthBytes + msglen]
-                self._buffer = self._buffer[headerLengthBytes + msglen:]
-        return msg
-
-    def _parseMessages(self):
-        s = self._readMessage()
-        while not self._stopped and s:
-            self._parseBody(s)
-            s = self._readMessage()
+        except socket.timeout:
+            self._onChannelTimeout()
+
+    def _readLine(self):
+        newline = self._buffer.find('\n')
+        while not self._stopped and newline < 0:
+            self._readBuffer()
+            newline = self._buffer.find('\n')
+        if newline >= 0:
+            newline += 1 # include the EOL char.
+            line = self._buffer[:newline]
+            self._buffer = self._buffer[newline:]
+        else:
+            line = None
+        return line
+
+    def _parseLine(self, line):
+        name = None
+        args = None
+        # line is None after stop() is called and the socket is closed.
+        if line:
+            try:
+                args = json.loads(line.decode('utf8'))
+                name = args['__name__']
+                del args['__name__']
+            except:
+                self.log.exception("Error parsing: %s" % (line))
+        return (name, args)
 
     def run(self):
         self._stopped = False
         try:
             if not self._connect():
                 return
-            self.sendHcCmdToDesktop('refresh')
+            self._forward('refresh')
             self._buffer = ''
             while not self._stopped:
-                self._readBuffer()
-                self._parseMessages()
+                (message, args) = self._parseLine(self._readLine())
+                if message:
+                    self._agentTimestamp = time.time()
+                    self._handleMessage(message, args)
         except:
             if not self._stopped:
                 self.log.error("Unexpected exception: " + traceback.format_exc())
diff --git a/vdsm/vm.py b/vdsm/vm.py
index 941ea0a..ab88375 100644
--- a/vdsm/vm.py
+++ b/vdsm/vm.py
@@ -667,8 +667,7 @@ class Vm(object):
                 self._guestEventTime = now
                 self._guestEvent = 'Powering down'
                 self.log.debug('guestAgent shutdown called')
-                guest_message = 'shutdown,' + timeout + ',' + message
-                self.guestAgent.sendHcCmdToDesktop(guest_message)
+                self.guestAgent.desktopShutdown(timeout, message)
                 agent_timeout = int(timeout) + config.getint('vars', 'sys_shutdown_timeout')
                 timer = threading.Timer(agent_timeout, self._timedShutdown)
                 timer.start()
diff --git a/vdsm_cli/vdsClient.py b/vdsm_cli/vdsClient.py
old mode 100755
new mode 100644
index d29cc4a..426b958
--- a/vdsm_cli/vdsClient.py
+++ b/vdsm_cli/vdsClient.py
@@ -332,12 +332,6 @@ class service:
         print response['status']['message']
         sys.exit(response['status']['code'])
 
-    def sendHcCmd (self, args):
-        vmId, message = tuple(args)
-        response = self.s.sendHcCmdToDesktop(vmId, message)
-        print response['status']['message']
-        sys.exit(response['status']['code'])
-
 ######## IRS methods ####################
     def createStorageDomain(self, args):
         validateArgTypes(args, [int, str, str, str, int, int])
@@ -1894,10 +1888,6 @@ if __name__ == '__main__':
                            ('<vmId>',
                              'Logoff current user'
                             )),
-            'sendHcCmd': ( serv.sendHcCmd,
-                           ('<vmId> <message>',
-                             'Sends a message to a specific VM through Hypercall channel'
-                            )),
             'hibernate': ( serv.hibernate,
                            ('<vmId>',
                              'Hibernates the desktop'
@@ -1962,7 +1952,7 @@ if __name__ == '__main__':
             if o == "-s" or o == "--SSL":
                 serv.useSSL = True
             if o == "--truststore":
-		serv.truststore = v
+                serv.truststore = v
         if len(args) < 2:
             raise Exception("Need at least two arguments")
         server, command = args[0:2]




More information about the vdsm-patches mailing list