Package Products :: Package ZenEvents :: Module MySqlSendEvent |
[hide private]
[frames] | no frames]

# Source Code for Module Products.ZenEvents.MySqlSendEvent

      1  ###########################################################################
      2  #
      3  # This program is part of Zenoss Core, an open source monitoring platform.
      4  # Copyright (C) 2008, Zenoss Inc.
      5  #
      6  # This program is free software; you can redistribute it and/or modify it
      7  # under the terms of the GNU General Public License version 2 or (at your
      8  # option) any later version as published by the Free Software Foundation.
      9  #
     10  # For complete information please visit: http://www.zenoss.com/oss/
     11  #
     12  ###########################################################################
     13
     14  __doc__ = """MySqlSendEvent
     15  Populate the events database with incoming events
     16  """
     17
     18  import logging
     19  log = logging.getLogger("zen.Events")
     20
     21  from zope.component import getUtility
     22
     23  import Products.ZenUtils.guid as guid
     24  from Event import buildEventFromDict
     25  from ZenEventClasses import Heartbeat, Unknown
     26  from Products.ZenMessaging.queuemessaging.interfaces import IEventPublisher, IQueuePublisher
     27  from zenoss.protocols.protobufs.zep_pb2 import DaemonHeartbeat
     28
     29 -class MySqlSendEventMixin:
     30      """
     31      Mix-in class that takes a MySQL db connection and builds inserts that
     32      sends the event to the backend.
     33      """
     34 -    def sendEvents(self, events):
     35          """
     36          Sends multiple events using a single publisher. This prevents
     37          using a new connection for each event.
     38          """
     39          publisher = None
     40          try:
     41              publisher = getUtility(IEventPublisher, 'batch')
     42              count = 0
     43              for event in events:
     44                  try:
     45                      self._publishEvent(event, publisher)
     46                      count += 1
     47                  except Exception, ex:
     48                      log.exception(ex)
     49              return count
     50          finally:
     51              if publisher:
     52                  publisher.close()
     53
     54 -    def sendEvent(self, event):
     55          """
     56          Send an event to the backend.
     57
     58          @param event: an event
     59          @type event: Event class
     60          @return: event id or None
     61          @rtype: string
     62          """
     63          event = self._publishEvent(event)
     64          return event.evid if event else None
     65
     66 -    def _publishEvent(self, event, publisher=None):
     67          """
     68          Sends this event to the event fan out queue
     69          """
     70          if publisher is None:
     71              publisher = getUtility(IEventPublisher)
     72          if log.isEnabledFor(logging.DEBUG):
     73              log.debug('%s%s%s' % ('=' * 15, '  incoming event  ', '=' * 15))
     74          if isinstance(event, dict):
     75              event = buildEventFromDict(event)
     76
     77          if getattr(event, 'eventClass', Unknown) == Heartbeat:
     78              log.debug("Got a %s %s heartbeat event (timeout %s sec).",
     79                        getattr(event, 'device', 'Unknown'),
     80                        getattr(event, 'component', 'Unknown'),
     81                        getattr(event, 'timeout', 'Unknown'))
     82              return self._sendHeartbeat(event)
     83
     84          event.evid = guid.generate(1)
     85          publisher.publish(event)
     86          return event
     87
     88 -    def _sendHeartbeat(self, event):
     89          """
     90          Publishes a heartbeat message to the queue.
     91
     92          @param event: event
     93          @type event: Event class
     94          """
     95          try:
     96              heartbeat = DaemonHeartbeat(monitor = event.device,
     97                                          daemon = getattr(event, "component", ""),
     98                                          timeout_seconds = int(event.timeout))
     99              publisher = getUtility(IQueuePublisher)
    100              publisher.publish('$Heartbeats', 'zenoss.heartbeat.%s' % heartbeat.monitor, heartbeat)
    101          except (ValueError, AttributeError) as e:
    102              log.error("Unable to send heartbeat: %s", event)
    103              log.exception(e)
    104