Closes #1012: monitoring log rotation

Closes #1027: monitoring log in the arbiter - still to be improved

Author: mohierf

alignak/brok.py

@@ -59,6 +59,9 @@ class Brok(object):
     """A Brok is a piece of information exported by Alignak to the Broker.
     Broker can do whatever he wants with it.
 
+    A specific type of Brok exists when the type is monitoring_log. This Brok contains
+    a monitoring event (alert, notification, ...) information
+
     Broks types:
     - log
     - monitoring_log
@@ -105,11 +108,24 @@ def __init__(self, params, parsing=True):
         else:
             self.data = serialize(params['data'])
 
-    def __repr__(self):  # pragma: no cover
+    def __repr__(self):
         ct = datetime.fromtimestamp(self.creation_time).strftime("%Y-%m-%d %H:%M:%S.%f")
         return "Brok %s (%s) '%s': %s" % (self.uuid, ct, self.type, self.data)
     __str__ = __repr__
 
+    def get_event(self):
+        """This function returns an Event from a Brok
+
+        If the type is monitoring_log then the Brok contains a monitoring event
+        (alert, notification, ...) information. This function will return a tuple
+        with the creation time, the level and message information
+
+        :return: tuple with date, level and message
+        :rtype: tuple
+        """
+        self.prepare()
+        return (self.creation_time, self.data['level'], self.data['message'])
+
     def serialize(self):
         """This function serialize into a simple dict object.
         It is used when transferring data to other daemons over the network (http)
@@ -130,7 +146,7 @@ def prepare(self):
 
         :return: None
         """
-        # Maybe the brok is a old daemon one or was already prepared
+        # Maybe the Brok is a old daemon one or was already prepared
         # if so, the data is already ok
         if hasattr(self, 'prepared') and not self.prepared:
             try:

alignak/daemon.py

@@ -626,7 +626,7 @@ def __init__(self, name, **kwargs):
         self.check_dir(os.path.dirname(self.log_filename))
 
         # Specific monitoring log directory
-        self.check_dir(os.path.join(os.path.dirname(self.log_filename), 'monitoring-log'))
+        # self.check_dir(os.path.join(os.path.dirname(self.log_filename), 'monitoring-log'))
 
         if 'log_filename' not in kwargs or not kwargs['log_filename']:
             # Log file name is not overridden, the logger will use the configured default one
@@ -1181,8 +1181,8 @@ def do_load_modules(self, modules):
         statsmgr.timer('modules.load-time', time.time() - _ts)
 
     def add(self, elt):
-        """ Abstract method for adding brok
-         It is overridden in subclasses of Daemon
+        """ Abstract method for adding brok, external commands, messages, ...
+         It is overridden in subclasses (Satellite) of Daemon
 
         :param elt: element to add
         :type elt:
@@ -1485,21 +1485,21 @@ def do_daemon_init_and_start(self, set_proc_title=True):
             # Else, I set my own pid as the reference one
             self.write_pid(os.getpid())
 
-        # TODO: check if really necessary!
-        # -------
-        # Set ownership on some default log files. It may happen that these default
-        # files are owned by a privileged user account
-        try:
-            for log_file in ['alignak.log', 'monitoring-logs.log']:
-                if os.path.exists('/tmp/%s' % log_file):
-                    with open('/tmp/%s' % log_file, "w") as file_log_file:
-                        os.fchown(file_log_file.fileno(), self.uid, self.gid)
-                if os.path.exists('/tmp/monitoring-log/%s' % log_file):
-                    with open('/tmp/monitoring-log/%s' % log_file, "w") as file_log_file:
-                        os.fchown(file_log_file.fileno(), self.uid, self.gid)
-        except Exception as exp:  # pylint: disable=broad-except
-            #  pragma: no cover
-            print("Could not set default log files ownership, exception: %s" % str(exp))
+        # # TODO: check if really necessary!
+        # # -------
+        # # Set ownership on some default log files. It may happen that these default
+        # # files are owned by a privileged user account
+        # try:
+        #     for log_file in ['alignak.log', 'alignak-events.log']:
+        #         if os.path.exists('/tmp/%s' % log_file):
+        #             with open('/tmp/%s' % log_file, "w") as file_log_file:
+        #                 os.fchown(file_log_file.fileno(), self.uid, self.gid)
+        #         if os.path.exists('/tmp/monitoring-log/%s' % log_file):
+        #             with open('/tmp/monitoring-log/%s' % log_file, "w") as file_log_file:
+        #                 os.fchown(file_log_file.fileno(), self.uid, self.gid)
+        # except Exception as exp:  # pylint: disable=broad-except
+        #     #  pragma: no cover
+        #     print("Could not set default log files ownership, exception: %s" % str(exp))
 
         # Configure the daemon logger
         self.setup_alignak_logger()
@@ -1910,8 +1910,7 @@ def watch_for_new_conf(self, timeout=0):
         return any(self.new_conf)
 
     def hook_point(self, hook_name, handle=None):
-        """Used to call module function that may define a hook function
-        for hook_name
+        """Used to call module function that may define a hook function for hook_name
 
         Available hook points:
         - `tick`, called on each daemon loop turn

alignak/daemons/arbiterdaemon.py

@@ -173,8 +173,15 @@ def __init__(self, **kwargs):
             except ValueError:  # pragma: no cover, simple protection
                 pass
 
+        # This because it is the Satellite that has these properties and I am a Daemon
+        # todo: change this?
+        # My own broks
         self.broks = []
         self.broks_lock = threading.RLock()
+        # My own monitoring events
+        self.events = []
+        self.events_lock = threading.RLock()
+
         self.is_master = False
         self.link_to_myself = None
         self.instance_id = None
@@ -210,9 +217,16 @@ def add(self, elt):
         :return: None
         """
         if isinstance(elt, Brok):
+            # For brok, we tag the brok with our instance_id
             elt.instance_id = self.instance_id
-            with self.broks_lock:
-                self.broks.append(elt)
+            if elt.type == 'monitoring_log':
+                # The brok is a monitoring event
+                with self.events_lock:
+                    self.events.append(elt)
+                statsmgr.counter('events', 1)
+            else:
+                with self.broks_lock:
+                    self.broks.append(elt)
             statsmgr.counter('broks.added', 1)
         elif isinstance(elt, ExternalCommand):
             logger.debug("Queuing an external command '%s'", str(elt.__dict__))
@@ -1159,14 +1173,21 @@ def daemons_reachability_check(self):
 
         _t0 = time.time()
         logger.debug("Alignak daemons reachability check")
-        self.dispatcher.check_reachable()
+        result = self.dispatcher.check_reachable()
         statsmgr.timer('dispatcher.check-alive', time.time() - _t0)
 
         _t0 = time.time()
         logger.debug("Alignak daemons status get")
-        result = self.dispatcher.check_status()
-        statsmgr.timer('dispatcher.check-status', time.time() - _t0)
-        logger.debug("Getting daemons status duration: %.2f seconds", time.time() - _t0)
+        events = self.dispatcher.check_status_and_get_events()
+        duration = time.time() - _t0
+        statsmgr.timer('dispatcher.check-status', duration)
+        logger.debug("Getting daemons status duration: %.2f seconds", duration)
+
+        # Send the collected events to the Alignak logger
+        for event in events:
+            event.prepare()
+            make_monitoring_log(event.data['level'], event.data['message'],
+                                timestamp=event.creation_time, to_logger=True)
 
         # Set the last check as now
         self.daemons_last_reachable_check = start
@@ -2060,11 +2081,9 @@ def main(self):
                         _ts = time.time()
                         logger.warning('--- Reloading configuration...')
                         self.load_monitoring_config_file()
-                        brok = make_monitoring_log('info', 'CONFIGURATION RELOAD')
-                        if self.conf.monitoring_log_broks:
-                            self.add(brok)
-                        logger.warning('--- Configuration reloaded, %.2f seconds',
-                                       time.time() - _ts)
+                        duration = int(time.time() - _ts)
+                        self.add(make_monitoring_log('info', 'CONFIGURATION RELOAD;%d' % duration))
+                        logger.warning('--- Configuration reloaded, %d seconds', duration)
 
                         # Make a pause to let our satellites get ready...
                         pause = max(1, self.conf.daemons_new_conf_timeout)

alignak/daemons/brokerdaemon.py

@@ -136,10 +136,16 @@ def add(self, elt):
         :return: None
         """
         if isinstance(elt, Brok):
-            # We tag the broks with our instance_id
+            # For brok, we tag the brok with our instance_id
             elt.instance_id = self.instance_id
-            with self.broks_lock:
-                self.internal_broks.append(elt)
+            if elt.type == 'monitoring_log':
+                # The brok is a monitoring event
+                with self.events_lock:
+                    self.events.append(elt)
+                statsmgr.counter('events', 1)
+            else:
+                with self.broks_lock:
+                    self.broks.append(elt)
             statsmgr.counter('broks.added', 1)
         elif isinstance(elt, ExternalCommand):
             logger.debug("Queuing an external command '%s'", str(elt.__dict__))
@@ -319,7 +325,7 @@ def setup_new_conf(self):
                     new_link = SatelliteLink.get_a_satellite_link(link_type[:-1],
                                                                   rs_conf)
                     my_satellites[new_link.uuid] = new_link
-                    logger.info("I got a new %s satellite: %s", link_type, new_link)
+                    logger.info("I got a new %s satellite: %s", link_type[:-1], new_link)
                     # print("My new %s satellite: %s" % (link_type, new_link))
 
                     new_link.running_id = running_id

alignak/daemons/receiverdaemon.py

@@ -121,10 +121,16 @@ def add(self, elt):
             elt = ExternalCommand(elt['cmd_line'], elt['creation_timestamp'])
 
         if isinstance(elt, Brok):
-            # We tag the broks with our instance_id
+            # For brok, we tag the brok with our instance_id
             elt.instance_id = self.instance_id
-            with self.broks_lock:
-                self.broks.append(elt)
+            if elt.type == 'monitoring_log':
+                # The brok is a monitoring event
+                with self.events_lock:
+                    self.events.append(elt)
+                statsmgr.counter('events', 1)
+            else:
+                with self.broks_lock:
+                    self.broks.append(elt)
             statsmgr.counter('broks.added', 1)
         elif isinstance(elt, ExternalCommand):
             logger.debug("Queuing an external command: %s", str(ExternalCommand.__dict__))

alignak/daemons/schedulerdaemon.py

@@ -113,6 +113,10 @@ def __init__(self, **kwargs):
         self.reactionners = {}
         self.receivers = {}
 
+        # This because it is the Satellite that has thes properties and I am a Satellite
+        # todo: change this?
+        # Broks are stored in each broker link, not locally
+        # self.broks = []
         self.broks_lock = threading.RLock()
 
         # Modules are only loaded one time
@@ -287,14 +291,19 @@ def get_managed_configurations(self):
         hash, push_flavor and configuration identifier as values
         :rtype: dict
         """
+        # for scheduler_link in list(self.schedulers.values()):
+        #     res[scheduler_link.instance_id] = {
+        #         'hash': scheduler_link.hash,
+        #         'push_flavor': scheduler_link.push_flavor,
+        #         'managed_conf_id': scheduler_link.managed_conf_id
+        #     }
+
         res = {}
-        if self.sched.pushed_conf and self.cur_conf:
-            res = {
-                self.cur_conf['instance_id']: {
-                    'hash': self.cur_conf['hash'],
-                    'push_flavor': self.cur_conf['push_flavor'],
-                    'managed_conf_id': self.cur_conf['managed_conf_id']
-                }
+        if self.sched.pushed_conf and self.cur_conf and 'instance_id' in self.cur_conf:
+            res[self.cur_conf['instance_id']] = {
+                'hash': self.cur_conf['hash'],
+                'push_flavor': self.cur_conf['push_flavor'],
+                'managed_conf_id': self.cur_conf['managed_conf_id']
             }
         logger.debug("Get managed configuration: %s", res)
         return res
@@ -386,7 +395,7 @@ def setup_new_conf(self):
                     new_link = SatelliteLink.get_a_satellite_link(link_type[:-1],
                                                                   rs_conf)
                     my_satellites[new_link.uuid] = new_link
-                    logger.info("I got a new %s satellite: %s", link_type, new_link)
+                    logger.info("I got a new %s satellite: %s", link_type[:-1], new_link)
                     # print("My new %s satellite: %s" % (link_type, new_link))
 
                     new_link.running_id = running_id

alignak/dispatcher.py

@@ -265,14 +265,14 @@ def check_reachable(self, forced=False, test=False):
                 self.not_configured.append(daemon_link)
 
         if self.not_configured and self.new_to_dispatch and not self.first_dispatch_done:
-            logger.info("Dispatcher, those daemons are not configured: %s, "
+            logger.info("Dispatcher, these daemons are not configured: %s, "
                         "and a configuration is ready to dispatch, run the dispatching...",
                         ','.join(d.name for d in self.not_configured))
             self.dispatch_ok = False
             self.dispatch(test=test)
 
         elif self.not_configured and self.first_dispatch_done:
-            logger.info("Dispatcher, those daemons are not configured: %s, "
+            logger.info("Dispatcher, these daemons are not configured: %s, "
                         "and a configuration has yet been dispatched dispatch, "
                         "a new dispatch is required...",
                         ','.join(d.name for d in self.not_configured))
@@ -284,15 +284,16 @@ def check_reachable(self, forced=False, test=False):
 
         return all_ok
 
-    def check_status(self):
+    def check_status_and_get_events(self):
         # pylint: disable=too-many-branches
         """Get all the daemons status
 
 
-        :return: Fictionary with all the daemons returned information
+        :return: Dictionary with all the daemons returned information
         :rtype: dict
         """
         statistics = {}
+        events = []
         for daemon_link in self.all_daemons_links:
             if daemon_link == self.arbiter_link:
                 # I exclude myself from the polling, sure I am reachable ;)
@@ -312,7 +313,15 @@ def check_status(self):
             except LinkError:
                 logger.warning("Daemon connection failed, I could not get statistics.")
 
-        return statistics
+            try:
+                got = daemon_link.get_events()
+                if got:
+                    events.extend(got)
+                    logger.debug("Daemon %s has %d events: %s", daemon_link.name, len(got), got)
+            except LinkError:
+                logger.warning("Daemon connection failed, I could not get events.")
+
+        return events
 
     def check_dispatch(self):
         """Check that all active satellites have a configuration dispatched