Package Products :: Package ZenHub :: Module PBDaemon |
[hide private]
[frames] | no frames]

# Source Code for Module Products.ZenHub.PBDaemon

      1  ##########################################################################
      2  #
      3  # This program is part of Zenoss Core, an open source monitoring platform.
      4  # Copyright (C) 2007, 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 as published by
      8  # the Free Software Foundation.
      9  #
     10  # For complete information please visit: http://www.zenoss.com/oss/
     11  #
     12  ###########################################################################
     13
     14  __doc__ = """PBDaemon
     15
     16  Base for daemons that connect to zenhub
     17
     18  """
     19
     20  import sys
     21  import time
     22  import traceback
     23
     24  import Globals
     25
     26  from Products.ZenUtils.ZenDaemon import ZenDaemon
     27  from Products.ZenEvents.ZenEventClasses import Heartbeat
     28  from Products.ZenUtils.PBUtil import ReconnectingPBClientFactory
     29  from Products.ZenUtils.DaemonStats import DaemonStats
     30  from Products.ZenUtils.Driver import drive
     31  from Products.ZenEvents.ZenEventClasses import App_Start, App_Stop, \
     32                                                  Clear, Warning
     33
     34  from twisted.cred import credentials
     35  from twisted.internet import reactor, defer
     36  from twisted.internet.error import ConnectionLost
     37  from twisted.spread import pb
     38  from twisted.python.failure import Failure
     39  import twisted.python.log
     40
     41  from ZODB.POSException import ConflictError
     42
     43 -class RemoteException(Exception, pb.Copyable, pb.RemoteCopy):
     44      "Exception that can cross the PB barrier"
     45 -    def __init__(self, msg, tb):
     46          Exception.__init__(self, msg)
     47          self.traceback = tb
     48 -    def __str__(self):
     49          return Exception.__str__(self) + self.traceback
     50
     51  pb.setUnjellyableForClass(RemoteException, RemoteException)
     52
     53  # ZODB conflicts
     54 -class RemoteConflictError(RemoteException): pass
     55  pb.setUnjellyableForClass(RemoteConflictError, RemoteConflictError)
     56
     57  # Invalid monitor specified
     58 -class RemoteBadMonitor(RemoteException): pass
     59
     60 -def translateError(callable):
     61      """
     62      Decorator function to wrap remote exceptions into something
     63      understandable by our daemon.
     64
     65      @parameter callable: function to wrap
     66      @type callable: function
     67      @return: function's return or an exception
     68      @rtype: various
     69      """
     70      def inner(*args, **kw):
     71          """
     72          Interior decorator
     73          """
     74          try:
     75              return callable(*args, **kw)
     76          except ConflictError, ex:
     77              raise RemoteConflictError(
     78                  'Remote exception: %s: %s' % (ex.__class__, ex),
     79                  traceback.format_exc())
     80          except Exception, ex:
     81              raise RemoteException(
     82                  'Remote exception: %s: %s' % (ex.__class__, ex),
     83                  traceback.format_exc())
     84      return inner
     85
     86
     87  PB_PORT = 8789
     88
     89  startEvent = {
     90      'eventClass': App_Start,
     91      'summary': 'started',
     92      'severity': Clear,
     93      }
     94
     95  stopEvent = {
     96      'eventClass':App_Stop,
     97      'summary': 'stopped',
     98      'severity': Warning,
     99      }
    100
    101
    102  DEFAULT_HUB_HOST = 'localhost'
    103  DEFAULT_HUB_PORT = PB_PORT
    104  DEFAULT_HUB_USERNAME = 'admin'
    105  DEFAULT_HUB_PASSWORD = 'zenoss'
    106  DEFAULT_HUB_MONITOR = 'localhost'
    107
    108 -class HubDown(Exception): pass
    109
    110 -class FakeRemote:
    111 -    def callRemote(self, *unused):
    112          ex = HubDown("ZenHub is down")
    113          return defer.fail(ex)
    114
    115 -class PBDaemon(ZenDaemon, pb.Referenceable):
    116
    117      name = 'pbdaemon'
    118      initialServices = ['EventService']
    119      heartbeatEvent = {'eventClass':Heartbeat}
    120      heartbeatTimeout = 60*3
    121      _customexitcode = 0
    122      _sendingEvents = False
    123
    124 -    def __init__(self, noopts=0, keeproot=False, name=None):
    125          # if we were provided our collector name via the constructor instead of
    126          # via code, be sure to store it correctly.
    127          if name is not None:
    128              self.name = name
    129              self.mname = name
    130
    131          try:
    132              ZenDaemon.__init__(self, noopts, keeproot)
    133
    134          except IOError:
    135              import traceback
    136              self.log.critical( traceback.format_exc( 0 ) )
    137              sys.exit(1)
    138
    139          self.rrdStats = DaemonStats()
    140          self.lastStats = 0
    141          self.perspective = None
    142          self.services = {}
    143          self.eventQueue = []
    144          self.startEvent = startEvent.copy()
    145          self.stopEvent = stopEvent.copy()
    146          details = dict(component=self.name, device=self.options.monitor)
    147          for evt in self.startEvent, self.stopEvent, self.heartbeatEvent:
    148              evt.update(details)
    149          self.initialConnect = defer.Deferred()
    150          self.stopped = False
    151          self._eventStatus = {}
    152
    153 -    def gotPerspective(self, perspective):
    154          """
    155          This gets called every time we reconnect.
    156
    157          @parameter perspective: Twisted perspective object
    158          @type perspective: Twisted perspective object
    159          """
    160          self.log.info("Connected to ZenHub")
    161          self.perspective = perspective
    162          d2 = self.getInitialServices()
    163          if self.initialConnect:
    164              self.log.debug('Chaining getInitialServices with d2')
    165              self.initialConnect, d = None, self.initialConnect
    166              d2.chainDeferred(d)
    167
    168
    169 -    def connect(self):
    170          factory = ReconnectingPBClientFactory()
    171          self.log.info("Connecting to %s:%d" % (self.options.hubhost,
    172              self.options.hubport))
    173          reactor.connectTCP(self.options.hubhost, self.options.hubport, factory)
    174          username = self.options.hubusername
    175          password = self.options.hubpassword
    176          self.log.debug("Logging in as %s" % username)
    177          c = credentials.UsernamePassword(username, password)
    178          factory.gotPerspective = self.gotPerspective
    179          factory.startLogin(c)
    180          def timeout(d):
    181              if not d.called:
    182                  self.connectTimeout()
    183          reactor.callLater(self.options.hubtimeout, timeout, self.initialConnect)
    184          return self.initialConnect
    185
    186 -    def connectTimeout(self):
    187          self.log.error('Timeout connecting to zenhub: is it running?')
    188          pass
    189
    190 -    def eventService(self):
    191          return self.getServiceNow('EventService')
    192
    193
    194 -    def getServiceNow(self, svcName):
    195          if not self.services.has_key(svcName):
    196              self.log.warning('No service %s named: ZenHub may be disconnected' % svcName)
    197          return self.services.get(svcName, None) or FakeRemote()
    198
    199
    200 -    def getService(self, serviceName, serviceListeningInterface=None):
    201          """
    202          Attempt to get a service from zenhub.  Returns a deferred.
    203          When service is retrieved it is stashed in self.services with
    204          serviceName as the key.  When getService is called it will first
    205          check self.services and if serviceName is already there it will return
    206          the entry from self.services wrapped in a defer.succeed
    207          """
    208          if self.services.has_key(serviceName):
    209              return defer.succeed(self.services[serviceName])
    210
    211          def removeService(ignored):
    212              self.log.debug('Removing service %s' % serviceName)
    213              if serviceName in self.services:
    214                  del self.services[serviceName]
    215
    216          def callback(result, serviceName):
    217              self.log.debug('Loaded service %s from zenhub' % serviceName)
    218              self.services[serviceName] = result
    219              result.notifyOnDisconnect(removeService)
    220              return result
    221
    222          def errback(error, serviceName):
    223              self.log.debug('errback after getting service %s' % serviceName)
    224              self.log.error('Could not retrieve service %s' % serviceName)
    225              if serviceName in self.services:
    226                  del self.services[serviceName]
    227              return error
    228
    229          d = self.perspective.callRemote('getService',
    230                                          serviceName,
    231                                          self.options.monitor,
    232                                          serviceListeningInterface or self)
    233          d.addCallback(callback, serviceName)
    234          d.addErrback(errback, serviceName)
    235          return d
    236
    237 -    def getInitialServices(self):
    238          """
    239          After connecting to zenhub, gather our initial list of services.
    240          """
    241          def errback(error):
    242              if isinstance(error, Failure):
    243                  self.log.critical( "Invalid monitor: %s" % self.options.monitor)
    244                  reactor.stop()
    245                  return defer.fail(RemoteBadMonitor(
    246                             "Invalid monitor: %s" % self.options.monitor, ''))
    247              return error
    248
    249          self.log.debug('Setting up initial services: %s' % \
    250                  ', '.join(self.initialServices))
    251          d = defer.DeferredList(
    252              [self.getService(name) for name in self.initialServices],
    253              fireOnOneErrback=True, consumeErrors=True)
    254          d.addErrback(errback)
    255          return d
    256
    257
    258 -    def connected(self):
    259          pass
    260
    261 -    def run(self):
    262          self.log.debug('Starting PBDaemon initialization')
    263          d = self.connect()
    264          def callback(result):
    265              self.sendEvent(self.startEvent)
    266              self.pushEventsLoop()
    267              self.log.debug('Calling connected.')
    268              self.connected()
    269              return result
    270          d.addCallback(callback)
    271          d.addErrback(twisted.python.log.err)
    272          reactor.run()
    273          if self._customexitcode:
    274              sys.exit(self._customexitcode)
    275
    276 -    def sigTerm(self, signum=None, frame=None):
    277          try:
    278              ZenDaemon.sigTerm(self, signum, frame)
    279          except SystemExit:
    280              pass
    281
    282 -    def setExitCode(self, exitcode):
    283          self._customexitcode = exitcode
    284
    285 -    def stop(self, ignored=''):
    286          def stopNow(ignored):
    287              if reactor.running:
    288                  reactor.stop()
    289          if reactor.running and not self.stopped:
    290              self.stopped = True
    291              if 'EventService' in self.services:
    292                  # send stop event if we don't have an implied --cycle,
    293                  # or if --cycle has been specified
    294                  if not hasattr(self.options, 'cycle') or \
    295                     getattr(self.options, 'cycle', True):
    296                      self.sendEvent(self.stopEvent)
    297                  # give the reactor some time to send the shutdown event
    298                  drive(self.pushEvents).addBoth(stopNow)
    299                  self.log.debug( "Sent a 'stop' event" )
    300              else:
    301                  self.log.debug( "No event sent as no EventService available." )
    302              # but not too much time
    303              reactor.callLater(1, stopNow, True) # requires bogus arg
    304          else:
    305              self.log.debug( "stop() called when not running" )
    306
    307 -    def sendEvents(self, events):
    308          map(self.sendEvent, events)
    309
    310 -    def sendEvent(self, event, **kw):
    311          ''' Add event to queue of events to be sent.  If we have an event
    312          service then process the queue.
    313          '''
    314          if not reactor.running: return
    315          event = event.copy()
    316          event['agent'] = self.name
    317          event['monitor'] = self.options.monitor
    318          event['manager'] = self.fqdn
    319          event.update(kw)
    320          if not self.options.allowduplicateclears:
    321              statusKey = ( event['device'],
    322                            event.get('component', None),
    323                            event.get('eventKey', None),
    324                            event.get('eventClass', None) )
    325              severity = event.get('severity', None)
    326              status = self._eventStatus.get(statusKey, None)
    327              self._eventStatus[statusKey] = severity
    328              if severity == Clear and status == Clear:
    329                  self.log.debug("Dropping useless clear event %r", event)
    330                  return
    331          self.log.debug("Queueing event %r", event)
    332          self.eventQueue.append(event)
    333          self.log.debug("Total of %d queued events" % len(self.eventQueue))
    334
    335 -    def pushEventsLoop(self):
    336          """Periodially, wake up and flush events to ZenHub.
    337          """
    338          reactor.callLater(self.options.eventflushseconds, self.pushEventsLoop)
    339          drive(self.pushEvents)
    340
    341          # Record the number of events in the queue every 5 minutes.
    342          now = time.time()
    343          if self.rrdStats.name and now >= (self.lastStats + 300):
    344              self.lastStats = now
    345              self.sendEvents(self.rrdStats.gauge('eventQueueLength',
    346                  300, len(self.eventQueue)))
    347
    348 -    def pushEvents(self, driver):
    349          """Flush events to ZenHub.
    350          """
    351          try:
    352              # Set a maximum size on the eventQueue to avoid consuming all RAM.
    353              queueLen = len(self.eventQueue)
    354              if queueLen > self.options.maxqueuelen:
    355                  self.log.warn('Queue exceeded maximum length: %d/%d. Trimming',
    356                      queueLen, self.options.maxqueuelen)
    357                  diff = queueLen - self.options.maxqueuelen
    358                  self.eventQueue = self.eventQueue[diff:]
    359
    360              # are we already shutting down?
    361              if not reactor.running:
    362                  return
    363              if self._sendingEvents:
    364                  return
    365              # try to send everything we have, serially
    366              self._sendingEvents = True
    367              while self.eventQueue:
    368                  # are still connected to ZenHub?
    369                  evtSvc = self.services.get('EventService', None)
    370                  if not evtSvc: break
    371                  # send the events in large bundles, carefully reducing
    372                  # the eventQueue in case we get in here more than once
    373                  chunkSize = self.options.eventflushchunksize
    374                  events = self.eventQueue[:chunkSize]
    375                  self.eventQueue = self.eventQueue[chunkSize:]
    376                  # send the events and wait for the response
    377                  yield evtSvc.callRemote('sendEvents', events)
    378                  try:
    379                      driver.next()
    380                  except ConnectionLost, ex:
    381                      self.log.error('Error sending event: %s' % ex)
    382                      self.eventQueue = events + self.eventQueue
    383                      break
    384              self._sendingEvents = False
    385          except Exception, ex:
    386              self._sendingEvents = False
    387              self.log.exception(ex)
    388
    389 -    def heartbeat(self):
    390          'if cycling, send a heartbeat, else, shutdown'
    391          if not self.options.cycle:
    392              self.stop()
    393              return
    394          self.sendEvent(self.heartbeatEvent, timeout=self.heartbeatTimeout)
    395          # heartbeat is normally 3x cycle time
    396          self.niceDoggie(self.heartbeatTimeout / 3)
    397
    398
    399 -    def remote_getName(self):
    400          return self.name
    401
    402
    403 -    def remote_shutdown(self, unused):
    404          self.stop()
    405          self.sigTerm()
    406
    407
    408 -    def remote_setPropertyItems(self, items):
    409          pass
    410
    411
    412      @translateError
    413 -    def remote_updateThresholdClasses(self, classes):
    414          from Products.ZenUtils.Utils import importClass
    415          self.log.debug("Loading classes %s", classes)
    416          for c in classes:
    417              try:
    418                  importClass(c)
    419              except ImportError:
    420                  self.log.error("Unable to import class %s", c)
    421
    422
    423 -    def buildOptions(self):
    424          self.parser.add_option('--hubhost',
    425                                  dest='hubhost',
    426                                  default=DEFAULT_HUB_HOST,
    427                                  help='Host of zenhub daemon.'
    428                                  ' Default is %s.' % DEFAULT_HUB_HOST)
    429          self.parser.add_option('--hubport',
    430                                  dest='hubport',
    431                                  type='int',
    432                                  default=DEFAULT_HUB_PORT,
    433                                  help='Port zenhub listens on.'
    434                                      'Default is %s.' % DEFAULT_HUB_PORT)
    435          self.parser.add_option('--hubusername',
    436                                  dest='hubusername',
    437                                  default=DEFAULT_HUB_USERNAME,
    438                                  help='Username for zenhub login.'
    439                                      ' Default is %s.' % DEFAULT_HUB_USERNAME)
    440          self.parser.add_option('--hubpassword',
    441                                  dest='hubpassword',
    442                                  default=DEFAULT_HUB_PASSWORD,
    443                                  help='Password for zenhub login.'
    444                                      ' Default is %s.' % DEFAULT_HUB_PASSWORD)
    445          self.parser.add_option('--monitor',
    446                                  dest='monitor',
    447                                  default=DEFAULT_HUB_MONITOR,
    448                                  help='Name of monitor instance to use for'
    449                                      ' configuration.  Default is %s.'
    450                                      % DEFAULT_HUB_MONITOR)
    451          self.parser.add_option('--initialHubTimeout',
    452                                 dest='hubtimeout',
    453                                 type='int',
    454                                 default=30,
    455                                 help='Initial time to wait for a ZenHub '
    456                                      'connection')
    457          self.parser.add_option('--allowduplicateclears',
    458                                 dest='allowduplicateclears',
    459                                 default=False,
    460                                 action='store_true',
    461                                 help='Send clear events even when the most '
    462                                 'recent event was also a clear event.')
    463
    464          self.parser.add_option('--eventflushseconds',
    465                                 dest='eventflushseconds',
    466                                 default=5.,
    467                                 type='float',
    468                                 help='Seconds between attempts to flush '
    469                                 'events to ZenHub.')
    470
    471          self.parser.add_option('--eventflushchunksize',
    472                                 dest='eventflushchunksize',
    473                                 default=50,
    474                                 type='int',
    475                                 help='Number of events to send to ZenHub'
    476                                 'at one time')
    477
    478          self.parser.add_option('--maxqueuelen',
    479                                 dest='maxqueuelen',
    480                                 default=5000,
    481                                 type='int',
    482                                 help='Maximum number of events to queue')
    483
    484
    485          ZenDaemon.buildOptions(self)
    486