ovsdb-idl: Support for readonly columns that are fetched on-demand

There is currently no mechanism in IDL to fetch specific column values
on-demand without having to register them for monitoring. In the case
where the column represent a frequently changing entity (e.g. counter),
and the reads are relatively infrequent (e.g. CLI client), there is a
significant overhead in replication.

This patch adds support in the Python IDL to register a subset of the
columns of a table as "readonly". Readonly columns are not replicated.
Users may "fetch" the readonly columns of a row on-demand. Once fetched,
the columns are not updated until the next fetch by the user. Writes by
the user to readonly columns does not change the value (both locally or
on the server).

The two main user visible changes in this patch are:
  - The SchemaHelper.register_columns() method now takes an optionaly
    argument to specify the subset of readonly column(s)
  - A new Row.fetch(columns) method to fetch values of readonly columns(s)

Usage:
------

    # Schema file includes all columns, including readonly
    schema_helper = ovs.db.idl.SchemaHelper(schema_file)

    # Register interest in columns with 'r' and 's' as readonly
    schema_helper.register_columns("simple", [i, r, s], [r, s])

    # Create Idl and jsonrpc, and wait for update, as usual
    ...

    # Fetch value of column 'r' for a specific row
    row.fetch('r')
    txn.commit_block()

    print row.r
    print getattr(row, 'r')

    # Writing to readonly column has no effect (locally or on server)
    row.r = 3
    print row.r     # prints fetched value not 3

Signed-off-by: Shad Ansari <shad.ansari@hp.com>
Signed-off-by: Ben Pfaff <blp@ovn.org>

1 files changed, 82 insertions, 4 deletions

diff --git a/python/ovs/db/idl.py b/python/ovs/db/idl.py
index f074dbfce..c8990c71e 100644
--- a/python/ovs/db/idl.py
+++ b/python/ovs/db/idl.py
@@ -107,6 +107,7 @@ class Idl(object):
         schema = schema.get_idl_schema()
 
         self.tables = schema.tables
+        self.readonly = schema.readonly
         self._db = schema
         self._session = ovs.jsonrpc.Session.open(remote)
         self._monitor_request_id = None
@@ -338,7 +339,13 @@ class Idl(object):
     def __send_monitor_request(self):
         monitor_requests = {}
         for table in self.tables.itervalues():
-            monitor_requests[table.name] = {"columns": table.columns.keys()}
+            columns = []
+            for column in table.columns.keys():
+                if ((table.name not in self.readonly) or
+                    (table.name in self.readonly) and
+                    (column not in self.readonly[table.name])):
+                    columns.append(column)
+            monitor_requests[table.name] = {"columns": columns}
         msg = ovs.jsonrpc.Message.create_request(
             "monitor", [self._db.name, None, monitor_requests])
         self._monitor_request_id = msg.id
@@ -571,7 +578,11 @@ class Row(object):
             if self._data is None:
                 raise AttributeError("%s instance has no attribute '%s'" %
                                      (self.__class__.__name__, column_name))
-            datum = self._data[column_name]
+            if column_name in self._data:
+                datum = self._data[column_name]
+            else:
+                raise AttributeError("%s instance has no attribute '%s'" %
+                                     (self.__class__.__name__, column_name))
 
         return datum.to_python(_uuid_to_row)
 
@@ -579,6 +590,11 @@ class Row(object):
         assert self._changes is not None
         assert self._idl.txn
 
+        if ((self._table.name in self._idl.readonly) and
+            (column_name in self._idl.readonly[self._table.name])):
+            vlog.warn("attempting to write to readonly column %s" % column_name)
+            return
+
         column = self._table.columns[column_name]
         try:
             datum = ovs.db.data.Datum.from_python(column.type, value,
@@ -655,6 +671,9 @@ class Row(object):
         self.__dict__["_changes"] = None
         del self._table.rows[self.uuid]
 
+    def fetch(self, column_name):
+        self._idl.txn._fetch(self, column_name)
+
     def increment(self, column_name):
         """Causes the transaction, when committed, to increment the value of
         'column_name' within this row by 1.  'column_name' must have an integer
@@ -777,10 +796,12 @@ class Transaction(object):
         self._inc_row = None
         self._inc_column = None
 
+        self._fetch_requests = []
+
         self._inserted_rows = {}  # Map from UUID to _InsertedRow
 
     def add_comment(self, comment):
-        """Appens 'comment' to the comments that will be passed to the OVSDB
+        """Appends 'comment' to the comments that will be passed to the OVSDB
         server when this transaction is committed.  (The comment will be
         committed to the OVSDB log, which "ovsdb-tool show-log" can print in a
         relatively human-readable form.)"""
@@ -947,6 +968,16 @@ class Transaction(object):
                 if row._data is None or row_json:
                     operations.append(op)
 
+        if self._fetch_requests:
+            for fetch in self._fetch_requests:
+                fetch["index"] = len(operations) - 1
+                operations.append({"op": "select",
+                                   "table": fetch["row"]._table.name,
+                                   "where": self._substitute_uuids(
+                                       _where_uuid_equals(fetch["row"].uuid)),
+                                   "columns": [fetch["column_name"]]})
+            any_updates = True
+
         # Add increment.
         if self._inc_row and any_updates:
             self._inc_index = len(operations) - 1
@@ -1057,6 +1088,9 @@ class Transaction(object):
         self._inc_row = row
         self._inc_column = column
 
+    def _fetch(self, row, column_name):
+        self._fetch_requests.append({"row":row, "column_name":column_name})
+
     def _write(self, row, column, datum):
         assert row._changes is not None
 
@@ -1139,6 +1173,11 @@ class Transaction(object):
             if not soft_errors and not hard_errors and not lock_errors:
                 if self._inc_row and not self.__process_inc_reply(ops):
                     hard_errors = True
+                if self._fetch_requests:
+                    if self.__process_fetch_reply(ops):
+                        self.idl.change_seqno += 1
+                    else:
+                        hard_errors = True
 
                 for insert in self._inserted_rows.itervalues():
                     if not self.__process_insert_reply(insert, ops):
@@ -1166,6 +1205,38 @@ class Transaction(object):
         else:
             return True
 
+    def __process_fetch_reply(self, ops):
+        update = False
+        for fetch_request in self._fetch_requests:
+            row = fetch_request["row"]
+            column_name = fetch_request["column_name"]
+            index = fetch_request["index"]
+            table = row._table
+
+            select = ops[index]
+            fetched_rows = select.get("rows")
+            if not Transaction.__check_json_type(fetched_rows, (list, tuple),
+                                                 '"select" reply "rows"'):
+                return False
+            if len(fetched_rows) != 1:
+                # XXX rate-limit
+                vlog.warn('"select" reply "rows" has %d elements '
+                          'instead of 1' % len(rows))
+                continue
+            fetched_row = fetched_rows[0]
+            if not Transaction.__check_json_type(fetched_row, (dict,),
+                                                 '"select" reply row'):
+                continue
+
+            column = table.columns.get(column_name)
+            datum_json = fetched_row.get(column_name)
+            datum = ovs.db.data.Datum.from_json(column.type, datum_json)
+
+            row._data[column_name] = datum
+            update = True
+
+        return update
+
     def __process_inc_reply(self, ops):
         if self._inc_index + 2 > len(ops):
             # XXX rate-limit
@@ -1261,16 +1332,21 @@ class SchemaHelper(object):
 
         self.schema_json = schema_json
         self._tables = {}
+        self._readonly = {}
         self._all = False
 
-    def register_columns(self, table, columns):
+    def register_columns(self, table, columns, readonly=[]):
         """Registers interest in the given 'columns' of 'table'.  Future calls
         to get_idl_schema() will include 'table':column for each column in
         'columns'. This function automatically avoids adding duplicate entries
         to the schema.
+        A subset of 'columns' can be specified as 'readonly'. The readonly
+        columns are not replicated but can be fetched on-demand by the user
+        with Row.fetch().
 
         'table' must be a string.
         'columns' must be a list of strings.
+        'readonly' must be a list of strings.
         """
 
         assert type(table) is str
@@ -1278,6 +1354,7 @@ class SchemaHelper(object):
 
         columns = set(columns) | self._tables.get(table, set())
         self._tables[table] = columns
+        self._readonly[table] = readonly
 
     def register_table(self, table):
         """Registers interest in the given all columns of 'table'. Future calls
@@ -1307,6 +1384,7 @@ class SchemaHelper(object):
                     self._keep_table_columns(schema, table, columns))
 
             schema.tables = schema_tables
+        schema.readonly = self._readonly
         return schema
 
     def _keep_table_columns(self, schema, table_name, columns):