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