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

Source Code for Module Products.ZenHub.PBDaemon