Merge "Scale fixes in Alarmgen. - Timeout for releasing partition own…

…ership is extended from 60s to 120s - Took out printing of UVE Keys in log messages - these lists can get too big - We only process upto 200 UVE updates at at time. Unlimited processing can starve other partitions - Changed log levels of some libpartition logs - Fixed discovery client ID of alarmgen to accomodate multiple alarmgens per analytics node - Increased partitions from 15 to 30 Closes-Bug: 1596795 Change-Id: Id4a8f9a7c411e8e1a9c223afd7be3455abfff855" into R3.0

src/analytics/contrail-collector.conf

@@ -24,6 +24,7 @@
 
 # List of IP:port for kafka brokers
 # kafka_broker_list=127.0.0.1:9092
+partitions=30
 
 # IP address of analytics node. Resolved IP of 'hostname'
 # hostip=

src/libpartition/libpartition.py

@@ -186,13 +186,13 @@ def _acquire_partition_ownership(self):
         # list of partitions for which locks have to be released
         release_lock_list = []
 
-        self._logger.error("known servers: %s" % self._con_hash.get_all_nodes())
+        self._logger.info("known servers: %s" % self._con_hash.get_all_nodes())
 
         for part in range(0, self._max_partition):
             if (part in self._target_part_ownership_list):
                 if (part in self._curr_part_ownership_list):
                     # do nothing, I already have ownership of this partition
-                    self._logger.error("No need to acquire ownership of:" +
+                    self._logger.info("No need to acquire ownership of:" +
                             str(part))
                 else:
                     # I need to acquire lock for this partition before I own
@@ -274,13 +274,13 @@ def update_cluster_list(self, cluster_list):
         deleted_servers = list(set(self._cluster_list).difference(
             new_cluster_list)) 
         self._cluster_list = set(cluster_list)
-        self._logger.error("deleted servers:" + str(deleted_servers))
-        self._logger.error("new servers:" + str(new_servers))
 
         # update the hash structure
         if new_servers:
+            self._logger.error("new servers:" + str(new_servers))
             self._con_hash.add_nodes(new_servers)
         if deleted_servers:
+            self._logger.error("deleted servers:" + str(deleted_servers))
             self._con_hash.del_nodes(deleted_servers)
 
         # update target partition ownership list

src/opserver/alarmgen.py

@@ -565,7 +565,8 @@ def __init__(self, conf, test_logger=None):
             self.disc = client.DiscoveryClient(
                 self._conf.discovery()['server'],
                 self._conf.discovery()['port'],
-                ModuleNames[Module.ALARM_GENERATOR])
+                ModuleNames[Module.ALARM_GENERATOR],
+                '%s-%s' % (self._hostname, self._instance_id))
 
         is_collector = True
         if test_logger is not None:
@@ -1009,9 +1010,16 @@ def run_uve_processing(self):
 
                     # Allow the partition handlers to queue new UVEs without
                     # interfering with the work of processing the current UVEs
-                    pendingset[part] = copy.deepcopy(self._uveq[part])
-                    self._uveq[part] = {}
-
+                    # Process no more than 200 keys at a time
+                    pendingset[part] = {}
+                    icount = 0
+                    while (len(self._uveq[part]) > 0) and icount < 200:
+                        kp,vp = self._uveq[part].popitem()
+                        pendingset[part][kp] = vp
+                        icount += 1
+                    self._logger.info("UVE Process for %d : %d, %d remain" % \
+                            (part, len(pendingset[part]), len(self._uveq[part])))
+
                     gevs[part] = gevent.spawn(self.handle_uve_notif,part,\
                         pendingset[part])
                 if len(gevs):
@@ -1489,7 +1497,7 @@ def partition_change(self, partno, enl):
                 ph.start()
                 self._workers[partno] = ph
                 self._uvestats[partno] = {}
-                tout = 600
+                tout = 1200
                 idx = 0
                 while idx < tout:
                     # When this partitions starts,
@@ -1523,7 +1531,7 @@ def partition_change(self, partno, enl):
                 del self._workers[partno]
                 del self._uvestats[partno]
 
-                tout = 600
+                tout = 1200
                 idx = 0
                 while idx < tout:
                     # When this partitions stop.s

src/opserver/contrail-alarm-gen.conf

@@ -7,6 +7,7 @@ log_level = SYS_NOTICE
 #log_category = 
 log_file = /var/log/contrail/contrail-alarm-gen.log
 #kafka_broker_list = xx.xx.xx.xx:9092
+partitions=30
 #zk_list = xx.xx.xx.xx:2181
 
 [DISCOVERY]

src/opserver/contrail-analytics-api.conf

@@ -12,6 +12,7 @@ log_file = /var/log/contrail/contrail-analytics-api.log
 # Sandesh send rate limit can be used to throttle system logs transmitted per
 # second. System logs are dropped if the sending rate is exceeded
 #sandesh_send_rate_limit = 100
+partitions=30
 
 [DISCOVERY]
 #disc_server_ip = 127.0.0.1

src/opserver/partition_handler.py

@@ -750,8 +750,8 @@ def start_partition(self, cbdb):
                         # TODO: for loading only specific types:
                         #       uves[kk][typ] = contents
 
-        self._logger.error("Starting part %d UVEs %s" % \
-                          (self._partno, str(uves.keys())))
+        self._logger.error("Starting part %d UVEs %d" % \
+                           (self._partno, len(uves)))
         self._callback(self._partno, uves)
 
     def contents(self):