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

# Source Code for Module Products.ZenEvents.MySqlSendEvent

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