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

# Source Code for Module Products.ZenEvents.zeneventd

      1  ##############################################################################
      2  #
      3  # Copyright (C) Zenoss, Inc. 2010, 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  from twisted.internet import reactor
     12  from twisted.internet import defer
     13
     14  import time
     15  from datetime import datetime, timedelta
     16
     17  import Globals
     18  from zope.component import getUtility, adapter
     19  from zope.interface import implements
     20  from zope.component.event import objectEventNotify
     21
     22  from Products.ZenCollector.utils.maintenance import MaintenanceCycle, maintenanceBuildOptions, QueueHeartbeatSender
     23  from Products.ZenMessaging.queuemessaging.interfaces import IQueueConsumerTask
     24  from Products.ZenUtils.ZCmdBase import ZCmdBase
     25  from Products.ZenUtils.guid import guid
     26  from zenoss.protocols.interfaces import IAMQPConnectionInfo, IQueueSchema
     27  from zenoss.protocols.protobufs.zep_pb2 import ZepRawEvent, Event, STATUS_DROPPED
     28  from zenoss.protocols.jsonformat import from_dict, to_dict
     29  from zenoss.protocols import hydrateQueueMessage
     30  from Products.ZenMessaging.queuemessaging.QueueConsumer import QueueConsumer
     31  from Products.ZenEvents.events2.processing import (Manager, EventPluginPipe, CheckInputPipe, IdentifierPipe,
     32      AddDeviceContextAndTagsPipe, TransformAndReidentPipe, TransformPipe, UpdateDeviceContextAndTagsPipe,
     33      AssignDefaultEventClassAndTagPipe, FingerprintPipe, SerializeContextPipe, ClearClassRefreshPipe,
     34      EventContext, DropEvent, ProcessingException)
     35  from Products.ZenEvents.interfaces import IPreEventPlugin, IPostEventPlugin
     36  from Products.ZenEvents.daemonlifecycle import DaemonCreatedEvent, SigTermEvent, SigUsr1Event
     37  from Products.ZenEvents.daemonlifecycle import DaemonStartRunEvent, BuildOptionsEvent
     38
     39  import logging
     40  log = logging.getLogger("zen.eventd")
     41
     42  EXCHANGE_ZEP_ZEN_EVENTS = '$ZepZenEvents'
     43  QUEUE_RAW_ZEN_EVENTS = '$RawZenEvents'
     44
     45 -class EventPipelineProcessor(object):
     46
     47      SYNC_EVERY_EVENT = False
     48
     49 -    def __init__(self, dmd):
     50          self.dmd = dmd
     51          self._manager = Manager(self.dmd)
     52          self._pipes = (
     53              EventPluginPipe(self._manager, IPreEventPlugin, 'PreEventPluginPipe'),
     54              CheckInputPipe(self._manager),
     55              IdentifierPipe(self._manager),
     56              AddDeviceContextAndTagsPipe(self._manager),
     57              TransformAndReidentPipe(self._manager,
     58                  TransformPipe(self._manager),
     59                  [
     60                  IdentifierPipe(self._manager),
     61                  UpdateDeviceContextAndTagsPipe(self._manager),
     62                  ]),
     63              AssignDefaultEventClassAndTagPipe(self._manager),
     64              FingerprintPipe(self._manager),
     65              SerializeContextPipe(self._manager),
     66              EventPluginPipe(self._manager, IPostEventPlugin, 'PostEventPluginPipe'),
     67              ClearClassRefreshPipe(self._manager),
     68          )
     69
     70          if not self.SYNC_EVERY_EVENT:
     71              # don't call sync() more often than 1 every 0.5 sec - helps throughput
     72              # when receiving events in bursts
     73              self.nextSync = datetime.now()
     74              self.syncInterval = timedelta(0,0,500000)
     75
     76 -    def processMessage(self, message):
     77          """
     78          Handles a queue message, can call "acknowledge" on the Queue Consumer
     79          class when it is done with the message
     80          """
     81
     82          if self.SYNC_EVERY_EVENT:
     83              doSync = True
     84          else:
     85              # sync() db if it has been longer than self.syncInterval since the last time
     86              currentTime = datetime.now()
     87              doSync = currentTime > self.nextSync
     88              self.nextSync = currentTime + self.syncInterval
     89
     90          if doSync:
     91              self.dmd._p_jar.sync()
     92
     93          try:
     94              retry = True
     95              processed = False
     96              while not processed:
     97                  try:
     98                      # extract event from message body
     99                      zepevent = ZepRawEvent()
    100                      zepevent.event.CopyFrom(message)
    101                      if log.isEnabledFor(logging.DEBUG):
    102                          log.debug("Received event: %s", to_dict(zepevent.event))
    103
    104                      eventContext = EventContext(log, zepevent)
    105
    106                      for pipe in self._pipes:
    107                          eventContext = pipe(eventContext)
    108                          if log.isEnabledFor(logging.DEBUG):
    109                              log.debug('After pipe %s, event context is %s' % ( pipe.name, to_dict(eventContext.zepRawEvent) ))
    110                          if eventContext.event.status == STATUS_DROPPED:
    111                              raise DropEvent('Dropped by %s' % pipe, eventContext.event)
    112
    113                      processed = True
    114
    115                  except AttributeError:
    116                      # _manager throws Attribute errors if connection to zope is lost - reset
    117                      # and retry ONE time
    118                      if retry:
    119                          retry=False
    120                          log.debug("Resetting connection to catalogs")
    121                          self._manager.reset()
    122                      else:
    123                          raise
    124
    125          except DropEvent:
    126              # we want these to propagate out
    127              raise
    128          except Exception as e:
    129              log.info("Failed to process event, forward original raw event: %s", to_dict(zepevent.event))
    130              # Pipes and plugins may raise ProcessingException's for their own reasons - only log unexpected
    131              # exceptions of other type (will insert stack trace in log)
    132              if not isinstance(e, ProcessingException):
    133                  log.exception(e)
    134
    135              # construct wrapper event to report this event processing failure (including content of the
    136              # original event)
    137              origzepevent = ZepRawEvent()
    138              origzepevent.event.CopyFrom(message)
    139              failReportEvent = dict(
    140                  uuid = guid.generate(),
    141                  created_time = int(time.time()*1000),
    142                  fingerprint='|'.join(['zeneventd', 'processMessage', repr(e)]),
    143                  # Don't send the *same* event class or we trash and and crash endlessly
    144                  eventClass='/',
    145                  summary='Internal exception processing event: %r' % e,
    146                  message='Internal exception processing event: %r/%s' % (e, to_dict(origzepevent.event)),
    147                  severity=4,
    148              )
    149              zepevent = ZepRawEvent()
    150              zepevent.event.CopyFrom(from_dict(Event, failReportEvent))
    151              eventContext = EventContext(log, zepevent)
    152              eventContext.eventProxy.device = 'zeneventd'
    153              eventContext.eventProxy.component = 'processMessage'
    154
    155          if log.isEnabledFor(logging.DEBUG):
    156              log.debug("Publishing event: %s", to_dict(eventContext.zepRawEvent))
    157
    158          return eventContext.zepRawEvent
    159
    160 -class BaseQueueConsumerTask(object):
    161
    162      implements(IQueueConsumerTask)
    163
    164 -    def __init__(self, processor):
    165          self.processor = processor
    166          self._queueSchema = getUtility(IQueueSchema)
    167          self.dest_routing_key_prefix = 'zenoss.zenevent'
    168          self._dest_exchange = self._queueSchema.getExchange(EXCHANGE_ZEP_ZEN_EVENTS)
    169
    170 -    def _routing_key(self, event):
    171          return (self.dest_routing_key_prefix +
    172                  event.event.event_class.replace('/', '.').lower())
    173
    174 -class TwistedQueueConsumerTask(BaseQueueConsumerTask):
    175
    176 -    def __init__(self, processor):
    177          BaseQueueConsumerTask.__init__(self, processor)
    178          self.queue = self._queueSchema.getQueue(QUEUE_RAW_ZEN_EVENTS)
    179
    180      @defer.inlineCallbacks
    181 -    def processMessage(self, message):
    182          try:
    183              hydrated = hydrateQueueMessage(message, self._queueSchema)
    184          except Exception as e:
    185              log.error("Failed to hydrate raw event: %s", e)
    186              yield self.queueConsumer.acknowledge(message)
    187          else:
    188              try:
    189                  zepRawEvent = self.processor.processMessage(hydrated)
    190                  if log.isEnabledFor(logging.DEBUG):
    191                      log.debug("Publishing event: %s", to_dict(zepRawEvent))
    192                  yield self.queueConsumer.publishMessage(EXCHANGE_ZEP_ZEN_EVENTS,
    193                      self._routing_key(zepRawEvent), zepRawEvent, declareExchange=False)
    194                  yield self.queueConsumer.acknowledge(message)
    195              except DropEvent as e:
    196                  if log.isEnabledFor(logging.DEBUG):
    197                      log.debug('%s - %s' % (e.message, to_dict(e.event)))
    198                  yield self.queueConsumer.acknowledge(message)
    199              except ProcessingException as e:
    200                  log.error('%s - %s' % (e.message, to_dict(e.event)))
    201                  log.exception(e)
    202                  yield self.queueConsumer.reject(message)
    203              except Exception as e:
    204                  log.exception(e)
    205                  yield self.queueConsumer.reject(message)
    206
    207
    208 -class EventDTwistedWorker(object):
    209 -    def __init__(self, dmd):
    210          super(EventDTwistedWorker, self).__init__()
    211          self._amqpConnectionInfo = getUtility(IAMQPConnectionInfo)
    212          self._queueSchema = getUtility(IQueueSchema)
    213          self._consumer_task = TwistedQueueConsumerTask(EventPipelineProcessor(dmd))
    214          self._consumer = QueueConsumer(self._consumer_task, dmd)
    215
    216 -    def run(self):
    217          reactor.callWhenRunning(self._start)
    218          reactor.run()
    219
    220 -    def _start(self):
    221          reactor.addSystemEventTrigger('before', 'shutdown', self._shutdown)
    222          self._consumer.run()
    223
    224      @defer.inlineCallbacks
    225 -    def _shutdown(self):
    226          if self._consumer:
    227              yield self._consumer.shutdown()
    228
    229 -class ZenEventD(ZCmdBase):
    230
    231 -    def __init__(self, *args, **kwargs):
    232          super(ZenEventD, self).__init__(*args, **kwargs)
    233          EventPipelineProcessor.SYNC_EVERY_EVENT = self.options.syncEveryEvent
    234          self._heartbeatSender = QueueHeartbeatSender('localhost',
    235                                                       'zeneventd',
    236                                                       self.options.maintenancecycle *3)
    237          self._maintenanceCycle = MaintenanceCycle(self.options.maintenancecycle,
    238                                    self._heartbeatSender)
    239          objectEventNotify(DaemonCreatedEvent(self))
    240
    241 -    def sigTerm(self, signum=None, frame=None):
    242          log.info("Shutting down...")
    243          self._maintenanceCycle.stop()
    244          objectEventNotify(SigTermEvent(self))
    245          super(ZenEventD, self).sigTerm(signum, frame)
    246
    247 -    def run(self):
    248          if self.options.daemon:
    249              self._maintenanceCycle.start()
    250          objectEventNotify(DaemonStartRunEvent(self))
    251
    252 -    def sighandler_USR1(self, signum, frame):
    253          super(ZenEventD, self).sighandler_USR1(signum, frame)
    254          log.debug('sighandler_USR1 called %s' % signum)
    255          objectEventNotify(SigUsr1Event(self, signum))
    256
    257 -    def buildOptions(self):
    258          super(ZenEventD, self).buildOptions()
    259          maintenanceBuildOptions(self.parser)
    260          self.parser.add_option('--synceveryevent', dest='syncEveryEvent',
    261                      action="store_true", default=False,
    262                      help='Force sync() before processing every event; default is to sync() no more often '
    263                      'than once every 1/2 second.')
    264          objectEventNotify(BuildOptionsEvent(self))
    265
    266
    267  @adapter(ZenEventD, DaemonStartRunEvent)
    268 -def onDaemonStartRun(daemon, event):
    269      """
    270      Start up an EventDWorker.
    271      """
    272      EventDTwistedWorker(daemon.dmd).run()
    273
    274  if __name__ == '__main__':
    275      # explicit import of ZenEventD to activate enterprise extensions
    276      from Products.ZenEvents.zeneventd import ZenEventD
    277      zed = ZenEventD()
    278      zed.run()
    279