Clean arbiter/scheduler interface

Author: mohierf

alignak/daemons/schedulerdaemon.py

@@ -473,6 +473,9 @@ def setup_new_conf(self):
                 # Ok now we can load the retention data
                 self.sched.retention_load()
 
+                # Log hosts/services initial states
+                self.sched.log_initial_states()
+
             # Create brok new conf
             brok = Brok({'type': 'new_conf', 'data': {}})
             self.sched.add_brok(brok)

alignak/dispatcher.py

@@ -304,7 +304,8 @@ def check_status_and_get_events(self):
                 continue
 
             try:
-                daemon_link.statistics = daemon_link.get_daemon_stats(details=True)
+                # Do not get the details to avoid overloading the communication
+                daemon_link.statistics = daemon_link.get_daemon_stats(details=False)
                 if daemon_link.statistics:
                     daemon_link.statistics['_freshness'] = int(time.time())
                     statistics[daemon_link.name] = daemon_link.statistics

alignak/http/arbiter_interface.py

@@ -429,8 +429,17 @@ def monitoring_problems(self):
         :rtype: dict
         """
         res = self.get_id()
-        res.update(self.get_start_time())
-        res.update(self.app.get_monitoring_problems())
+        res['problems'] = {}
+        for scheduler_link in self.app.conf.schedulers:
+            print("Scheduler link: %s" % scheduler_link)
+            sched_res = scheduler_link.con.get('monitoring_problems', wait=True)
+            res['problems'][scheduler_link.name] = {}
+            if '_freshness' in sched_res:
+                res['problems'][scheduler_link.name].update({'_freshness': sched_res['_freshness']})
+            if 'problems' in sched_res:
+                res['problems'][scheduler_link.name].update({'problems': sched_res['problems']})
+        res['_freshness'] = int(time.time())
+
         return res
 
     @cherrypy.expose
@@ -596,7 +605,8 @@ def object(self, o_type, o_name='None'):
         :rtype: str
         """
         for scheduler_link in self.app.conf.schedulers:
-            o_found = scheduler_link.get_object(o_type, o_name)
-            if isinstance(o_found, dict) and 'content' in o_found:
-                return o_found
+            sched_res = scheduler_link.con.get('object', {'o_type': o_type, 'o_name': o_name},
+                                               wait=True)
+            if isinstance(sched_res, dict) and 'content' in sched_res:
+                return sched_res
         return {'_status': u'ERR', '_message': u'Required %s not found.' % o_type}

alignak/http/scheduler_interface.py

@@ -163,7 +163,7 @@ def fill_initial_broks(self, broker_name):
         """
         with self.app.conf_lock:
             logger.info("A new broker just connected : %s", broker_name)
-            return self.app.sched.fill_initial_broks(broker_name, with_logs=True)
+            return self.app.sched.fill_initial_broks(broker_name)
 
     @cherrypy.expose
     @cherrypy.tools.json_in()

alignak/modules/inner_retention.py

@@ -139,12 +139,12 @@ def hook_load_retention(self, scheduler):
             else:
                 for host in response.values():
                     hostname = host['name']
-                    service_key = 'key'
+                    service_key = 'services'
                     if 'retention_services' in host:
                         service_key = 'retention_services'
                     if service_key in host:
                         for service in host[service_key]:
-                            all_data[service_key][(host['host'], service)] = \
+                            all_data['services'][(host['name'], service)] = \
                                 host[service_key][service]
                     all_data['hosts'][hostname] = host
 

alignak/objects/satellitelink.py

@@ -781,7 +781,7 @@ def get_initial_broks(self, broker_name):
         :type: bool
         """
         logger.debug("Getting initial broks for %s, %s %s", self.name, self.alive, self.reachable)
-        return self.con.get('fill_initial_broks', {'broker_name': broker_name}, wait='long')
+        return self.con.get('fill_initial_broks', {'broker_name': broker_name}, wait=True)
 
     @valid_connection()
     @communicate()
@@ -814,7 +814,7 @@ def put_conf(self, configuration, test=False):
             return True
         # ----------
 
-        return self.con.post('push_configuration', {'conf': configuration}, wait='long')
+        return self.con.post('push_configuration', {'conf': configuration}, wait=True)
 
     @valid_connection()
     @communicate()
@@ -885,7 +885,7 @@ def push_broks(self, broks):
         :rtype: bool
         """
         logger.debug("[%s] Pushing %d broks", self.name, len(broks))
-        return self.con.post('push_broks', {'broks': broks}, wait='long')
+        return self.con.post('push_broks', {'broks': broks}, wait=True)
 
     @valid_connection()
     @communicate()
@@ -903,7 +903,7 @@ def push_actions(self, actions, scheduler_instance_id):
         logger.debug("Pushing %d actions from %s", len(actions), scheduler_instance_id)
         return self.con.post('push_actions', {'actions': actions,
                                               'scheduler_instance_id': scheduler_instance_id},
-                             wait='long')
+                             wait=True)
 
     @valid_connection()
     @communicate()
@@ -921,7 +921,7 @@ def push_results(self, results, scheduler_name):
         """
         logger.debug("Pushing %d results", len(results))
         result = self.con.post('put_results', {'results': results, 'from': scheduler_name},
-                               wait='long')
+                               wait=True)
         return result
 
     @valid_connection()
@@ -936,7 +936,7 @@ def push_external_commands(self, commands):
         :rtype: bool
         """
         logger.debug("Pushing %d external commands", len(commands))
-        return self.con.post('run_external_commands', {'cmds': commands}, wait='long')
+        return self.con.post('run_external_commands', {'cmds': commands}, wait=True)
 
     @valid_connection()
     @communicate()
@@ -947,7 +947,7 @@ def get_external_commands(self):
         :return: External Command list on success, [] on failure
         :rtype: list
         """
-        res = self.con.get('get_external_commands', wait=True)
+        res = self.con.get('get_external_commands', wait=False)
         logger.debug("Got %d external commands from %s: %s", len(res), self.name, res)
         return unserialize(res, True)
 
@@ -963,7 +963,7 @@ def get_broks(self, broker_name):
         :return: Broks list on success, [] on failure
         :rtype: list
         """
-        res = self.con.get('get_broks', {'broker_name': broker_name}, wait=True)
+        res = self.con.get('get_broks', {'broker_name': broker_name}, wait=False)
         logger.debug("Got broks from %s: %s", self.name, res)
         return unserialize(res, True)
 
@@ -976,7 +976,7 @@ def get_events(self):
         :return: Broks list on success, [] on failure
         :rtype: list
         """
-        res = self.con.get('get_events', wait=True)
+        res = self.con.get('get_events', wait=False)
         logger.debug("Got events from %s: %s", self.name, res)
         return unserialize(res, True)
 
@@ -991,7 +991,7 @@ def get_results(self, scheduler_instance_id):
         :rtype: list
         """
         res = self.con.get('get_results', {'scheduler_instance_id': scheduler_instance_id},
-                           wait='long')
+                           wait=True)
         logger.debug("Got %d results from %s: %s", len(res), self.name, res)
         return res
 
@@ -1010,23 +1010,6 @@ def get_actions(self, params):
         logger.debug("Got actions from %s: %s", self.name, res)
         return unserialize(res, True)
 
-    @valid_connection()
-    def get_object(self, o_type, o_name):
-        """Send a HTTP request to the scheduler (GET /get_host)
-        Get host information from the scheduler.
-        Un-serialize data received.
-
-        :param o_type: searched object type
-        :type o_type: str
-        :param o_name: searched object name (or uuid)
-        :type o_name: str
-        :return: serialized object information
-        :rtype: str
-        """
-        res = self.con.get('object', {'o_type': o_type, 'o_name': o_name}, wait=True)
-        logger.debug("Got an %s object from %s: %s", o_type, self.name, res)
-        return res
-
 
 class SatelliteLinks(Items):
     """Class to handle serveral SatelliteLink"""

alignak/objects/service.py

@@ -661,9 +661,9 @@ def raise_initial_state(self):
         :return: None
         """
         log_level = 'info'
-        if self.state == 'WARNING':
+        if self.state in ['WARNING', 'UNREACHABLE']:
             log_level = 'warning'
-        if self.state == 'CRITICAL':
+        if self.state in ['CRITICAL', 'UNKNOWN']:
             log_level = 'error'
         if self.__class__.log_initial_states:
             brok = make_monitoring_log(

alignak/scheduler.py

@@ -1393,6 +1393,16 @@ def retention_load(self, forced=False):
         self.add(make_monitoring_log('INFO', 'RETENTION LOAD: %s' % self.my_daemon.name))
         logger.info('Retention data loaded: %.2f seconds', time.time() - _t0)
 
+    def log_initial_states(self):
+        """Log hosts and services initial state
+
+        :return: None
+        """
+        if getattr(self.my_daemon, 'log_initial_states', False):
+            # Raise current state log
+            for elt in self.all_my_hosts_and_services():
+                elt.raise_initial_state()
+
     def get_retention_data(self):  # pylint: disable=too-many-branches,too-many-statements
         # pylint: disable=too-many-locals
         """Get all hosts and services data to be sent to the retention storage.
@@ -1571,14 +1581,12 @@ def restore_retention_data_item(self, data, item):
         item.notified_contacts = new_notified_contacts
         item.notified_contacts_ids = new_notified_contacts_ids
 
-    def fill_initial_broks(self, broker_name, with_logs=False):
+    def fill_initial_broks(self, broker_name):
         # pylint: disable=too-many-branches
         """Create initial broks for a specific broker
 
         :param broker_name: broker name
         :type broker_name: str
-        :param with_logs: tell if we raise some broks to log the initial states for hosts/services
-        :type with_logs: bool
         :return: number of created broks
         """
         broker_uuid = None
@@ -1632,14 +1640,6 @@ def fill_initial_broks(self, broker_name, with_logs=False):
                     brok = item.get_initial_status_brok(member_items)
                     self.add_brok(brok, broker_uuid)
 
-        # Only raises the all logs at the scheduler startup
-        if with_logs:
-            # Ask for INITIAL logs for services and hosts
-            for item in self.hosts:
-                item.raise_initial_state()
-            for item in self.services:
-                item.raise_initial_state()
-
         # Add a brok to say that we finished all initial_pass
         brok = Brok({'type': 'initial_broks_done', 'data': {'instance_id': self.instance_id}})
         self.add_brok(brok, broker_uuid)
@@ -2107,13 +2107,15 @@ def get_scheduler_stats(self, details=False):  # pylint: disable=unused-argument
              'hosts': len(self.hosts),
              'services': len(self.services),
              'commands': [{'cmd': c, 'u_time': u_time, 's_time': s_time}, ...] (10 first)
+             'livesynthesis': {...}
            }
 
         :rtype: dict
         """
         m_solver = MacroResolver()
 
         res = {
+            '_freshness': int(time.time()),
             'counters': {},
             'latency': self.stats['latency'],
             'monitored_objects': {},

tests/cfg/cfg_default_retention.cfg

@@ -1,5 +1,7 @@
 cfg_dir=default
 
+log_initial_states=true
+
 ; --------------------------------------------------------------------
 ; Retention configuration
 ; ---

tests/test_a_properties_default.py

@@ -266,7 +266,7 @@ class TestConfig(PropertiesTester, AlignakTest):
         ('daemons_start_timeout', 1),
         ('daemons_dispatch_timeout', 5),
         ('daemons_new_conf_timeout', 1),
-        ('daemons_stop_timeout', 15),
+        ('daemons_stop_timeout', 5),
         ('daemons_failure_kill', True),
 
         ('alignak_env', []),

tests/test_config.py

@@ -190,7 +190,7 @@ def test_config_conf_inner_properties(self):
         # assert self._arbiter.conf.daemons_start_timeout == 1
         # Changed to 5 seconds for tests purpose
         assert self._arbiter.conf.daemons_start_timeout == 1
-        assert self._arbiter.conf.daemons_stop_timeout == 15
+        assert self._arbiter.conf.daemons_stop_timeout == 5
 
     def test_config_conf_inner_properties_named_alignak(self):
         """ Default configuration with an alignak_name property