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, ReactorNotRunning
     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                  try:
    289                      reactor.stop()
    290                  except ReactorNotRunning:
    291                      self.log.debug("Tried to stop reactor that was stopped")
    292          if reactor.running and not self.stopped:
    293              self.stopped = True
    294              if 'EventService' in self.services:
    295                  # send stop event if we don't have an implied --cycle,
    296                  # or if --cycle has been specified
    297                  if not hasattr(self.options, 'cycle') or \
    298                     getattr(self.options, 'cycle', True):
    299                      self.sendEvent(self.stopEvent)
    300                  # give the reactor some time to send the shutdown event
    301                  drive(self.pushEvents).addBoth(stopNow)
    302                  self.log.debug( "Sent a 'stop' event" )
    303              else:
    304                  self.log.debug( "No event sent as no EventService available." )
    305                  # but not too much time
    306                  reactor.callLater(1, stopNow, True) # requires bogus arg
    307          else:
    308              self.log.debug( "stop() called when not running" )
    309
    310 -    def sendEvents(self, events):
    311          map(self.sendEvent, events)
    312
    313 -    def sendEvent(self, event, **kw):
    314          ''' Add event to queue of events to be sent.  If we have an event
    315          service then process the queue.
    316          '''
    317          if not reactor.running: return
    318          event = event.copy()
    319          event['agent'] = self.name
    320          event['monitor'] = self.options.monitor
    321          event['manager'] = self.fqdn
    322          event.update(kw)
    323          if not self.options.allowduplicateclears:
    324              statusKey = ( event['device'],
    325                            event.get('component', None),
    326                            event.get('eventKey', None),
    327                            event.get('eventClass', None) )
    328              severity = event.get('severity', None)
    329              status = self._eventStatus.get(statusKey, None)
    330              self._eventStatus[statusKey] = severity
    331              if severity == Clear and status == Clear:
    332                  self.log.debug("Dropping useless clear event %r", event)
    333                  return
    334          self.log.debug("Queueing event %r", event)
    335          self.eventQueue.append(event)
    336          self.log.debug("Total of %d queued events" % len(self.eventQueue))
    337
    338 -    def pushEventsLoop(self):
    339          """Periodially, wake up and flush events to ZenHub.
    340          """
    341          reactor.callLater(self.options.eventflushseconds, self.pushEventsLoop)
    342          drive(self.pushEvents)
    343
    344          # Record the number of events in the queue every 5 minutes.
    345          now = time.time()
    346          if self.rrdStats.name and now >= (self.lastStats + 300):
    347              self.lastStats = now
    348              self.sendEvents(self.rrdStats.gauge('eventQueueLength',
    349                  300, len(self.eventQueue)))
    350
    351 -    def pushEvents(self, driver):
    352          """Flush events to ZenHub.
    353          """
    354          try:
    355              # Set a maximum size on the eventQueue to avoid consuming all RAM.
    356              queueLen = len(self.eventQueue)
    357              if queueLen > self.options.maxqueuelen:
    358                  self.log.warn('Queue exceeded maximum length: %d/%d. Trimming',
    359                      queueLen, self.options.maxqueuelen)
    360                  diff = queueLen - self.options.maxqueuelen
    361                  self.eventQueue = self.eventQueue[diff:]
    362
    363              # are we already shutting down?
    364              if not reactor.running:
    365                  return
    366              if self._sendingEvents:
    367                  return
    368              # try to send everything we have, serially
    369              self._sendingEvents = True
    370              while self.eventQueue:
    371                  # are still connected to ZenHub?
    372                  evtSvc = self.services.get('EventService', None)
    373                  if not evtSvc: break
    374                  # send the events in large bundles, carefully reducing
    375                  # the eventQueue in case we get in here more than once
    376                  chunkSize = self.options.eventflushchunksize
    377                  events = self.eventQueue[:chunkSize]
    378                  self.eventQueue = self.eventQueue[chunkSize:]
    379                  # send the events and wait for the response
    380                  yield evtSvc.callRemote('sendEvents', events)
    381                  try:
    382                      driver.next()
    383                  except ConnectionLost, ex:
    384                      self.log.error('Error sending event: %s' % ex)
    385                      self.eventQueue = events + self.eventQueue
    386                      break
    387              self._sendingEvents = False
    388          except Exception, ex:
    389              self._sendingEvents = False
    390              self.log.exception(ex)
    391
    392 -    def heartbeat(self):
    393          'if cycling, send a heartbeat, else, shutdown'
    394          if not self.options.cycle:
    395              self.stop()
    396              return
    397          self.sendEvent(self.heartbeatEvent, timeout=self.heartbeatTimeout)
    398          # heartbeat is normally 3x cycle time
    399          self.niceDoggie(self.heartbeatTimeout / 3)
    400
    401
    402 -    def remote_getName(self):
    403          return self.name
    404
    405
    406 -    def remote_shutdown(self, unused):
    407          self.stop()
    408          self.sigTerm()
    409
    410
    411 -    def remote_setPropertyItems(self, items):
    412          pass
    413
    414
    415      @translateError
    416 -    def remote_updateThresholdClasses(self, classes):
    417          from Products.ZenUtils.Utils import importClass
    418          self.log.debug("Loading classes %s", classes)
    419          for c in classes:
    420              try:
    421                  importClass(c)
    422              except ImportError:
    423                  self.log.error("Unable to import class %s", c)
    424
    425
    426 -    def buildOptions(self):
    427          self.parser.add_option('--hubhost',
    428                                  dest='hubhost',
    429                                  default=DEFAULT_HUB_HOST,
    430                                  help='Host of zenhub daemon.'
    431                                  ' Default is %s.' % DEFAULT_HUB_HOST)
    432          self.parser.add_option('--hubport',
    433                                  dest='hubport',
    434                                  type='int',
    435                                  default=DEFAULT_HUB_PORT,
    436                                  help='Port zenhub listens on.'
    437                                      'Default is %s.' % DEFAULT_HUB_PORT)
    438          self.parser.add_option('--hubusername',
    439                                  dest='hubusername',
    440                                  default=DEFAULT_HUB_USERNAME,
    441                                  help='Username for zenhub login.'
    442                                      ' Default is %s.' % DEFAULT_HUB_USERNAME)
    443          self.parser.add_option('--hubpassword',
    444                                  dest='hubpassword',
    445                                  default=DEFAULT_HUB_PASSWORD,
    446                                  help='Password for zenhub login.'
    447                                      ' Default is %s.' % DEFAULT_HUB_PASSWORD)
    448          self.parser.add_option('--monitor',
    449                                  dest='monitor',
    450                                  default=DEFAULT_HUB_MONITOR,
    451                                  help='Name of monitor instance to use for'
    452                                      ' configuration.  Default is %s.'
    453                                      % DEFAULT_HUB_MONITOR)
    454          self.parser.add_option('--initialHubTimeout',
    455                                 dest='hubtimeout',
    456                                 type='int',
    457                                 default=30,
    458                                 help='Initial time to wait for a ZenHub '
    459                                      'connection')
    460          self.parser.add_option('--allowduplicateclears',
    461                                 dest='allowduplicateclears',
    462                                 default=False,
    463                                 action='store_true',
    464                                 help='Send clear events even when the most '
    465                                 'recent event was also a clear event.')
    466
    467          self.parser.add_option('--eventflushseconds',
    468                                 dest='eventflushseconds',
    469                                 default=5.,
    470                                 type='float',
    471                                 help='Seconds between attempts to flush '
    472                                 'events to ZenHub.')
    473
    474          self.parser.add_option('--eventflushchunksize',
    475                                 dest='eventflushchunksize',
    476                                 default=50,
    477                                 type='int',
    478                                 help='Number of events to send to ZenHub'
    479                                 'at one time')
    480
    481          self.parser.add_option('--maxqueuelen',
    482                                 dest='maxqueuelen',
    483                                 default=5000,
    484                                 type='int',
    485                                 help='Maximum number of events to queue')
    486
    487
    488          ZenDaemon.buildOptions(self)
    489