Add support for monitor_cond_since / update3 to python-ovs to allow more efficient reconnections when connecting to clustered OVSDB servers.
Signed-off-by: Terry Wilson <twil...@redhat.com> --- python/ovs/db/idl.py | 245 ++++++++++++++++++++++++++++++++++++------- tests/ovsdb-idl.at | 2 +- 2 files changed, 211 insertions(+), 36 deletions(-) diff --git a/python/ovs/db/idl.py b/python/ovs/db/idl.py index 60e58b03e..0d5e00208 100644 --- a/python/ovs/db/idl.py +++ b/python/ovs/db/idl.py @@ -13,6 +13,7 @@ # limitations under the License. import collections +import enum import functools import uuid @@ -36,6 +37,7 @@ ROW_DELETE = "delete" OVSDB_UPDATE = 0 OVSDB_UPDATE2 = 1 +OVSDB_UPDATE3 = 2 CLUSTERED = "clustered" RELAY = "relay" @@ -75,6 +77,65 @@ class ColumnDefaultDict(dict): return item in self.keys() +class Monitor(enum.IntEnum): + monitor = OVSDB_UPDATE + monitor_cond = OVSDB_UPDATE2 + monitor_cond_since = OVSDB_UPDATE3 + + +class ConditionState(object): + def __init__(self): + self._ack_cond = None + self._req_cond = None + self._new_cond = [True] + + def __iter__(self): + return iter([self._new_cond, self._req_cond, self._ack_cond]) + + @property + def new(self): + """The latest freshly initialized condition change""" + return self._new_cond + + @property + def acked(self): + """The last condition change that has been accepted by the server""" + return self._ack_cond + + @property + def requested(self): + """A condition that's been requested, but not acked by the server""" + return self._req_cond + + @property + def latest(self): + """The most recent condition change""" + return next(cond for cond in self if cond is not None) + + @staticmethod + def is_true(condition): + return condition == [True] + + def init(self, cond): + """Signal that a condition change is being initiated""" + self._new_cond = cond + + def ack(self): + """Signal that a condition change has been acked""" + if self._req_cond is not None: + self._ack_cond, self._req_cond = (self._req_cond, None) + + def request(self): + """Signal that a condition change has been requested""" + if self._new_cond is not None: + self._req_cond, self._new_cond = (self._new_cond, None) + + def reset(self): + """Reset a requested condition change back to new""" + if self._req_cond is not None and self._new_cond is None: + self._new_cond, self._req_cond = (self._req_cond, None) + + class Idl(object): """Open vSwitch Database Interface Definition Language (OVSDB IDL). @@ -132,7 +193,13 @@ class Idl(object): IDL_S_SERVER_MONITOR_REQUESTED = 2 IDL_S_DATA_MONITOR_REQUESTED = 3 IDL_S_DATA_MONITOR_COND_REQUESTED = 4 - IDL_S_MONITORING = 5 + IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED = 5 + IDL_S_MONITORING = 6 + + monitor_map = { + Monitor.monitor: IDL_S_SERVER_MONITOR_REQUESTED, + Monitor.monitor_cond: IDL_S_DATA_MONITOR_COND_REQUESTED, + Monitor.monitor_cond_since: IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED} def __init__(self, remote, schema_helper, probe_interval=None, leader_only=True): @@ -176,10 +243,12 @@ class Idl(object): remotes = self._parse_remotes(remote) self._session = ovs.jsonrpc.Session.open_multiple(remotes, probe_interval=probe_interval) + self._request_id = None self._monitor_request_id = None self._last_seqno = None self.change_seqno = 0 self.uuid = uuid.uuid1() + self.last_id = str(uuid.UUID(int=0)) # Server monitor. self._server_schema_request_id = None @@ -206,6 +275,9 @@ class Idl(object): self.txn = None self._outstanding_txns = {} + self.cond_changed = False + self.cond_seqno = 0 + for table in schema.tables.values(): for column in table.columns.values(): if not hasattr(column, 'alert'): @@ -213,8 +285,7 @@ class Idl(object): table.need_table = False table.rows = custom_index.IndexedRows(table) table.idl = self - table.condition = [True] - table.cond_changed = False + table.condition = ConditionState() def _parse_remotes(self, remote): # If remote is - @@ -252,6 +323,38 @@ class Idl(object): update.""" self._session.close() + def ack_conditions(self): + """Mark all requested table conditions as acked""" + for table in self.tables.values(): + table.condition.ack() + + def sync_conditions(self): + """Synchronize condition state when the FSM is restarted + + If a non-zero last_id is available for the DB, then upon reconnect + the IDL should first request acked conditions to avoid missing updates + about records that were added before the transaction with + txn-id == last_id. If there were requested condition changes in flight + and the IDL client didn't set new conditions, then reset the requested + conditions to new to trigger a follow-up monitor_cond_change request + """ + ack_all = self.last_id == str(uuid.UUID(int=0)) + for table in self.tables.values(): + if ack_all: + table.condition.request() + table.condition.ack() + else: + table.condition.reset() + self.cond_changed = True + + def restart_fsm(self): + # Resync data DB table conditions to avoid missing updated due to + # conditions that were in flight or changed locally while the + # connection was down. + self.sync_conditions() + self.__send_server_schema_request() + self.state = self.IDL_S_SERVER_SCHEMA_REQUESTED + def run(self): """Processes a batch of messages from the database server. Returns True if the database as seen through the IDL changed, False if it did @@ -286,7 +389,7 @@ class Idl(object): if seqno != self._last_seqno: self._last_seqno = seqno self.__txn_abort_all() - self.__send_server_schema_request() + self.restart_fsm() if self.lock_name: self.__send_lock_request() break @@ -294,8 +397,20 @@ class Idl(object): msg = self._session.recv() if msg is None: break + is_response = msg.type in (ovs.jsonrpc.Message.T_REPLY, + ovs.jsonrpc.Message.T_ERROR) + + if is_response and self._request_id and self._request_id == msg.id: + self._request_id = None + # process_response follows if (msg.type == ovs.jsonrpc.Message.T_NOTIFY + and msg.method == "update3" + and len(msg.params) == 3): + # Database contents changed. + self.__parse_update(msg.params[2], OVSDB_UPDATE3) + self.last_id = msg.params[1] + elif (msg.type == ovs.jsonrpc.Message.T_NOTIFY and msg.method == "update2" and len(msg.params) == 2): # Database contents changed. @@ -320,11 +435,18 @@ class Idl(object): try: self.change_seqno += 1 self._monitor_request_id = None - self.__clear() - if self.state == self.IDL_S_DATA_MONITOR_COND_REQUESTED: + if (self.state == + self.IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED): + # If 'found' is false, clear table rows for new dump + if not msg.result[0]: + self.__clear() + self.__parse_update(msg.result[2], OVSDB_UPDATE3) + elif self.state == self.IDL_S_DATA_MONITOR_COND_REQUESTED: + self.__clear() self.__parse_update(msg.result, OVSDB_UPDATE2) else: assert self.state == self.IDL_S_DATA_MONITOR_REQUESTED + self.__clear() self.__parse_update(msg.result, OVSDB_UPDATE) self.state = self.IDL_S_MONITORING @@ -398,11 +520,17 @@ class Idl(object): elif msg.type == ovs.jsonrpc.Message.T_NOTIFY and msg.id == "echo": # Reply to our echo request. Ignore it. pass + elif (msg.type == ovs.jsonrpc.Message.T_ERROR and + self.state == ( + self.IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED) and + self._monitor_request_id == msg.id): + if msg.error == "unknown method": + self.__send_monitor_request(Monitor.monitor_cond) elif (msg.type == ovs.jsonrpc.Message.T_ERROR and self.state == self.IDL_S_DATA_MONITOR_COND_REQUESTED and self._monitor_request_id == msg.id): if msg.error == "unknown method": - self.__send_monitor_request() + self.__send_monitor_request(Monitor.monitor) elif (msg.type == ovs.jsonrpc.Message.T_ERROR and self._server_schema_request_id is not None and self._server_schema_request_id == msg.id): @@ -418,6 +546,13 @@ class Idl(object): and self.__txn_process_reply(msg)): # __txn_process_reply() did everything needed. pass + elif (msg.type == ovs.jsonrpc.Message.T_REPLY and + self.state == self.IDL_S_MONITORING): + # Mark the last requested conditions as acked and if further + # condition changes were pending, send them now. + self.ack_conditions() + self.send_cond_change() + self.cond_seqno += 1 else: # This can happen if a transaction is destroyed before we # receive the reply, so keep the log level low. @@ -427,14 +562,36 @@ class Idl(object): return initial_change_seqno != self.change_seqno - def send_cond_change(self): - if not self._session.is_connected(): + def compose_cond_change(self): + if not self.cond_changed: return + change_requests = {} for table in self.tables.values(): - if table.cond_changed: - self.__send_cond_change(table, table.condition) - table.cond_changed = False + # Always use the most recent conditions set by the IDL client when + # requesting monitor_cond_change + if table.condition.new is not None: + change_requests[table.name] = [ + {"where": table.condition.new}] + table.condition.request() + + if not change_requests: + return + + self.cond_changed = False + old_uuid = str(self.uuid) + self.uuid = uuid.uuid1() + params = [old_uuid, str(self.uuid), change_requests] + return ovs.jsonrpc.Message.create_request( + "monitor_cond_change", params) + + def send_cond_change(self): + if not self._session.is_connected() or self._request_id is not None: + return + + msg = self.compose_cond_change() + if msg: + self.send_request(msg) def cond_change(self, table_name, cond): """Sets the condition for 'table_name' to 'cond', which should be a @@ -450,13 +607,28 @@ class Idl(object): if cond == []: cond = [False] - if table.condition != cond: - table.condition = cond - table.cond_changed = True + + # Compare the new condition to the last known condition + if table.condition.latest != cond: + table.condition.init(cond) + self.cond_changed = True + + # New condition will be sent out after all already requested ones + # are acked. + if table.condition.new: + any_reqs = any(t.condition.request for t in self.tables.values()) + return self.cond_seqno + int(any_reqs) + 1 + + # Already requested conditions should be up to date at + # self.cond_seqno + 1 while acked conditions are already up to date + return self.cond_seqno + int(bool(table.condition.requested)) def wait(self, poller): """Arranges for poller.block() to wake up when self.run() has something to do or when activity occurs on a transaction on 'self'.""" + if self.cond_changed: + poller.immediate_wake() + return self._session.wait(poller) self._session.recv_wait(poller) @@ -531,14 +703,6 @@ class Idl(object): to doing nothing to avoid overhead where it is not needed. """ - def __send_cond_change(self, table, cond): - monitor_cond_change = {table.name: [{"where": cond}]} - old_uuid = str(self.uuid) - self.uuid = uuid.uuid1() - params = [old_uuid, str(self.uuid), monitor_cond_change] - msg = ovs.jsonrpc.Message.create_request("monitor_cond_change", params) - self._session.send(msg) - def __clear(self): changed = False @@ -547,6 +711,8 @@ class Idl(object): changed = True table.rows = custom_index.IndexedRows(table) + self.cond_seqno = 0 + if changed: self.change_seqno += 1 @@ -601,11 +767,18 @@ class Idl(object): self._db_change_aware_request_id = msg.id self._session.send(msg) - def __send_monitor_request(self): - if (self.state in [self.IDL_S_SERVER_MONITOR_REQUESTED, - self.IDL_S_INITIAL]): + def send_request(self, request): + self._request_id = request.id + if self._session.is_connected(): + return self._session.send(request) + + def __send_monitor_request(self, max_version=Monitor.monitor_cond_since): + if self.state == self.IDL_S_INITIAL: self.state = self.IDL_S_DATA_MONITOR_COND_REQUESTED method = "monitor_cond" + elif self.state == self.IDL_S_SERVER_MONITOR_REQUESTED: + self.state = self.monitor_map[Monitor(max_version)] + method = Monitor(max_version).name else: self.state = self.IDL_S_DATA_MONITOR_REQUESTED method = "monitor" @@ -619,22 +792,24 @@ class Idl(object): (column not in self.readonly[table.name])): columns.append(column) monitor_request = {"columns": columns} - if method == "monitor_cond" and table.condition != [True]: - monitor_request["where"] = table.condition - table.cond_change = False + if method in ("monitor_cond", "monitor_cond_since") and ( + not ConditionState.is_true(table.condition.acked)): + monitor_request["where"] = table.condition.acked monitor_requests[table.name] = [monitor_request] - msg = ovs.jsonrpc.Message.create_request( - method, [self._db.name, str(self.uuid), monitor_requests]) + args = [self._db.name, str(self.uuid), monitor_requests] + if method == "monitor_cond_since": + args.append(str(self.last_id)) + msg = ovs.jsonrpc.Message.create_request(method, args) self._monitor_request_id = msg.id - self._session.send(msg) + self.send_request(msg) def __send_server_schema_request(self): self.state = self.IDL_S_SERVER_SCHEMA_REQUESTED msg = ovs.jsonrpc.Message.create_request( "get_schema", [self._server_db_name, str(self.uuid)]) self._server_schema_request_id = msg.id - self._session.send(msg) + self.send_request(msg) def __send_server_monitor_request(self): self.state = self.IDL_S_SERVER_MONITOR_REQUESTED @@ -654,7 +829,7 @@ class Idl(object): str(self.server_monitor_uuid), monitor_requests]) self._server_monitor_request_id = msg.id - self._session.send(msg) + self.send_request(msg) def __parse_update(self, update, version, tables=None): try: @@ -698,7 +873,7 @@ class Idl(object): self.cooperative_yield() - if version == OVSDB_UPDATE2: + if version in (OVSDB_UPDATE2, OVSDB_UPDATE3): changes = self.__process_update2(table, uuid, row_update) if changes: notices.append(changes) diff --git a/tests/ovsdb-idl.at b/tests/ovsdb-idl.at index 3adb9d638..86a75f920 100644 --- a/tests/ovsdb-idl.at +++ b/tests/ovsdb-idl.at @@ -2319,7 +2319,7 @@ m4_define([OVSDB_CHECK_CLUSTER_IDL], # Checks that monitor_cond_since works fine when disconnects happen # with cond_change requests in flight (i.e., IDL is properly updated). -OVSDB_CHECK_CLUSTER_IDL_C([simple idl, monitor_cond_since, cluster disconnect], +OVSDB_CHECK_CLUSTER_IDL([simple idl, monitor_cond_since, cluster disconnect], 3, [['["idltest", {"op": "insert", -- 2.31.1 _______________________________________________ dev mailing list d...@openvswitch.org https://mail.openvswitch.org/mailman/listinfo/ovs-dev