vdsm-patches April 2015

vdsm-patches@lists.fedorahosted.org

51 participants
798 discussions

Change in vdsm[master]: [WIP] EVENTS
by smizrahi＠redhat.com 10 Jun '15

10 Jun '15

Saggi Mizrahi has uploaded a new change for review. Change subject: [WIP] EVENTS ...................................................................... [WIP] EVENTS Change-Id: Id27b5ca1773139932eb5cb16921d5abec4991c5e Signed-off-by: Saggi Mizrahi <smizrahi(a)redhat.com> --- M lib/stompClient.py M lib/vdsm/config.py.in M lib/vdsm/sslutils.py M lib/yajsonrpc/betterAsyncore.py M lib/yajsonrpc/stomp.py M lib/yajsonrpc/stompReactor.py M tests/jsonRpcHelper.py M tests/jsonRpcTests.py M tests/miscTests.py M tests/protocoldetectorTests.py M vdsm.spec.in M vdsm/API.py M vdsm/clientIF.py M vdsm/protocoldetector.py M vdsm/rpc/BindingJsonRpc.py 15 files changed, 336 insertions(+), 141 deletions(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/69/38069/1 diff --git a/lib/stompClient.py b/lib/stompClient.py index d3cd0ab..5a6d7f5 100644 --- a/lib/stompClient.py +++ b/lib/stompClient.py @@ -1,6 +1,12 @@ #!/usr/bin/python import yajsonrpc as yjrpc -import yajsonrpc.stompReactor as sr +from yajsonrpc.stompReactor import ( + _DEFAULT_REQUEST_DESTINATION, + _DEFAULT_RESPONSE_DESTINATION, + ClientRpcTransportAdapter, + ServerRpcContextAdapter, + StompReactor, +) import socket from threading import Thread, Lock import time @@ -14,7 +20,7 @@ if _reactor is None: with _reactorLock: if _reactor is None: - _reactor = sr.StompReactor() + _reactor = StompReactor() t = Thread(target=_reactor.process_requests) t.setDaemon(True) t.start() @@ -42,95 +48,38 @@ t.setDaemon(True) t.start() sub = stomp_client.subscribe( - sr._DEFAULT_REQUEST_DESTINATION, + _DEFAULT_REQUEST_DESTINATION, message_handler=ServerRpcContextAdapter.subscription_handler(server) ) server._sub_ = sub return server -class ServerRpcContextAdapter(object): - @classmethod - def subscription_handler(cls, server): - def handler(sub, frame): - server.queueRequest( - ( - ServerRpcContextAdapter(sub.client, frame), - frame.body - ) - ) - - return handler - - def __init__(self, client, request_frame): - self._client = client - self._reply_to = request_frame.headers.get('reply-to', None) - - def get_local_address(self, *args, **kwargs): - return "" - - def send(self, data): - if self._reply_to is None: - return - - self._client.send( - self._reply_to, - data, - { - "content-type": "application/json", - } - ) - - -class ClientRpcTransportAdapter(object): - def __init__(self, sub, destination, client): - self._sub = sub - sub.set_message_handler(self._handle_message) - self._destination = destination - self._client = client - self._message_handler = lambda arg: None - - def setMessageHandler(self, handler): - self._message_handler = handler - - def send(self, data): - headers = { - "content-type": "application/json", - "reply-to": self._sub.destination, - } - self._client.send( - data, - self._destination, - headers, - ) - - def _handle_message(self, sub, frame): - self._message_handler((self, frame.body)) - - def close(self): - self._sub.unsubscribe() - - def connect(address): sock = socket.create_connection(address) reactor = get_reactor() stomp_client = reactor.createClient(sock) - subscription = stomp_client.subscribe(sr._DEFAULT_RESPONSE_DESTINATIOM) + subscription = stomp_client.subscribe( + _DEFAULT_RESPONSE_DESTINATION, + sub_id="__vdsm_fake_broker__", + ) client = yjrpc.JsonRpcClient( ClientRpcTransportAdapter( subscription, - sr._DEFAULT_REQUEST_DESTINATION, + _DEFAULT_REQUEST_DESTINATION, stomp_client, ) ) return client -BROKER_ADDRESS = ("127.0.0.1", 5445) +BROKER_ADDRESS = ("127.0.0.1", 54321) -server = dummy_server(BROKER_ADDRESS) +# server = dummy_server(BROKER_ADDRESS) time.sleep(2) client = connect(BROKER_ADDRESS) -client.callMethod("echo", ["123"], 1) +client.callMethod("Host.ping", [], 1) +client.callMethod("Host.ping", [], 1) +client.callMethod("Host.ping", [], 1) diff --git a/lib/vdsm/config.py.in b/lib/vdsm/config.py.in index 94c6782..fb0150b 100644 --- a/lib/vdsm/config.py.in +++ b/lib/vdsm/config.py.in @@ -325,6 +325,20 @@ ('guests_gateway_ip', '', None), + ('broker_address', '127.0.0.1', + 'Address where the broker is listening at. Use an empty string ' + 'for none'), + + ('broker_port', '5445', + 'Port where the broker is listening at.'), + + ('request_queues', + 'jms.topic.vdsm_requests, jms.topic.vdsm_irs_requests', + 'Queues to subscribe to for RPC requests'), + + ('topic_prefix', 'jms.topic.', + 'Prefix to use for events'), + ]), # Section: [devel] diff --git a/lib/vdsm/sslutils.py b/lib/vdsm/sslutils.py index 74555ee..6568842 100644 --- a/lib/vdsm/sslutils.py +++ b/lib/vdsm/sslutils.py @@ -35,7 +35,7 @@ class SSLSocket(object): def __init__(self, connection): self.connection = connection - self._data = None + self._data = "" def gettimeout(self): return self.connection.socket.gettimeout() @@ -57,12 +57,14 @@ def read(self, size=4096, flag=None): result = None if flag == socket.MSG_PEEK: - self._data = self.connection.read(size) + bytes_left = size - len(self._data) + if bytes_left > 0: + self._data += self.connection.read(bytes_left) result = self._data else: if self._data: result = self._data - self._data = None + self._data = "" else: result = self.connection.read(size) return result diff --git a/lib/yajsonrpc/betterAsyncore.py b/lib/yajsonrpc/betterAsyncore.py index af88877..8656df8 100644 --- a/lib/yajsonrpc/betterAsyncore.py +++ b/lib/yajsonrpc/betterAsyncore.py @@ -18,14 +18,17 @@ # while enabling compositing instead of inheritance. import asyncore import socket -import types from errno import EWOULDBLOCK +import types from vdsm.infra.eventfd import EventFD class Dispatcher(asyncore.dispatcher): def __init__(self, impl=None, sock=None, map=None): + # This has to be done before the super initialization because + # dispatcher implements __getattr__. + self.__impl = None asyncore.dispatcher.__init__(self, sock=sock, map=map) if impl is not None: self.switch_implementation(impl) @@ -82,8 +85,9 @@ Note that this value is a reccomendation only. """ + impl = self.__impl return getattr( - self.__impl, + impl, "next_check_interval", lambda d: None )(self) @@ -100,7 +104,7 @@ self.handle_read() if hasattr(self.socket, "pending"): - while self.socket.pending() > 0: + while self.socket.pending() > 0 and self.connected: self.handle_read() def recv(self, buffer_size): @@ -115,7 +119,9 @@ return data except socket.error, why: # winsock sometimes raises ENOTCONN - if why.args[0] in asyncore._DISCONNECTED: + if why.args[0] == EWOULDBLOCK: + return None + elif why.args[0] in asyncore._DISCONNECTED: self.handle_close() return '' else: diff --git a/lib/yajsonrpc/stomp.py b/lib/yajsonrpc/stomp.py index a3d63f3..cf24a7f 100644 --- a/lib/yajsonrpc/stomp.py +++ b/lib/yajsonrpc/stomp.py @@ -16,11 +16,12 @@ import logging import os import socket -from threading import Timer from uuid import uuid4 from collections import deque import re + +FAKE_SUB_ID = "__vdsm_fake_broker__" _RE_ESCAPE_SEQUENCE = re.compile(r"\\(.)") @@ -75,7 +76,10 @@ class StompError(RuntimeError): def __init__(self, frame): - RuntimeError.__init__(self, frame.body) + self.frame = frame + super(RuntimeError, self).__init__( + self.body, + ) class _HeartBeatFrame(object): @@ -441,10 +445,9 @@ def _process_message(self, frame, dispatcher): sub_id = frame.headers.get(Headers.SUBSCRIPTION) if sub_id is None: - self.log.warning( - "Got message without a subscription" - ) - return + # This is actually an error, but because of the fake broker we + # injects a fake sub id + sub_id = "__vdsm_fake_broker__" sub = self._subscriptions.get(sub_id) if sub is None: @@ -464,7 +467,8 @@ def send(self, destination, data="", headers=None): final_headers = {"destination": destination} - final_headers.update(headers) + if headers is not None: + final_headers.update(headers) self.queue_frame(Frame( Command.SEND, final_headers, @@ -552,8 +556,3 @@ {"id": str(subid)}) client.put(frame) self._valid = False - - def __del__(self): - # Using a timer because unsubscribe action might involve taking locks. - if self._valid: - Timer(0, self.unsubscribe) diff --git a/lib/yajsonrpc/stompReactor.py b/lib/yajsonrpc/stompReactor.py index 5a1b595..44fbcbb 100644 --- a/lib/yajsonrpc/stompReactor.py +++ b/lib/yajsonrpc/stompReactor.py @@ -20,15 +20,17 @@ from betterAsyncore import Dispatcher, Reactor from vdsm.sslutils import SSLSocket +from . import ( + JsonRpcClient, + JsonRpcServer, +) _STATE_LEN = "Waiting for message length" _STATE_MSG = "Waiting for message" -_DEFAULT_RESPONSE_DESTINATIOM = "jms.topic.vdsm_legacy_responses" +_DEFAULT_RESPONSE_DESTINATION = "jms.topic.vdsm_legacy_responses" _DEFAULT_REQUEST_DESTINATION = "jms.topic.vdsm_legacy_requests" - -_FAKE_SUB_ID = "__vdsm_fake_broker__" def parseHeartBeatHeader(v): @@ -177,13 +179,16 @@ def check_read(self): self._stompConn._dispatcher.handle_read_event() - def send(self, message): + def send( + self, + message, + destination=_DEFAULT_RESPONSE_DESTINATION, + ): self.log.debug("Sending response") res = stomp.Frame( stomp.Command.MESSAGE, { - stomp.Headers.DESTINATION: _DEFAULT_RESPONSE_DESTINATIOM, - stomp.Headers.SUBSCRIPTION: _FAKE_SUB_ID, + stomp.Headers.DESTINATION: destination, stomp.Headers.CONTENT_TYPE: "application/json", }, message @@ -325,3 +330,101 @@ dispatcher.del_channel() self.json_binding.add_socket(self._reactor, client_socket) self.log.debug("Stomp detected from %s", socket_address) + + +class ServerRpcContextAdapter(object): + @classmethod + def subscription_handler(cls, server): + def handler(sub, frame): + server.queueRequest( + ( + ServerRpcContextAdapter(sub.client, frame), + frame.body + ) + ) + + return handler + + def __init__(self, client, request_frame): + self._client = client + self._reply_to = request_frame.headers.get('reply-to', None) + + def get_local_address(self, *args, **kwargs): + return "" + + def send(self, data): + if self._reply_to is None: + return + + self._client.send( + self._reply_to, + data, + { + "content-type": "application/json", + } + ) + + +class ClientRpcTransportAdapter(object): + def __init__(self, sub, destination, client): + self._sub = sub + sub.set_message_handler(self._handle_message) + self._destination = destination + self._client = client + self._message_handler = lambda arg: None + + def setMessageHandler(self, handler): + self._message_handler = handler + + def send(self, data): + headers = { + "content-type": "application/json", + "reply-to": self._sub.destination, + } + self._client.send( + data, + self._destination, + headers, + ) + + def _handle_message(self, sub, frame): + self._message_handler((self, frame.body)) + + def close(self): + self._sub.unsubscribe() + + +def StompRpcClient( + stomp_client, + request_queue, + response_queue +): + sub_id = None + if request_queue == stomp.FAKE_SUB_ID: + sub_id = stomp.FAKE_SUB_ID + + return JsonRpcClient( + ClientRpcTransportAdapter( + stomp_client.subscribe(response_queue, sub_id=sub_id), + request_queue, + stomp_client, + ) + ) + + +def StompRpcServer( + bridge, + stomp_client, + request_queue, +): + server = JsonRpcServer( + bridge, + ) + + sub = stomp_client.subscribe( + request_queue, + message_handler=ServerRpcContextAdapter.subscription_handler(server) + ) + + server._sub_ = sub + return server diff --git a/tests/jsonRpcHelper.py b/tests/jsonRpcHelper.py index b3eb1fd..2fb7cfb 100644 --- a/tests/jsonRpcHelper.py +++ b/tests/jsonRpcHelper.py @@ -27,9 +27,17 @@ from itertools import product from M2Crypto import SSL from rpc.BindingXMLRPC import BindingXMLRPC, XmlDetector -from yajsonrpc.stompReactor import StompDetector +from yajsonrpc.betterAsyncore import ( + Reactor, +) +from yajsonrpc.stompReactor import ( + StompDetector, + StompRpcClient, +) +from yajsonrpc.stomp import ( + FAKE_SUB_ID, +) from protocoldetector import MultiProtocolAcceptor -from yajsonrpc import JsonRpcClient from rpc.BindingJsonRpc import BindingJsonRpc from sslhelper import DEAFAULT_SSL_CONTEXT @@ -60,7 +68,13 @@ @contextmanager def constructAcceptor(log, ssl, jsonBridge): sslctx = DEAFAULT_SSL_CONTEXT if ssl else None - acceptor = MultiProtocolAcceptor("127.0.0.1", 0, sslctx) + reactor = Reactor() + acceptor = MultiProtocolAcceptor( + reactor, + "127.0.0.1", + 0, + sslctx, + ) cif = FakeClientIf() xml_binding = BindingXMLRPC(cif, cif.log) @@ -73,8 +87,10 @@ stompDetector = StompDetector(json_binding) acceptor.add_detector(stompDetector) - thread = threading.Thread(target=acceptor.serve_forever, - name='Detector thread') + thread = threading.Thread( + target=reactor.process_requests, + name='Detector thread', + ) thread.setDaemon(True) thread.start() @@ -103,9 +119,12 @@ reactor = handler._reactor if not client: - client = lambda client_socket: ( - JsonRpcClient(reactor.createClient(client_socket)) - ) + def client(client_socket): + return StompRpcClient( + reactor.createClient(client_socket), + FAKE_SUB_ID, + FAKE_SUB_ID, + ) def clientFactory(): return client(create_socket( diff --git a/tests/jsonRpcTests.py b/tests/jsonRpcTests.py index 230b61a..3254393 100644 --- a/tests/jsonRpcTests.py +++ b/tests/jsonRpcTests.py @@ -83,12 +83,7 @@ @contextmanager def _client(self, clientFactory): client = clientFactory() - client.setTimeout(CALL_TIMEOUT) - client.connect() - try: - yield client - finally: - client.close() + yield client @MonkeyPatch(clientIF, 'getInstance', getInstance) @permutations(PERMUTATIONS) diff --git a/tests/miscTests.py b/tests/miscTests.py index 1fa572e..f98f539 100644 --- a/tests/miscTests.py +++ b/tests/miscTests.py @@ -991,6 +991,7 @@ openFds = openFdNum() self.testStdOut() gc.collect() + print gc.garbage self.assertEquals(len(gc.garbage), 0) self.assertEquals(openFdNum(), openFds) diff --git a/tests/protocoldetectorTests.py b/tests/protocoldetectorTests.py index ea64619..c64a1e2 100644 --- a/tests/protocoldetectorTests.py +++ b/tests/protocoldetectorTests.py @@ -23,9 +23,12 @@ import ssl import threading import time -from contextlib import contextmanager, closing +from contextlib import contextmanager import protocoldetector +from yajsonrpc.betterAsyncore import ( + Reactor, +) from vdsm import sslutils from sslhelper import KEY_FILE, CRT_FILE @@ -59,12 +62,13 @@ def run(): try: - assert client_socket.gettimeout() is None - rfile = client_socket.makefile('rb', -1) - with closing(rfile): - request = rfile.readline() - response = self.response(request) - client_socket.sendall(response) + request = "" + while "\n" not in request: + request += dispatcher.recv(1024) + + response = self.response(request) + client_socket.setblocking(1) + client_socket.sendall(response) finally: client_socket.shutdown(socket.SHUT_RDWR) client_socket.close() @@ -182,14 +186,14 @@ def check_slow_client(self, use_ssl): with self.connect(use_ssl) as client: - time.sleep(self.acceptor.CLEANUP_INTERVAL - self.GRACETIME) + time.sleep(self.acceptor._handshake_timeout - self.GRACETIME) data = "echo let me in\n" client.sendall(data) self.assertEqual(client.recv(self.BUFSIZE), data) def check_very_slow_client(self, use_ssl): with self.connect(use_ssl) as client: - time.sleep(self.acceptor.CLEANUP_INTERVAL * 2 + self.GRACETIME) + time.sleep(self.acceptor._handshake_timeout * 2 + self.GRACETIME) client.sendall("echo too slow probably\n") self.check_disconnected(client) @@ -204,12 +208,18 @@ # Helpers def start_acceptor(self, use_ssl): - self.acceptor = TestingAcceptor( - '127.0.0.1', 0, sslctx=self.SSLCTX if use_ssl else None) + self.reactor = Reactor() + self.acceptor = protocoldetector.MultiProtocolAcceptor( + self.reactor, + '127.0.0.1', + 0, + sslctx=self.SSLCTX if use_ssl else None + ) + self.acceptor._handshake_timeout = 1 self.acceptor.add_detector(Echo()) self.acceptor.add_detector(Uppercase()) - self.acceptor_address = self.acceptor._socket.getsockname() - t = threading.Thread(target=self.acceptor.serve_forever) + self.acceptor_address = self.acceptor._acceptor.socket.getsockname() + t = threading.Thread(target=self.reactor.process_requests) t.deamon = True t.start() diff --git a/vdsm.spec.in b/vdsm.spec.in index 0740530..845ddf0 100644 --- a/vdsm.spec.in +++ b/vdsm.spec.in @@ -1606,6 +1606,7 @@ %{python_sitelib}/yajsonrpc/stompReactor.py* %files infra +%{python_sitelib}/%{vdsm_name}/infra/eventfd/__init__.py* %{python_sitelib}/%{vdsm_name}/infra/filecontrol/__init__.py* %{python_sitelib}/%{vdsm_name}/infra/sigutils/__init__.py* %{python_sitelib}/%{vdsm_name}/infra/zombiereaper/__init__.py* diff --git a/vdsm/API.py b/vdsm/API.py index 42c471e..fdf2c62 100644 --- a/vdsm/API.py +++ b/vdsm/API.py @@ -1267,6 +1267,9 @@ def ping(self): "Ping the server. Useful for tests" updateTimestamp() + n = self._cif.create_notification("vdsm.ping") + n.emit(pinged=True) + return {'status': doneCode} def getRoute(self, ip): @@ -1286,7 +1289,6 @@ c = caps.get() c['netConfigDirty'] = str(self._cif._netConfigDirty) c = hooks.after_get_caps(c) - return {'status': doneCode, 'info': c} def getHardwareInfo(self): diff --git a/vdsm/clientIF.py b/vdsm/clientIF.py index d29f8c2..46e7f41 100644 --- a/vdsm/clientIF.py +++ b/vdsm/clientIF.py @@ -23,6 +23,8 @@ import time import threading import uuid +import socket +import json from functools import partial from weakref import proxy @@ -54,6 +56,35 @@ _glusterEnabled = True except ImportError: _glusterEnabled = False + +from yajsonrpc.betterAsyncore import ( + Reactor, +) + +from yajsonrpc.stompReactor import ( + StompClient, + StompRpcServer, +) + + +class Notification(object): + def __init__(self, event_id, cb): + self._event_id = event_id + self._cb = cb + + def emit(self, **kwargs): + notification = json.dumps( + { + 'jsonrpc': '2.0', + 'method': self._event_id, + 'params': kwargs, + } + ) + + self._cb( + notification, + self._event_id + ) class clientIF(object): @@ -87,6 +118,7 @@ self._generationID = str(uuid.uuid4()) self.mom = None self.bindings = {} + self._clients = [] if _glusterEnabled: self.gluster = gapi.GlusterApi(self, log) else: @@ -109,9 +141,11 @@ host = config.get('addresses', 'management_ip') port = config.getint('addresses', 'management_port') + self._createReactor() self._createAcceptor(host, port) self._prepareXMLRPCBinding() self._prepareJSONRPCBinding() + self._connectToBroker() except: self.log.error('failed to init clientIF, ' 'shutting down storage dispatcher') @@ -124,6 +158,21 @@ @property def ready(self): return (self.irs is None or self.irs.ready) and not self._recovery + + def create_notification(self, event_id): + prefix = config.get('addresses', 'topic_prefix') + return Notification( + prefix + event_id, + self._send_notification, + ) + + def _send_notification(self, message, destination): + if self._broker_client: + self._broker_client.send(message, destination) + + self.log.warn("@@@@@") + for client in self._clients[:]: + client.send(message, destination) def contEIOVms(self, sdUUID, isDomainStateValid): # This method is called everytime the onDomainStateChange @@ -158,10 +207,46 @@ cls._instance = clientIF(irs, log) return cls._instance + def _createReactor(self): + self._reactor = Reactor() + def _createAcceptor(self, host, port): sslctx = self._createSSLContext() - self._acceptor = MultiProtocolAcceptor(host, port, sslctx) + self._acceptor = MultiProtocolAcceptor( + self._reactor, + host, + port, + sslctx + ) + + def _connectToBroker(self): + broker_address = config.get('addresses', 'broker_address') + broker_port = config.getint('addresses', 'broker_port') + request_queues = config.get('addresses', 'request_queues') + self._broker_client = None + if broker_address: + sslctx = self._createSSLContext() + sock = socket.socket() + sock.connect((broker_address, broker_port)) + if sslctx: + sock = sslctx.wrapSocket(sock) + + stomp_client = StompClient(sock, self._reactor) + subs = [] + for destination in request_queues.split(","): + destination = destination.strip() + if destination: + subs.append( + StompRpcServer( + self.bindings['jsonrpc'].server, + stomp_client, + destination, + ) + ) + + self._subscriptions = subs + self._broker_client = stomp_client def _createSSLContext(self): sslctx = None @@ -200,7 +285,7 @@ 'Please make sure it is installed.') else: bridge = Bridge.DynamicBridge() - json_binding = BindingJsonRpc(bridge) + json_binding = BindingJsonRpc(bridge, self._clients) self.bindings['jsonrpc'] = json_binding stomp_detector = StompDetector(json_binding) self._acceptor.add_detector(stomp_detector) @@ -243,8 +328,10 @@ def start(self): for binding in self.bindings.values(): binding.start() - self.thread = threading.Thread(target=self._acceptor.serve_forever, - name='Detector thread') + self.thread = threading.Thread( + target=self._reactor.process_requests, + name='Reactor thread' + ) self.thread.setDaemon(True) self.thread.start() diff --git a/vdsm/protocoldetector.py b/vdsm/protocoldetector.py index 76d5a00..20a0b41 100644 --- a/vdsm/protocoldetector.py +++ b/vdsm/protocoldetector.py @@ -23,11 +23,9 @@ from M2Crypto import SSL -from vdsm.utils import traceback import vdsm.infra.filecontrol as filecontrol from yajsonrpc.betterAsyncore import ( Dispatcher, - Reactor, ) from vdsm.utils import ( @@ -48,8 +46,6 @@ filecontrol.set_close_on_exec(server_socket.fileno()) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind(addr[0][4]) - - server_socket.setblocking(0) return server_socket @@ -75,6 +71,7 @@ except socket.error: pass + client.setblocking(0) self.log.info("Accepting connection from %s:%d", *client.getpeername()) self._dispatcher_factory(client) @@ -97,8 +94,16 @@ sock = dispatcher.socket try: data = sock.recv(self._required_size, socket.MSG_PEEK) - except socket.error: + except socket.error, why: + if why.args[0] == socket.EWOULDBLOCK: + return + dispatcher.handle_error() + return + + if monotonic_time() > self._give_up_at: + self.log.debug("Timed out while waiting for data") + dispatcher.close() return if len(data) < self._required_size: @@ -210,13 +215,15 @@ def __init__( self, + reactor, host, port, sslctx=None, ssl_hanshake_timeout=SSL_HANDSHAKE_TIMEOUT, ): self._sslctx = sslctx - self._reactor = Reactor() + self._reactor = reactor + sock = _create_socket(host, port) self._host, self._port = sock.getsockname() self.log.info("Listening at %s:%d", self._host, self._port) @@ -261,13 +268,6 @@ ) return dispatcher - - @traceback(on=log.name) - def serve_forever(self): - self.log.debug("Running") - required_size = max(h.REQUIRED_SIZE for h in self._handlers) - self.log.debug("Using required_size=%d", required_size) - self._reactor.process_requests() def add_detector(self, detector): self.log.debug("Adding detector %s", detector) diff --git a/vdsm/rpc/BindingJsonRpc.py b/vdsm/rpc/BindingJsonRpc.py index a5b5b4a..eeea8ba 100644 --- a/vdsm/rpc/BindingJsonRpc.py +++ b/vdsm/rpc/BindingJsonRpc.py @@ -29,14 +29,21 @@ class BindingJsonRpc(object): log = logging.getLogger('BindingJsonRpc') - def __init__(self, bridge): + def __init__(self, bridge, clients=None): self._server = JsonRpcServer(bridge, _simpleThreadFactory) + self._clients = clients self._reactors = [] + + @property + def server(self): + return self._server def add_socket(self, reactor, client_socket): reactor.createListener(client_socket, self._onAccept) def _onAccept(self, listener, client): + if self._clients is not None: + self._clients.append(client) client.setMessageHandler(self._server.queueRequest) def createStompReactor(self): -- To view, visit http://gerrit.ovirt.org/38069 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: Id27b5ca1773139932eb5cb16921d5abec4991c5e Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Saggi Mizrahi <smizrahi(a)redhat.com>

9 102

Change in vdsm[master]: gluster: fix volume name parsing in getVmVolumeInfo
by Federico Simoncelli 10 Jun '15

10 Jun '15

Federico Simoncelli has uploaded a new change for review. Change subject: gluster: fix volume name parsing in getVmVolumeInfo ...................................................................... gluster: fix volume name parsing in getVmVolumeInfo It is permitted to use slashes '/' in gluster mount strings even though they're prohibited in the volume name. In order to recognize and strip them out we should use getRealPath instead of getRemotePath that translates slashes into permitted underscores. It is now allowed to mount gluster volumes with slashes (serv:/volume) and correctly identifying the volume name and its information. Change-Id: Icfdbc573ec2aec31b78323253f8178425a07302c Signed-off-by: Federico Simoncelli <fsimonce(a)redhat.com> --- M vdsm/storage/glusterVolume.py 1 file changed, 3 insertions(+), 4 deletions(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/37/36237/1 diff --git a/vdsm/storage/glusterVolume.py b/vdsm/storage/glusterVolume.py index 647b1ec..8f701ba 100644 --- a/vdsm/storage/glusterVolume.py +++ b/vdsm/storage/glusterVolume.py @@ -19,10 +19,9 @@ """ Send info to represent Gluster volume as a network block device """ - rpath = sdCache.produce(self.sdUUID).getRemotePath() - rpath_list = rpath.rsplit(":", 1) - volfileServer = rpath_list[0] - volname = rpath_list[1] + rpath = sdCache.produce(self.sdUUID).getRealPath() + volfileServer, volname = rpath.rsplit(":", 1) + volname = volname.strip('/') # Volume transport to Libvirt transport mapping VOLUME_TRANS_MAP = {'TCP': 'tcp', 'RDMA': 'rdma'} -- To view, visit http://gerrit.ovirt.org/36237 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: Icfdbc573ec2aec31b78323253f8178425a07302c Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Federico Simoncelli <fsimonce(a)redhat.com>

8 13

Change in vdsm[master]: vdsm: implement NUMA aware KSM policy
by dudi＠redhat.com 10 Jun '15

10 Jun '15

Dudi Maroshi has uploaded a new change for review. Change subject: vdsm: implement NUMA aware KSM policy ...................................................................... vdsm: implement NUMA aware KSM policy Add monitored parameter ksm_merge_across_nodes to: 1. momIf.py class - append to status report 2. superVdsmServer file - enable writing to the kernel flag Update policy logic (a simple state machine) to 03-ksm.policy merge_across_nodes flag allowed to be changed only after unmerging all pages. This unmerge completion is tested on pages_shared flag == 0. Update policy parameter ksmMergeAcrossNodes to 00-defines.policy Change-Id: Iec607f9ef3284c1448bfc2831d125fc4d81b28d2 Related-to: https://bugzilla.redhat.com/show_bug.cgi?id=840114 Signed-off-by: Dudi Maroshi <dudi(a)redhat.com> --- M vdsm/mom.d/00-defines.policy M vdsm/mom.d/03-ksm.policy M vdsm/momIF.py M vdsm/supervdsmServer 4 files changed, 73 insertions(+), 53 deletions(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/29/40129/1 diff --git a/vdsm/mom.d/00-defines.policy b/vdsm/mom.d/00-defines.policy index faa6d90..718cf04 100644 --- a/vdsm/mom.d/00-defines.policy +++ b/vdsm/mom.d/00-defines.policy @@ -4,6 +4,7 @@ (defvar True 1) # Define variables for configurable options here +(defvar ksmMergeAcrossNodes 1) (defvar ksmEnabled 1) (defvar balloonEnabled 0) (defvar cpuTuneEnabled 1) diff --git a/vdsm/mom.d/03-ksm.policy b/vdsm/mom.d/03-ksm.policy index 5a3b1bd..9706e5d 100644 --- a/vdsm/mom.d/03-ksm.policy +++ b/vdsm/mom.d/03-ksm.policy @@ -22,6 +22,10 @@ # KSM will be started to try and free up some memory. (defvar ksm_free_percent 0.20) +# KSM service can unmerged shared pages (ksm_run = 2) +# When this happens the KSM is affectivley non responsive +(defvar ksm_umerging_pages 0) + ### Helper functions (def change_npages (delta) { @@ -31,6 +35,43 @@ (Host.Control "ksm_pages_to_scan" newval) }) +(def enable_ksmMergeAcrossNodes() +{ + (debug "entry: enable_ksmMergeAcrossNodes") + (debug Host.ksm_run "=ksm_run") + (debug Host.ksm_merge_across_nodes "=ksm_merge_across_nodes") + (debug ksmMergeAcrossNodes "=ksmMergeAcrossNodes") + (if (!= Host.ksm_merge_across_nodes ksmMergeAcrossNodes){ + (debug "ksm_merge_across_nodes policy change") + (if (== Host.ksm_run 0){ + (debug "ksm_run == 0 : Update ksm_merge_across_nodes") + (Host.Control "ksm_merge_across_nodes" ksmMergeAcrossNodes) + (debug "ksm_merge_across_nodes configured")} + 6) + (if (== Host.ksm_run 1){ + (debug "ksm_run == 1 : unmerge KSM pages") + (setq ksm_umerging_pages 1) + (Host.Control "ksm_run" 2) + (debug "ksm_run unmerge")} + 7) + (if (== Host.ksm_run 2){ + (debug "ksm_run == 2 : if all pages unmerged, update ksm_merge_across_nodes. Rest KSM") + (Host.Control "ksm_run" 2) + (if (== Host.ksm_pages_shared 0) { + (debug "ksm_pages_shared == 0") + (Host.Control "ksm_merge_across_nodes" ksmMergeAcrossNodes) + (Host.Control "ksm_run" 0) + (setq ksm_umerging_pages 0) + } + { #else + (debug "ksm_pages_shared != 0 : continue with unmerging") + (setq ksm_umerging_pages 1) + }) + (debug "ksm_merge_across_nodes configured")} + 9) + } 10) + (debug "exit: enable_ksmMergeAcrossNodes") +}) ### Main Script # Methodology: Since running KSM does incur some overhead, try to run it only # when necessary. If the amount of committed KSM shareable memory is high or if @@ -40,16 +81,19 @@ (defvar ksm_pressure_threshold (* Host.mem_available ksm_free_percent)) (defvar ksm_committed Host.ksm_shareable) - -(if (or (and (< (+ ksm_pressure_threshold ksm_committed) Host.mem_available) - (> (Host.StatAvg "mem_free") ksm_pressure_threshold)) (not ksmEnabled)) - (Host.Control "ksm_run" 0) - { # else - (Host.Control "ksm_run" 1) - (Host.Control "ksm_sleep_millisecs" - (/ (* ksm_sleep_ms_baseline 16777216) Host.mem_available)) - (if (< (Host.StatAvg "mem_free") ksm_pressure_threshold) - (change_npages ksm_pages_boost) - (change_npages ksm_pages_decay)) - } -) +(enable_ksmMergeAcrossNodes) +(if (== ksm_umerging_pages 0) + (if (or (and (< (+ ksm_pressure_threshold ksm_committed) Host.mem_available) + (> (Host.StatAvg "mem_free") ksm_pressure_threshold)) + (not ksmEnabled)) + (Host.Control "ksm_run" 0) + { # else + (Host.Control "ksm_run" 1) + (Host.Control "ksm_sleep_millisecs" + (/ (* ksm_sleep_ms_baseline 16777216) Host.mem_available)) + (if (< (Host.StatAvg "mem_free") ksm_pressure_threshold) + (change_npages ksm_pages_boost) + (change_npages ksm_pages_decay)) + } + ) +0) \ No newline at end of file diff --git a/vdsm/momIF.py b/vdsm/momIF.py index d70d826..e6481c2 100644 --- a/vdsm/momIF.py +++ b/vdsm/momIF.py @@ -56,6 +56,7 @@ stats = self._mom.getStatistics()['host'] ret = {} ret['ksmState'] = bool(stats['ksm_run']) + ret['ksmMergeAcrossNodes'] = bool(stats['ksm_merge_across_nodes']) ret['ksmPages'] = stats['ksm_pages_to_scan'] ret['memShared'] = stats['ksm_pages_sharing'] * PAGE_SIZE_BYTES ret['memShared'] /= Mbytes diff --git a/vdsm/supervdsmServer b/vdsm/supervdsmServer index 3cdbf71..43ed1e6 100755 --- a/vdsm/supervdsmServer +++ b/vdsm/supervdsmServer @@ -308,25 +308,14 @@ of /dev/vfio/$iommu_group to qemu:qemu. This method should be called when detaching a device from the host. """ - rule_file = _UDEV_RULE_FILE_NAME_VFIO % iommu_group - - if not os.path.isfile(rule_file): - # If the file exists, different device from the same group has - # already been detached and we therefore can skip overwriting the - # file. Also, this file should only be created/removed via the - # means of supervdsm. - - rule = ('KERNEL=="{}", SUBSYSTEM=="vfio" RUN+="{} {}:{} ' - '/dev/vfio/{}"').format(iommu_group, EXT_CHOWN, - QEMU_PROCESS_USER, - QEMU_PROCESS_GROUP, - iommu_group) - - with open(rule_file, "w") as rf: - self.log.debug("Creating rule %s: %r", rule_file, rule) - rf.write(rule) - - self.udevTrigger(iommu_group) + ruleFile = _UDEV_RULE_FILE_NAME_VFIO % iommu_group + rule = ('KERNEL=="{}", SUBSYSTEM=="vfio" RUN+="{} {}:{} ' + '/dev/vfio/{}"').format(iommu_group, EXT_CHOWN, + QEMU_PROCESS_USER, QEMU_PROCESS_GROUP, + iommu_group) + with open(ruleFile, "w") as rf: + self.log.debug("Creating rule %s: %r", ruleFile, rule) + rf.write(rule) @logDecorator def rmAppropriateIommuGroup(self, iommu_group): @@ -334,26 +323,10 @@ Remove udev rule in /etc/udev/rules.d/ created by vfioAppropriateDevice. """ - rule_file = os.path.join(_UDEV_RULE_FILE_DIR, _UDEV_RULE_FILE_PREFIX + - "iommu_group_" + iommu_group + - _UDEV_RULE_FILE_EXT) - error = False - - try: - os.remove(rule_file) - except OSError as e: - if e.errno == errno.ENOENT: - # OSError with ENOENT errno here means that the rule file does - # not exist - this is expected when multiple devices in one - # iommu group were passed through. - error = True - else: - raise - else: - self.log.debug("Removing rule %s", rule_file) - - if not error: - self.udevTrigger(iommu_group) + rule = os.path.join(_UDEV_RULE_FILE_DIR, _UDEV_RULE_FILE_PREFIX + + "iommu_group_" + iommu_group + _UDEV_RULE_FILE_EXT) + self.log.debug("Removing rule %s", rule) + os.remove(rule) @logDecorator def ksmTune(self, tuningParams): @@ -362,7 +335,8 @@ when it's lauched by vdsm. So it needs supervdsm's assistance to tune KSM's parameters. ''' - KSM_PARAMS = {'run': 3, 'sleep_millisecs': 0x100000000, + KSM_PARAMS = {'run': 3, 'merge_across_nodes': 3, + 'sleep_millisecs': 0x100000000, 'pages_to_scan': 0x100000000} for (k, v) in tuningParams.iteritems(): if k not in KSM_PARAMS.iterkeys(): -- To view, visit https://gerrit.ovirt.org/40129 To unsubscribe, visit https://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: Iec607f9ef3284c1448bfc2831d125fc4d81b28d2 Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Dudi Maroshi <dudi(a)redhat.com>

7 38

Change in vdsm[ovirt-3.5]: network: Force blocking DHCP when restoring networks
by ibarkan＠redhat.com 09 Jun '15

09 Jun '15

Ido Barkan has uploaded a new change for review. Change subject: network: Force blocking DHCP when restoring networks ...................................................................... network: Force blocking DHCP when restoring networks This is done right before restoring the network configuration, and before calling setupNetworks. It forces the configurator to wait for an IP address to be configured on the devices before restoration is completed. This prevents VDSM from possibly report missing IP addresses on interfaces that had been restored right before it was started. Change-Id: Ibc8b84a82794ac97eba6d34ec9d54430e387b659 Signed-off-by: Ido Barkan <ibarkan(a)redhat.com> --- M lib/vdsm/utils.py M tests/functional/dhcp.py M tests/functional/firewall.py M tests/functional/networkTests.py M vdsm/vdsm-restore-net-config 5 files changed, 179 insertions(+), 37 deletions(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/29/39529/1 diff --git a/lib/vdsm/utils.py b/lib/vdsm/utils.py index 4619b74..8b8e6ec 100644 --- a/lib/vdsm/utils.py +++ b/lib/vdsm/utils.py @@ -27,6 +27,7 @@ Contains a reverse dictionary pointing from error string to its error code. """ from collections import namedtuple, deque +from contextlib import contextmanager from fnmatch import fnmatch from SimpleXMLRPCServer import SimpleXMLRPCRequestHandler from SimpleXMLRPCServer import SimpleXMLRPCDispatcher @@ -1289,6 +1290,15 @@ self._finally.insert(0, (func, args, kwargs)) +@contextmanager +def running(runnable): + runnable.start() + try: + yield runnable + finally: + runnable.stop() + + def get_selinux_enforce_mode(): """ Returns the SELinux mode as reported by kernel. diff --git a/tests/functional/dhcp.py b/tests/functional/dhcp.py index 28d1220..b913061 100644 --- a/tests/functional/dhcp.py +++ b/tests/functional/dhcp.py @@ -19,7 +19,9 @@ import logging import os -from time import sleep +from signal import SIGKILL, SIGTERM +from time import sleep, time +from errno import ENOENT, ESRCH from nose.plugins.skip import SkipTest @@ -32,6 +34,7 @@ _NM_CLI_BINARY = CommandPath('nmcli', '/usr/bin/nmcli') _START_CHECK_TIMEOUT = 0.5 _DHCLIENT_TIMEOUT = 10 +_WAIT_FOR_STOP_TIMEOUT = 2 class DhcpError(Exception): @@ -42,52 +45,117 @@ def __init__(self): self.proc = None - def start(self, interface, dhcpRangeFrom, dhcpRangeTo, router=None): - # --dhcp-option=3,<router> advertise specific router (can be None) - # -O 6 don't reply with any DNS servers either - # -d do not daemonize and log to stderr - # -p 0 disable all the dnsmasq dns functionality - self.proc = execCmd([ + def start(self, interface, dhcp_range_from, dhcp_range_to, + dhcpv6_range_from=None, dhcpv6_range_to=None, router=None, + bind_dynamic=True): + # -p 0 don't act as a DNS server + # --dhcp-option=3,<router> advertise a specific gateway (or None) + # --dhcp-option=6 don't reply with any DNS servers + # -d don't daemonize and log to stderr + # --bind-dynamic bind only the testing veth iface + # (a better, and quiet, version of --bind-interfaces, but not on EL6) + command = [ _DNSMASQ_BINARY.cmd, '--dhcp-authoritative', - '-p', '0', '--dhcp-range=' + dhcpRangeFrom + ',' + - dhcpRangeTo + ',2m', - '--dhcp-option=3' + ',%s' % (router,) if router else '', - '-O', '6', + '-p', '0', + '--dhcp-range={0},{1},2m'.format(dhcp_range_from, dhcp_range_to), + '--dhcp-option=3' + (',{0}'.format(router) if router else ''), + '--dhcp-option=6', '-i', interface, '-I', 'lo', '-d', - '--bind-interfaces'], sync=False) + '--bind-dynamic' if bind_dynamic else '--bind-interfaces'] + if dhcpv6_range_from and dhcpv6_range_to: + command += ['--dhcp-range={0},{1},2m'.format(dhcpv6_range_from, + dhcpv6_range_to)] + self.proc = execCmd(command, sync=False) sleep(_START_CHECK_TIMEOUT) if self.proc.returncode: - raise DhcpError('Failed to start dnsmasq DHCP server.' + - ''.join(self.proc.stderr)) + raise DhcpError('Failed to start dnsmasq DHCP server.\n%s\n%s' % + (''.join(self.proc.stderr), ' '.join(command))) def stop(self): self.proc.kill() logging.debug(''.join(self.proc.stderr)) -def runDhclient(interface, tmpDir, dateFormat): +class ProcessCannotBeKilled(Exception): + pass + + +class DhclientRunner(object): """On the interface, dhclient is run to obtain a DHCP lease. - In the working directory (tmpDir), which is managed by the caller, - a lease file is created and a path to it is returned. - dhclient accepts the following dateFormats: 'default' and 'local'. + In the working directory (tmp_dir), which is managed by the caller. + dhclient accepts the following date_formats: 'default' and 'local'. """ - confFile = os.path.join(tmpDir, 'test.conf') - pidFile = os.path.join(tmpDir, 'test.pid') - leaseFile = os.path.join(tmpDir, 'test.lease') + def __init__(self, interface, family, tmp_dir, date_format, + default_route=False): + self._interface = interface + self._family = family + self._date_format = date_format + self._conf_file = os.path.join(tmp_dir, 'test.conf') + self._pid_file = os.path.join(tmp_dir, 'test.pid') + self.pid = None + self.lease_file = os.path.join(tmp_dir, 'test.lease') + cmd = [_DHCLIENT_BINARY.cmd, '-' + str(family), '-1', '-v', + '-timeout', str(_DHCLIENT_TIMEOUT), '-cf', self._conf_file, + '-pf', self._pid_file, '-lf', self.lease_file + ] + if not default_route: + # Instruct Fedora/EL's dhclient-script not to set gateway on iface + cmd += ['-e', 'DEFROUTE=no'] + self._cmd = cmd + [self._interface] - with open(confFile, 'w') as f: - f.write('db-time-format {0};'.format(dateFormat)) + def _create_conf(self): + with open(self._conf_file, 'w') as f: + if self._date_format: + f.write('db-time-format {0};'.format(self._date_format)) - rc, out, err = execCmd([_DHCLIENT_BINARY.cmd, '-lf', leaseFile, - '-pf', pidFile, '-timeout', str(_DHCLIENT_TIMEOUT), - '-1', '-cf', confFile, interface]) + def start(self): + self._create_conf() + rc, out, err = execCmd(self._cmd) - if rc: # == 2 - logging.debug(''.join(err)) - raise DhcpError('dhclient failed to obtain a lease: %d', rc) + if rc: # == 2 + logging.debug(''.join(err)) + raise DhcpError('dhclient failed to obtain a lease: %d', rc) - return leaseFile + with open(self._pid_file) as pid_file: + self.pid = int(pid_file.readline()) + + def stop(self): + if self._try_kill(SIGTERM): + return + if self._try_kill(SIGKILL): + return + raise ProcessCannotBeKilled('cmd=%s, pid=%s' % (' '.join(self._cmd), + self.pid)) + + def _try_kill(self, signal, timeout=_WAIT_FOR_STOP_TIMEOUT): + now = time() + deadline = now + timeout + while now < deadline: + try: + os.kill(self.pid, signal) + except OSError as err: + if err.errno != ESRCH: + raise + return True # no such process + + sleep(0.5) + if not self._is_running(): + return True + now = time() + + return False + + def _is_running(self): + executable_link = '/proc/{0}/exe'.format(self.pid) + try: + executable = os.readlink(executable_link) + except OSError as err: + if err.errno == ENOENT: + return False # no such pid + else: + raise + return executable == _DHCLIENT_BINARY.cmd def addNMplaceholderConnection(interface, connection): diff --git a/tests/functional/firewall.py b/tests/functional/firewall.py index ee7bb9f..b703603 100644 --- a/tests/functional/firewall.py +++ b/tests/functional/firewall.py @@ -48,10 +48,9 @@ _execCmdChecker([_FIREWALLD_BINARY.cmd, '--zone=trusted', '--change-interface=' + veth]) else: - raise SkipTest('No firewall service detected.') + logging.info('No firewall service detected.') except FirewallError as e: - raise SkipTest('Failed to allow dhcp traffic in firewall because of ' - '%s' % e) + raise SkipTest('Failed to allow DHCP traffic in firewall: %s' % e) def stopAllowingDhcp(veth): diff --git a/tests/functional/networkTests.py b/tests/functional/networkTests.py index 075fac4..ef02d3f 100644 --- a/tests/functional/networkTests.py +++ b/tests/functional/networkTests.py @@ -41,7 +41,7 @@ getLinks, routeShowTable) from vdsm.constants import EXT_BRCTL, EXT_IFUP -from vdsm.utils import RollbackContext, execCmd +from vdsm.utils import RollbackContext, execCmd, running from vdsm.netinfo import (bridges, operstate, prefix2netmask, getRouteDeviceTo, getDhclientIfaces, BONDING_SLAVES, BONDING_MASTERS, NET_CONF_PREF) @@ -63,6 +63,8 @@ IP_GATEWAY = '240.0.0.254' DHCP_RANGE_FROM = '240.0.0.10' DHCP_RANGE_TO = '240.0.0.100' +DHCPv6_RANGE_FROM = 'fdb3:84e5:4ff4:55e3::a' +DHCPv6_RANGE_TO = 'fdb3:84e5:4ff4:55e3::64' CUSTOM_PROPS = {'linux': 'rules', 'vdsm': 'as well'} IPv6_ADDRESS = 'fdb3:84e5:4ff4:55e3::1/64' @@ -91,12 +93,18 @@ @contextmanager -def dnsmasqDhcp(interface): - """Manages the life cycle of dnsmasq as a DHCP server.""" +def dnsmasqDhcp(interface, el6=False): + """Manages the life cycle of dnsmasq as a DHCP server. + + 'el6' parameter serves to disable DHCPv6 functionality on EL6 where it is + not supported, and avoids warning on --bind-interfaces switch elsewhere.""" dhcpServer = dhcp.Dnsmasq() try: + dhcpv6_range_from, dhcpv6_range_to = ( + (None, None) if el6 else (DHCPv6_RANGE_FROM, DHCPv6_RANGE_TO)) dhcpServer.start(interface, DHCP_RANGE_FROM, DHCP_RANGE_TO, - router=IP_GATEWAY) + dhcpv6_range_from, dhcpv6_range_to, router=IP_GATEWAY, + bind_dynamic=not el6) except dhcp.DhcpError as e: raise SkipTest(e) @@ -176,6 +184,12 @@ if state != originalState: raise OperStateChangedError('%s operstate changed: %s -> %r' % (device, originalState, changes)) + + +def _system_is_el6(): + # REQUIRED_FOR: el6 + return (caps.getos() in (caps.OSName.RHEVH, caps.OSName.RHEL) + and caps.osversion()['version'].startswith('6')) @expandPermutations @@ -1483,6 +1497,41 @@ self.vdsm_net.save_config() @cleanupNet + @RequireVethMod + @ValidateRunningAsRoot + def testRestoreToBlockingDHCP(self): + def _get_blocking_dhcp(net_name): + self.vdsm_net.refreshNetinfo() + return self.vdsm_net.config.networks[net_name].get('blockingdhcp') + + with vethIf() as (server, client): + veth.setIP(server, IP_ADDRESS, IP_CIDR) + veth.setLinkUp(server) + dhcp_async_net = {'nic': client, 'bridged': False, + 'bootproto': 'dhcp', 'blockingdhcp': False} + status, msg = self.vdsm_net.setupNetworks( + {NETWORK_NAME: dhcp_async_net}, {}, NOCHK) + self.assertEquals(status, SUCCESS, msg) + + self.assertNetworkExists(NETWORK_NAME) + self.assertFalse(_get_blocking_dhcp(NETWORK_NAME)) + + self.vdsm_net.save_config() + + with dnsmasqDhcp(server, _system_is_el6()): + with namedTemporaryDir(dir='/var/lib/dhclient') as dhdir: + dhclient_runner = dhcp.DhclientRunner(client, 4, dhdir, + 'default') + with running(dhclient_runner): + self.vdsm_net.restoreNetConfig() + self.assertTrue(_get_blocking_dhcp(NETWORK_NAME)) + + # cleanup + status, msg = self.vdsm_net.setupNetworks( + {NETWORK_NAME: {'remove': True}}, {}, NOCHK) + self.assertEquals(status, SUCCESS, msg) + + @cleanupNet @permutations([[True], [False]]) @RequireDummyMod @ValidateRunningAsRoot diff --git a/vdsm/vdsm-restore-net-config b/vdsm/vdsm-restore-net-config index d7cb8c7..515a03a 100755 --- a/vdsm/vdsm-restore-net-config +++ b/vdsm/vdsm-restore-net-config @@ -73,11 +73,27 @@ persistentConfig = PersistentConfig() nets, bonds = _filter_nets_bonds(persistentConfig.networks, persistentConfig.bonds) + _convert_to_blocking_dhcp(nets) logging.debug('Calling setupNetworks with networks (%s) and bond (%s).', nets, bonds) setupNetworks(nets, bonds, connectivityCheck=False, _inRollback=True) +def _convert_to_blocking_dhcp(networks): + """ + This function changes DHCP configuration, if present, to be blocking. + + This is done right before restoring the network configuration, and forces + the configurator to wait for an IP address to be configured on the devices + before restoration is completed. This prevents VDSM to possibly report + missing IP address on interfaces that had been restored right before it was + started. + """ + for net, net_attr in networks.iteritems(): + if net_attr.get('bootproto') == 'dhcp': + net_attr['blockingdhcp'] = True + + def _filter_nets_bonds(nets, bonds): """Returns only nets and bonds that can be configured with the devices present in the system""" -- To view, visit https://gerrit.ovirt.org/39529 To unsubscribe, visit https://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: Ibc8b84a82794ac97eba6d34ec9d54430e387b659 Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: ovirt-3.5 Gerrit-Owner: Ido Barkan <ibarkan(a)redhat.com>

4 13

Change in vdsm[master]: network: wait for all udev events after configuring SRIOV
by ibarkan＠redhat.com 09 Jun '15

09 Jun '15

Ido Barkan has uploaded a new change for review. Change subject: network: wait for all udev events after configuring SRIOV ...................................................................... network: wait for all udev events after configuring SRIOV After changing the number of virtual functions on an SRIOV device udev asynchronously adds them and then renames them from 'eth0' according to it's naming policy. Since that process is asynchronous and might take time, we have to wait for it to finish since subsequent reporting of hardware by vdsm must return a stable state. Change-Id: I04d63ca35aab585051c3d0a5f2652dbf2f91b080 Signed-off-by: Ido Barkan <ibarkan(a)redhat.com> --- M vdsm/network/api.py 1 file changed, 6 insertions(+), 0 deletions(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/00/40400/1 diff --git a/vdsm/network/api.py b/vdsm/network/api.py index 2d31adc..eae43b3 100755 --- a/vdsm/network/api.py +++ b/vdsm/network/api.py @@ -32,6 +32,7 @@ from vdsm import netinfo from vdsm import utils from vdsm import ipwrapper +from vdsm import udevadm from .configurators import libvirt from .errors import ConfigNetworkError @@ -538,6 +539,11 @@ # https://www.kernel.org/doc/Documentation/PCI/pci-iov-howto.txt f.write('0') f.write(str(num_vfs)) + # This is a blocking call that should wait for all udev events to be + # handled. Since those include renaming devices, we should wait until + # udev finishes in order to report a stable hardware state in + # subsequent engine calls. + udevadm.settle(timeout=10) def _persist_num_vfs(device_name, numvfs): -- To view, visit https://gerrit.ovirt.org/40400 To unsubscribe, visit https://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: I04d63ca35aab585051c3d0a5f2652dbf2f91b080 Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Ido Barkan <ibarkan(a)redhat.com>

6 62

Change in vdsm[master]: network: restore sriov devices number of vfs
by ibarkan＠redhat.com 09 Jun '15

09 Jun '15

Ido Barkan has uploaded a new change for review. Change subject: network: restore sriov devices number of vfs ...................................................................... network: restore sriov devices number of vfs vdsm-restore-net-config will attempt to restore the number of vfs from the last successful change. If no sriov devices are found the function only tries to collect garbage. This is done before network restoration so that networks can be configured based on the restored vfs. The patch also introduces a 'safe' changing of number of vfs of an sriov card which also checks if the connectivity to the engine was lost as a consequence. Change-Id: I76b898019840ffe65939ffad4a1e98829ad3c887 Signed-off-by: Ido Barkan <ibarkan(a)redhat.com> --- M vdsm/network/api.py M vdsm/vdsm-restore-net-config 2 files changed, 85 insertions(+), 5 deletions(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/88/40088/1 diff --git a/vdsm/network/api.py b/vdsm/network/api.py index bc2dbd2..07f2640 100755 --- a/vdsm/network/api.py +++ b/vdsm/network/api.py @@ -40,6 +40,7 @@ import hooks # TODO: Turn into parent package import when vdsm is a package CONNECTIVITY_TIMEOUT_DEFAULT = 4 +_SYSFS_SRIOV_NUMVFS = '/sys/bus/pci/devices/{}/sriov_numvfs' def _getConfiguratorClass(): @@ -541,17 +542,51 @@ 'connectivity check failed') -def changeNumvfs(device_name, numvfs): - with open('/sys/bus/pci/devices/{}/sriov_numvfs'.format( - device_name), 'w', 0) as f: +def _update_num_vfs(device_name, num_vfs): + with open(_SYSFS_SRIOV_NUMVFS.format(device_name), 'w', 0) as f: # Zero needs to be written first in order to remove previous VFs. # Trying to just write the number (if n > 0 VF's existed before) # results in 'write error: Device or resource busy' # https://www.kernel.org/doc/Documentation/PCI/pci-iov-howto.txt f.write('0') + f.write(str(num_vfs)) + + +def _persist_num_vfs(device_name, numvfs): + dir_path = os.path.join(constants.P_VDSM_LIB, 'virtual_functions') + try: + os.makedirs(dir_path) + except OSError as ose: + if errno.EEXIST != ose.errno: + raise + with open(dir_path, device_name, 'w') as f: f.write(str(numvfs)) +def changeNumvfs(device_name, num_vfs): + """Change number of virtual functions of a device in a 'safe' way. If the + connectivity to engine is lost as a consequence, the value is restored. + Since calling this verb is currently blocked by engine if this device is + already in use, the logic does not attempt to do anything further, such as + restoring possibly lost networks. + The persistence is stored in a place which is writable also in ovirt-node. + Garbage is will be collected during restoration. + """ + original_value = _get_num_vfs(device_name) + _update_num_vfs(device_name, num_vfs) + if not clientSeen(): + _update_num_vfs(device_name, original_value) + raise ConfigNetworkError( + ne.ERR_LOST_CONNECTION, 'connectivity check failed') + + _persist_num_vfs(device_name, num_vfs) + + +def _get_num_vfs(device_name): + with open(_SYSFS_SRIOV_NUMVFS.format(device_name), 'r', 0) as f: + return int(f.read()) + + def _validateNetworkSetup(networks, bondings): for network, networkAttrs in networks.iteritems(): if networkAttrs.get('remove', False): diff --git a/vdsm/vdsm-restore-net-config b/vdsm/vdsm-restore-net-config index 21d6592..afe4abd 100755 --- a/vdsm/vdsm-restore-net-config +++ b/vdsm/vdsm-restore-net-config @@ -27,21 +27,64 @@ from vdsm.config import config from vdsm import netinfo -from vdsm.constants import P_VDSM_RUN +from vdsm.constants import P_VDSM_RUN, P_VDSM_LIB +import hostdev # Ifcfg persistence restoration from network.configurators import ifcfg # Unified persistence restoration -from network.api import setupNetworks +from network.api import setupNetworks, changeNumvfs from network import configurators from vdsm.netconfpersistence import RunningConfig, PersistentConfig import pkgutil _NETS_RESTORED_MARK = os.path.join(P_VDSM_RUN, 'nets_restored') +_VIRTUAL_FUNCTIONS_PATH = os.path.join(P_VDSM_LIB, 'virtual_functions') + + +def _get_sriov_devices(): + devices = hostdev.list_by_caps() + sriov_pci_addrs = [addr for addr, device in devices.iteritems() + if 'totalvfs' in device['params']] + net_devs = [device for device in devices.itervalues() + if device['params']['capability'] == 'net'] + return [device['params']['interface'] for device in net_devs + if device['params']['parent'] in sriov_pci_addrs] + + +def _get_persisted_numvfs(existing_sriov_devices): + num_vfs_by_device = {} + missing_devices = [] + for file_name in os.listdir(_VIRTUAL_FUNCTIONS_PATH): + if file_name not in existing_sriov_devices: + missing_devices.append(file_name) + else: + with open(os.path.join( + _VIRTUAL_FUNCTIONS_PATH, file_name), 'r') as f: + num_vfs_by_device[file_name] = int(f.read().strip()) + + return num_vfs_by_device, missing_devices + + +def _remove_missing_devices_persistence(missing_devices): + for file_name in missing_devices: + os.remove(os.path.join(_VIRTUAL_FUNCTIONS_PATH, file_name)) + + +def _restore_sriov_num_vfs(): + sriov_devices = _get_sriov_devices() + persisted_num_vfs, missing_devices = _get_persisted_numvfs(sriov_devices) + + for device in sriov_devices: + changeNumvfs(device, persisted_num_vfs[device]) + + _remove_missing_devices_persistence(missing_devices) def ifcfg_restoration(): + _restore_sriov_num_vfs() + configWriter = ifcfg.ConfigWriter() configWriter.restorePersistentBackup() @@ -51,6 +94,8 @@ Builds a setupNetworks command from the persistent configuration to set it as running configuration. """ + _restore_sriov_num_vfs() + runningConfig = RunningConfig() removeNetworks = {} removeBonds = {} -- To view, visit https://gerrit.ovirt.org/40088 To unsubscribe, visit https://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: I76b898019840ffe65939ffad4a1e98829ad3c887 Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Ido Barkan <ibarkan(a)redhat.com>

6 97

Change in vdsm[master]: packaging: debian: making code compatible with upstream M2Cr...
by stirabos＠redhat.com 09 Jun '15

09 Jun '15

Simone Tiraboschi has uploaded a new change for review. Change subject: packaging: debian: making code compatible with upstream M2Crypto ...................................................................... packaging: debian: making code compatible with upstream M2Crypto Upstream M2Crypto still doesn't provide settimeout method in the M2Crypto.SSL.Connection class: https://github.com/M2Crypto/M2Crypto/blob/master/M2Crypto/SSL/Connection.py The method is indeed just a wrapper of the same method on the internal socket object. Using the method on the external class if available otherwise fall back and directly use it on the socket object. Change-Id: Id1808bc9e27dde72b018c3be413f5b2066127982 Signed-off-by: Simone Tiraboschi <stirabos(a)redhat.com> --- M lib/vdsm/sslutils.py 1 file changed, 7 insertions(+), 4 deletions(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/46/37746/1 diff --git a/lib/vdsm/sslutils.py b/lib/vdsm/sslutils.py index 74555ee..4f2d4f5 100644 --- a/lib/vdsm/sslutils.py +++ b/lib/vdsm/sslutils.py @@ -118,11 +118,14 @@ client = SSL.Connection(self.context, client) client.addr = address try: - client.setup_ssl() - client.set_accept_state() - client.settimeout(self.accept_timeout) + settimeout = getattr( + client, + 'settimeout', + client.socket.settimeout + ) + settimeout(self.accept_timeout) client.accept_ssl() - client.settimeout(None) + settimeout(None) except SSL.SSLError as e: raise SSL.SSLError("%s, client %s" % (e, address[0])) -- To view, visit http://gerrit.ovirt.org/37746 To unsubscribe, visit http://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: Id1808bc9e27dde72b018c3be413f5b2066127982 Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Simone Tiraboschi <stirabos(a)redhat.com>

6 38

Change in vdsm[master]: vdscliTests: add a test for connect with a timeout
by ykaplan＠redhat.com 08 Jun '15

08 Jun '15

Yeela Kaplan has uploaded a new change for review. Change subject: vdscliTests: add a test for connect with a timeout ...................................................................... vdscliTests: add a test for connect with a timeout Change-Id: I21793986338a00caa1163c7f9297d160a9d1fc45 Signed-off-by: Yeela Kaplan <ykaplan(a)redhat.com> --- M tests/vdsClientTests.py 1 file changed, 19 insertions(+), 1 deletion(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/06/38806/1 diff --git a/tests/vdsClientTests.py b/tests/vdsClientTests.py index 3556018..c928f98 100644 --- a/tests/vdsClientTests.py +++ b/tests/vdsClientTests.py @@ -23,9 +23,10 @@ import sys from tempfile import mkstemp from contextlib import contextmanager +from _ssl import SSLError from testlib import VdsmTestCase as TestCaseBase -from testValidation import brokentest +from testValidation import brokentest, ValidateRunningAsRoot from monkeypatch import MonkeyPatch from vdsm import vdscli @@ -521,6 +522,23 @@ 'key3': 'val3'}) +class ConnectTest(TestCaseBase): + + @ValidateRunningAsRoot + def testTimeout(self): + c = vdscli.connect(useSSL=True, + tsPath='/etc/pki/vdsm', + timeout=0.01) + self.assertRaises(SSLError, c.getVdsStats) + + @ValidateRunningAsRoot + def testNoTimeout(self): + c = vdscli.connect(useSSL=True, + tsPath='/etc/pki/vdsm', + timeout=3) + self.assertNotRaises(c.getVdsStats) + + class CannonizeHostPortTest(TestCaseBase): def testNoArguments(self): -- To view, visit https://gerrit.ovirt.org/38806 To unsubscribe, visit https://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: I21793986338a00caa1163c7f9297d160a9d1fc45 Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Yeela Kaplan <ykaplan(a)redhat.com>

6 47

Change in vdsm[master]: vdscli: add a timeout option for plaintext connections
by ykaplan＠redhat.com 08 Jun '15

08 Jun '15

Yeela Kaplan has uploaded a new change for review. Change subject: vdscli: add a timeout option for plaintext connections ...................................................................... vdscli: add a timeout option for plaintext connections Change-Id: I38ebcb5bfb8fe93a04b8693cfcc17a6c0f6785ce Signed-off-by: Yeela Kaplan <ykaplan(a)redhat.com> --- M lib/vdsm/vdscli.py 1 file changed, 10 insertions(+), 6 deletions(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/47/38947/1 diff --git a/lib/vdsm/vdscli.py b/lib/vdsm/vdscli.py index 8339766..d40b8c0 100644 --- a/lib/vdsm/vdscli.py +++ b/lib/vdsm/vdscli.py @@ -53,10 +53,17 @@ connection. This class reverts the change to avoid the concurrency issues.''' + def __init__(self, timeout=sslutils.SOCKET_DEFAULT_TIMEOUT): + xmlrpclib.Transport.__init__(self) + self.timeout = timeout + def make_connection(self, host): '''Creates a new HTTPConnection to the host.''' + self._connection = None - return xmlrpclib.Transport.make_connection(self, host) + httpCon = xmlrpclib.Transport.make_connection(self, host) + httpCon._conn.timeout = self.timeout + return httpCon def __guessDefaults(): @@ -116,11 +123,8 @@ server = xmlrpclib.ServerProxy('https://%s' % hostPort, wrap_transport(transport)) else: - # TODO: add timeout option for HTTP connection - if timeout is not sslutils.SOCKET_DEFAULT_TIMEOUT: - raise NotImplementedError('timeout is not supported in' - ' plaintext connection') - transport = wrap_transport(SingleRequestTransport()) + transport = wrap_transport( + SingleRequestTransport(timeout=timeout)) server = xmlrpclib.Server('http://%s' % hostPort, transport) return server -- To view, visit https://gerrit.ovirt.org/38947 To unsubscribe, visit https://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: I38ebcb5bfb8fe93a04b8693cfcc17a6c0f6785ce Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Yeela Kaplan <ykaplan(a)redhat.com>

7 25

Change in vdsm[master]: jsonrpc: ignore not needed transport
by Piotr Kliczewski 08 Jun '15

08 Jun '15

Piotr Kliczewski has uploaded a new change for review. Change subject: jsonrpc: ignore not needed transport ...................................................................... jsonrpc: ignore not needed transport Change-Id: Icecf5c06e216151bd95f56f6fef9ed9ccba9074d Signed-off-by: pkliczew <piotr.kliczewski(a)gmail.com> --- M lib/yajsonrpc/__init__.py 1 file changed, 1 insertion(+), 1 deletion(-) git pull ssh://gerrit.ovirt.org:29418/vdsm refs/changes/73/40173/1 diff --git a/lib/yajsonrpc/__init__.py b/lib/yajsonrpc/__init__.py index e1c7848..5704a42 100644 --- a/lib/yajsonrpc/__init__.py +++ b/lib/yajsonrpc/__init__.py @@ -360,7 +360,7 @@ return ("result" in obj or "error" in obj) def _handleMessage(self, req): - transport, message = req + _, message = req try: mobj = json.loads(message) isResponse = self._isResponse(mobj) -- To view, visit https://gerrit.ovirt.org/40173 To unsubscribe, visit https://gerrit.ovirt.org/settings Gerrit-MessageType: newchange Gerrit-Change-Id: Icecf5c06e216151bd95f56f6fef9ed9ccba9074d Gerrit-PatchSet: 1 Gerrit-Project: vdsm Gerrit-Branch: master Gerrit-Owner: Piotr Kliczewski <piotr.kliczewski(a)gmail.com>

8 42

← Newer
1
...
19
20
21
22
23
24
25
...
80
Older →

Jump to page:

2024

2023

2022

2021

2020

2019

2018

2017

2016

2015

2014

2013

2012

2011

vdsm-patches April 2015