[ovs-dev,v2] Monitor Database table to manage lifecycle of IDL client.

Message ID 1546568184-202648-1-git-send-email-ted.elhourani@nutanix.com
State Changes Requested
Headers show
Series
  • [ovs-dev,v2] Monitor Database table to manage lifecycle of IDL client.
Related show

Commit Message

Ted Elhourani Jan. 4, 2019, 2:16 a.m.
The Python IDL implementation supports ovsdb cluster connections.
This patch is a follow up to commit 31e434fc98, it adds the option of
connecting to the leader (the default) in the Raft-based cluster. It mimics
the exisiting C IDL support for clusters introduced in commit 1b1d2e6daa.

The _Server database schema is first requested, then a monitor of the
Database table in the _Server Database. Method __check_server_db verifies
the eligibility of the server. If the attempt to obtain a monitor of the
_Server database fails and a cluster id was not provided this implementation
proceeds to request the data monitor. If a cluster id was provided via the
set_cluster_id method then the connection is aborted and a connection to a
different node is instead attempted, until a valid cluster node is found.
Thus, when supplied, cluster id is interpreted as the intention to only
allow connections to a clustered database. If not supplied, connections to
standalone nodes, or nodes that do not have the _Server database are
allowed. change_seqno is not incremented in the case of Database table
updates.
Signed-off-by: Ted Elhourani <ted.elhourani@nutanix.com>
---
 python/ovs/db/idl.py    | 217 ++++++++++++++++++++++++++++++++++++++++++++----
 python/ovs/reconnect.py |   3 +
 tests/ovsdb-idl.at      |  62 +++++++-------
 3 files changed, 237 insertions(+), 45 deletions(-)

Comments

Numan Siddique Jan. 4, 2019, 11:56 a.m. | #1
Hi Ted,

This patch is failing the below test cases for me. All are python3 related.
Can you please
take a look if that's the case with you as well ?
******
2139: simple idl, writing via IDL with unicode - Python3 FAILED (
ovsdb-idl.at:456)
2141: simple idl, writing via IDL with unicode - Python3 - register_columns
FAILED (ovsdb-idl.at:456)
2143: simple idl, writing via IDL with unicode - Python3 - tcp FAILED (
ovsdb-idl.at:456)
2145: simple idl, writing via IDL with unicode - Python3 (multiple remotes)
- tcp FAILED (ovsdb-idl.at:456)
2147: simple idl, writing via IDL with unicode - Python3 - tcp6 FAILED (
ovsdb-idl.at:456)
2149: simple idl, writing via IDL with unicode - Python3 (multiple remotes)
- tcp6 FAILED (ovsdb-idl.at:456)
2151: simple idl, writing via IDL with unicode - Python3 - SSL FAILED (
ovsdb-idl.at:456)
2153: simple idl, writing large data via IDL with unicode - Python3 FAILED (
ovsdb-idl.at:490)
**********

Please see some comments below.

Thanks
Numan

On Fri, Jan 4, 2019 at 7:47 AM Ted Elhourani <ted.elhourani@nutanix.com>
wrote:

> The Python IDL implementation supports ovsdb cluster connections.
> This patch is a follow up to commit 31e434fc98, it adds the option of
> connecting to the leader (the default) in the Raft-based cluster. It mimics
> the exisiting C IDL support for clusters introduced in commit 1b1d2e6daa.
>
> The _Server database schema is first requested, then a monitor of the
> Database table in the _Server Database. Method __check_server_db verifies
> the eligibility of the server. If the attempt to obtain a monitor of the
> _Server database fails and a cluster id was not provided this
> implementation
> proceeds to request the data monitor. If a cluster id was provided via the
> set_cluster_id method then the connection is aborted and a connection to a
> different node is instead attempted, until a valid cluster node is found.
> Thus, when supplied, cluster id is interpreted as the intention to only
> allow connections to a clustered database. If not supplied, connections to
> standalone nodes, or nodes that do not have the _Server database are
> allowed. change_seqno is not incremented in the case of Database table
> updates.
> Signed-off-by: Ted Elhourani <ted.elhourani@nutanix.com>
> ---
>  python/ovs/db/idl.py    | 217
> ++++++++++++++++++++++++++++++++++++++++++++----
>  python/ovs/reconnect.py |   3 +
>  tests/ovsdb-idl.at      |  62 +++++++-------
>  3 files changed, 237 insertions(+), 45 deletions(-)
>
> diff --git a/python/ovs/db/idl.py b/python/ovs/db/idl.py
> index 250e897..f989548 100644
> --- a/python/ovs/db/idl.py
> +++ b/python/ovs/db/idl.py
> @@ -38,6 +38,7 @@ ROW_DELETE = "delete"
>  OVSDB_UPDATE = 0
>  OVSDB_UPDATE2 = 1
>
> +CLUSTERED = "clustered"
>
>  class Idl(object):
>      """Open vSwitch Database Interface Definition Language (OVSDB IDL).
> @@ -92,10 +93,12 @@ class Idl(object):
>  """
>
>      IDL_S_INITIAL = 0
> -    IDL_S_MONITOR_REQUESTED = 1
> -    IDL_S_MONITOR_COND_REQUESTED = 2
> +    IDL_S_SERVER_SCHEMA_REQUESTED = 1
> +    IDL_S_SERVER_MONITOR_REQUESTED = 2
> +    IDL_S_DATA_MONITOR_REQUESTED = 3
> +    IDL_S_DATA_MONITOR_COND_REQUESTED = 4
>
> -    def __init__(self, remote, schema_helper, probe_interval=None):
> +    def __init__(self, remote, schema_helper, leader_only=True,
> probe_interval=None):
>

Can you please add the "leader_only" arg at the end. Older clients of
python idl would
break if they were creating Idl object as
my_idl = Idl(remote, helper, 1000)


         """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
> @@ -119,6 +122,9 @@ class Idl(object):
>
>          The IDL uses and modifies 'schema' directly.
>
> +        If 'leader_only' is set to True (default value) the IDL will only
> monitor
> +        and transact with the leader of the cluster.
> +
>          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.
> @@ -137,6 +143,20 @@ class Idl(object):
>          self._last_seqno = None
>          self.change_seqno = 0
>          self.uuid = uuid.uuid1()
> +
> +        # Server monitor.
> +        self._server_schema_request_id = None
> +        self._server_monitor_request_id = None
> +        self._db_change_aware_request_id = None
> +        self._server_db_name = '_Server'
> +        self._server_db_table = 'Database'
> +        self.server_tables = None
> +        self._server_db = None
> +        self.server_monitor_uuid = uuid.uuid1()
> +        self.leader_only = leader_only
> +        self.cluster_id = None
> +        self._min_index = 0
> +
>          self.state = self.IDL_S_INITIAL
>
>          # Database locking.
> @@ -172,6 +192,15 @@ class Idl(object):
>                  remotes.append(r)
>          return remotes
>
> +    def set_cluster_id(self, cluster_id):
> +        """Set the id of the cluster that this idl must connect to."""
> +        if cluster_id:
> +            self.cluster_id = str(cluster_id)
> +        else:
> +            self.cluster_id = None
>

Instead of the "if - else", how about just
self.cluster_id = cluster_id

Can cluster_id be passed as an integer by the caller ?



> +        if self.state != self.IDL_S_INITIAL:
> +            self.force_reconnect()
> +
>      def index_create(self, table, name):
>          """Create a named multi-column index on a table"""
>          return self.tables[table].rows.index_create(name)
> @@ -222,7 +251,7 @@ class Idl(object):
>              if seqno != self._last_seqno:
>                  self._last_seqno = seqno
>                  self.__txn_abort_all()
> -                self.__send_monitor_request()
> +                self.__send_server_schema_request()
>                  if self.lock_name:
>                      self.__send_lock_request()
>                  break
> @@ -230,6 +259,7 @@ class Idl(object):
>              msg = self._session.recv()
>              if msg is None:
>                  break
> +
>              if (msg.type == ovs.jsonrpc.Message.T_NOTIFY
>                      and msg.method == "update2"
>                      and len(msg.params) == 2):
> @@ -239,7 +269,15 @@ class Idl(object):
>                      and msg.method == "update"
>                      and len(msg.params) == 2):
>                  # Database contents changed.
> -                self.__parse_update(msg.params[1], OVSDB_UPDATE)
> +                if msg.params[0] == str(self.server_monitor_uuid):
> +                    self.__parse_update(msg.params[1], OVSDB_UPDATE,
> +                                        tables=self.server_tables)
> +                    self.change_seqno = initial_change_seqno
> +                    if not self.__check_server_db():
> +                        self.force_reconnect()
> +                        break
> +                else:
> +                    self.__parse_update(msg.params[1], OVSDB_UPDATE)
>              elif (msg.type == ovs.jsonrpc.Message.T_REPLY
>                    and self._monitor_request_id is not None
>                    and self._monitor_request_id == msg.id):
> @@ -248,17 +286,66 @@ class Idl(object):
>                      self.change_seqno += 1
>                      self._monitor_request_id = None
>                      self.__clear()
> -                    if self.state == self.IDL_S_MONITOR_COND_REQUESTED:
> +                    if self.state ==
> self.IDL_S_DATA_MONITOR_COND_REQUESTED:
>                          self.__parse_update(msg.result, OVSDB_UPDATE2)
>                      else:
> -                        assert self.state == self.IDL_S_MONITOR_REQUESTED
> +                        assert self.state ==
> self.IDL_S_DATA_MONITOR_REQUESTED
>                          self.__parse_update(msg.result, OVSDB_UPDATE)
> -
>                  except error.Error as e:
>                      vlog.err("%s: parse error in received schema: %s"
>                               % (self._session.get_name(), e))
>                      self.__error()
>              elif (msg.type == ovs.jsonrpc.Message.T_REPLY
> +                  and self._server_schema_request_id is not None
> +                  and self._server_schema_request_id == msg.id):
> +                # Reply to our "get_schema" of _Server request.
> +                try:
> +                    self._server_schema_request_id = None
> +                    sh = SchemaHelper(None, msg.result)
> +                    sh.register_table(self._server_db_table)
> +                    schema = sh.get_idl_schema()
> +                    self._server_db = schema
> +                    self.server_tables = schema.tables
> +                    self.__send_server_monitor_request()
> +                except error.Error as e:
> +                    vlog.err("%s: error receiving server schema: %s"
> +                             % (self._session.get_name(), e))
> +                    if self.cluster_id:
> +                        self.__error()
> +                        break
> +                    else:
> +                        self.change_seqno = initial_change_seqno
> +                        self.__send_monitor_request()
> +            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
> +                  and self._server_monitor_request_id is not None
> +                  and self._server_monitor_request_id == msg.id):
> +                # Reply to our "monitor" of _Server request.
> +                try:
> +                    self._server_monitor_request_id = None
> +                    self.__parse_update(msg.result, OVSDB_UPDATE,
> +                                        tables=self.server_tables)
> +                    self.change_seqno = initial_change_seqno
> +                    if self.__check_server_db():
> +                        self.__send_monitor_request()
> +                        self.__send_db_change_aware()
> +                    else:
> +                        self.force_reconnect()
> +                        break
> +                except error.Error as e:
> +                    vlog.err("%s: parse error in received schema: %s"
> +                             % (self._session.get_name(), e))
> +                    if self.cluster_id:
> +                        self.__error()
> +                        break
> +                    else:
> +                        self.change_seqno = initial_change_seqno
> +                        self.__send_monitor_request()
> +            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
> +                  and self._db_change_aware_request_id is not None
> +                  and self._db_change_aware_request_id == msg.id):
> +                # Reply to us notifying the server of our change awarness.
> +                self._db_change_aware_request_id = None
> +            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
>                    and self._lock_request_id is not None
>                    and self._lock_request_id == msg.id):
>                  # Reply to our "lock" request.
> @@ -275,10 +362,20 @@ class Idl(object):
>                  # Reply to our echo request.  Ignore it.
>                  pass
>              elif (msg.type == ovs.jsonrpc.Message.T_ERROR and
> -                  self.state == self.IDL_S_MONITOR_COND_REQUESTED 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()
> +            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):
> +                self._server_schema_request_id = None
> +                if self.cluster_id:
> +                    self.force_reconnect()
> +                    break
> +                else:
> +                    self.change_seqno = initial_change_seqno
> +                    self.__send_monitor_request()
>              elif (msg.type in (ovs.jsonrpc.Message.T_ERROR,
>                                 ovs.jsonrpc.Message.T_REPLY)
>                    and self.__txn_process_reply(msg)):
> @@ -440,12 +537,19 @@ class Idl(object):
>              if not new_has_lock:
>                  self.is_lock_contended = True
>
> +    def __send_db_change_aware(self):
> +        msg = ovs.jsonrpc.Message.create_request("set_db_change_aware",
> +                                                 [True])
> +        self._db_change_aware_request_id = msg.id
> +        self._session.send(msg)
> +
>      def __send_monitor_request(self):
> -        if self.state == self.IDL_S_INITIAL:
> -            self.state = self.IDL_S_MONITOR_COND_REQUESTED
> +        if (self.state in [self.IDL_S_SERVER_MONITOR_REQUESTED,
> +                           self.IDL_S_INITIAL]):
> +            self.state = self.IDL_S_DATA_MONITOR_COND_REQUESTED
>              method = "monitor_cond"
>          else:
> -            self.state = self.IDL_S_MONITOR_REQUESTED
> +            self.state = self.IDL_S_DATA_MONITOR_REQUESTED
>              method = "monitor"
>
>          monitor_requests = {}
> @@ -467,20 +571,50 @@ class Idl(object):
>          self._monitor_request_id = msg.id
>          self._session.send(msg)
>
> -    def __parse_update(self, update, version):
> +    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
> +        res = self._session.send(msg)
> +
> +    def __send_server_monitor_request(self):
> +        self.state = self.IDL_S_SERVER_MONITOR_REQUESTED
> +        monitor_requests = {}
> +        table = self.server_tables[self._server_db_table]
> +        columns = [column for column in six.iterkeys(table.columns)]
> +        for column in six.itervalues(table.columns):
> +            if not hasattr(column, 'alert'):
> +                column.alert = True
> +        table.rows = custom_index.IndexedRows(table)
> +        table.need_table = False
> +        table.idl = self
> +        monitor_request = {"columns": columns}
> +        monitor_requests[table.name] = [monitor_request]
> +        msg = ovs.jsonrpc.Message.create_request(
> +            'monitor', [self._server_db.name,
> +                             str(self.server_monitor_uuid),
> +                             monitor_requests])
> +        self._server_monitor_request_id = msg.id
> +        self._session.send(msg)
> +
> +    def __parse_update(self, update, version, tables=None):
>          try:
> -            self.__do_parse_update(update, version)
> +            if not tables:
> +                self.__do_parse_update(update, version, self.tables)
> +            else:
> +                self.__do_parse_update(update, version, tables)
>          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):
> +    def __do_parse_update(self, table_updates, version, tables):
>          if not isinstance(table_updates, dict):
>              raise error.Error("<table-updates> is not an object",
>                                table_updates)
>
>          for table_name, table_update in six.iteritems(table_updates):
> -            table = self.tables.get(table_name)
> +            table = tables.get(table_name)
>              if not table:
>                  raise error.Error('<table-updates> includes unknown '
>                                    'table "%s"' % table_name)
> @@ -605,6 +739,57 @@ class Idl(object):
>                  self.notify(op, row, Row.from_json(self, table, uuid,
> old))
>          return changed
>
> +    def __check_server_db(self):
> +        """Returns True if this is a valid ovsdb server, False
> otherwise."""
>
The function name seems a bit confusing.
This function   returns True if the server has "_Server" table and if it
is a clustered db, additional checks are done.
May be this function can be broken in 2 separate functions ?
One to check if the server supports '_Server' db and other to check
the status of cluster.


+        session_name = self._session.get_name()
> +
> +        if self._server_db_table not in self.server_tables:
> +            vlog.info("%s: server does not have %s table in its %s
> database"
> +                      % (session_name, self._server_db_table,
> self._server_db_name))
> +            return False
> +
> +        rows = self.server_tables[self._server_db_table].rows
> +
> +        database = None
> +        for row in six.itervalues(rows):
> +            if self.cluster_id:
> +                if self.cluster_id in \
> +                   map(lambda x: str(x)[:4], row.cid):
> +                    database = row
> +                    break
> +            elif row.name == self._db.name:
> +                database = row
> +                break
> +
> +        if not database:
> +            vlog.info("%s: server does not have %s database"
> +                      % (session_name, self._db.name))
> +            return False
> +
> +        if (database.model == CLUSTERED and
> +            self._session.get_num_of_remotes() > 1):
> +            if not database.schema:
> +                vlog.info('%s: clustered database server has not yet
> joined '
> +                          'cluster; trying another server' % session_name)
> +                return False
> +            if not database.connected:
> +                vlog.info('%s: clustered database server is disconnected
> '
> +                          'from cluster; trying another server' %
> session_name)
> +                return False
> +            if (self.leader_only and
> +                not database.leader):
> +                vlog.info('%s: clustered database server is not cluster '
> +                          'leader; trying another server' % session_name)
> +                return False
> +            if database.index:
> +                if database.index[0] < self._min_index:
> +                    vlog.warn('%s: clustered database server has stale
> data; '
> +                              'trying another server' % session_name)
> +                    return False
> +                self._min_index = database.index[0]
> +
> +        return True
> +
>      def __column_name(self, column):
>          if column.type.key.type == ovs.db.types.UuidType:
>              return ovs.ovsuuid.to_json(column.type.key.type.default)
> diff --git a/python/ovs/reconnect.py b/python/ovs/reconnect.py
> index 34cc769..afbe445 100644
> --- a/python/ovs/reconnect.py
> +++ b/python/ovs/reconnect.py
> @@ -344,6 +344,9 @@ class Reconnect(object):
>                  else:
>                      self.info_level("%s: error listening for connections"
>                                      % self.name)
> +            elif self.state == Reconnect.Reconnect:
> +                self.info_level("%s: connection closed by client"
> +                                % self.name)
>              elif self.backoff < self.max_backoff:
>                  if self.passive:
>                      type_ = "listen"
> diff --git a/tests/ovsdb-idl.at b/tests/ovsdb-idl.at
> index 8981b5e..46b047f 100644
> --- a/tests/ovsdb-idl.at
> +++ b/tests/ovsdb-idl.at
> @@ -1466,40 +1466,44 @@ OVSDB_CHECK_IDL_NOTIFY([simple idl verify notify],
>         "where": [["i", "==", 0]]}]' \
>      'reconnect']],
>    [[000: empty
> -001:
> {"error":null,"result":[{"uuid":["uuid","<0>"]},{"uuid":["uuid","<1>"]}]}
> -002: event:create, row={i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[]
> ua=[] uuid=<1>}, updates=None
> -002: event:create, row={i=1 r=2 b=true s=mystring u=<3> ia=[1 2 3]
> ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates=None
> -002: i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
> -002: i=1 r=2 b=true s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<4> <5>] uuid=<0>
> +000: event:create, row={uuid=<0>}, updates=None
> +000: event:create, row={uuid=<1>}, updates=None
> +001:
> {"error":null,"result":[{"uuid":["uuid","<2>"]},{"uuid":["uuid","<3>"]}]}
> +002: event:create, row={i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[]
> ua=[] uuid=<3>}, updates=None
> +002: event:create, row={i=1 r=2 b=true s=mystring u=<5> ia=[1 2 3]
> ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates=None
> +002: i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
> +002: i=1 r=2 b=true s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<6> <7>] uuid=<2>
>  003: {"error":null,"result":[{"count":2}]}
> -004: event:update, row={i=1 r=2 b=false s=mystring u=<3> ia=[1 2 3]
> ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates={b=true
> uuid=<0>}
> -004: i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
> -004: i=1 r=2 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<4> <5>] uuid=<0>
> +004: event:update, row={i=1 r=2 b=false s=mystring u=<5> ia=[1 2 3]
> ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates={b=true
> uuid=<2>}
> +004: i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
> +004: i=1 r=2 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<6> <7>] uuid=<2>
>  005: {"error":null,"result":[{"count":2}]}
> -006: event:update, row={i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[]
> sa=[] ua=[] uuid=<1>}, updates={r=0 uuid=<1>}
> -006: event:update, row={i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3]
> ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates={r=2
> uuid=<0>}
> -006: i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
> -006: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<4> <5>] uuid=<0>
> -007: {"error":null,"result":[{"uuid":["uuid","<6>"]}]}
> -008: event:create, row={i=-1 r=125 b=false s= u=<2> ia=[1] ra=[1.5]
> ba=[false] sa=[] ua=[] uuid=<6>}, updates=None
> -008: i=-1 r=125 b=false s= u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[]
> uuid=<6>
> -008: i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
> -008: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<4> <5>] uuid=<0>
> +006: event:update, row={i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[]
> sa=[] ua=[] uuid=<3>}, updates={r=0 uuid=<3>}
> +006: event:update, row={i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3]
> ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates={r=2
> uuid=<2>}
> +006: i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
> +006: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<6> <7>] uuid=<2>
> +007: {"error":null,"result":[{"uuid":["uuid","<8>"]}]}
> +008: event:create, row={i=-1 r=125 b=false s= u=<4> ia=[1] ra=[1.5]
> ba=[false] sa=[] ua=[] uuid=<8>}, updates=None
> +008: i=-1 r=125 b=false s= u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[]
> uuid=<8>
> +008: i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
> +008: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<6> <7>] uuid=<2>
>  009: {"error":null,"result":[{"count":2}]}
> -010: event:update, row={i=-1 r=125 b=false s=newstring u=<2> ia=[1]
> ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>}, updates={s= uuid=<6>}
> -010: event:update, row={i=0 r=123.5 b=false s=newstring u=<2> ia=[] ra=[]
> ba=[] sa=[] ua=[] uuid=<1>}, updates={s= uuid=<1>}
> -010: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false]
> sa=[] ua=[] uuid=<6>
> -010: i=0 r=123.5 b=false s=newstring u=<2> ia=[] ra=[] ba=[] sa=[] ua=[]
> uuid=<1>
> -010: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<4> <5>] uuid=<0>
> +010: event:update, row={i=-1 r=125 b=false s=newstring u=<4> ia=[1]
> ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>}, updates={s= uuid=<8>}
> +010: event:update, row={i=0 r=123.5 b=false s=newstring u=<4> ia=[] ra=[]
> ba=[] sa=[] ua=[] uuid=<3>}, updates={s= uuid=<3>}
> +010: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false]
> sa=[] ua=[] uuid=<8>
> +010: i=0 r=123.5 b=false s=newstring u=<4> ia=[] ra=[] ba=[] sa=[] ua=[]
> uuid=<3>
> +010: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<6> <7>] uuid=<2>
>  011: {"error":null,"result":[{"count":1}]}
> -012: event:delete, row={i=0 r=123.5 b=false s=newstring u=<2> ia=[] ra=[]
> ba=[] sa=[] ua=[] uuid=<1>}, updates=None
> -012: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false]
> sa=[] ua=[] uuid=<6>
> -012: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<4> <5>] uuid=<0>
> +012: event:delete, row={i=0 r=123.5 b=false s=newstring u=<4> ia=[] ra=[]
> ba=[] sa=[] ua=[] uuid=<3>}, updates=None
> +012: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false]
> sa=[] ua=[] uuid=<8>
> +012: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<6> <7>] uuid=<2>
>  013: reconnect
> -014: event:create, row={i=-1 r=125 b=false s=newstring u=<2> ia=[1]
> ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>}, updates=None
> -014: event:create, row={i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3]
> ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates=None
> -014: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false]
> sa=[] ua=[] uuid=<6>
> -014: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<4> <5>] uuid=<0>
> +014: event:create, row={i=-1 r=125 b=false s=newstring u=<4> ia=[1]
> ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>}, updates=None
> +014: event:create, row={i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3]
> ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates=None
> +014: event:create, row={uuid=<0>}, updates=None
> +014: event:create, row={uuid=<1>}, updates=None
> +014: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false]
> sa=[] ua=[] uuid=<8>
> +014: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
> sa=[abc def] ua=[<6> <7>] uuid=<2>
>  015: done
>  ]])
>
>
I think it would be good if we have few test cases which starts a cluster
of ovsdb-server's and then
   -verify if the client has connected to the cluster leader
   - bring down the leader and check if the client connects to the new
leader.


Thanks
Numan
Numan Siddique Jan. 11, 2019, 10:23 a.m. | #2
On Fri, Jan 4, 2019 at 5:26 PM Numan Siddique <nusiddiq@redhat.com> wrote:

> Hi Ted,
>
> This patch is failing the below test cases for me. All are python3
> related. Can you please
> take a look if that's the case with you as well ?
> ******
> 2139: simple idl, writing via IDL with unicode - Python3 FAILED (
> ovsdb-idl.at:456)
> 2141: simple idl, writing via IDL with unicode - Python3 -
> register_columns FAILED (ovsdb-idl.at:456)
> 2143: simple idl, writing via IDL with unicode - Python3 - tcp FAILED (
> ovsdb-idl.at:456)
> 2145: simple idl, writing via IDL with unicode - Python3 (multiple
> remotes) - tcp FAILED (ovsdb-idl.at:456)
> 2147: simple idl, writing via IDL with unicode - Python3 - tcp6 FAILED (
> ovsdb-idl.at:456)
> 2149: simple idl, writing via IDL with unicode - Python3 (multiple
> remotes) - tcp6 FAILED (ovsdb-idl.at:456)
> 2151: simple idl, writing via IDL with unicode - Python3 - SSL FAILED (
> ovsdb-idl.at:456)
> 2153: simple idl, writing large data via IDL with unicode - Python3 FAILED
> (ovsdb-idl.at:490)
> **********
>

These failures are not related to this patch. Please ignore. There seems to
be some issues with python3.7 with unicode
in Fedora 29 and I think Timothy  is planning to submit a patch to fix
these failures.

Thanks
Numan


> Please see some comments below.
>
> Thanks
> Numan
>
> On Fri, Jan 4, 2019 at 7:47 AM Ted Elhourani <ted.elhourani@nutanix.com>
> wrote:
>
>> The Python IDL implementation supports ovsdb cluster connections.
>> This patch is a follow up to commit 31e434fc98, it adds the option of
>> connecting to the leader (the default) in the Raft-based cluster. It
>> mimics
>> the exisiting C IDL support for clusters introduced in commit 1b1d2e6daa.
>>
>> The _Server database schema is first requested, then a monitor of the
>> Database table in the _Server Database. Method __check_server_db verifies
>> the eligibility of the server. If the attempt to obtain a monitor of the
>> _Server database fails and a cluster id was not provided this
>> implementation
>> proceeds to request the data monitor. If a cluster id was provided via the
>> set_cluster_id method then the connection is aborted and a connection to a
>> different node is instead attempted, until a valid cluster node is found.
>> Thus, when supplied, cluster id is interpreted as the intention to only
>> allow connections to a clustered database. If not supplied, connections to
>> standalone nodes, or nodes that do not have the _Server database are
>> allowed. change_seqno is not incremented in the case of Database table
>> updates.
>> Signed-off-by: Ted Elhourani <ted.elhourani@nutanix.com>
>> ---
>>  python/ovs/db/idl.py    | 217
>> ++++++++++++++++++++++++++++++++++++++++++++----
>>  python/ovs/reconnect.py |   3 +
>>  tests/ovsdb-idl.at      |  62 +++++++-------
>>  3 files changed, 237 insertions(+), 45 deletions(-)
>>
>> diff --git a/python/ovs/db/idl.py b/python/ovs/db/idl.py
>> index 250e897..f989548 100644
>> --- a/python/ovs/db/idl.py
>> +++ b/python/ovs/db/idl.py
>> @@ -38,6 +38,7 @@ ROW_DELETE = "delete"
>>  OVSDB_UPDATE = 0
>>  OVSDB_UPDATE2 = 1
>>
>> +CLUSTERED = "clustered"
>>
>>  class Idl(object):
>>      """Open vSwitch Database Interface Definition Language (OVSDB IDL).
>> @@ -92,10 +93,12 @@ class Idl(object):
>>  """
>>
>>      IDL_S_INITIAL = 0
>> -    IDL_S_MONITOR_REQUESTED = 1
>> -    IDL_S_MONITOR_COND_REQUESTED = 2
>> +    IDL_S_SERVER_SCHEMA_REQUESTED = 1
>> +    IDL_S_SERVER_MONITOR_REQUESTED = 2
>> +    IDL_S_DATA_MONITOR_REQUESTED = 3
>> +    IDL_S_DATA_MONITOR_COND_REQUESTED = 4
>>
>> -    def __init__(self, remote, schema_helper, probe_interval=None):
>> +    def __init__(self, remote, schema_helper, leader_only=True,
>> probe_interval=None):
>>
>
> Can you please add the "leader_only" arg at the end. Older clients of
> python idl would
> break if they were creating Idl object as
> my_idl = Idl(remote, helper, 1000)
>
>
>          """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
>> @@ -119,6 +122,9 @@ class Idl(object):
>>
>>          The IDL uses and modifies 'schema' directly.
>>
>> +        If 'leader_only' is set to True (default value) the IDL will
>> only monitor
>> +        and transact with the leader of the cluster.
>> +
>>          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.
>> @@ -137,6 +143,20 @@ class Idl(object):
>>          self._last_seqno = None
>>          self.change_seqno = 0
>>          self.uuid = uuid.uuid1()
>> +
>> +        # Server monitor.
>> +        self._server_schema_request_id = None
>> +        self._server_monitor_request_id = None
>> +        self._db_change_aware_request_id = None
>> +        self._server_db_name = '_Server'
>> +        self._server_db_table = 'Database'
>> +        self.server_tables = None
>> +        self._server_db = None
>> +        self.server_monitor_uuid = uuid.uuid1()
>> +        self.leader_only = leader_only
>> +        self.cluster_id = None
>> +        self._min_index = 0
>> +
>>          self.state = self.IDL_S_INITIAL
>>
>>          # Database locking.
>> @@ -172,6 +192,15 @@ class Idl(object):
>>                  remotes.append(r)
>>          return remotes
>>
>> +    def set_cluster_id(self, cluster_id):
>> +        """Set the id of the cluster that this idl must connect to."""
>> +        if cluster_id:
>> +            self.cluster_id = str(cluster_id)
>> +        else:
>> +            self.cluster_id = None
>>
>
> Instead of the "if - else", how about just
> self.cluster_id = cluster_id
>
> Can cluster_id be passed as an integer by the caller ?
>
>
>
>> +        if self.state != self.IDL_S_INITIAL:
>> +            self.force_reconnect()
>> +
>>      def index_create(self, table, name):
>>          """Create a named multi-column index on a table"""
>>          return self.tables[table].rows.index_create(name)
>> @@ -222,7 +251,7 @@ class Idl(object):
>>              if seqno != self._last_seqno:
>>                  self._last_seqno = seqno
>>                  self.__txn_abort_all()
>> -                self.__send_monitor_request()
>> +                self.__send_server_schema_request()
>>                  if self.lock_name:
>>                      self.__send_lock_request()
>>                  break
>> @@ -230,6 +259,7 @@ class Idl(object):
>>              msg = self._session.recv()
>>              if msg is None:
>>                  break
>> +
>>              if (msg.type == ovs.jsonrpc.Message.T_NOTIFY
>>                      and msg.method == "update2"
>>                      and len(msg.params) == 2):
>> @@ -239,7 +269,15 @@ class Idl(object):
>>                      and msg.method == "update"
>>                      and len(msg.params) == 2):
>>                  # Database contents changed.
>> -                self.__parse_update(msg.params[1], OVSDB_UPDATE)
>> +                if msg.params[0] == str(self.server_monitor_uuid):
>> +                    self.__parse_update(msg.params[1], OVSDB_UPDATE,
>> +                                        tables=self.server_tables)
>> +                    self.change_seqno = initial_change_seqno
>> +                    if not self.__check_server_db():
>> +                        self.force_reconnect()
>> +                        break
>> +                else:
>> +                    self.__parse_update(msg.params[1], OVSDB_UPDATE)
>>              elif (msg.type == ovs.jsonrpc.Message.T_REPLY
>>                    and self._monitor_request_id is not None
>>                    and self._monitor_request_id == msg.id):
>> @@ -248,17 +286,66 @@ class Idl(object):
>>                      self.change_seqno += 1
>>                      self._monitor_request_id = None
>>                      self.__clear()
>> -                    if self.state == self.IDL_S_MONITOR_COND_REQUESTED:
>> +                    if self.state ==
>> self.IDL_S_DATA_MONITOR_COND_REQUESTED:
>>                          self.__parse_update(msg.result, OVSDB_UPDATE2)
>>                      else:
>> -                        assert self.state == self.IDL_S_MONITOR_REQUESTED
>> +                        assert self.state ==
>> self.IDL_S_DATA_MONITOR_REQUESTED
>>                          self.__parse_update(msg.result, OVSDB_UPDATE)
>> -
>>                  except error.Error as e:
>>                      vlog.err("%s: parse error in received schema: %s"
>>                               % (self._session.get_name(), e))
>>                      self.__error()
>>              elif (msg.type == ovs.jsonrpc.Message.T_REPLY
>> +                  and self._server_schema_request_id is not None
>> +                  and self._server_schema_request_id == msg.id):
>> +                # Reply to our "get_schema" of _Server request.
>> +                try:
>> +                    self._server_schema_request_id = None
>> +                    sh = SchemaHelper(None, msg.result)
>> +                    sh.register_table(self._server_db_table)
>> +                    schema = sh.get_idl_schema()
>> +                    self._server_db = schema
>> +                    self.server_tables = schema.tables
>> +                    self.__send_server_monitor_request()
>> +                except error.Error as e:
>> +                    vlog.err("%s: error receiving server schema: %s"
>> +                             % (self._session.get_name(), e))
>> +                    if self.cluster_id:
>> +                        self.__error()
>> +                        break
>> +                    else:
>> +                        self.change_seqno = initial_change_seqno
>> +                        self.__send_monitor_request()
>> +            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
>> +                  and self._server_monitor_request_id is not None
>> +                  and self._server_monitor_request_id == msg.id):
>> +                # Reply to our "monitor" of _Server request.
>> +                try:
>> +                    self._server_monitor_request_id = None
>> +                    self.__parse_update(msg.result, OVSDB_UPDATE,
>> +                                        tables=self.server_tables)
>> +                    self.change_seqno = initial_change_seqno
>> +                    if self.__check_server_db():
>> +                        self.__send_monitor_request()
>> +                        self.__send_db_change_aware()
>> +                    else:
>> +                        self.force_reconnect()
>> +                        break
>> +                except error.Error as e:
>> +                    vlog.err("%s: parse error in received schema: %s"
>> +                             % (self._session.get_name(), e))
>> +                    if self.cluster_id:
>> +                        self.__error()
>> +                        break
>> +                    else:
>> +                        self.change_seqno = initial_change_seqno
>> +                        self.__send_monitor_request()
>> +            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
>> +                  and self._db_change_aware_request_id is not None
>> +                  and self._db_change_aware_request_id == msg.id):
>> +                # Reply to us notifying the server of our change
>> awarness.
>> +                self._db_change_aware_request_id = None
>> +            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
>>                    and self._lock_request_id is not None
>>                    and self._lock_request_id == msg.id):
>>                  # Reply to our "lock" request.
>> @@ -275,10 +362,20 @@ class Idl(object):
>>                  # Reply to our echo request.  Ignore it.
>>                  pass
>>              elif (msg.type == ovs.jsonrpc.Message.T_ERROR and
>> -                  self.state == self.IDL_S_MONITOR_COND_REQUESTED 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()
>> +            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):
>> +                self._server_schema_request_id = None
>> +                if self.cluster_id:
>> +                    self.force_reconnect()
>> +                    break
>> +                else:
>> +                    self.change_seqno = initial_change_seqno
>> +                    self.__send_monitor_request()
>>              elif (msg.type in (ovs.jsonrpc.Message.T_ERROR,
>>                                 ovs.jsonrpc.Message.T_REPLY)
>>                    and self.__txn_process_reply(msg)):
>> @@ -440,12 +537,19 @@ class Idl(object):
>>              if not new_has_lock:
>>                  self.is_lock_contended = True
>>
>> +    def __send_db_change_aware(self):
>> +        msg = ovs.jsonrpc.Message.create_request("set_db_change_aware",
>> +                                                 [True])
>> +        self._db_change_aware_request_id = msg.id
>> +        self._session.send(msg)
>> +
>>      def __send_monitor_request(self):
>> -        if self.state == self.IDL_S_INITIAL:
>> -            self.state = self.IDL_S_MONITOR_COND_REQUESTED
>> +        if (self.state in [self.IDL_S_SERVER_MONITOR_REQUESTED,
>> +                           self.IDL_S_INITIAL]):
>> +            self.state = self.IDL_S_DATA_MONITOR_COND_REQUESTED
>>              method = "monitor_cond"
>>          else:
>> -            self.state = self.IDL_S_MONITOR_REQUESTED
>> +            self.state = self.IDL_S_DATA_MONITOR_REQUESTED
>>              method = "monitor"
>>
>>          monitor_requests = {}
>> @@ -467,20 +571,50 @@ class Idl(object):
>>          self._monitor_request_id = msg.id
>>          self._session.send(msg)
>>
>> -    def __parse_update(self, update, version):
>> +    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
>> +        res = self._session.send(msg)
>> +
>> +    def __send_server_monitor_request(self):
>> +        self.state = self.IDL_S_SERVER_MONITOR_REQUESTED
>> +        monitor_requests = {}
>> +        table = self.server_tables[self._server_db_table]
>> +        columns = [column for column in six.iterkeys(table.columns)]
>> +        for column in six.itervalues(table.columns):
>> +            if not hasattr(column, 'alert'):
>> +                column.alert = True
>> +        table.rows = custom_index.IndexedRows(table)
>> +        table.need_table = False
>> +        table.idl = self
>> +        monitor_request = {"columns": columns}
>> +        monitor_requests[table.name] = [monitor_request]
>> +        msg = ovs.jsonrpc.Message.create_request(
>> +            'monitor', [self._server_db.name,
>> +                             str(self.server_monitor_uuid),
>> +                             monitor_requests])
>> +        self._server_monitor_request_id = msg.id
>> +        self._session.send(msg)
>> +
>> +    def __parse_update(self, update, version, tables=None):
>>          try:
>> -            self.__do_parse_update(update, version)
>> +            if not tables:
>> +                self.__do_parse_update(update, version, self.tables)
>> +            else:
>> +                self.__do_parse_update(update, version, tables)
>>          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):
>> +    def __do_parse_update(self, table_updates, version, tables):
>>          if not isinstance(table_updates, dict):
>>              raise error.Error("<table-updates> is not an object",
>>                                table_updates)
>>
>>          for table_name, table_update in six.iteritems(table_updates):
>> -            table = self.tables.get(table_name)
>> +            table = tables.get(table_name)
>>              if not table:
>>                  raise error.Error('<table-updates> includes unknown '
>>                                    'table "%s"' % table_name)
>> @@ -605,6 +739,57 @@ class Idl(object):
>>                  self.notify(op, row, Row.from_json(self, table, uuid,
>> old))
>>          return changed
>>
>> +    def __check_server_db(self):
>> +        """Returns True if this is a valid ovsdb server, False
>> otherwise."""
>>
> The function name seems a bit confusing.
> This function   returns True if the server has "_Server" table and if it
> is a clustered db, additional checks are done.
> May be this function can be broken in 2 separate functions ?
> One to check if the server supports '_Server' db and other to check
> the status of cluster.
>
>
> +        session_name = self._session.get_name()
>> +
>> +        if self._server_db_table not in self.server_tables:
>> +            vlog.info("%s: server does not have %s table in its %s
>> database"
>> +                      % (session_name, self._server_db_table,
>> self._server_db_name))
>> +            return False
>> +
>> +        rows = self.server_tables[self._server_db_table].rows
>> +
>> +        database = None
>> +        for row in six.itervalues(rows):
>> +            if self.cluster_id:
>> +                if self.cluster_id in \
>> +                   map(lambda x: str(x)[:4], row.cid):
>> +                    database = row
>> +                    break
>> +            elif row.name == self._db.name:
>> +                database = row
>> +                break
>> +
>> +        if not database:
>> +            vlog.info("%s: server does not have %s database"
>> +                      % (session_name, self._db.name))
>> +            return False
>> +
>> +        if (database.model == CLUSTERED and
>> +            self._session.get_num_of_remotes() > 1):
>> +            if not database.schema:
>> +                vlog.info('%s: clustered database server has not yet
>> joined '
>> +                          'cluster; trying another server' %
>> session_name)
>> +                return False
>> +            if not database.connected:
>> +                vlog.info('%s: clustered database server is
>> disconnected '
>> +                          'from cluster; trying another server' %
>> session_name)
>> +                return False
>> +            if (self.leader_only and
>> +                not database.leader):
>> +                vlog.info('%s: clustered database server is not cluster
>> '
>> +                          'leader; trying another server' % session_name)
>> +                return False
>> +            if database.index:
>> +                if database.index[0] < self._min_index:
>> +                    vlog.warn('%s: clustered database server has stale
>> data; '
>> +                              'trying another server' % session_name)
>> +                    return False
>> +                self._min_index = database.index[0]
>> +
>> +        return True
>> +
>>      def __column_name(self, column):
>>          if column.type.key.type == ovs.db.types.UuidType:
>>              return ovs.ovsuuid.to_json(column.type.key.type.default)
>> diff --git a/python/ovs/reconnect.py b/python/ovs/reconnect.py
>> index 34cc769..afbe445 100644
>> --- a/python/ovs/reconnect.py
>> +++ b/python/ovs/reconnect.py
>> @@ -344,6 +344,9 @@ class Reconnect(object):
>>                  else:
>>                      self.info_level("%s: error listening for connections"
>>                                      % self.name)
>> +            elif self.state == Reconnect.Reconnect:
>> +                self.info_level("%s: connection closed by client"
>> +                                % self.name)
>>              elif self.backoff < self.max_backoff:
>>                  if self.passive:
>>                      type_ = "listen"
>> diff --git a/tests/ovsdb-idl.at b/tests/ovsdb-idl.at
>> index 8981b5e..46b047f 100644
>> --- a/tests/ovsdb-idl.at
>> +++ b/tests/ovsdb-idl.at
>> @@ -1466,40 +1466,44 @@ OVSDB_CHECK_IDL_NOTIFY([simple idl verify notify],
>>         "where": [["i", "==", 0]]}]' \
>>      'reconnect']],
>>    [[000: empty
>> -001:
>> {"error":null,"result":[{"uuid":["uuid","<0>"]},{"uuid":["uuid","<1>"]}]}
>> -002: event:create, row={i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[]
>> ua=[] uuid=<1>}, updates=None
>> -002: event:create, row={i=1 r=2 b=true s=mystring u=<3> ia=[1 2 3]
>> ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates=None
>> -002: i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
>> -002: i=1 r=2 b=true s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<4> <5>] uuid=<0>
>> +000: event:create, row={uuid=<0>}, updates=None
>> +000: event:create, row={uuid=<1>}, updates=None
>> +001:
>> {"error":null,"result":[{"uuid":["uuid","<2>"]},{"uuid":["uuid","<3>"]}]}
>> +002: event:create, row={i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[]
>> ua=[] uuid=<3>}, updates=None
>> +002: event:create, row={i=1 r=2 b=true s=mystring u=<5> ia=[1 2 3]
>> ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates=None
>> +002: i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
>> +002: i=1 r=2 b=true s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<6> <7>] uuid=<2>
>>  003: {"error":null,"result":[{"count":2}]}
>> -004: event:update, row={i=1 r=2 b=false s=mystring u=<3> ia=[1 2 3]
>> ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates={b=true
>> uuid=<0>}
>> -004: i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
>> -004: i=1 r=2 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<4> <5>] uuid=<0>
>> +004: event:update, row={i=1 r=2 b=false s=mystring u=<5> ia=[1 2 3]
>> ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates={b=true
>> uuid=<2>}
>> +004: i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
>> +004: i=1 r=2 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<6> <7>] uuid=<2>
>>  005: {"error":null,"result":[{"count":2}]}
>> -006: event:update, row={i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[]
>> sa=[] ua=[] uuid=<1>}, updates={r=0 uuid=<1>}
>> -006: event:update, row={i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3]
>> ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates={r=2
>> uuid=<0>}
>> -006: i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
>> -006: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<4> <5>] uuid=<0>
>> -007: {"error":null,"result":[{"uuid":["uuid","<6>"]}]}
>> -008: event:create, row={i=-1 r=125 b=false s= u=<2> ia=[1] ra=[1.5]
>> ba=[false] sa=[] ua=[] uuid=<6>}, updates=None
>> -008: i=-1 r=125 b=false s= u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[]
>> uuid=<6>
>> -008: i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
>> -008: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<4> <5>] uuid=<0>
>> +006: event:update, row={i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[]
>> sa=[] ua=[] uuid=<3>}, updates={r=0 uuid=<3>}
>> +006: event:update, row={i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3]
>> ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates={r=2
>> uuid=<2>}
>> +006: i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
>> +006: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<6> <7>] uuid=<2>
>> +007: {"error":null,"result":[{"uuid":["uuid","<8>"]}]}
>> +008: event:create, row={i=-1 r=125 b=false s= u=<4> ia=[1] ra=[1.5]
>> ba=[false] sa=[] ua=[] uuid=<8>}, updates=None
>> +008: i=-1 r=125 b=false s= u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[]
>> uuid=<8>
>> +008: i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
>> +008: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<6> <7>] uuid=<2>
>>  009: {"error":null,"result":[{"count":2}]}
>> -010: event:update, row={i=-1 r=125 b=false s=newstring u=<2> ia=[1]
>> ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>}, updates={s= uuid=<6>}
>> -010: event:update, row={i=0 r=123.5 b=false s=newstring u=<2> ia=[]
>> ra=[] ba=[] sa=[] ua=[] uuid=<1>}, updates={s= uuid=<1>}
>> -010: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false]
>> sa=[] ua=[] uuid=<6>
>> -010: i=0 r=123.5 b=false s=newstring u=<2> ia=[] ra=[] ba=[] sa=[] ua=[]
>> uuid=<1>
>> -010: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<4> <5>] uuid=<0>
>> +010: event:update, row={i=-1 r=125 b=false s=newstring u=<4> ia=[1]
>> ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>}, updates={s= uuid=<8>}
>> +010: event:update, row={i=0 r=123.5 b=false s=newstring u=<4> ia=[]
>> ra=[] ba=[] sa=[] ua=[] uuid=<3>}, updates={s= uuid=<3>}
>> +010: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false]
>> sa=[] ua=[] uuid=<8>
>> +010: i=0 r=123.5 b=false s=newstring u=<4> ia=[] ra=[] ba=[] sa=[] ua=[]
>> uuid=<3>
>> +010: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<6> <7>] uuid=<2>
>>  011: {"error":null,"result":[{"count":1}]}
>> -012: event:delete, row={i=0 r=123.5 b=false s=newstring u=<2> ia=[]
>> ra=[] ba=[] sa=[] ua=[] uuid=<1>}, updates=None
>> -012: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false]
>> sa=[] ua=[] uuid=<6>
>> -012: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<4> <5>] uuid=<0>
>> +012: event:delete, row={i=0 r=123.5 b=false s=newstring u=<4> ia=[]
>> ra=[] ba=[] sa=[] ua=[] uuid=<3>}, updates=None
>> +012: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false]
>> sa=[] ua=[] uuid=<8>
>> +012: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<6> <7>] uuid=<2>
>>  013: reconnect
>> -014: event:create, row={i=-1 r=125 b=false s=newstring u=<2> ia=[1]
>> ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>}, updates=None
>> -014: event:create, row={i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3]
>> ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates=None
>> -014: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false]
>> sa=[] ua=[] uuid=<6>
>> -014: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<4> <5>] uuid=<0>
>> +014: event:create, row={i=-1 r=125 b=false s=newstring u=<4> ia=[1]
>> ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>}, updates=None
>> +014: event:create, row={i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3]
>> ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates=None
>> +014: event:create, row={uuid=<0>}, updates=None
>> +014: event:create, row={uuid=<1>}, updates=None
>> +014: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false]
>> sa=[] ua=[] uuid=<8>
>> +014: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true]
>> sa=[abc def] ua=[<6> <7>] uuid=<2>
>>  015: done
>>  ]])
>>
>>
> I think it would be good if we have few test cases which starts a cluster
> of ovsdb-server's and then
>    -verify if the client has connected to the cluster leader
>    - bring down the leader and check if the client connects to the new
> leader.
>
>
> Thanks
> Numan
>
> --
>> 1.8.3.1
>>
>> _______________________________________________
>> dev mailing list
>> dev@openvswitch.org
>> https://mail.openvswitch.org/mailman/listinfo/ovs-dev
>>
>
Ted Elhourani Jan. 16, 2019, 7:13 p.m. | #3
Hi Numan,

Thanks for the review, please find responses inline.

Ted

On Jan 11, 2019, at 2:23 AM, Numan Siddique <nusiddiq@redhat.com<mailto:nusiddiq@redhat.com>> wrote:



On Fri, Jan 4, 2019 at 5:26 PM Numan Siddique <nusiddiq@redhat.com<mailto:nusiddiq@redhat.com>> wrote:
Hi Ted,

This patch is failing the below test cases for me. All are python3 related. Can you please
take a look if that's the case with you as well ?
******
2139: simple idl, writing via IDL with unicode - Python3 FAILED (ovsdb-idl.at:456 [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at-3A456&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=O_FLff8wdsNhNgmdTrYVvYi3QYV2Jfdxca-_zhURfYI&e=>)
2141: simple idl, writing via IDL with unicode - Python3 - register_columns FAILED (ovsdb-idl.at:456 [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at-3A456&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=O_FLff8wdsNhNgmdTrYVvYi3QYV2Jfdxca-_zhURfYI&e=>)
2143: simple idl, writing via IDL with unicode - Python3 - tcp FAILED (ovsdb-idl.at:456 [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at-3A456&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=O_FLff8wdsNhNgmdTrYVvYi3QYV2Jfdxca-_zhURfYI&e=>)
2145: simple idl, writing via IDL with unicode - Python3 (multiple remotes) - tcp FAILED (ovsdb-idl.at:456 [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at-3A456&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=O_FLff8wdsNhNgmdTrYVvYi3QYV2Jfdxca-_zhURfYI&e=>)
2147: simple idl, writing via IDL with unicode - Python3 - tcp6 FAILED (ovsdb-idl.at:456 [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at-3A456&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=O_FLff8wdsNhNgmdTrYVvYi3QYV2Jfdxca-_zhURfYI&e=>)
2149: simple idl, writing via IDL with unicode - Python3 (multiple remotes) - tcp6 FAILED (ovsdb-idl.at:456 [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at-3A456&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=O_FLff8wdsNhNgmdTrYVvYi3QYV2Jfdxca-_zhURfYI&e=>)
2151: simple idl, writing via IDL with unicode - Python3 - SSL FAILED (ovsdb-idl.at:456 [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at-3A456&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=O_FLff8wdsNhNgmdTrYVvYi3QYV2Jfdxca-_zhURfYI&e=>)
2153: simple idl, writing large data via IDL with unicode - Python3 FAILED (ovsdb-idl.at:490 [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at-3A490&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=d-KSVXlW0TVOTieNWZtsAIAx54VB7zPs-vp5FgMcwOg&e=>)
**********

These failures are not related to this patch. Please ignore. There seems to be some issues with python3.7 with unicode
in Fedora 29 and I think Timothy  is planning to submit a patch to fix these failures.

Thanks
Numan

All tests are passing with Python 3.4.8 & Python 2.7.5 on my machine, no skips except for daemon.at<http://daemon.at/> (windows skip).



Please see some comments below.

Thanks
Numan

On Fri, Jan 4, 2019 at 7:47 AM Ted Elhourani <ted.elhourani@nutanix.com<mailto:ted.elhourani@nutanix.com>> wrote:
The Python IDL implementation supports ovsdb cluster connections.
This patch is a follow up to commit 31e434fc98, it adds the option of
connecting to the leader (the default) in the Raft-based cluster. It mimics
the exisiting C IDL support for clusters introduced in commit 1b1d2e6daa.

The _Server database schema is first requested, then a monitor of the
Database table in the _Server Database. Method __check_server_db verifies
the eligibility of the server. If the attempt to obtain a monitor of the
_Server database fails and a cluster id was not provided this implementation
proceeds to request the data monitor. If a cluster id was provided via the
set_cluster_id method then the connection is aborted and a connection to a
different node is instead attempted, until a valid cluster node is found.
Thus, when supplied, cluster id is interpreted as the intention to only
allow connections to a clustered database. If not supplied, connections to
standalone nodes, or nodes that do not have the _Server database are
allowed. change_seqno is not incremented in the case of Database table
updates.
Signed-off-by: Ted Elhourani <ted.elhourani@nutanix.com<mailto:ted.elhourani@nutanix.com>>
---
 python/ovs/db/idl.py    | 217 ++++++++++++++++++++++++++++++++++++++++++++----
 python/ovs/reconnect.py |   3 +
 tests/ovsdb-idl.at [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=4YuL3pa8C9yzpOxISvjLrpCF6emBAefujjcMeXTtzQA&e=>      |  62 +++++++-------
 3 files changed, 237 insertions(+), 45 deletions(-)

diff --git a/python/ovs/db/idl.py b/python/ovs/db/idl.py
index 250e897..f989548 100644
--- a/python/ovs/db/idl.py
+++ b/python/ovs/db/idl.py
@@ -38,6 +38,7 @@ ROW_DELETE = "delete"
 OVSDB_UPDATE = 0
 OVSDB_UPDATE2 = 1

+CLUSTERED = "clustered"

 class Idl(object):
     """Open vSwitch Database Interface Definition Language (OVSDB IDL).
@@ -92,10 +93,12 @@ class Idl(object):
 """

     IDL_S_INITIAL = 0
-    IDL_S_MONITOR_REQUESTED = 1
-    IDL_S_MONITOR_COND_REQUESTED = 2
+    IDL_S_SERVER_SCHEMA_REQUESTED = 1
+    IDL_S_SERVER_MONITOR_REQUESTED = 2
+    IDL_S_DATA_MONITOR_REQUESTED = 3
+    IDL_S_DATA_MONITOR_COND_REQUESTED = 4

-    def __init__(self, remote, schema_helper, probe_interval=None):
+    def __init__(self, remote, schema_helper, leader_only=True, probe_interval=None):

Can you please add the "leader_only" arg at the end. Older clients of python idl would
break if they were creating Idl object as
my_idl = Idl(remote, helper, 1000)

Will fix in next revision.



         """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
@@ -119,6 +122,9 @@ class Idl(object):

         The IDL uses and modifies 'schema' directly.

+        If 'leader_only' is set to True (default value) the IDL will only monitor
+        and transact with the leader of the cluster.
+
         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.
@@ -137,6 +143,20 @@ class Idl(object):
         self._last_seqno = None
         self.change_seqno = 0
         self.uuid = uuid.uuid1()
+
+        # Server monitor.
+        self._server_schema_request_id = None
+        self._server_monitor_request_id = None
+        self._db_change_aware_request_id = None
+        self._server_db_name = '_Server'
+        self._server_db_table = 'Database'
+        self.server_tables = None
+        self._server_db = None
+        self.server_monitor_uuid = uuid.uuid1()
+        self.leader_only = leader_only
+        self.cluster_id = None
+        self._min_index = 0
+
         self.state = self.IDL_S_INITIAL

         # Database locking.
@@ -172,6 +192,15 @@ class Idl(object):
                 remotes.append(r)
         return remotes

+    def set_cluster_id(self, cluster_id):
+        """Set the id of the cluster that this idl must connect to."""
+        if cluster_id:
+            self.cluster_id = str(cluster_id)
+        else:
+            self.cluster_id = None

Instead of the "if - else", how about just
self.cluster_id = cluster_id

Can cluster_id be passed as an integer by the caller ?

Since it is a substring of the UUID we can expect any client code to pass it as a string. I’ll fix in the next revision.



+        if self.state != self.IDL_S_INITIAL:
+            self.force_reconnect()
+
     def index_create(self, table, name):
         """Create a named multi-column index on a table"""
         return self.tables[table].rows.index_create(name)
@@ -222,7 +251,7 @@ class Idl(object):
             if seqno != self._last_seqno:
                 self._last_seqno = seqno
                 self.__txn_abort_all()
-                self.__send_monitor_request()
+                self.__send_server_schema_request()
                 if self.lock_name:
                     self.__send_lock_request()
                 break
@@ -230,6 +259,7 @@ class Idl(object):
             msg = self._session.recv()
             if msg is None:
                 break
+
             if (msg.type == ovs.jsonrpc.Message.T_NOTIFY
                     and msg.method == "update2"
                     and len(msg.params) == 2):
@@ -239,7 +269,15 @@ class Idl(object):
                     and msg.method == "update"
                     and len(msg.params) == 2):
                 # Database contents changed.
-                self.__parse_update(msg.params[1], OVSDB_UPDATE)
+                if msg.params[0] == str(self.server_monitor_uuid):
+                    self.__parse_update(msg.params[1], OVSDB_UPDATE,
+                                        tables=self.server_tables)
+                    self.change_seqno = initial_change_seqno
+                    if not self.__check_server_db():
+                        self.force_reconnect()
+                        break
+                else:
+                    self.__parse_update(msg.params[1], OVSDB_UPDATE)
             elif (msg.type == ovs.jsonrpc.Message.T_REPLY
                   and self._monitor_request_id is not None
                   and self._monitor_request_id == msg.id [msg.id]<https://urldefense.proofpoint.com/v2/url?u=http-3A__msg.id&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=ipPX8qg6xX9LFHpiOIoYNtWpxlILFrRBvra7dbxF2hg&e=>):
@@ -248,17 +286,66 @@ class Idl(object):
                     self.change_seqno += 1
                     self._monitor_request_id = None
                     self.__clear()
-                    if self.state == self.IDL_S_MONITOR_COND_REQUESTED:
+                    if self.state == self.IDL_S_DATA_MONITOR_COND_REQUESTED:
                         self.__parse_update(msg.result, OVSDB_UPDATE2)
                     else:
-                        assert self.state == self.IDL_S_MONITOR_REQUESTED
+                        assert self.state == self.IDL_S_DATA_MONITOR_REQUESTED
                         self.__parse_update(msg.result, OVSDB_UPDATE)
-
                 except error.Error as e:
                     vlog.err("%s: parse error in received schema: %s"
                              % (self._session.get_name(), e))
                     self.__error()
             elif (msg.type == ovs.jsonrpc.Message.T_REPLY
+                  and self._server_schema_request_id is not None
+                  and self._server_schema_request_id == msg.id [msg.id]<https://urldefense.proofpoint.com/v2/url?u=http-3A__msg.id&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=ipPX8qg6xX9LFHpiOIoYNtWpxlILFrRBvra7dbxF2hg&e=>):
+                # Reply to our "get_schema" of _Server request.
+                try:
+                    self._server_schema_request_id = None
+                    sh = SchemaHelper(None, msg.result)
+                    sh.register_table(self._server_db_table)
+                    schema = sh.get_idl_schema()
+                    self._server_db = schema
+                    self.server_tables = schema.tables
+                    self.__send_server_monitor_request()
+                except error.Error as e:
+                    vlog.err("%s: error receiving server schema: %s"
+                             % (self._session.get_name(), e))
+                    if self.cluster_id:
+                        self.__error()
+                        break
+                    else:
+                        self.change_seqno = initial_change_seqno
+                        self.__send_monitor_request()
+            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
+                  and self._server_monitor_request_id is not None
+                  and self._server_monitor_request_id == msg.id [msg.id]<https://urldefense.proofpoint.com/v2/url?u=http-3A__msg.id&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=ipPX8qg6xX9LFHpiOIoYNtWpxlILFrRBvra7dbxF2hg&e=>):
+                # Reply to our "monitor" of _Server request.
+                try:
+                    self._server_monitor_request_id = None
+                    self.__parse_update(msg.result, OVSDB_UPDATE,
+                                        tables=self.server_tables)
+                    self.change_seqno = initial_change_seqno
+                    if self.__check_server_db():
+                        self.__send_monitor_request()
+                        self.__send_db_change_aware()
+                    else:
+                        self.force_reconnect()
+                        break
+                except error.Error as e:
+                    vlog.err("%s: parse error in received schema: %s"
+                             % (self._session.get_name(), e))
+                    if self.cluster_id:
+                        self.__error()
+                        break
+                    else:
+                        self.change_seqno = initial_change_seqno
+                        self.__send_monitor_request()
+            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
+                  and self._db_change_aware_request_id is not None
+                  and self._db_change_aware_request_id == msg.id [msg.id]<https://urldefense.proofpoint.com/v2/url?u=http-3A__msg.id&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=ipPX8qg6xX9LFHpiOIoYNtWpxlILFrRBvra7dbxF2hg&e=>):
+                # Reply to us notifying the server of our change awarness.
+                self._db_change_aware_request_id = None
+            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
                   and self._lock_request_id is not None
                   and self._lock_request_id == msg.id [msg.id]<https://urldefense.proofpoint.com/v2/url?u=http-3A__msg.id&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=ipPX8qg6xX9LFHpiOIoYNtWpxlILFrRBvra7dbxF2hg&e=>):
                 # Reply to our "lock" request.
@@ -275,10 +362,20 @@ class Idl(object):
                 # Reply to our echo request.  Ignore it.
                 pass
             elif (msg.type == ovs.jsonrpc.Message.T_ERROR and
-                  self.state == self.IDL_S_MONITOR_COND_REQUESTED and
+                  self.state == self.IDL_S_DATA_MONITOR_COND_REQUESTED and
                   self._monitor_request_id == msg.id [msg.id]<https://urldefense.proofpoint.com/v2/url?u=http-3A__msg.id&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=ipPX8qg6xX9LFHpiOIoYNtWpxlILFrRBvra7dbxF2hg&e=>):
                 if msg.error == "unknown method":
                     self.__send_monitor_request()
+            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 [msg.id]<https://urldefense.proofpoint.com/v2/url?u=http-3A__msg.id&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=ipPX8qg6xX9LFHpiOIoYNtWpxlILFrRBvra7dbxF2hg&e=>):
+                self._server_schema_request_id = None
+                if self.cluster_id:
+                    self.force_reconnect()
+                    break
+                else:
+                    self.change_seqno = initial_change_seqno
+                    self.__send_monitor_request()
             elif (msg.type in (ovs.jsonrpc.Message.T_ERROR,
                                ovs.jsonrpc.Message.T_REPLY)
                   and self.__txn_process_reply(msg)):
@@ -440,12 +537,19 @@ class Idl(object):
             if not new_has_lock:
                 self.is_lock_contended = True

+    def __send_db_change_aware(self):
+        msg = ovs.jsonrpc.Message.create_request("set_db_change_aware",
+                                                 [True])
+        self._db_change_aware_request_id = msg.id [msg.id]<https://urldefense.proofpoint.com/v2/url?u=http-3A__msg.id&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=ipPX8qg6xX9LFHpiOIoYNtWpxlILFrRBvra7dbxF2hg&e=>
+        self._session.send(msg)
+
     def __send_monitor_request(self):
-        if self.state == self.IDL_S_INITIAL:
-            self.state = self.IDL_S_MONITOR_COND_REQUESTED
+        if (self.state in [self.IDL_S_SERVER_MONITOR_REQUESTED,
+                           self.IDL_S_INITIAL]):
+            self.state = self.IDL_S_DATA_MONITOR_COND_REQUESTED
             method = "monitor_cond"
         else:
-            self.state = self.IDL_S_MONITOR_REQUESTED
+            self.state = self.IDL_S_DATA_MONITOR_REQUESTED
             method = "monitor"

         monitor_requests = {}
@@ -467,20 +571,50 @@ class Idl(object):
         self._monitor_request_id = msg.id [msg.id]<https://urldefense.proofpoint.com/v2/url?u=http-3A__msg.id&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=ipPX8qg6xX9LFHpiOIoYNtWpxlILFrRBvra7dbxF2hg&e=>
         self._session.send(msg)

-    def __parse_update(self, update, version):
+    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 [msg.id]<https://urldefense.proofpoint.com/v2/url?u=http-3A__msg.id&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=ipPX8qg6xX9LFHpiOIoYNtWpxlILFrRBvra7dbxF2hg&e=>
+        res = self._session.send(msg)
+
+    def __send_server_monitor_request(self):
+        self.state = self.IDL_S_SERVER_MONITOR_REQUESTED
+        monitor_requests = {}
+        table = self.server_tables[self._server_db_table]
+        columns = [column for column in six.iterkeys(table.columns)]
+        for column in six.itervalues(table.columns):
+            if not hasattr(column, 'alert'):
+                column.alert = True
+        table.rows = custom_index.IndexedRows(table)
+        table.need_table = False
+        table.idl = self
+        monitor_request = {"columns": columns}
+        monitor_requests[table.name [table.name]<https://urldefense.proofpoint.com/v2/url?u=http-3A__table.name&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=CT-ATV90ancafNslReWTWUlAP4U2yOEOTDyh5N9ChEA&e=>] = [monitor_request]
+        msg = ovs.jsonrpc.Message.create_request(
+            'monitor', [self._server_db.name [server_db.name]<https://urldefense.proofpoint.com/v2/url?u=http-3A__server-5Fdb.name&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=wr8oQebCMEzbper7UCClxjT2UqYIFPTKJcWn5aTVxr4&e=>,
+                             str(self.server_monitor_uuid),
+                             monitor_requests])
+        self._server_monitor_request_id = msg.id [msg.id]<https://urldefense.proofpoint.com/v2/url?u=http-3A__msg.id&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=ipPX8qg6xX9LFHpiOIoYNtWpxlILFrRBvra7dbxF2hg&e=>
+        self._session.send(msg)
+
+    def __parse_update(self, update, version, tables=None):
         try:
-            self.__do_parse_update(update, version)
+            if not tables:
+                self.__do_parse_update(update, version, self.tables)
+            else:
+                self.__do_parse_update(update, version, tables)
         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):
+    def __do_parse_update(self, table_updates, version, tables):
         if not isinstance(table_updates, dict):
             raise error.Error("<table-updates> is not an object",
                               table_updates)

         for table_name, table_update in six.iteritems(table_updates):
-            table = self.tables.get(table_name)
+            table = tables.get(table_name)
             if not table:
                 raise error.Error('<table-updates> includes unknown '
                                   'table "%s"' % table_name)
@@ -605,6 +739,57 @@ class Idl(object):
                 self.notify(op, row, Row.from_json(self, table, uuid, old))
         return changed

+    def __check_server_db(self):
+        """Returns True if this is a valid ovsdb server, False otherwise."""
The function name seems a bit confusing.
This function   returns True if the server has "_Server" table and if it
is a clustered db, additional checks are done.
May be this function can be broken in 2 separate functions ?
One to check if the server supports '_Server' db and other to check
the status of cluster.

We are already checking for _Server support in the state machine, with the get-schema transaction (send_server_schema_request). This function is supposed to be identical to
https://github.com/openvswitch/ovs/blob/master/lib/ovsdb-idl.c#L1815
If the server supports  _Server then the next step is to check if the Database table contains a row with a matching cid or db name if the client code did not set the cid (then of course the other columns for validity, e.g. leader). The table check "if self._server_db_table not in self.server_tables” adds to the confusion, and might be unnecessary. This function is checking the content of the Database table to determine the admissibility of the server. I’ll be happy to re-arrange if you still see a reason to split the check in two.



+        session_name = self._session.get_name()
+
+        if self._server_db_table not in self.server_tables:
+            vlog.info [vlog.info]<https://urldefense.proofpoint.com/v2/url?u=http-3A__vlog.info&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=6qLDhyx-IO4Gc5C3CnZcIAQADJSLT3fQMx6JHpH43ic&e=>("%s: server does not have %s table in its %s database"
+                      % (session_name, self._server_db_table, self._server_db_name))
+            return False
+
+        rows = self.server_tables[self._server_db_table].rows
+
+        database = None
+        for row in six.itervalues(rows):
+            if self.cluster_id:
+                if self.cluster_id in \
+                   map(lambda x: str(x)[:4], row.cid):
+                    database = row
+                    break
+            elif row.name [row.name]<https://urldefense.proofpoint.com/v2/url?u=http-3A__row.name&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=5_G5MzWIKTCLvLO7rzkOmVGA2rKjTF7NS2Xtg8fMYwE&e=> == self._db.name [db.name]<https://urldefense.proofpoint.com/v2/url?u=http-3A__db.name&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=Fnu7uRGwOKN3Lf2iporNo7OUbi7kT-zH1Sv9XcJoevU&e=>:
+                database = row
+                break
+
+        if not database:
+            vlog.info [vlog.info]<https://urldefense.proofpoint.com/v2/url?u=http-3A__vlog.info&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=6qLDhyx-IO4Gc5C3CnZcIAQADJSLT3fQMx6JHpH43ic&e=>("%s: server does not have %s database"
+                      % (session_name, self._db.name [db.name]<https://urldefense.proofpoint.com/v2/url?u=http-3A__db.name&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=Fnu7uRGwOKN3Lf2iporNo7OUbi7kT-zH1Sv9XcJoevU&e=>))
+            return False
+
+        if (database.model == CLUSTERED and
+            self._session.get_num_of_remotes() > 1):
+            if not database.schema:
+                vlog.info [vlog.info]<https://urldefense.proofpoint.com/v2/url?u=http-3A__vlog.info&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=6qLDhyx-IO4Gc5C3CnZcIAQADJSLT3fQMx6JHpH43ic&e=>('%s: clustered database server has not yet joined '
+                          'cluster; trying another server' % session_name)
+                return False
+            if not database.connected:
+                vlog.info [vlog.info]<https://urldefense.proofpoint.com/v2/url?u=http-3A__vlog.info&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=6qLDhyx-IO4Gc5C3CnZcIAQADJSLT3fQMx6JHpH43ic&e=>('%s: clustered database server is disconnected '
+                          'from cluster; trying another server' % session_name)
+                return False
+            if (self.leader_only and
+                not database.leader):
+                vlog.info [vlog.info]<https://urldefense.proofpoint.com/v2/url?u=http-3A__vlog.info&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=6qLDhyx-IO4Gc5C3CnZcIAQADJSLT3fQMx6JHpH43ic&e=>('%s: clustered database server is not cluster '
+                          'leader; trying another server' % session_name)
+                return False
+            if database.index:
+                if database.index[0] < self._min_index:
+                    vlog.warn('%s: clustered database server has stale data; '
+                              'trying another server' % session_name)
+                    return False
+                self._min_index = database.index[0]
+
+        return True
+
     def __column_name(self, column):
         if column.type.key.type == ovs.db.types.UuidType:
             return ovs.ovsuuid.to_json(column.type.key.type.default)
diff --git a/python/ovs/reconnect.py b/python/ovs/reconnect.py
index 34cc769..afbe445 100644
--- a/python/ovs/reconnect.py
+++ b/python/ovs/reconnect.py
@@ -344,6 +344,9 @@ class Reconnect(object):
                 else:
                     self.info_level("%s: error listening for connections"
                                     % self.name [self.name]<https://urldefense.proofpoint.com/v2/url?u=http-3A__self.name&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=uoKThZ5fYJU0gn-r7DWLvQd6I-AB-poCZd0ROFidptU&e=>)
+            elif self.state == Reconnect.Reconnect:
+                self.info_level("%s: connection closed by client"
+                                % self.name [self.name]<https://urldefense.proofpoint.com/v2/url?u=http-3A__self.name&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=uoKThZ5fYJU0gn-r7DWLvQd6I-AB-poCZd0ROFidptU&e=>)
             elif self.backoff < self.max_backoff:
                 if self.passive:
                     type_ = "listen"
diff --git a/tests/ovsdb-idl.at [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=4YuL3pa8C9yzpOxISvjLrpCF6emBAefujjcMeXTtzQA&e=> b/tests/ovsdb-idl.at [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=4YuL3pa8C9yzpOxISvjLrpCF6emBAefujjcMeXTtzQA&e=>
index 8981b5e..46b047f 100644
--- a/tests/ovsdb-idl.at [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=4YuL3pa8C9yzpOxISvjLrpCF6emBAefujjcMeXTtzQA&e=>
+++ b/tests/ovsdb-idl.at [ovsdb-idl.at]<https://urldefense.proofpoint.com/v2/url?u=http-3A__ovsdb-2Didl.at&d=DwMFaQ&c=s883GpUCOChKOHiocYtGcg&r=mB3ItZGjEYI9nn4mb7sBHd5RJ9XbfVZnYVRmdST65Cg&m=mnAThoC2lT5Ia3WO-qtYlzsSvtKGm0f60ORMPJ1apUc&s=4YuL3pa8C9yzpOxISvjLrpCF6emBAefujjcMeXTtzQA&e=>
@@ -1466,40 +1466,44 @@ OVSDB_CHECK_IDL_NOTIFY([simple idl verify notify],
        "where": [["i", "==", 0]]}]' \
     'reconnect']],
   [[000: empty
-001: {"error":null,"result":[{"uuid":["uuid","<0>"]},{"uuid":["uuid","<1>"]}]}
-002: event:create, row={i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>}, updates=None
-002: event:create, row={i=1 r=2 b=true s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates=None
-002: i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
-002: i=1 r=2 b=true s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+000: event:create, row={uuid=<0>}, updates=None
+000: event:create, row={uuid=<1>}, updates=None
+001: {"error":null,"result":[{"uuid":["uuid","<2>"]},{"uuid":["uuid","<3>"]}]}
+002: event:create, row={i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>}, updates=None
+002: event:create, row={i=1 r=2 b=true s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates=None
+002: i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
+002: i=1 r=2 b=true s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 003: {"error":null,"result":[{"count":2}]}
-004: event:update, row={i=1 r=2 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates={b=true uuid=<0>}
-004: i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
-004: i=1 r=2 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+004: event:update, row={i=1 r=2 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates={b=true uuid=<2>}
+004: i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
+004: i=1 r=2 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 005: {"error":null,"result":[{"count":2}]}
-006: event:update, row={i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>}, updates={r=0 uuid=<1>}
-006: event:update, row={i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates={r=2 uuid=<0>}
-006: i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
-006: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
-007: {"error":null,"result":[{"uuid":["uuid","<6>"]}]}
-008: event:create, row={i=-1 r=125 b=false s= u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>}, updates=None
-008: i=-1 r=125 b=false s= u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>
-008: i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
-008: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+006: event:update, row={i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>}, updates={r=0 uuid=<3>}
+006: event:update, row={i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates={r=2 uuid=<2>}
+006: i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
+006: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
+007: {"error":null,"result":[{"uuid":["uuid","<8>"]}]}
+008: event:create, row={i=-1 r=125 b=false s= u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>}, updates=None
+008: i=-1 r=125 b=false s= u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>
+008: i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
+008: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 009: {"error":null,"result":[{"count":2}]}
-010: event:update, row={i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>}, updates={s= uuid=<6>}
-010: event:update, row={i=0 r=123.5 b=false s=newstring u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>}, updates={s= uuid=<1>}
-010: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>
-010: i=0 r=123.5 b=false s=newstring u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
-010: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+010: event:update, row={i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>}, updates={s= uuid=<8>}
+010: event:update, row={i=0 r=123.5 b=false s=newstring u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>}, updates={s= uuid=<3>}
+010: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>
+010: i=0 r=123.5 b=false s=newstring u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
+010: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 011: {"error":null,"result":[{"count":1}]}
-012: event:delete, row={i=0 r=123.5 b=false s=newstring u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>}, updates=None
-012: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>
-012: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+012: event:delete, row={i=0 r=123.5 b=false s=newstring u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>}, updates=None
+012: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>
+012: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 013: reconnect
-014: event:create, row={i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>}, updates=None
-014: event:create, row={i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates=None
-014: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>
-014: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+014: event:create, row={i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>}, updates=None
+014: event:create, row={i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates=None
+014: event:create, row={uuid=<0>}, updates=None
+014: event:create, row={uuid=<1>}, updates=None
+014: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>
+014: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 015: done
 ]])


I think it would be good if we have few test cases which starts a cluster of ovsdb-server's and then
   -verify if the client has connected to the cluster leader
   - bring down the leader and check if the client connects to the new leader.


I wrote 2 tests, I’ll include them in the next revision.


Thanks
Numan

--
1.8.3.1
Ben Pfaff Jan. 16, 2019, 8:29 p.m. | #4
On Fri, Jan 04, 2019 at 02:16:36AM +0000, Ted Elhourani wrote:
> The Python IDL implementation supports ovsdb cluster connections.
> This patch is a follow up to commit 31e434fc98, it adds the option of
> connecting to the leader (the default) in the Raft-based cluster. It mimics
> the exisiting C IDL support for clusters introduced in commit 1b1d2e6daa.

I'm getting the following reports from flake8 when I apply this patch.
Can you fix those up for v3?

../python/ovs/db/idl.py:43:1: E302 expected 2 blank lines, found 1
../python/ovs/db/idl.py:101:80: E501 line too long (85 > 79 characters)
../python/ovs/db/idl.py:125:80: E501 line too long (81 > 79 characters)
../python/ovs/db/idl.py:579:9: F841 local variable 'res' is assigned to but never used
../python/ovs/db/idl.py:748:80: E501 line too long (84 > 79 characters)

Thanks,

Ben.
Ted Elhourani Jan. 16, 2019, 9:38 p.m. | #5
Sure, I fixed and verified with “make flake8-check”.

Ted
> On Jan 16, 2019, at 12:29 PM, Ben Pfaff <blp@ovn.org> wrote:
> 
> On Fri, Jan 04, 2019 at 02:16:36AM +0000, Ted Elhourani wrote:
>> The Python IDL implementation supports ovsdb cluster connections.
>> This patch is a follow up to commit 31e434fc98, it adds the option of
>> connecting to the leader (the default) in the Raft-based cluster. It mimics
>> the exisiting C IDL support for clusters introduced in commit 1b1d2e6daa.
> 
> I'm getting the following reports from flake8 when I apply this patch.
> Can you fix those up for v3?
> 
> ../python/ovs/db/idl.py:43:1: E302 expected 2 blank lines, found 1
> ../python/ovs/db/idl.py:101:80: E501 line too long (85 > 79 characters)
> ../python/ovs/db/idl.py:125:80: E501 line too long (81 > 79 characters)
> ../python/ovs/db/idl.py:579:9: F841 local variable 'res' is assigned to but never used
> ../python/ovs/db/idl.py:748:80: E501 line too long (84 > 79 characters)
> 
> Thanks,
> 
> Ben.

Patch

diff --git a/python/ovs/db/idl.py b/python/ovs/db/idl.py
index 250e897..f989548 100644
--- a/python/ovs/db/idl.py
+++ b/python/ovs/db/idl.py
@@ -38,6 +38,7 @@  ROW_DELETE = "delete"
 OVSDB_UPDATE = 0
 OVSDB_UPDATE2 = 1
 
+CLUSTERED = "clustered"
 
 class Idl(object):
     """Open vSwitch Database Interface Definition Language (OVSDB IDL).
@@ -92,10 +93,12 @@  class Idl(object):
 """
 
     IDL_S_INITIAL = 0
-    IDL_S_MONITOR_REQUESTED = 1
-    IDL_S_MONITOR_COND_REQUESTED = 2
+    IDL_S_SERVER_SCHEMA_REQUESTED = 1
+    IDL_S_SERVER_MONITOR_REQUESTED = 2
+    IDL_S_DATA_MONITOR_REQUESTED = 3
+    IDL_S_DATA_MONITOR_COND_REQUESTED = 4
 
-    def __init__(self, remote, schema_helper, probe_interval=None):
+    def __init__(self, remote, schema_helper, leader_only=True, probe_interval=None):
         """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
@@ -119,6 +122,9 @@  class Idl(object):
 
         The IDL uses and modifies 'schema' directly.
 
+        If 'leader_only' is set to True (default value) the IDL will only monitor
+        and transact with the leader of the cluster.
+
         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.
@@ -137,6 +143,20 @@  class Idl(object):
         self._last_seqno = None
         self.change_seqno = 0
         self.uuid = uuid.uuid1()
+
+        # Server monitor.
+        self._server_schema_request_id = None
+        self._server_monitor_request_id = None
+        self._db_change_aware_request_id = None
+        self._server_db_name = '_Server'
+        self._server_db_table = 'Database'
+        self.server_tables = None
+        self._server_db = None
+        self.server_monitor_uuid = uuid.uuid1()
+        self.leader_only = leader_only
+        self.cluster_id = None
+        self._min_index = 0
+
         self.state = self.IDL_S_INITIAL
 
         # Database locking.
@@ -172,6 +192,15 @@  class Idl(object):
                 remotes.append(r)
         return remotes
 
+    def set_cluster_id(self, cluster_id):
+        """Set the id of the cluster that this idl must connect to."""
+        if cluster_id:
+            self.cluster_id = str(cluster_id)
+        else:
+            self.cluster_id = None
+        if self.state != self.IDL_S_INITIAL:
+            self.force_reconnect()
+
     def index_create(self, table, name):
         """Create a named multi-column index on a table"""
         return self.tables[table].rows.index_create(name)
@@ -222,7 +251,7 @@  class Idl(object):
             if seqno != self._last_seqno:
                 self._last_seqno = seqno
                 self.__txn_abort_all()
-                self.__send_monitor_request()
+                self.__send_server_schema_request()
                 if self.lock_name:
                     self.__send_lock_request()
                 break
@@ -230,6 +259,7 @@  class Idl(object):
             msg = self._session.recv()
             if msg is None:
                 break
+
             if (msg.type == ovs.jsonrpc.Message.T_NOTIFY
                     and msg.method == "update2"
                     and len(msg.params) == 2):
@@ -239,7 +269,15 @@  class Idl(object):
                     and msg.method == "update"
                     and len(msg.params) == 2):
                 # Database contents changed.
-                self.__parse_update(msg.params[1], OVSDB_UPDATE)
+                if msg.params[0] == str(self.server_monitor_uuid):
+                    self.__parse_update(msg.params[1], OVSDB_UPDATE,
+                                        tables=self.server_tables)
+                    self.change_seqno = initial_change_seqno
+                    if not self.__check_server_db():
+                        self.force_reconnect()
+                        break
+                else:
+                    self.__parse_update(msg.params[1], OVSDB_UPDATE)
             elif (msg.type == ovs.jsonrpc.Message.T_REPLY
                   and self._monitor_request_id is not None
                   and self._monitor_request_id == msg.id):
@@ -248,17 +286,66 @@  class Idl(object):
                     self.change_seqno += 1
                     self._monitor_request_id = None
                     self.__clear()
-                    if self.state == self.IDL_S_MONITOR_COND_REQUESTED:
+                    if self.state == self.IDL_S_DATA_MONITOR_COND_REQUESTED:
                         self.__parse_update(msg.result, OVSDB_UPDATE2)
                     else:
-                        assert self.state == self.IDL_S_MONITOR_REQUESTED
+                        assert self.state == self.IDL_S_DATA_MONITOR_REQUESTED
                         self.__parse_update(msg.result, OVSDB_UPDATE)
-
                 except error.Error as e:
                     vlog.err("%s: parse error in received schema: %s"
                              % (self._session.get_name(), e))
                     self.__error()
             elif (msg.type == ovs.jsonrpc.Message.T_REPLY
+                  and self._server_schema_request_id is not None
+                  and self._server_schema_request_id == msg.id):
+                # Reply to our "get_schema" of _Server request.
+                try:
+                    self._server_schema_request_id = None
+                    sh = SchemaHelper(None, msg.result)
+                    sh.register_table(self._server_db_table)
+                    schema = sh.get_idl_schema()
+                    self._server_db = schema
+                    self.server_tables = schema.tables
+                    self.__send_server_monitor_request()
+                except error.Error as e:
+                    vlog.err("%s: error receiving server schema: %s"
+                             % (self._session.get_name(), e))
+                    if self.cluster_id:
+                        self.__error()
+                        break
+                    else:
+                        self.change_seqno = initial_change_seqno
+                        self.__send_monitor_request()
+            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
+                  and self._server_monitor_request_id is not None
+                  and self._server_monitor_request_id == msg.id):
+                # Reply to our "monitor" of _Server request.
+                try:
+                    self._server_monitor_request_id = None
+                    self.__parse_update(msg.result, OVSDB_UPDATE,
+                                        tables=self.server_tables)
+                    self.change_seqno = initial_change_seqno
+                    if self.__check_server_db():
+                        self.__send_monitor_request()
+                        self.__send_db_change_aware()
+                    else:
+                        self.force_reconnect()
+                        break
+                except error.Error as e:
+                    vlog.err("%s: parse error in received schema: %s"
+                             % (self._session.get_name(), e))
+                    if self.cluster_id:
+                        self.__error()
+                        break
+                    else:
+                        self.change_seqno = initial_change_seqno
+                        self.__send_monitor_request()
+            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
+                  and self._db_change_aware_request_id is not None
+                  and self._db_change_aware_request_id == msg.id):
+                # Reply to us notifying the server of our change awarness.
+                self._db_change_aware_request_id = None
+            elif (msg.type == ovs.jsonrpc.Message.T_REPLY
                   and self._lock_request_id is not None
                   and self._lock_request_id == msg.id):
                 # Reply to our "lock" request.
@@ -275,10 +362,20 @@  class Idl(object):
                 # Reply to our echo request.  Ignore it.
                 pass
             elif (msg.type == ovs.jsonrpc.Message.T_ERROR and
-                  self.state == self.IDL_S_MONITOR_COND_REQUESTED 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()
+            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):
+                self._server_schema_request_id = None
+                if self.cluster_id:
+                    self.force_reconnect()
+                    break
+                else:
+                    self.change_seqno = initial_change_seqno
+                    self.__send_monitor_request()
             elif (msg.type in (ovs.jsonrpc.Message.T_ERROR,
                                ovs.jsonrpc.Message.T_REPLY)
                   and self.__txn_process_reply(msg)):
@@ -440,12 +537,19 @@  class Idl(object):
             if not new_has_lock:
                 self.is_lock_contended = True
 
+    def __send_db_change_aware(self):
+        msg = ovs.jsonrpc.Message.create_request("set_db_change_aware",
+                                                 [True])
+        self._db_change_aware_request_id = msg.id
+        self._session.send(msg)
+
     def __send_monitor_request(self):
-        if self.state == self.IDL_S_INITIAL:
-            self.state = self.IDL_S_MONITOR_COND_REQUESTED
+        if (self.state in [self.IDL_S_SERVER_MONITOR_REQUESTED,
+                           self.IDL_S_INITIAL]):
+            self.state = self.IDL_S_DATA_MONITOR_COND_REQUESTED
             method = "monitor_cond"
         else:
-            self.state = self.IDL_S_MONITOR_REQUESTED
+            self.state = self.IDL_S_DATA_MONITOR_REQUESTED
             method = "monitor"
 
         monitor_requests = {}
@@ -467,20 +571,50 @@  class Idl(object):
         self._monitor_request_id = msg.id
         self._session.send(msg)
 
-    def __parse_update(self, update, version):
+    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
+        res = self._session.send(msg)
+
+    def __send_server_monitor_request(self):
+        self.state = self.IDL_S_SERVER_MONITOR_REQUESTED
+        monitor_requests = {}
+        table = self.server_tables[self._server_db_table]
+        columns = [column for column in six.iterkeys(table.columns)]
+        for column in six.itervalues(table.columns):
+            if not hasattr(column, 'alert'):
+                column.alert = True
+        table.rows = custom_index.IndexedRows(table)
+        table.need_table = False
+        table.idl = self
+        monitor_request = {"columns": columns}
+        monitor_requests[table.name] = [monitor_request]
+        msg = ovs.jsonrpc.Message.create_request(
+            'monitor', [self._server_db.name,
+                             str(self.server_monitor_uuid),
+                             monitor_requests])
+        self._server_monitor_request_id = msg.id
+        self._session.send(msg)
+
+    def __parse_update(self, update, version, tables=None):
         try:
-            self.__do_parse_update(update, version)
+            if not tables:
+                self.__do_parse_update(update, version, self.tables)
+            else:
+                self.__do_parse_update(update, version, tables)
         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):
+    def __do_parse_update(self, table_updates, version, tables):
         if not isinstance(table_updates, dict):
             raise error.Error("<table-updates> is not an object",
                               table_updates)
 
         for table_name, table_update in six.iteritems(table_updates):
-            table = self.tables.get(table_name)
+            table = tables.get(table_name)
             if not table:
                 raise error.Error('<table-updates> includes unknown '
                                   'table "%s"' % table_name)
@@ -605,6 +739,57 @@  class Idl(object):
                 self.notify(op, row, Row.from_json(self, table, uuid, old))
         return changed
 
+    def __check_server_db(self):
+        """Returns True if this is a valid ovsdb server, False otherwise."""
+        session_name = self._session.get_name()
+
+        if self._server_db_table not in self.server_tables:
+            vlog.info("%s: server does not have %s table in its %s database"
+                      % (session_name, self._server_db_table, self._server_db_name))
+            return False
+
+        rows = self.server_tables[self._server_db_table].rows
+
+        database = None
+        for row in six.itervalues(rows):
+            if self.cluster_id:
+                if self.cluster_id in \
+                   map(lambda x: str(x)[:4], row.cid):
+                    database = row
+                    break
+            elif row.name == self._db.name:
+                database = row
+                break
+
+        if not database:
+            vlog.info("%s: server does not have %s database"
+                      % (session_name, self._db.name))
+            return False
+
+        if (database.model == CLUSTERED and
+            self._session.get_num_of_remotes() > 1):
+            if not database.schema:
+                vlog.info('%s: clustered database server has not yet joined '
+                          'cluster; trying another server' % session_name)
+                return False
+            if not database.connected:
+                vlog.info('%s: clustered database server is disconnected '
+                          'from cluster; trying another server' % session_name)
+                return False
+            if (self.leader_only and
+                not database.leader):
+                vlog.info('%s: clustered database server is not cluster '
+                          'leader; trying another server' % session_name)
+                return False
+            if database.index:
+                if database.index[0] < self._min_index:
+                    vlog.warn('%s: clustered database server has stale data; '
+                              'trying another server' % session_name)
+                    return False
+                self._min_index = database.index[0]
+
+        return True
+
     def __column_name(self, column):
         if column.type.key.type == ovs.db.types.UuidType:
             return ovs.ovsuuid.to_json(column.type.key.type.default)
diff --git a/python/ovs/reconnect.py b/python/ovs/reconnect.py
index 34cc769..afbe445 100644
--- a/python/ovs/reconnect.py
+++ b/python/ovs/reconnect.py
@@ -344,6 +344,9 @@  class Reconnect(object):
                 else:
                     self.info_level("%s: error listening for connections"
                                     % self.name)
+            elif self.state == Reconnect.Reconnect:
+                self.info_level("%s: connection closed by client"
+                                % self.name)
             elif self.backoff < self.max_backoff:
                 if self.passive:
                     type_ = "listen"
diff --git a/tests/ovsdb-idl.at b/tests/ovsdb-idl.at
index 8981b5e..46b047f 100644
--- a/tests/ovsdb-idl.at
+++ b/tests/ovsdb-idl.at
@@ -1466,40 +1466,44 @@  OVSDB_CHECK_IDL_NOTIFY([simple idl verify notify],
        "where": [["i", "==", 0]]}]' \
     'reconnect']],
   [[000: empty
-001: {"error":null,"result":[{"uuid":["uuid","<0>"]},{"uuid":["uuid","<1>"]}]}
-002: event:create, row={i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>}, updates=None
-002: event:create, row={i=1 r=2 b=true s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates=None
-002: i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
-002: i=1 r=2 b=true s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+000: event:create, row={uuid=<0>}, updates=None
+000: event:create, row={uuid=<1>}, updates=None
+001: {"error":null,"result":[{"uuid":["uuid","<2>"]},{"uuid":["uuid","<3>"]}]}
+002: event:create, row={i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>}, updates=None
+002: event:create, row={i=1 r=2 b=true s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates=None
+002: i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
+002: i=1 r=2 b=true s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 003: {"error":null,"result":[{"count":2}]}
-004: event:update, row={i=1 r=2 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates={b=true uuid=<0>}
-004: i=0 r=0 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
-004: i=1 r=2 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+004: event:update, row={i=1 r=2 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates={b=true uuid=<2>}
+004: i=0 r=0 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
+004: i=1 r=2 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 005: {"error":null,"result":[{"count":2}]}
-006: event:update, row={i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>}, updates={r=0 uuid=<1>}
-006: event:update, row={i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates={r=2 uuid=<0>}
-006: i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
-006: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
-007: {"error":null,"result":[{"uuid":["uuid","<6>"]}]}
-008: event:create, row={i=-1 r=125 b=false s= u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>}, updates=None
-008: i=-1 r=125 b=false s= u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>
-008: i=0 r=123.5 b=false s= u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
-008: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+006: event:update, row={i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>}, updates={r=0 uuid=<3>}
+006: event:update, row={i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates={r=2 uuid=<2>}
+006: i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
+006: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
+007: {"error":null,"result":[{"uuid":["uuid","<8>"]}]}
+008: event:create, row={i=-1 r=125 b=false s= u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>}, updates=None
+008: i=-1 r=125 b=false s= u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>
+008: i=0 r=123.5 b=false s= u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
+008: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 009: {"error":null,"result":[{"count":2}]}
-010: event:update, row={i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>}, updates={s= uuid=<6>}
-010: event:update, row={i=0 r=123.5 b=false s=newstring u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>}, updates={s= uuid=<1>}
-010: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>
-010: i=0 r=123.5 b=false s=newstring u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>
-010: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+010: event:update, row={i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>}, updates={s= uuid=<8>}
+010: event:update, row={i=0 r=123.5 b=false s=newstring u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>}, updates={s= uuid=<3>}
+010: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>
+010: i=0 r=123.5 b=false s=newstring u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>
+010: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 011: {"error":null,"result":[{"count":1}]}
-012: event:delete, row={i=0 r=123.5 b=false s=newstring u=<2> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<1>}, updates=None
-012: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>
-012: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+012: event:delete, row={i=0 r=123.5 b=false s=newstring u=<4> ia=[] ra=[] ba=[] sa=[] ua=[] uuid=<3>}, updates=None
+012: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>
+012: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 013: reconnect
-014: event:create, row={i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>}, updates=None
-014: event:create, row={i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>}, updates=None
-014: i=-1 r=125 b=false s=newstring u=<2> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<6>
-014: i=1 r=123.5 b=false s=mystring u=<3> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<4> <5>] uuid=<0>
+014: event:create, row={i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>}, updates=None
+014: event:create, row={i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>}, updates=None
+014: event:create, row={uuid=<0>}, updates=None
+014: event:create, row={uuid=<1>}, updates=None
+014: i=-1 r=125 b=false s=newstring u=<4> ia=[1] ra=[1.5] ba=[false] sa=[] ua=[] uuid=<8>
+014: i=1 r=123.5 b=false s=mystring u=<5> ia=[1 2 3] ra=[-0.5] ba=[true] sa=[abc def] ua=[<6> <7>] uuid=<2>
 015: done
 ]])