code.grnet.gr Git - ganeti-local/blob - lib/luxi.py

   1 #
   2 #
   3
   4 # Copyright (C) 2006, 2007, 2011, 2012 Google Inc.
   5 #
   6 # This program is free software; you can redistribute it and/or modify
   7 # it under the terms of the GNU General Public License as published by
   8 # the Free Software Foundation; either version 2 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # This program is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU General Public License
  17 # along with this program; if not, write to the Free Software
  18 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  19 # 02110-1301, USA.
  20
  21
  22 """Module for the unix socket protocol
  23
  24 This module implements the local unix socket protocol. You only need
  25 this module and the opcodes module in the client program in order to
  26 communicate with the master.
  27
  28 The module is also used by the master daemon.
  29
  30 """
  31
  32 import socket
  33 import collections
  34 import time
  35 import errno
  36 import logging
  37
  38 from ganeti import serializer
  39 from ganeti import constants
  40 from ganeti import errors
  41 from ganeti import utils
  42 from ganeti import objects
  43
  44
  45 KEY_METHOD = "method"
  46 KEY_ARGS = "args"
  47 KEY_SUCCESS = "success"
  48 KEY_RESULT = "result"
  49 KEY_VERSION = "version"
  50
  51 REQ_SUBMIT_JOB = "SubmitJob"
  52 REQ_SUBMIT_MANY_JOBS = "SubmitManyJobs"
  53 REQ_WAIT_FOR_JOB_CHANGE = "WaitForJobChange"
  54 REQ_CANCEL_JOB = "CancelJob"
  55 REQ_ARCHIVE_JOB = "ArchiveJob"
  56 REQ_AUTO_ARCHIVE_JOBS = "AutoArchiveJobs"
  57 REQ_QUERY = "Query"
  58 REQ_QUERY_FIELDS = "QueryFields"
  59 REQ_QUERY_JOBS = "QueryJobs"
  60 REQ_QUERY_INSTANCES = "QueryInstances"
  61 REQ_QUERY_NODES = "QueryNodes"
  62 REQ_QUERY_GROUPS = "QueryGroups"
  63 REQ_QUERY_EXPORTS = "QueryExports"
  64 REQ_QUERY_CONFIG_VALUES = "QueryConfigValues"
  65 REQ_QUERY_CLUSTER_INFO = "QueryClusterInfo"
  66 REQ_QUERY_TAGS = "QueryTags"
  67 REQ_SET_DRAIN_FLAG = "SetDrainFlag"
  68 REQ_SET_WATCHER_PAUSE = "SetWatcherPause"
  69
  70 #: List of all LUXI requests
  71 REQ_ALL = frozenset([
  72   REQ_ARCHIVE_JOB,
  73   REQ_AUTO_ARCHIVE_JOBS,
  74   REQ_CANCEL_JOB,
  75   REQ_QUERY,
  76   REQ_QUERY_CLUSTER_INFO,
  77   REQ_QUERY_CONFIG_VALUES,
  78   REQ_QUERY_EXPORTS,
  79   REQ_QUERY_FIELDS,
  80   REQ_QUERY_GROUPS,
  81   REQ_QUERY_INSTANCES,
  82   REQ_QUERY_JOBS,
  83   REQ_QUERY_NODES,
  84   REQ_QUERY_TAGS,
  85   REQ_SET_DRAIN_FLAG,
  86   REQ_SET_WATCHER_PAUSE,
  87   REQ_SUBMIT_JOB,
  88   REQ_SUBMIT_MANY_JOBS,
  89   REQ_WAIT_FOR_JOB_CHANGE,
  90   ])
  91
  92 DEF_CTMO = 10
  93 DEF_RWTO = 60
  94
  95 # WaitForJobChange timeout
  96 WFJC_TIMEOUT = (DEF_RWTO - 1) / 2
  97
  98
  99 class ProtocolError(errors.LuxiError):
 100   """Denotes an error in the LUXI protocol."""
 101
 102
 103 class ConnectionClosedError(ProtocolError):
 104   """Connection closed error."""
 105
 106
 107 class TimeoutError(ProtocolError):
 108   """Operation timeout error."""
 109
 110
 111 class RequestError(ProtocolError):
 112   """Error on request.
 113
 114   This signifies an error in the request format or request handling,
 115   but not (e.g.) an error in starting up an instance.
 116
 117   Some common conditions that can trigger this exception:
 118     - job submission failed because the job data was wrong
 119     - query failed because required fields were missing
 120
 121   """
 122
 123
 124 class NoMasterError(ProtocolError):
 125   """The master cannot be reached.
 126
 127   This means that the master daemon is not running or the socket has
 128   been removed.
 129
 130   """
 131
 132
 133 class PermissionError(ProtocolError):
 134   """Permission denied while connecting to the master socket.
 135
 136   This means the user doesn't have the proper rights.
 137
 138   """
 139
 140
 141 class Transport:
 142   """Low-level transport class.
 143
 144   This is used on the client side.
 145
 146   This could be replace by any other class that provides the same
 147   semantics to the Client. This means:
 148     - can send messages and receive messages
 149     - safe for multithreading
 150
 151   """
 152
 153   def __init__(self, address, timeouts=None):
 154     """Constructor for the Client class.
 155
 156     Arguments:
 157       - address: a valid address the the used transport class
 158       - timeout: a list of timeouts, to be used on connect and read/write
 159
 160     There are two timeouts used since we might want to wait for a long
 161     time for a response, but the connect timeout should be lower.
 162
 163     If not passed, we use a default of 10 and respectively 60 seconds.
 164
 165     Note that on reading data, since the timeout applies to an
 166     invidual receive, it might be that the total duration is longer
 167     than timeout value passed (we make a hard limit at twice the read
 168     timeout).
 169
 170     """
 171     self.address = address
 172     if timeouts is None:
 173       self._ctimeout, self._rwtimeout = DEF_CTMO, DEF_RWTO
 174     else:
 175       self._ctimeout, self._rwtimeout = timeouts
 176
 177     self.socket = None
 178     self._buffer = ""
 179     self._msgs = collections.deque()
 180
 181     try:
 182       self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
 183
 184       # Try to connect
 185       try:
 186         utils.Retry(self._Connect, 1.0, self._ctimeout,
 187                     args=(self.socket, address, self._ctimeout))
 188       except utils.RetryTimeout:
 189         raise TimeoutError("Connect timed out")
 190
 191       self.socket.settimeout(self._rwtimeout)
 192     except (socket.error, NoMasterError):
 193       if self.socket is not None:
 194         self.socket.close()
 195       self.socket = None
 196       raise
 197
 198   @staticmethod
 199   def _Connect(sock, address, timeout):
 200     sock.settimeout(timeout)
 201     try:
 202       sock.connect(address)
 203     except socket.timeout, err:
 204       raise TimeoutError("Connect timed out: %s" % str(err))
 205     except socket.error, err:
 206       error_code = err.args[0]
 207       if error_code in (errno.ENOENT, errno.ECONNREFUSED):
 208         raise NoMasterError(address)
 209       elif error_code in (errno.EPERM, errno.EACCES):
 210         raise PermissionError(address)
 211       elif error_code == errno.EAGAIN:
 212         # Server's socket backlog is full at the moment
 213         raise utils.RetryAgain()
 214       raise
 215
 216   def _CheckSocket(self):
 217     """Make sure we are connected.
 218
 219     """
 220     if self.socket is None:
 221       raise ProtocolError("Connection is closed")
 222
 223   def Send(self, msg):
 224     """Send a message.
 225
 226     This just sends a message and doesn't wait for the response.
 227
 228     """
 229     if constants.LUXI_EOM in msg:
 230       raise ProtocolError("Message terminator found in payload")
 231
 232     self._CheckSocket()
 233     try:
 234       # TODO: sendall is not guaranteed to send everything
 235       self.socket.sendall(msg + constants.LUXI_EOM)
 236     except socket.timeout, err:
 237       raise TimeoutError("Sending timeout: %s" % str(err))
 238
 239   def Recv(self):
 240     """Try to receive a message from the socket.
 241
 242     In case we already have messages queued, we just return from the
 243     queue. Otherwise, we try to read data with a _rwtimeout network
 244     timeout, and making sure we don't go over 2x_rwtimeout as a global
 245     limit.
 246
 247     """
 248     self._CheckSocket()
 249     etime = time.time() + self._rwtimeout
 250     while not self._msgs:
 251       if time.time() > etime:
 252         raise TimeoutError("Extended receive timeout")
 253       while True:
 254         try:
 255           data = self.socket.recv(4096)
 256         except socket.timeout, err:
 257           raise TimeoutError("Receive timeout: %s" % str(err))
 258         except socket.error, err:
 259           if err.args and err.args[0] == errno.EAGAIN:
 260             continue
 261           raise
 262         break
 263       if not data:
 264         raise ConnectionClosedError("Connection closed while reading")
 265       new_msgs = (self._buffer + data).split(constants.LUXI_EOM)
 266       self._buffer = new_msgs.pop()
 267       self._msgs.extend(new_msgs)
 268     return self._msgs.popleft()
 269
 270   def Call(self, msg):
 271     """Send a message and wait for the response.
 272
 273     This is just a wrapper over Send and Recv.
 274
 275     """
 276     self.Send(msg)
 277     return self.Recv()
 278
 279   def Close(self):
 280     """Close the socket"""
 281     if self.socket is not None:
 282       self.socket.close()
 283       self.socket = None
 284
 285
 286 def ParseRequest(msg):
 287   """Parses a LUXI request message.
 288
 289   """
 290   try:
 291     request = serializer.LoadJson(msg)
 292   except ValueError, err:
 293     raise ProtocolError("Invalid LUXI request (parsing error): %s" % err)
 294
 295   logging.debug("LUXI request: %s", request)
 296
 297   if not isinstance(request, dict):
 298     logging.error("LUXI request not a dict: %r", msg)
 299     raise ProtocolError("Invalid LUXI request (not a dict)")
 300
 301   method = request.get(KEY_METHOD, None) # pylint: disable=E1103
 302   args = request.get(KEY_ARGS, None) # pylint: disable=E1103
 303   version = request.get(KEY_VERSION, None) # pylint: disable=E1103
 304
 305   if method is None or args is None:
 306     logging.error("LUXI request missing method or arguments: %r", msg)
 307     raise ProtocolError(("Invalid LUXI request (no method or arguments"
 308                          " in request): %r") % msg)
 309
 310   return (method, args, version)
 311
 312
 313 def ParseResponse(msg):
 314   """Parses a LUXI response message.
 315
 316   """
 317   # Parse the result
 318   try:
 319     data = serializer.LoadJson(msg)
 320   except KeyboardInterrupt:
 321     raise
 322   except Exception, err:
 323     raise ProtocolError("Error while deserializing response: %s" % str(err))
 324
 325   # Validate response
 326   if not (isinstance(data, dict) and
 327           KEY_SUCCESS in data and
 328           KEY_RESULT in data):
 329     raise ProtocolError("Invalid response from server: %r" % data)
 330
 331   return (data[KEY_SUCCESS], data[KEY_RESULT],
 332           data.get(KEY_VERSION, None)) # pylint: disable=E1103
 333
 334
 335 def FormatResponse(success, result, version=None):
 336   """Formats a LUXI response message.
 337
 338   """
 339   response = {
 340     KEY_SUCCESS: success,
 341     KEY_RESULT: result,
 342     }
 343
 344   if version is not None:
 345     response[KEY_VERSION] = version
 346
 347   logging.debug("LUXI response: %s", response)
 348
 349   return serializer.DumpJson(response)
 350
 351
 352 def FormatRequest(method, args, version=None):
 353   """Formats a LUXI request message.
 354
 355   """
 356   # Build request
 357   request = {
 358     KEY_METHOD: method,
 359     KEY_ARGS: args,
 360     }
 361
 362   if version is not None:
 363     request[KEY_VERSION] = version
 364
 365   # Serialize the request
 366   return serializer.DumpJson(request)
 367
 368
 369 def CallLuxiMethod(transport_cb, method, args, version=None):
 370   """Send a LUXI request via a transport and return the response.
 371
 372   """
 373   assert callable(transport_cb)
 374
 375   request_msg = FormatRequest(method, args, version=version)
 376
 377   # Send request and wait for response
 378   response_msg = transport_cb(request_msg)
 379
 380   (success, result, resp_version) = ParseResponse(response_msg)
 381
 382   # Verify version if there was one in the response
 383   if resp_version is not None and resp_version != version:
 384     raise errors.LuxiError("LUXI version mismatch, client %s, response %s" %
 385                            (version, resp_version))
 386
 387   if success:
 388     return result
 389
 390   errors.MaybeRaise(result)
 391   raise RequestError(result)
 392
 393
 394 class Client(object):
 395   """High-level client implementation.
 396
 397   This uses a backing Transport-like class on top of which it
 398   implements data serialization/deserialization.
 399
 400   """
 401   def __init__(self, address=None, timeouts=None, transport=Transport):
 402     """Constructor for the Client class.
 403
 404     Arguments:
 405       - address: a valid address the the used transport class
 406       - timeout: a list of timeouts, to be used on connect and read/write
 407       - transport: a Transport-like class
 408
 409
 410     If timeout is not passed, the default timeouts of the transport
 411     class are used.
 412
 413     """
 414     if address is None:
 415       address = constants.MASTER_SOCKET
 416     self.address = address
 417     self.timeouts = timeouts
 418     self.transport_class = transport
 419     self.transport = None
 420     self._InitTransport()
 421
 422   def _InitTransport(self):
 423     """(Re)initialize the transport if needed.
 424
 425     """
 426     if self.transport is None:
 427       self.transport = self.transport_class(self.address,
 428                                             timeouts=self.timeouts)
 429
 430   def _CloseTransport(self):
 431     """Close the transport, ignoring errors.
 432
 433     """
 434     if self.transport is None:
 435       return
 436     try:
 437       old_transp = self.transport
 438       self.transport = None
 439       old_transp.Close()
 440     except Exception: # pylint: disable=W0703
 441       pass
 442
 443   def _SendMethodCall(self, data):
 444     # Send request and wait for response
 445     try:
 446       self._InitTransport()
 447       return self.transport.Call(data)
 448     except Exception:
 449       self._CloseTransport()
 450       raise
 451
 452   def Close(self):
 453     """Close the underlying connection.
 454
 455     """
 456     self._CloseTransport()
 457
 458   def CallMethod(self, method, args):
 459     """Send a generic request and return the response.
 460
 461     """
 462     if not isinstance(args, (list, tuple)):
 463       raise errors.ProgrammerError("Invalid parameter passed to CallMethod:"
 464                                    " expected list, got %s" % type(args))
 465     return CallLuxiMethod(self._SendMethodCall, method, args,
 466                           version=constants.LUXI_VERSION)
 467
 468   def SetQueueDrainFlag(self, drain_flag):
 469     return self.CallMethod(REQ_SET_DRAIN_FLAG, (drain_flag, ))
 470
 471   def SetWatcherPause(self, until):
 472     return self.CallMethod(REQ_SET_WATCHER_PAUSE, (until, ))
 473
 474   def SubmitJob(self, ops):
 475     ops_state = map(lambda op: op.__getstate__(), ops)
 476     return self.CallMethod(REQ_SUBMIT_JOB, (ops_state, ))
 477
 478   def SubmitManyJobs(self, jobs):
 479     jobs_state = []
 480     for ops in jobs:
 481       jobs_state.append([op.__getstate__() for op in ops])
 482     return self.CallMethod(REQ_SUBMIT_MANY_JOBS, (jobs_state, ))
 483
 484   def CancelJob(self, job_id):
 485     return self.CallMethod(REQ_CANCEL_JOB, (job_id, ))
 486
 487   def ArchiveJob(self, job_id):
 488     return self.CallMethod(REQ_ARCHIVE_JOB, (job_id, ))
 489
 490   def AutoArchiveJobs(self, age):
 491     timeout = (DEF_RWTO - 1) / 2
 492     return self.CallMethod(REQ_AUTO_ARCHIVE_JOBS, (age, timeout))
 493
 494   def WaitForJobChangeOnce(self, job_id, fields,
 495                            prev_job_info, prev_log_serial,
 496                            timeout=WFJC_TIMEOUT):
 497     """Waits for changes on a job.
 498
 499     @param job_id: Job ID
 500     @type fields: list
 501     @param fields: List of field names to be observed
 502     @type prev_job_info: None or list
 503     @param prev_job_info: Previously received job information
 504     @type prev_log_serial: None or int/long
 505     @param prev_log_serial: Highest log serial number previously received
 506     @type timeout: int/float
 507     @param timeout: Timeout in seconds (values larger than L{WFJC_TIMEOUT} will
 508                     be capped to that value)
 509
 510     """
 511     assert timeout >= 0, "Timeout can not be negative"
 512     return self.CallMethod(REQ_WAIT_FOR_JOB_CHANGE,
 513                            (job_id, fields, prev_job_info,
 514                             prev_log_serial,
 515                             min(WFJC_TIMEOUT, timeout)))
 516
 517   def WaitForJobChange(self, job_id, fields, prev_job_info, prev_log_serial):
 518     while True:
 519       result = self.WaitForJobChangeOnce(job_id, fields,
 520                                          prev_job_info, prev_log_serial)
 521       if result != constants.JOB_NOTCHANGED:
 522         break
 523     return result
 524
 525   def Query(self, what, fields, qfilter):
 526     """Query for resources/items.
 527
 528     @param what: One of L{constants.QR_VIA_LUXI}
 529     @type fields: List of strings
 530     @param fields: List of requested fields
 531     @type qfilter: None or list
 532     @param qfilter: Query filter
 533     @rtype: L{objects.QueryResponse}
 534
 535     """
 536     result = self.CallMethod(REQ_QUERY, (what, fields, qfilter))
 537     return objects.QueryResponse.FromDict(result)
 538
 539   def QueryFields(self, what, fields):
 540     """Query for available fields.
 541
 542     @param what: One of L{constants.QR_VIA_LUXI}
 543     @type fields: None or list of strings
 544     @param fields: List of requested fields
 545     @rtype: L{objects.QueryFieldsResponse}
 546
 547     """
 548     result = self.CallMethod(REQ_QUERY_FIELDS, (what, fields))
 549     return objects.QueryFieldsResponse.FromDict(result)
 550
 551   def QueryJobs(self, job_ids, fields):
 552     return self.CallMethod(REQ_QUERY_JOBS, (job_ids, fields))
 553
 554   def QueryInstances(self, names, fields, use_locking):
 555     return self.CallMethod(REQ_QUERY_INSTANCES, (names, fields, use_locking))
 556
 557   def QueryNodes(self, names, fields, use_locking):
 558     return self.CallMethod(REQ_QUERY_NODES, (names, fields, use_locking))
 559
 560   def QueryGroups(self, names, fields, use_locking):
 561     return self.CallMethod(REQ_QUERY_GROUPS, (names, fields, use_locking))
 562
 563   def QueryExports(self, nodes, use_locking):
 564     return self.CallMethod(REQ_QUERY_EXPORTS, (nodes, use_locking))
 565
 566   def QueryClusterInfo(self):
 567     return self.CallMethod(REQ_QUERY_CLUSTER_INFO, ())
 568
 569   def QueryConfigValues(self, fields):
 570     return self.CallMethod(REQ_QUERY_CONFIG_VALUES, (fields, ))
 571
 572   def QueryTags(self, kind, name):
 573     return self.CallMethod(REQ_QUERY_TAGS, (kind, name))