Convert instance_os_import rpc to new style result

[ganeti-local] / lib / luxi.py

diff --git a/lib/luxi.py b/lib/luxi.py
index ff4ac75..1c3ca6d 100644 (file)
--- a/lib/luxi.py
+++ b/lib/luxi.py
@@ -21,11 +21,11 @@
 
 """Module for the unix socket protocol
 
 
 """Module for the unix socket protocol
 

-This module implements the local unix socket protocl. You only need
+This module implements the local unix socket protocol. You only need

 this module and the opcodes module in the client program in order to
 communicate with the master.
 
 this module and the opcodes module in the client program in order to
 communicate with the master.
 

-The module is also be used by the master daemon.
+The module is also used by the master daemon.

 
 """
 
 
 """
 

@@ -36,6 +36,7 @@ import errno
 
 from ganeti import serializer
 from ganeti import constants
 
 from ganeti import serializer
 from ganeti import constants

+from ganeti import errors

 
 
 KEY_METHOD = 'method'
 
 
 KEY_METHOD = 'method'

@@ -44,9 +45,18 @@ KEY_SUCCESS = "success"
 KEY_RESULT = "result"
 
 REQ_SUBMIT_JOB = "SubmitJob"
 KEY_RESULT = "result"
 
 REQ_SUBMIT_JOB = "SubmitJob"

+REQ_SUBMIT_MANY_JOBS = "SubmitManyJobs"
+REQ_WAIT_FOR_JOB_CHANGE = "WaitForJobChange"

 REQ_CANCEL_JOB = "CancelJob"
 REQ_ARCHIVE_JOB = "ArchiveJob"
 REQ_CANCEL_JOB = "CancelJob"
 REQ_ARCHIVE_JOB = "ArchiveJob"

+REQ_AUTOARCHIVE_JOBS = "AutoArchiveJobs"

 REQ_QUERY_JOBS = "QueryJobs"
 REQ_QUERY_JOBS = "QueryJobs"

+REQ_QUERY_INSTANCES = "QueryInstances"
+REQ_QUERY_NODES = "QueryNodes"
+REQ_QUERY_EXPORTS = "QueryExports"
+REQ_QUERY_CONFIG_VALUES = "QueryConfigValues"
+REQ_QUERY_CLUSTER_INFO = "QueryClusterInfo"
+REQ_QUEUE_SET_DRAIN_FLAG = "SetDrainFlag"

 
 DEF_CTMO = 10
 DEF_RWTO = 60
 
 DEF_CTMO = 10
 DEF_RWTO = 60

@@ -150,7 +160,7 @@ class Transport:
       except socket.timeout, err:
         raise TimeoutError("Connect timed out: %s" % str(err))
       except socket.error, err:
       except socket.timeout, err:
         raise TimeoutError("Connect timed out: %s" % str(err))
       except socket.error, err:

-        if err.args[0] == errno.ENOENT:
+        if err.args[0] in (errno.ENOENT, errno.ECONNREFUSED):

           raise NoMasterError((address,))
         raise
       self.socket.settimeout(self._rwtimeout)
           raise NoMasterError((address,))
         raise
       self.socket.settimeout(self._rwtimeout)

@@ -244,7 +254,32 @@ class Client(object):
     """
     if address is None:
       address = constants.MASTER_SOCKET
     """
     if address is None:
       address = constants.MASTER_SOCKET

-    self.transport = transport(address, timeouts=timeouts)
+    self.address = address
+    self.timeouts = timeouts
+    self.transport_class = transport
+    self.transport = None
+    self._InitTransport()
+
+  def _InitTransport(self):
+    """(Re)initialize the transport if needed.
+
+    """
+    if self.transport is None:
+      self.transport = self.transport_class(self.address,
+                                            timeouts=self.timeouts)
+
+  def _CloseTransport(self):
+    """Close the transport, ignoring errors.
+
+    """
+    if self.transport is None:
+      return
+    try:
+      old_transp = self.transport
+      self.transport = None
+      old_transp.Close()
+    except Exception, err:
+      pass

 
   def CallMethod(self, method, args):
     """Send a generic request and return the response.
 
   def CallMethod(self, method, args):
     """Send a generic request and return the response.

@@ -256,8 +291,18 @@ class Client(object):
       KEY_ARGS: args,
       }
 
       KEY_ARGS: args,
       }
 

+    # Serialize the request
+    send_data = serializer.DumpJson(request, indent=False)
+

     # Send request and wait for response
     # Send request and wait for response

-    result = self.transport.Call(serializer.DumpJson(request, indent=False))
+    try:
+      self._InitTransport()
+      result = self.transport.Call(send_data)
+    except Exception:
+      self._CloseTransport()
+      raise
+
+    # Parse the result

     try:
       data = serializer.LoadJson(result)
     except Exception, err:
     try:
       data = serializer.LoadJson(result)
     except Exception, err:

@@ -269,23 +314,71 @@ class Client(object):
         KEY_RESULT not in data):
       raise DecodingError("Invalid response from server: %s" % str(data))
 
         KEY_RESULT not in data):
       raise DecodingError("Invalid response from server: %s" % str(data))
 

+    result = data[KEY_RESULT]
+

     if not data[KEY_SUCCESS]:
       # TODO: decide on a standard exception
     if not data[KEY_SUCCESS]:
       # TODO: decide on a standard exception

-      raise RequestError(data[KEY_RESULT])
+      if (isinstance(result, (tuple, list)) and len(result) == 2 and
+          isinstance(result[1], (tuple, list))):
+        # custom ganeti errors
+        err_class = errors.GetErrorClass(result[0])
+        if err_class is not None:
+          raise err_class, tuple(result[1])
+
+      raise RequestError(result)
+
+    return result

 
 

-    return data[KEY_RESULT]
+  def SetQueueDrainFlag(self, drain_flag):
+    return self.CallMethod(REQ_QUEUE_SET_DRAIN_FLAG, drain_flag)

 
   def SubmitJob(self, ops):
     ops_state = map(lambda op: op.__getstate__(), ops)
     return self.CallMethod(REQ_SUBMIT_JOB, ops_state)
 
 
   def SubmitJob(self, ops):
     ops_state = map(lambda op: op.__getstate__(), ops)
     return self.CallMethod(REQ_SUBMIT_JOB, ops_state)
 

+  def SubmitManyJobs(self, jobs):
+    jobs_state = []
+    for ops in jobs:
+      jobs_state.append([op.__getstate__() for op in ops])
+    return self.CallMethod(REQ_SUBMIT_MANY_JOBS, jobs_state)
+

   def CancelJob(self, job_id):
     return self.CallMethod(REQ_CANCEL_JOB, job_id)
 
   def ArchiveJob(self, job_id):
     return self.CallMethod(REQ_ARCHIVE_JOB, job_id)
 
   def CancelJob(self, job_id):
     return self.CallMethod(REQ_CANCEL_JOB, job_id)
 
   def ArchiveJob(self, job_id):
     return self.CallMethod(REQ_ARCHIVE_JOB, job_id)
 

+  def AutoArchiveJobs(self, age):
+    timeout = (DEF_RWTO - 1) / 2
+    return self.CallMethod(REQ_AUTOARCHIVE_JOBS, (age, timeout))
+
+  def WaitForJobChange(self, job_id, fields, prev_job_info, prev_log_serial):
+    timeout = (DEF_RWTO - 1) / 2
+    while True:
+      result = self.CallMethod(REQ_WAIT_FOR_JOB_CHANGE,
+                               (job_id, fields, prev_job_info,
+                                prev_log_serial, timeout))
+      if result != constants.JOB_NOTCHANGED:
+        break
+    return result
+

   def QueryJobs(self, job_ids, fields):
     return self.CallMethod(REQ_QUERY_JOBS, (job_ids, fields))
 
   def QueryJobs(self, job_ids, fields):
     return self.CallMethod(REQ_QUERY_JOBS, (job_ids, fields))
 

+  def QueryInstances(self, names, fields, use_locking):
+    return self.CallMethod(REQ_QUERY_INSTANCES, (names, fields, use_locking))
+
+  def QueryNodes(self, names, fields, use_locking):
+    return self.CallMethod(REQ_QUERY_NODES, (names, fields, use_locking))
+
+  def QueryExports(self, nodes, use_locking):
+    return self.CallMethod(REQ_QUERY_EXPORTS, (nodes, use_locking))
+
+  def QueryClusterInfo(self):
+    return self.CallMethod(REQ_QUERY_CLUSTER_INFO, ())
+
+  def QueryConfigValues(self, fields):
+    return self.CallMethod(REQ_QUERY_CONFIG_VALUES, fields)
+
+

 # TODO: class Server(object)
 # TODO: class Server(object)