Merge "contrail-topology starts to report process_status"

src/analytics/contrail-topology/SConscript

@@ -25,19 +25,25 @@ sources = [
     'contrail_topology/topology_uve.py',
     ]
 
-src_sandesh = env.SandeshGenPy('link.sandesh', 'contrail_topology/')
+src_sandesh = env.SandeshGenPy('link.sandesh', 'contrail_topology/sandesh/', False)
+analytics_pkg = env.SandeshGenPy('#controller/src/analytics/analytics.sandesh', 'contrail_topology/sandesh/', False)
+cpu_info_pkg = env.SandeshGenPy('#controller/src/base/sandesh/cpuinfo.sandesh', 'contrail_topology/sandesh/analytics/', False)
+process_info_pkg = env.SandeshGenPy('#controller/src/base/sandesh/process_info.sandesh', 'contrail_topology/sandesh/analytics/', False)
 
 version = '0.0.1'
 #with open("CHANGES.txt") as f:
 #    version = f.read().split(",")[0][1:]
 
+sdist_depends = [ analytics_pkg, cpu_info_pkg, process_info_pkg]
 cd_cmd = 'cd ' + Dir('.').path + ' && '
 sdist_gen = env.Command('dist/contrail-topology-' \
         + version + '.tar.gz',
         sources + src_sandesh, cd_cmd + 'python setup.py sdist')
+env.Depends(sdist_gen, sdist_depends)
 env.Alias('controller/src/analytics/contrail-topology/' + \
         'contrail_topology:sdist', sdist_gen)
 
+
 src_rules = [env.Install(Dir("."),
     '#controller/src/analytics/contrail-topology/' + f) for f in sources]
 src_rules.append(env.Install(Dir("."),
@@ -111,6 +117,7 @@ if 'install' in BUILD_TARGETS:
     install_cmd = env.Command(None, sources + src_sandesh,
                               cd_cmd + 'python setup.py install %s' %
                               env['PYTHON_INSTALL_OPT'])
+    env.Depends(install_cmd, sdist_depends)
     env.Alias('install', install_cmd)
 env.Alias('install', env.Install(env['INSTALL_CONF'],
             'contrail-topology.conf'))

src/analytics/contrail-topology/contrail_topology/__init__.py

@@ -1,3 +0,0 @@
-#
-# Copyright (c) 2015 Juniper Networks, Inc. All rights reserved.
-#

src/analytics/contrail-topology/contrail_topology/topology_uve.py

@@ -4,7 +4,8 @@
 import pprint, socket
 from pysandesh.sandesh_base import *
 from pysandesh.connection_info import ConnectionState
-from gen_py.link.ttypes import LinkEntry, PRouterLinkEntry, PRouterLinkUVE
+from opserver.sandesh.analytics.ttypes import NodeStatusUVE, NodeStatus
+from sandesh.link.ttypes import LinkEntry, PRouterLinkEntry, PRouterLinkUVE
 from sandesh_common.vns.ttypes import Module, NodeType
 from sandesh_common.vns.constants import ModuleNames, CategoryNames,\
      ModuleCategoryMap, Module2NodeType, NodeTypeNames, ModuleIds,\
@@ -27,19 +28,19 @@ def __init__(self, conf):
                                       self._conf.collectors(), 
                                       self._node_type_name,
                                       self._conf.http_port(),
-                                      ['contrail_topology.gen_py'])
+                                      ['contrail_topology.sandesh'])
         sandesh_global.set_logging_params(
             enable_local_log=self._conf.log_local(),
             category=self._conf.log_category(),
             level=self._conf.log_level(),
             file=self._conf.log_file(),
             enable_syslog=self._conf.use_syslog(),
             syslog_facility=self._conf.syslog_facility())
-        #ConnectionState.init(sandesh_global, self._hostname, self._moduleid,
-        #    self._instance_id,
-        #    staticmethod(ConnectionState.get_process_state_cb),
-        #    NodeStatusUVE, NodeStatus)
-        #
+        ConnectionState.init(sandesh_global, self._hostname, self._moduleid,
+            self._instance_id,
+            staticmethod(ConnectionState.get_process_state_cb),
+            NodeStatusUVE, NodeStatus)
+
         # generator_init()
 
     def send(self, data):

src/opserver/consistent_schdlr.py

@@ -6,11 +6,11 @@
 import hashlib
 import logging
 from kazoo.client import KazooClient
+from kazoo.client import KazooState
 from random import randint
 import struct
 import traceback
 
-
 class ConsistentScheduler(object):
     '''
         LibPartitionHelper abstract out workers and work_items, and their
@@ -44,15 +44,66 @@ def __init__(self, service_name=None, zookeeper='127.0.0.1:2181',
         self._zk_path = '/'.join(['/contrail_cs', self._service_name])
         self._zk = KazooClient(self._zookeeper_srvr)
         self._zk.add_listener(self._zk_lstnr)
-        self._zk.start()
+        self._conn_state = None
+        while True:
+            try:
+                self._zk.start()
+                break
+            except gevent.event.Timeout as e:
+                # Update connection info
+                self._sandesh_connection_info_update(status='DOWN',
+                                                     message=str(e))
+                gevent.sleep(1)
+            # Zookeeper is also throwing exception due to delay in master election
+            except Exception as e:
+                # Update connection info
+                self._sandesh_connection_info_update(status='DOWN',
+                                                     message=str(e))
+                gevent.sleep(1)
         self._pc = self._zk.SetPartitioner(path=self._zk_path,
                                            set=self._partition_set,
                                            partition_func=self._partitioner)
         self._wait_allocation = 0
         gevent.sleep(0)
 
+    def _sandesh_connection_info_update(self, status, message):
+        from pysandesh.connection_info import ConnectionState
+        from pysandesh.gen_py.process_info.ttypes import ConnectionStatus, \
+            ConnectionType
+
+        new_conn_state = getattr(ConnectionStatus, status)
+        ConnectionState.update(conn_type = ConnectionType.ZOOKEEPER,
+                name = 'Zookeeper', status = new_conn_state,
+                message = message,
+                server_addrs = self._zookeeper_srvr.split(','))
+
+        if ((self._conn_state and self._conn_state != ConnectionStatus.DOWN) and
+            new_conn_state == ConnectionStatus.DOWN):
+            msg = 'Connection to Zookeeper down: %s' %(message)
+            self._supress_log(msg)
+        if (self._conn_state and self._conn_state != new_conn_state and
+            new_conn_state == ConnectionStatus.UP):
+            msg = 'Connection to Zookeeper ESTABLISHED'
+            self._supress_log(msg)
+
+        self._conn_state = new_conn_state
+    # end _sandesh_connection_info_update
+
     def _zk_lstnr(self, state):
-        self._supress_log('zk state change to %s' % str(state))
+        if state == KazooState.CONNECTED:
+            # Update connection info
+            self._sandesh_connection_info_update(status='UP', message='')
+        elif state == KazooState.LOST:
+            # Lost the session with ZooKeeper Server
+            # Best of option we have is to exit the process and restart all 
+            # over again
+            self._sandesh_connection_info_update(status='DOWN',
+                                      message='Connection to Zookeeper lost')
+            os._exit(2)
+        elif state == KazooState.SUSPENDED:
+            # Update connection info
+            self._sandesh_connection_info_update(status='INIT',
+                message = 'Connection to zookeeper lost. Retrying')
 
     def schedule(self, items, lock_timeout=30):
         gevent.sleep(0)

src/sandesh/common/vns.sandesh

@@ -141,6 +141,7 @@ const list<string> NodeUVEImplementedServices = [
     SERVICE_COMPUTE_NODE_MGR,
     SERVICE_CONTROL_NODE_MGR,
     SERVICE_DATABASE_NODE_MGR,
+    SERVICE_TOPOLOGY,
 ]
 
 const list<string> BackupImplementedServices = [