When the IDL clears its local replica (on initial connection or when monitor_cond_since cannot provide incremental updates), clients currently receive ROW_CREATE for all rows in the new snapshot with no indication of which rows were deleted while disconnected.
Add a notify_on_clear option that generates ROW_DELETE notices for existing rows before they are cleared. These are reconciled with the subsequent ROW_CREATE notices via a new ReconciledNotices view: - Surviving rows with identical data are suppressed - Surviving rows with changed data become ROW_UPDATE - Rows only in the old snapshot become ROW_DELETE - Rows only in the new snapshot become ROW_CREATE Assisted-by: Claude Opus 4.6 Signed-off-by: Terry Wilson <[email protected]> --- python/automake.mk | 3 +- python/ovs/db/idl.py | 113 ++++++++- python/ovs/tests/test_reconciled_notices.py | 255 ++++++++++++++++++++ tests/ovsdb-idl.at | 16 ++ tests/test-ovsdb.py | 8 +- 5 files changed, 380 insertions(+), 15 deletions(-) create mode 100644 python/ovs/tests/test_reconciled_notices.py diff --git a/python/automake.mk b/python/automake.mk index d2589e6fe..4eb2ff6b5 100644 --- a/python/automake.mk +++ b/python/automake.mk @@ -49,7 +49,8 @@ ovs_pytests = \ python/ovs/tests/test_kv.py \ python/ovs/tests/test_list.py \ python/ovs/tests/test_odp.py \ - python/ovs/tests/test_ofp.py + python/ovs/tests/test_ofp.py \ + python/ovs/tests/test_reconciled_notices.py ovs_flowviz = \ python/ovs/flowviz/__init__.py \ diff --git a/python/ovs/db/idl.py b/python/ovs/db/idl.py index fe504a63a..c345370b9 100644 --- a/python/ovs/db/idl.py +++ b/python/ovs/db/idl.py @@ -47,6 +47,61 @@ Notice = collections.namedtuple('Notice', ('event', 'row', 'updates')) Notice.__new__.__defaults__ = (None,) # default updates=None +class ReconciledNotices: + """Lazy view over pending notices, reconciling reconnect events. + + All notifications pass through this view before reaching + bulk_notify() or notify(). When notify_on_clear is False (the + default), each UUID has a single notice and the view passes it + through unchanged. When notify_on_clear is True, __clear() + generates DELETE notices for all pre-existing rows before + __parse_update() generates INSERT notices for the new snapshot. + The view reconciles DELETE+INSERT pairs per UUID: + + - Same data: suppressed (row survived unchanged) + - Different data: emitted as UPDATE with only changed columns + - DELETE only: emitted (row was truly deleted) + - INSERT only: emitted (genuinely new row) + + Iteration yields Notice objects. __getitem__(uuid) returns the + reconciled Notice for a specific UUID. + """ + + def __init__(self, notices): + self._raw = notices + + def __getitem__(self, row_uuid): + return self._reconcile(row_uuid, self._raw[row_uuid]) + + def __iter__(self): + for row_uuid in self._raw: + notice = self._reconcile(row_uuid, self._raw[row_uuid]) + if notice is not None: + yield notice + + @staticmethod + def _reconcile(row_uuid, events): + if len(events) == 1: + return events[0] + if ( + len(events) == 2 and + events[0].event == ROW_DELETE and + events[1].event == ROW_CREATE + ): + old_row = events[0].row + new_row = events[1].row + old_data = {} + for col in old_row._table.columns: + if col in old_row._data and col in new_row._data: + if old_row._data[col] != new_row._data[col]: + old_data[col] = old_row._data[col] + if old_data: + return Notice(ROW_UPDATE, new_row, + Row(None, old_row._table, row_uuid, old_data)) + return None + assert False, "unexpected number of events for row %s" % row_uuid + + class ColumnDefaultDict(dict): """A column dictionary with on-demand generated default values @@ -238,7 +293,7 @@ class Idl(object): Monitor.monitor_cond_since: IDL_S_DATA_MONITOR_COND_SINCE_REQUESTED} def __init__(self, remote, schema_helper, probe_interval=None, - leader_only=True): + leader_only=True, notify_on_clear=False): """Creates and returns a connection to the database named 'db_name' on 'remote', which should be in a form acceptable to ovs.jsonrpc.session.open(). The connection will maintain an in-memory @@ -268,6 +323,20 @@ class Idl(object): If "probe_interval" is zero it disables the connection keepalive feature. If non-zero the value will be forced to at least 1000 milliseconds. If None it will just use the default value in OVS. + + If 'notify_on_clear' is True, the IDL generates ROW_DELETE + notifications for existing rows when the local database replica + is cleared (e.g. on initial connection or when a + monitor_cond_since reply indicates the server could not provide + incremental updates). These DELETE notices are reconciled with + the subsequent INSERT notices from the new snapshot before + delivery via bulk_notify(): rows that survived unchanged are + suppressed, rows with changed data are delivered as ROW_UPDATE + with only the changed columns, truly deleted rows are delivered + as ROW_DELETE, and genuinely new rows are delivered as + ROW_CREATE. When False (the default), all rows from the new + snapshot are delivered as ROW_CREATE, with no notification for + rows that no longer exist. """ assert isinstance(schema_helper, SchemaHelper) @@ -314,6 +383,8 @@ class Idl(object): self.cond_changed = False self.cond_seqno = 0 + self.notify_on_clear = notify_on_clear + self._pending_notices = collections.defaultdict(list) def _parse_remotes(self, remote): # If remote is - @@ -494,15 +565,20 @@ class Idl(object): # 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) + self.__parse_update(msg.result[2], OVSDB_UPDATE3, + initial=True) + else: + self.__parse_update(msg.result[2], OVSDB_UPDATE3) self.last_id = msg.result[1] elif self.state == self.IDL_S_DATA_MONITOR_COND_REQUESTED: self.__clear() - self.__parse_update(msg.result, OVSDB_UPDATE2) + self.__parse_update(msg.result, OVSDB_UPDATE2, + initial=True) else: assert self.state == self.IDL_S_DATA_MONITOR_REQUESTED self.__clear() - self.__parse_update(msg.result, OVSDB_UPDATE) + self.__parse_update(msg.result, OVSDB_UPDATE, + initial=True) self.state = self.IDL_S_MONITORING except error.Error as e: @@ -782,6 +858,10 @@ class Idl(object): :type updates: Row """ + def bulk_notify(self, notices, initial=False): + for notice in notices: + self.notify(*notice) + def cooperative_yield(self): """Hook for cooperatively yielding to eventlet/gevent/asyncio/etc. @@ -797,6 +877,10 @@ class Idl(object): for table in self.tables.values(): if table.rows: changed = True + if self.notify_on_clear: + for row_uuid, row in table.rows.items(): + self._pending_notices[row_uuid].append( + Notice(ROW_DELETE, row)) table.rows.clear() self.cond_seqno = 0 @@ -919,22 +1003,24 @@ class Idl(object): self._server_monitor_request_id = msg.id self.send_request(msg) - def __parse_update(self, update, version, tables=None): + def __parse_update(self, update, version, tables=None, initial=False): try: if not tables: - self.__do_parse_update(update, version, self.tables) + self.__do_parse_update(update, version, self.tables, + initial=initial) else: - self.__do_parse_update(update, version, tables) + self.__do_parse_update(update, version, tables, + initial=initial) except error.Error as e: vlog.err("%s: error parsing update: %s" % (self._session.get_name(), e)) - def __do_parse_update(self, table_updates, version, tables): + def __do_parse_update(self, table_updates, version, tables, + initial=False): if not isinstance(table_updates, dict): raise error.Error("<table-updates> is not an object", table_updates) - notices = [] for table_name, table_update in table_updates.items(): table = tables.get(table_name) if not table: @@ -964,7 +1050,7 @@ class Idl(object): if version in (OVSDB_UPDATE2, OVSDB_UPDATE3): changes = self.__process_update2(table, uuid, row_update) if changes and tables is not self.server_tables: - notices.append(changes) + self._pending_notices[uuid].append(changes) self.change_seqno += 1 continue @@ -979,10 +1065,11 @@ class Idl(object): changes = self.__process_update(table, uuid, old, new) if changes and tables is not self.server_tables: - notices.append(changes) + self._pending_notices[uuid].append(changes) self.change_seqno += 1 - for notice in notices: - self.notify(*notice) + self.bulk_notify(ReconciledNotices(self._pending_notices), + initial=initial) + self._pending_notices.clear() def __process_update2(self, table, uuid, row_update): """Returns Notice if a column changed, False otherwise.""" diff --git a/python/ovs/tests/test_reconciled_notices.py b/python/ovs/tests/test_reconciled_notices.py new file mode 100644 index 000000000..df099e949 --- /dev/null +++ b/python/ovs/tests/test_reconciled_notices.py @@ -0,0 +1,255 @@ +import collections +import uuid + +import pytest + +from ovs.db.idl import ( + Notice, + ReconciledNotices, + ROW_CREATE, + ROW_DELETE, + ROW_UPDATE, + Row, +) + + +class FakeTable: + def __init__(self, column_names): + self.columns = {name: None for name in column_names} + + +class FakeDatum: + """Minimal Datum-like object that supports equality comparison.""" + def __init__(self, value): + self.value = value + + def __eq__(self, other): + if isinstance(other, FakeDatum): + return self.value == other.value + return NotImplemented + + def __repr__(self): + return "FakeDatum(%r)" % self.value + + +def make_row(table, row_uuid, data): + return Row(None, table, row_uuid, data) + + +def make_notices(notice_list): + """Build defaultdict(list) keyed by UUID from (uuid, Notice) pairs.""" + raw = collections.defaultdict(list) + for row_uuid, notice in notice_list: + raw[row_uuid].append(notice) + return raw + + +TABLE = FakeTable(["name", "value", "tag"]) + + [email protected]( + "event", + [ROW_CREATE, ROW_DELETE, ROW_UPDATE], +) +def test_single_event_passthrough(event): + row_uuid = uuid.uuid4() + row = make_row(TABLE, row_uuid, {"name": FakeDatum("r1")}) + if event == ROW_UPDATE: + updates = make_row(TABLE, row_uuid, {"name": FakeDatum("old")}) + notice = Notice(event, row, updates) + else: + notice = Notice(event, row) + raw = make_notices([(row_uuid, notice)]) + + result = list(ReconciledNotices(raw)) + + assert len(result) == 1 + assert result[0] is notice + + [email protected]( + "old_data,new_data", + [ + ( + {"name": FakeDatum("same")}, + {"name": FakeDatum("same")}, + ), + ( + {"name": FakeDatum("n"), "value": FakeDatum(42), + "tag": FakeDatum("t")}, + {"name": FakeDatum("n"), "value": FakeDatum(42), + "tag": FakeDatum("t")}, + ), + ], +) +def test_delete_create_unchanged_suppressed(old_data, new_data): + row_uuid = uuid.uuid4() + raw = make_notices([ + (row_uuid, Notice(ROW_DELETE, + make_row(TABLE, row_uuid, old_data))), + (row_uuid, Notice(ROW_CREATE, + make_row(TABLE, row_uuid, new_data))), + ]) + + result = list(ReconciledNotices(raw)) + + assert len(result) == 0 + + +def test_delete_create_unchanged_getitem_returns_none(): + row_uuid = uuid.uuid4() + data = {"name": FakeDatum("same")} + raw = make_notices([ + (row_uuid, Notice(ROW_DELETE, + make_row(TABLE, row_uuid, dict(data)))), + (row_uuid, Notice(ROW_CREATE, + make_row(TABLE, row_uuid, dict(data)))), + ]) + + assert ReconciledNotices(raw)[row_uuid] is None + + [email protected]( + "old_data,new_data,expected_changed_cols", + [ + ( + {"name": FakeDatum("old_val")}, + {"name": FakeDatum("new_val")}, + {"name"}, + ), + ( + {"name": FakeDatum("n"), "value": FakeDatum(1), + "tag": FakeDatum("t")}, + {"name": FakeDatum("n"), "value": FakeDatum(2), + "tag": FakeDatum("t")}, + {"value"}, + ), + ( + {"name": FakeDatum("a"), "value": FakeDatum(1)}, + {"name": FakeDatum("b"), "value": FakeDatum(2)}, + {"name", "value"}, + ), + ], +) +def test_delete_create_changed_becomes_update(old_data, new_data, + expected_changed_cols): + row_uuid = uuid.uuid4() + old_row = make_row(TABLE, row_uuid, old_data) + new_row = make_row(TABLE, row_uuid, new_data) + raw = make_notices([ + (row_uuid, Notice(ROW_DELETE, old_row)), + (row_uuid, Notice(ROW_CREATE, new_row)), + ]) + + result = list(ReconciledNotices(raw)) + + assert len(result) == 1 + notice = result[0] + assert notice.event == ROW_UPDATE + assert notice.row is new_row + assert set(notice.updates._data.keys()) == expected_changed_cols + for col in expected_changed_cols: + assert notice.updates._data[col] == old_data[col] + + +def test_update_row_has_correct_uuid_and_table(): + row_uuid = uuid.uuid4() + old_row = make_row(TABLE, row_uuid, {"name": FakeDatum("old")}) + new_row = make_row(TABLE, row_uuid, {"name": FakeDatum("new")}) + raw = make_notices([ + (row_uuid, Notice(ROW_DELETE, old_row)), + (row_uuid, Notice(ROW_CREATE, new_row)), + ]) + + result = list(ReconciledNotices(raw)) + + assert result[0].updates.uuid == row_uuid + assert result[0].updates._table is TABLE + + [email protected]( + "old_data,new_data", + [ + ( + {"name": FakeDatum("n"), "value": FakeDatum("x")}, + {"name": FakeDatum("n")}, + ), + ( + {"name": FakeDatum("n")}, + {"name": FakeDatum("n"), "value": FakeDatum("x")}, + ), + ], +) +def test_column_only_in_one_row_not_compared(old_data, new_data): + row_uuid = uuid.uuid4() + raw = make_notices([ + (row_uuid, Notice(ROW_DELETE, + make_row(TABLE, row_uuid, old_data))), + (row_uuid, Notice(ROW_CREATE, + make_row(TABLE, row_uuid, new_data))), + ]) + + result = list(ReconciledNotices(raw)) + + assert len(result) == 0 + + +def test_mixed_rows(): + uuid_new = uuid.uuid4() + uuid_del = uuid.uuid4() + uuid_unchanged = uuid.uuid4() + uuid_changed = uuid.uuid4() + + row_new = make_row(TABLE, uuid_new, {"name": FakeDatum("new_row")}) + row_del = make_row(TABLE, uuid_del, {"name": FakeDatum("gone")}) + row_unch_old = make_row(TABLE, uuid_unchanged, {"name": FakeDatum("s")}) + row_unch_new = make_row(TABLE, uuid_unchanged, {"name": FakeDatum("s")}) + row_chg_old = make_row(TABLE, uuid_changed, {"name": FakeDatum("before")}) + row_chg_new = make_row(TABLE, uuid_changed, {"name": FakeDatum("after")}) + + raw = make_notices([ + (uuid_new, Notice(ROW_CREATE, row_new)), + (uuid_del, Notice(ROW_DELETE, row_del)), + (uuid_unchanged, Notice(ROW_DELETE, row_unch_old)), + (uuid_unchanged, Notice(ROW_CREATE, row_unch_new)), + (uuid_changed, Notice(ROW_DELETE, row_chg_old)), + (uuid_changed, Notice(ROW_CREATE, row_chg_new)), + ]) + + results = {n.row.uuid: n for n in ReconciledNotices(raw)} + + assert len(results) == 3 + assert results[uuid_new].event == ROW_CREATE + assert results[uuid_del].event == ROW_DELETE + assert results[uuid_changed].event == ROW_UPDATE + assert uuid_unchanged not in results + + +def test_all_rows_suppressed(): + uuids = [uuid.uuid4() for _ in range(3)] + raw = make_notices( + [(u, Notice(ROW_DELETE, make_row(TABLE, u, {"name": FakeDatum("x")}))) + for u in uuids] + + [(u, Notice(ROW_CREATE, + make_row(TABLE, u, {"name": FakeDatum("x")}))) + for u in uuids] + ) + + assert list(ReconciledNotices(raw)) == [] + + +def test_empty_notices(): + assert list(ReconciledNotices(collections.defaultdict(list))) == [] + + +def test_unexpected_event_count_asserts(): + row_uuid = uuid.uuid4() + row = make_row(TABLE, row_uuid, {"name": FakeDatum("x")}) + raw = make_notices([ + (row_uuid, Notice(ROW_DELETE, row)), + (row_uuid, Notice(ROW_CREATE, row)), + (row_uuid, Notice(ROW_UPDATE, row)), + ]) + + with pytest.raises(AssertionError, match="unexpected number of events"): + list(ReconciledNotices(raw)) diff --git a/tests/ovsdb-idl.at b/tests/ovsdb-idl.at index cd534e09d..f3ecfd932 100644 --- a/tests/ovsdb-idl.at +++ b/tests/ovsdb-idl.at @@ -2106,6 +2106,22 @@ OVSDB_CHECK_IDL_NOTIFY([simple idl verify notify], 015: done ]]) +OVSDB_CHECK_IDL_PY([notify_on_clear suppresses unchanged rows on reconnect], [], + [['track-notify-on-clear' \ + '["idltest", + {"op": "insert", + "table": "simple", + "row": {"i": 1, "s": "original"}}]' \ + 'reconnect']], + [[000: empty +001: {"error":null,"result":[{"uuid":["uuid","<0>"]}]} +002: event:create, row={i=1 r=0 b=false s=original u=<1> ia=[] ra=[] ba=[] sa=[] ua=[]}, uuid=<0>, updates=None +002: table simple: i=1 r=0 b=false s=original u=<1> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<0> +003: reconnect +004: table simple: i=1 r=0 b=false s=original u=<1> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<0> +005: done +]], [notify]) + OVSDB_CHECK_IDL_NOTIFY([indexed idl, modification and removal notify], [['track-notify' \ '["idltest", diff --git a/tests/test-ovsdb.py b/tests/test-ovsdb.py index 394897648..4580c09f4 100644 --- a/tests/test-ovsdb.py +++ b/tests/test-ovsdb.py @@ -770,9 +770,14 @@ def do_idl(schema_file, remote, *commands): ovs.stream.Stream.ssl_set_ca_cert_file(commands[2]) commands = commands[3:] + notify_on_clear = False if commands and commands[0] == "track-notify": commands = commands[1:] track_notify = True + elif commands and commands[0] == "track-notify-on-clear": + commands = commands[1:] + track_notify = True + notify_on_clear = True if commands and commands[0].startswith("?"): readonly = {} @@ -788,7 +793,8 @@ def do_idl(schema_file, remote, *commands): commands = commands[1:] else: schema_helper.register_all() - idl = ovs.db.idl.Idl(remote, schema_helper, leader_only=False) + idl = ovs.db.idl.Idl(remote, schema_helper, leader_only=False, + notify_on_clear=notify_on_clear) if "simple3" in idl.tables: idl.index_create("simple3", "simple3_by_name") if "indexed" in idl.tables: -- 2.49.0 _______________________________________________ dev mailing list [email protected] https://mail.openvswitch.org/mailman/listinfo/ovs-dev
