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

Source Code for Module Products.ZenHub.PBDaemon