Commit 912b2278 authored by Petr Pudlak's avatar Petr Pudlak

Split Luxi Client into a generic and a specific part

The generic part will be reused in WConfd.
Signed-off-by: default avatarPetr Pudlak <pudlak@google.com>
Reviewed-by: default avatarHelga Velroyen <helgav@google.com>
parent ff1012ef
......@@ -465,6 +465,7 @@ server_PYTHON = \
rpc_PYTHON = \
lib/rpc/__init__.py \
lib/rpc/client.py \
lib/rpc/errors.py \
lib/rpc/node.py \
lib/rpc/transport.py
......@@ -1444,6 +1445,7 @@ python_tests = \
test/py/ganeti.rapi.rlib2_unittest.py \
test/py/ganeti.rapi.testutils_unittest.py \
test/py/ganeti.rpc_unittest.py \
test/py/ganeti.rpc.client_unittest.py \
test/py/ganeti.runtime_unittest.py \
test/py/ganeti.serializer_unittest.py \
test/py/ganeti.server.rapi_unittest.py \
......
......@@ -29,14 +29,10 @@ The module is also used by the master daemon.
"""
import logging
from ganeti import serializer
from ganeti import constants
from ganeti import errors
from ganeti import objects
from ganeti import pathutils
from ganeti.rpc import transport as t
import ganeti.rpc.client as cl
from ganeti.rpc.transport import Transport
from ganeti.rpc.errors import (ProtocolError, ConnectionClosedError,
TimeoutError, RequestError, NoMasterError,
PermissionError)
......@@ -49,22 +45,10 @@ __all__ = [
"RequestError",
"NoMasterError",
"PermissionError",
"ParseRequest",
"ParseResponse",
"FormatRequest",
"FormatResponse",
"CallLuxiMethod",
# classes:
"Client"
]
KEY_METHOD = constants.LUXI_KEY_METHOD
KEY_ARGS = constants.LUXI_KEY_ARGS
KEY_SUCCESS = constants.LUXI_KEY_SUCCESS
KEY_RESULT = constants.LUXI_KEY_RESULT
KEY_VERSION = constants.LUXI_KEY_VERSION
REQ_SUBMIT_JOB = constants.LUXI_REQ_SUBMIT_JOB
REQ_SUBMIT_JOB_TO_DRAINED_QUEUE = constants.LUXI_REQ_SUBMIT_JOB_TO_DRAINED_QUEUE
REQ_SUBMIT_MANY_JOBS = constants.LUXI_REQ_SUBMIT_MANY_JOBS
......@@ -93,187 +77,20 @@ DEF_RWTO = constants.LUXI_DEF_RWTO
WFJC_TIMEOUT = constants.LUXI_WFJC_TIMEOUT
def ParseRequest(msg):
"""Parses a LUXI request message.
"""
try:
request = serializer.LoadJson(msg)
except ValueError, err:
raise ProtocolError("Invalid LUXI request (parsing error): %s" % err)
logging.debug("LUXI request: %s", request)
if not isinstance(request, dict):
logging.error("LUXI request not a dict: %r", msg)
raise ProtocolError("Invalid LUXI request (not a dict)")
method = request.get(KEY_METHOD, None) # pylint: disable=E1103
args = request.get(KEY_ARGS, None) # pylint: disable=E1103
version = request.get(KEY_VERSION, None) # pylint: disable=E1103
if method is None or args is None:
logging.error("LUXI request missing method or arguments: %r", msg)
raise ProtocolError(("Invalid LUXI request (no method or arguments"
" in request): %r") % msg)
return (method, args, version)
def ParseResponse(msg):
"""Parses a LUXI response message.
"""
# Parse the result
try:
data = serializer.LoadJson(msg)
except KeyboardInterrupt:
raise
except Exception, err:
raise ProtocolError("Error while deserializing response: %s" % str(err))
# Validate response
if not (isinstance(data, dict) and
KEY_SUCCESS in data and
KEY_RESULT in data):
raise ProtocolError("Invalid response from server: %r" % data)
return (data[KEY_SUCCESS], data[KEY_RESULT],
data.get(KEY_VERSION, None)) # pylint: disable=E1103
def FormatResponse(success, result, version=None):
"""Formats a LUXI response message.
"""
response = {
KEY_SUCCESS: success,
KEY_RESULT: result,
}
if version is not None:
response[KEY_VERSION] = version
logging.debug("LUXI response: %s", response)
return serializer.DumpJson(response)
def FormatRequest(method, args, version=None):
"""Formats a LUXI request message.
"""
# Build request
request = {
KEY_METHOD: method,
KEY_ARGS: args,
}
if version is not None:
request[KEY_VERSION] = version
# Serialize the request
return serializer.DumpJson(request)
def CallLuxiMethod(transport_cb, method, args, version=None):
"""Send a LUXI request via a transport and return the response.
"""
assert callable(transport_cb)
request_msg = FormatRequest(method, args, version=version)
# Send request and wait for response
response_msg = transport_cb(request_msg)
(success, result, resp_version) = ParseResponse(response_msg)
# Verify version if there was one in the response
if resp_version is not None and resp_version != version:
raise errors.LuxiError("LUXI version mismatch, client %s, response %s" %
(version, resp_version))
if success:
return result
errors.MaybeRaise(result)
raise RequestError(result)
class Client(object):
class Client(cl.AbstractClient):
"""High-level client implementation.
This uses a backing Transport-like class on top of which it
implements data serialization/deserialization.
"""
def __init__(self, address=None, timeouts=None, transport=t.Transport):
def __init__(self, address=None, timeouts=None, transport=Transport):
"""Constructor for the Client class.
Arguments:
- address: a valid address the the used transport class
- timeout: a list of timeouts, to be used on connect and read/write
- transport: a Transport-like class
If timeout is not passed, the default timeouts of the transport
class are used.
"""
if address is None:
address = pathutils.MASTER_SOCKET
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: # pylint: disable=W0703
pass
def _SendMethodCall(self, data):
# Send request and wait for response
try:
self._InitTransport()
return self.transport.Call(data)
except Exception:
self._CloseTransport()
raise
def Close(self):
"""Close the underlying connection.
"""
self._CloseTransport()
def CallMethod(self, method, args):
"""Send a generic request and return the response.
Arguments are the same as for L{AbstractClient}.
"""
if not isinstance(args, (list, tuple)):
raise errors.ProgrammerError("Invalid parameter passed to CallMethod:"
" expected list, got %s" % type(args))
return CallLuxiMethod(self._SendMethodCall, method, args,
version=constants.LUXI_VERSION)
super(Client, self).__init__(address, timeouts, transport)
def SetQueueDrainFlag(self, drain_flag):
return self.CallMethod(REQ_SET_DRAIN_FLAG, (drain_flag, ))
......
......@@ -36,6 +36,7 @@ from ganeti import server
from ganeti import utils
from ganeti import compat
from ganeti import luxi
import ganeti.rpc.client as rpccl
from ganeti import rapi
import ganeti.http.server # pylint: disable=W0611
......@@ -273,7 +274,7 @@ class _TestLuxiTransport:
raised. There is no return value.
"""
(method, _, _) = luxi.ParseRequest(data)
(method, _, _) = rpccl.ParseRequest(data)
# Take a note of called method
self._record_fn(method)
......
#
#
# Copyright (C) 2013 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
# 02110-1301, USA.
"""Module for generic RPC clients.
"""
import logging
from ganeti import pathutils
import ganeti.rpc.transport as t
from ganeti import constants
from ganeti import errors
from ganeti.rpc.errors import (ProtocolError, RequestError, LuxiError)
from ganeti import serializer
KEY_METHOD = constants.LUXI_KEY_METHOD
KEY_ARGS = constants.LUXI_KEY_ARGS
KEY_SUCCESS = constants.LUXI_KEY_SUCCESS
KEY_RESULT = constants.LUXI_KEY_RESULT
KEY_VERSION = constants.LUXI_KEY_VERSION
def ParseRequest(msg):
"""Parses a request message.
"""
try:
request = serializer.LoadJson(msg)
except ValueError, err:
raise ProtocolError("Invalid LUXI request (parsing error): %s" % err)
logging.debug("LUXI request: %s", request)
if not isinstance(request, dict):
logging.error("LUXI request not a dict: %r", msg)
raise ProtocolError("Invalid LUXI request (not a dict)")
method = request.get(KEY_METHOD, None) # pylint: disable=E1103
args = request.get(KEY_ARGS, None) # pylint: disable=E1103
version = request.get(KEY_VERSION, None) # pylint: disable=E1103
if method is None or args is None:
logging.error("LUXI request missing method or arguments: %r", msg)
raise ProtocolError(("Invalid LUXI request (no method or arguments"
" in request): %r") % msg)
return (method, args, version)
def ParseResponse(msg):
"""Parses a response message.
"""
# Parse the result
try:
data = serializer.LoadJson(msg)
except KeyboardInterrupt:
raise
except Exception, err:
raise ProtocolError("Error while deserializing response: %s" % str(err))
# Validate response
if not (isinstance(data, dict) and
KEY_SUCCESS in data and
KEY_RESULT in data):
raise ProtocolError("Invalid response from server: %r" % data)
return (data[KEY_SUCCESS], data[KEY_RESULT],
data.get(KEY_VERSION, None)) # pylint: disable=E1103
def FormatResponse(success, result, version=None):
"""Formats a response message.
"""
response = {
KEY_SUCCESS: success,
KEY_RESULT: result,
}
if version is not None:
response[KEY_VERSION] = version
logging.debug("LUXI response: %s", response)
return serializer.DumpJson(response)
def FormatRequest(method, args, version=None):
"""Formats a request message.
"""
# Build request
request = {
KEY_METHOD: method,
KEY_ARGS: args,
}
if version is not None:
request[KEY_VERSION] = version
# Serialize the request
return serializer.DumpJson(request)
def CallLuxiMethod(transport_cb, method, args, version=None):
"""Send a LUXI request via a transport and return the response.
"""
assert callable(transport_cb)
request_msg = FormatRequest(method, args, version=version)
# Send request and wait for response
response_msg = transport_cb(request_msg)
(success, result, resp_version) = ParseResponse(response_msg)
# Verify version if there was one in the response
if resp_version is not None and resp_version != version:
raise LuxiError("LUXI version mismatch, client %s, response %s" %
(version, resp_version))
if success:
return result
errors.MaybeRaise(result)
raise RequestError(result)
class AbstractClient(object):
"""High-level client abstraction.
This uses a backing Transport-like class on top of which it
implements data serialization/deserialization.
"""
def __init__(self, address=None, timeouts=None,
transport=t.Transport):
"""Constructor for the Client class.
Arguments:
- address: a valid address the the used transport class
- timeout: a list of timeouts, to be used on connect and read/write
- transport: a Transport-like class
If timeout is not passed, the default timeouts of the transport
class are used.
"""
if address is None:
address = pathutils.MASTER_SOCKET
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: # pylint: disable=W0703
pass
def _SendMethodCall(self, data):
# Send request and wait for response
try:
self._InitTransport()
return self.transport.Call(data)
except Exception:
self._CloseTransport()
raise
def Close(self):
"""Close the underlying connection.
"""
self._CloseTransport()
def close(self):
"""Same as L{Close}, to be used with contextlib.closing(...).
"""
self.Close()
def CallMethod(self, method, args):
"""Send a generic request and return the response.
"""
if not isinstance(args, (list, tuple)):
raise errors.ProgrammerError("Invalid parameter passed to CallMethod:"
" expected list, got %s" % type(args))
return CallLuxiMethod(self._SendMethodCall, method, args,
version=constants.LUXI_VERSION)
......@@ -53,6 +53,7 @@ from ganeti import errors
from ganeti import ssconf
from ganeti import workerpool
import ganeti.rpc.node as rpc
import ganeti.rpc.client as rpccl
from ganeti import bootstrap
from ganeti import netutils
from ganeti import objects
......@@ -92,7 +93,7 @@ class ClientRequestWorker(workerpool.BaseWorker):
client_ops = ClientOps(server)
try:
(method, args, ver) = luxi.ParseRequest(message)
(method, args, ver) = rpccl.ParseRequest(message)
except luxi.ProtocolError, err:
logging.error("Protocol Error: %s", err)
client.close_log()
......@@ -117,7 +118,7 @@ class ClientRequestWorker(workerpool.BaseWorker):
result = "Caught exception: %s" % str(err[1])
try:
reply = luxi.FormatResponse(success, result)
reply = rpccl.FormatResponse(success, result)
client.send_message(reply)
# awake the main thread so that it can write out the data.
server.awaker.signal()
......
#!/usr/bin/python
#
# Copyright (C) 2010 Google Inc.
# Copyright (C) 2010, 2013 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
......@@ -19,7 +19,9 @@
# 02110-1301, USA.
"""Script for unittesting the luxi module"""
"""Script for unittesting the luxi module.
Currently empty (after all the tests moved to ganeti.rpc.client_unittest.py)."""
import unittest
......@@ -30,224 +32,3 @@ from ganeti import luxi
from ganeti import serializer
import testutils
class TestLuxiParsing(testutils.GanetiTestCase):
def testParseRequest(self):
msg = serializer.DumpJson({
luxi.KEY_METHOD: "foo",
luxi.KEY_ARGS: ("bar", "baz", 123),
})
self.assertEqualValues(luxi.ParseRequest(msg),
("foo", ["bar", "baz", 123], None))
self.assertRaises(luxi.ProtocolError, luxi.ParseRequest,
"this\"is {invalid, ]json data")
# No dict
self.assertRaises(luxi.ProtocolError, luxi.ParseRequest,
serializer.DumpJson(123))
# Empty dict
self.assertRaises(luxi.ProtocolError, luxi.ParseRequest,
serializer.DumpJson({ }))
# No arguments
self.assertRaises(luxi.ProtocolError, luxi.ParseRequest,
serializer.DumpJson({ luxi.KEY_METHOD: "foo", }))
# No method
self.assertRaises(luxi.ProtocolError, luxi.ParseRequest,
serializer.DumpJson({ luxi.KEY_ARGS: [], }))
# No method or arguments
self.assertRaises(luxi.ProtocolError, luxi.ParseRequest,
serializer.DumpJson({ luxi.KEY_VERSION: 1, }))
def testParseRequestWithVersion(self):
msg = serializer.DumpJson({
luxi.KEY_METHOD: "version",
luxi.KEY_ARGS: (["some"], "args", 0, "here"),
luxi.KEY_VERSION: 20100101,
})
self.assertEqualValues(luxi.ParseRequest(msg),
("version", [["some"], "args", 0, "here"], 20100101))
def testParseResponse(self):
msg = serializer.DumpJson({
luxi.KEY_SUCCESS: True,
luxi.KEY_RESULT: None,
})
self.assertEqual(luxi.ParseResponse(msg), (True, None, None))
self.assertRaises(luxi.ProtocolError, luxi.ParseResponse,
"this\"is {invalid, ]json data")
# No dict
self.assertRaises(luxi.ProtocolError, luxi.ParseResponse,
serializer.DumpJson(123))
# Empty dict
self.assertRaises(luxi.ProtocolError, luxi.ParseResponse,
serializer.DumpJson({ }))
# No success
self.assertRaises(luxi.ProtocolError, luxi.ParseResponse,
serializer.DumpJson({ luxi.KEY_RESULT: True, }))
# No result
self.assertRaises(luxi.ProtocolError, luxi.ParseResponse,
serializer.DumpJson({ luxi.KEY_SUCCESS: True, }))
# No result or success
self.assertRaises(luxi.ProtocolError, luxi.ParseResponse,
serializer.DumpJson({ luxi.KEY_VERSION: 123, }))
def testParseResponseWithVersion(self):
msg = serializer.DumpJson({
luxi.KEY_SUCCESS: True,
luxi.KEY_RESULT: "Hello World",
luxi.KEY_VERSION: 19991234,
})
self.assertEqual(luxi.ParseResponse(msg), (True, "Hello World", 19991234))
def testFormatResponse(self):
for success, result in [(False, "error"), (True, "abc"),
(True, { "a": 123, "b": None, })]:
msg = luxi.FormatResponse(success, result)
msgdata = serializer.LoadJson(msg)
self.assert_(luxi.KEY_SUCCESS in msgdata)
self.assert_(luxi.KEY_RESULT in msgdata)
self.assert_(luxi.KEY_VERSION not in msgdata)
self.assertEqualValues(msgdata,
{ luxi.KEY_SUCCESS: success,
luxi.KEY_RESULT: result,
})
def testFormatResponseWithVersion(self):
for success, result, version in [(False, "error", 123), (True, "abc", 999),
(True, { "a": 123, "b": None, }, 2010)]:
msg = luxi.FormatResponse(success, result, version=version)
msgdata = serializer.LoadJson(msg)
self.assert_(luxi.KEY_SUCCESS in msgdata)
self.assert_(luxi.KEY_RESULT in msgdata)
self.assert_(luxi.KEY_VERSION in msgdata)
self.assertEqualValues(msgdata,
{ luxi.KEY_SUCCESS: success,
luxi.KEY_RESULT: result,
luxi.KEY_VERSION: version,
})