Products.ZenRRD.zencommand

1 ########################################################################### 2 # 3 # This program is part of Zenoss Core, an open source monitoring platform. 4 # Copyright (C) 2007, 2009, 2010 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 or (at your 8 # option) any later version as published by the Free Software Foundation. 9 # 10 # For complete information please visit: http://www.zenoss.com/oss/ 11 # 12 ########################################################################### 13 14 __doc__ = """ZenCommand 15 16 Run Command plugins periodically. 17 18 """ 19 20 import random 21 import time 22 from pprint import pformat 23 import logging 24 log = logging.getLogger("zen.zencommand") 25 import traceback 26 from copy import copy 27 28 from twisted.internet import reactor, defer, error 29 from twisted.internet.protocol import ProcessProtocol 30 from twisted.python.failure import Failure 31 32 from twisted.spread import pb 33 34 import Globals 35 import zope.interface 36 37 from Products.ZenUtils.Utils import unused, getExitMessage, readable_time 38 from Products.DataCollector.SshClient import SshClient 39 from Products.ZenEvents.ZenEventClasses import Clear, Error, Cmd_Fail, Cmd_Ok 40 from Products.ZenRRD.CommandParser import ParsedResults 41 42 from Products.ZenCollector.daemon import CollectorDaemon 43 from Products.ZenCollector.interfaces import ICollectorPreferences,\ 44 IDataService,\ 45 IEventService,\ 46 IScheduledTask 47 from Products.ZenCollector.tasks import SimpleTaskFactory,\ 48 SubConfigurationTaskSplitter,\ 49 TaskStates, \ 50 BaseTask 51 from Products.ZenCollector.pools import getPool 52 from Products.ZenEvents import Event 53 from Products.ZenUtils.Executor import TwistedExecutor 54 55 from Products.DataCollector import Plugins 56 unused(Plugins) 57 58 MAX_CONNECTIONS = 250 59 MAX_BACK_OFF_MINUTES = 20 60 61 # We retrieve our configuration data remotely via a Twisted PerspectiveBroker 62 # connection. To do so, we need to import the class that will be used by the 63 # configuration service to send the data over, i.e. DeviceProxy. 64 from Products.ZenCollector.services.config import DeviceProxy 65 unused(DeviceProxy) 66 67 COLLECTOR_NAME = "zencommand" 68 POOL_NAME = 'SshConfigs' 69

70 -class SshPerformanceCollectionPreferences(object):

71 zope.interface.implements(ICollectorPreferences) 72

73 - def __init__(self):

74 """ 75 Constructs a new SshPerformanceCollectionPreferences instance and 76 provides default values for needed attributes. 77 """ 78 self.collectorName = COLLECTOR_NAME 79 self.defaultRRDCreateCommand = None 80 self.configCycleInterval = 20 # minutes 81 self.cycleInterval = 5 * 60 # seconds 82 83 # The configurationService attribute is the fully qualified class-name 84 # of our configuration service that runs within ZenHub 85 self.configurationService = 'Products.ZenHub.services.CommandPerformanceConfig' 86 87 # Provide a reasonable default for the max number of tasks 88 self.maxTasks = 50 89 90 # Will be filled in based on buildOptions 91 self.options = None

92

93 - def buildOptions(self, parser):

94 parser.add_option('--showrawresults', 95 dest='showrawresults', 96 action="store_true", 97 default=False, 98 help="Show the raw RRD values. For debugging purposes only.") 99 100 parser.add_option('--maxbackoffminutes', 101 dest='maxbackoffminutes', 102 default=MAX_BACK_OFF_MINUTES, 103 help="When a device fails to respond, increase the time to" \ 104 " check on the device until this limit.") 105 106 parser.add_option('--showfullcommand', 107 dest='showfullcommand', 108 action="store_true", 109 default=False, 110 help="Display the entire command and command-line arguments, " \ 111 " including any passwords.")

112

113 - def postStartup(self):

114 pass

115 116

117 -class SshPerCycletimeTaskSplitter(SubConfigurationTaskSplitter):

118 subconfigName = 'datasources' 119

120 - def makeConfigKey(self, config, subconfig):

121 return (config.id, subconfig.cycleTime, 'Remote' if subconfig.useSsh else 'Local')

122 123

124 -class TimeoutError(Exception):

125 """ 126 Error for a defered call taking too long to complete 127 """

128 - def __init__(self, *args):

129 Exception.__init__(self) 130 self.args = args

131 132

133 -def timeoutCommand(deferred, seconds, obj):

134 "Cause an error on a deferred when it is taking too long to complete" 135 136 def _timeout(deferred, obj): 137 "took too long... call an errback" 138 deferred.errback(Failure(TimeoutError(obj)))

139 140 def _cb(arg, timer): 141 "the command finished, possibly by timing out" 142 if not timer.called: 143 timer.cancel() 144 return arg 145 146 timer = reactor.callLater(seconds, _timeout, deferred, obj) 147 deferred.mytimer = timer 148 deferred.addBoth(_cb, timer) 149 return deferred 150 151

152 -class ProcessRunner(ProcessProtocol):

153 """ 154 Provide deferred process execution for a *single* command 155 """ 156 stopped = None 157 exitCode = None 158 output = '' 159 stderr = '' 160

161 - def start(self, cmd):

162 """ 163 Kick off the process: run it local 164 """ 165 log.debug('Running %s', cmd.command.split()[0]) 166 167 self._cmd = cmd 168 shell = '/bin/sh' 169 self.cmdline = (shell, '-c', 'exec %s' % cmd.command) 170 self.command = ' '.join(self.cmdline) 171 172 reactor.spawnProcess(self, shell, self.cmdline, env=cmd.env) 173 174 d = timeoutCommand(defer.Deferred(), cmd.deviceConfig.zCommandCommandTimeout, cmd) 175 self.stopped = d 176 self.stopped.addErrback(self.timeout) 177 return d

178

179 - def timeout(self, value):

180 """ 181 Kill a process if it takes too long 182 """ 183 try: 184 self.transport.signalProcess('KILL') 185 except error.ProcessExitedAlready: 186 log.debug("Command already exited: %s", self.command.split()[0]) 187 return value

188

189 - def outReceived(self, data):

190 """ 191 Store up the output as it arrives from the process 192 """ 193 self.output += data

194

195 - def errReceived(self, data):

196 """ 197 Store up the output as it arrives from the process 198 """ 199 self.stderr += data

200

201 - def processEnded(self, reason):

202 """ 203 Notify the starter that their process is complete 204 """ 205 self.exitCode = reason.value.exitCode 206 if self.exitCode is not None: 207 msg = """Datasource: %s Received exit code: %s Output:\n%r""" 208 data = [self._cmd.ds, self.exitCode, self.output] 209 if self.stderr: 210 msg += "\nStandard Error:\n%r" 211 data.append(self.stderr) 212 log.debug(msg, *data) 213 214 if self.stopped: 215 d, self.stopped = self.stopped, None 216 if not d.called: 217 d.callback(self)

218 219

220 -class MySshClient(SshClient):

221 """ 222 Connection to SSH server at the remote device 223 """

224 - def __init__(self, *args, **kw):

225 SshClient.__init__(self, *args, **kw) 226 self.defers = {} 227 self._taskList = set()

228

229 - def addCommand(self, command):

230 """ 231 Run a command against the server 232 """ 233 d = defer.Deferred() 234 self.defers[command] = d 235 SshClient.addCommand(self, command) 236 return d

237

238 - def addResult(self, command, data, code):

239 """ 240 Forward the results of the command execution to the starter 241 """ 242 # don't call the CollectorClient.addResult which adds the result to a 243 # member variable for zenmodeler 244 d = self.defers.pop(command, None) 245 if d is None: 246 log.error("Internal error where deferred object not in dictionary." \ 247 " Command = '%s' Data = '%s' Code = '%s'", 248 command.split()[0], data, code) 249 elif not d.called: 250 d.callback((data, code))

251

252 - def clientConnectionLost(self, connector, reason):

253 connection_description = '%s:*****@%s:%s' % (self.username, self.ip, self.port) 254 # Connection was lost, but could be because we just closed it. Not necessarily cause for concern. 255 log.debug("Connection %s lost." % connection_description) 256 pool = getPool('SSH Connections') 257 poolkey = hash((self.username, self.password, self.ip, self.port)) 258 if poolkey in pool: 259 # Clean it up so the next time around the task will get a new connection 260 log.debug("Deleting connection %s from pool." % connection_description) 261 del pool[poolkey]

262

263 - def check(self, ip, timeout=2):

264 """ 265 Turn off blocking SshClient.test method 266 """ 267 return True

268

269 - def clientFinished(self):

270 """ 271 We don't need to track commands/results when they complete 272 """ 273 SshClient.clientFinished(self) 274 self.cmdmap = {} 275 self._commands = [] 276 self.results = []

277

278 - def clientConnectionFailed(self, connector, reason):

279 """ 280 If we didn't connect let the modeler know 281 282 @param connector: connector associated with this failure 283 @type connector: object 284 @param reason: failure object 285 @type reason: object 286 """ 287 self.clientFinished() 288 message= reason.getErrorMessage() 289 for task in list(self._taskList): 290 task.connectionFailed(message)

291 292

293 -class SshOptions:

294 loginTries=1 295 searchPath='' 296 existenceTest=None 297

298 - def __init__(self, username, password, loginTimeout, commandTimeout, 299 keyPath, concurrentSessions):

300 self.username = username 301 self.password = password 302 self.loginTimeout=loginTimeout 303 self.commandTimeout=commandTimeout 304 self.keyPath = keyPath 305 self.concurrentSessions = concurrentSessions

306 307

308 -class SshRunner(object):

309 """ 310 Run a single command across a cached SSH connection 311 """ 312

313 - def __init__(self, connection):

314 self._connection = connection 315 self.exitCode = None 316 self.output = None 317 self.stderr = None

318

319 - def start(self, cmd):

320 "Initiate a command on the remote device" 321 self.defer = defer.Deferred(canceller=self._canceller) 322 try: 323 d = timeoutCommand(self._connection.addCommand(cmd.command), 324 self._connection.commandTimeout, 325 cmd) 326 except Exception, ex: 327 log.warning('Error starting command: %s', ex) 328 return defer.fail(ex) 329 d.addErrback(self.timeout) 330 d.addBoth(self.processEnded) 331 return d

332

333 - def _canceller(self, deferToCancel):

334 if not deferToCancel.mytimer.called: 335 deferToCancel.mytimer.cancel() 336 return None

337

338 - def timeout(self, arg):

339 "Deal with slow executing command/connection (close it)" 340 # We could send a kill signal, but then we would need to track 341 # the command channel to send it. Just close the connection. 342 return arg

343

344 - def processEnded(self, value):

345 "Deliver ourselves to the starter with the proper attributes" 346 if isinstance(value, Failure): 347 return value 348 self.output, self.exitCode = value 349 return self

350 351

352 -class DataPointConfig(pb.Copyable, pb.RemoteCopy):

353 id = '' 354 component = '' 355 rrdPath = '' 356 rrdType = None 357 rrdCreateCommand = '' 358 rrdMin = None 359 rrdMax = None 360

361 - def __init__(self):

362 self.data = {}

363

364 - def __repr__(self):

365 return pformat((self.data, self.id))

366 367 pb.setUnjellyableForClass(DataPointConfig, DataPointConfig) 368

369 -class Cmd(pb.Copyable, pb.RemoteCopy):

370 """ 371 Holds the config of every command to be run 372 """ 373 device = '' 374 command = None 375 ds = '' 376 useSsh = False 377 cycleTime = None 378 eventClass = None 379 eventKey = None 380 severity = 3 381 lastStart = 0 382 lastStop = 0 383 result = None 384 env = None 385

386 - def __init__(self):

387 self.points = []

388

389 - def processCompleted(self, pr):

390 """ 391 Return back the datasource with the ProcessRunner/SshRunner stored in 392 the the 'result' attribute. 393 """ 394 self.result = pr 395 self.lastStop = time.time() 396 397 # Check for a condition that could cause zencommand to stop cycling. 398 # http://dev.zenoss.org/trac/ticket/4936 399 if self.lastStop < self.lastStart: 400 log.debug('System clock went back?') 401 self.lastStop = self.lastStart 402 403 if isinstance(pr, Failure): 404 return pr 405 406 log.debug('Process %s stopped (%s), %.2f seconds elapsed', 407 self.name, 408 pr.exitCode, 409 self.lastStop - self.lastStart) 410 return self

411

412 - def getEventKey(self, point):

413 # fetch datapoint name from filename path and add it to the event key 414 return self.eventKey + '|' + point.rrdPath.split('/')[-1]

415

416 - def commandKey(self):

417 "Provide a value that establishes the uniqueness of this command" 418 return '%'.join(map(str, [self.useSsh, self.cycleTime, 419 self.severity, self.command]))

420 - def __str__(self):

421 return ' '.join(map(str, [ 422 self.ds, 423 'useSSH=%s' % self.useSsh, 424 self.cycleTime, 425 ]))

426 427 pb.setUnjellyableForClass(Cmd, Cmd) 428 429 430 STATUS_EVENT = { 'eventClass' : Cmd_Fail, 431 'component' : 'command', 432 } 433

434 -class SshPerformanceCollectionTask(BaseTask):

435 """ 436 A task that performs periodic performance collection for devices providing 437 data via SSH connections. 438 """ 439 zope.interface.implements(IScheduledTask) 440 441 STATE_CONNECTING = 'CONNECTING' 442 STATE_FETCH_DATA = 'FETCH_DATA' 443 STATE_PARSE_DATA = 'PARSING_DATA' 444 STATE_STORE_PERF = 'STORE_PERF_DATA' 445

446 - def __init__(self, 447 taskName, 448 configId, 449 scheduleIntervalSeconds, 450 taskConfig):

451 """ 452 @param taskName: the unique identifier for this task 453 @type taskName: string 454 @param configId: configuration to watch 455 @type configId: string 456 @param scheduleIntervalSeconds: the interval at which this task will be 457 collected 458 @type scheduleIntervalSeconds: int 459 @param taskConfig: the configuration for this task 460 """ 461 super(SshPerformanceCollectionTask, self).__init__( 462 taskName, configId, 463 scheduleIntervalSeconds, taskConfig 464 ) 465 466 # Needed for interface 467 self.name = taskName 468 self.configId = configId 469 self.state = TaskStates.STATE_IDLE 470 self.interval = scheduleIntervalSeconds 471 472 # The taskConfig corresponds to a DeviceProxy 473 self._device = taskConfig 474 475 self._devId = self._device.id 476 self._manageIp = self._device.manageIp 477 478 self._dataService = zope.component.queryUtility(IDataService) 479 self._eventService = zope.component.queryUtility(IEventService) 480 481 self._preferences = zope.component.queryUtility(ICollectorPreferences, 482 COLLECTOR_NAME) 483 self._lastErrorMsg = '' 484 485 self._maxbackoffseconds = self._preferences.options.maxbackoffminutes * 60 486 487 self._concurrentSessions = taskConfig.zSshConcurrentSessions 488 self._executor = TwistedExecutor(self._concurrentSessions) 489 self._useSsh = taskConfig.datasources[0].useSsh 490 self._connection = None 491 492 self._datasources = taskConfig.datasources 493 self.pool = getPool('SSH Connections') 494 self.executed = 0

495

496 - def __str__(self):

497 return "COMMAND schedule Name: %s configId: %s Datasources: %d" % ( 498 self.name, self.configId, len(self._datasources))

499

500 - def cleanup(self):

501 self._cleanUpPool() 502 self._close()

503

504 - def _getPoolKey(self):

505 """ 506 Get the key under which the client should be stored in the pool. 507 """ 508 username = self._device.zCommandUsername 509 password = self._device.zCommandPassword 510 ip = self._manageIp 511 port = self._device.zCommandPort 512 return hash((username, password, ip, port))

513

514 - def _cleanUpPool(self):

515 """ 516 Close the connection currently associated with this task. 517 """ 518 poolkey = self._getPoolKey() 519 if poolkey in self.pool: 520 client = self.pool[poolkey] 521 tasklist = client._taskList 522 if not tasklist: 523 # No other tasks, so safe to clean up 524 transport = client.transport 525 if transport: 526 transport.loseConnection() 527 del self.pool[poolkey]

528

529 - def doTask(self):

530 """ 531 Contact to one device and return a deferred which gathers data from 532 the device. 533 534 @return: Deferred actions to run against a device configuration 535 @rtype: Twisted deferred object 536 """ 537 # See if we need to connect first before doing any collection 538 d = defer.maybeDeferred(self._connect) 539 d.addCallbacks(self._connectCallback, self._failure) 540 d.addCallback(self._fetchPerf) 541 542 # Call _finished for both success and error scenarios 543 d.addBoth(self._finished) 544 545 # Wait until the Deferred actually completes 546 return d

547

548 - def _connect(self):

549 """ 550 If a local datasource executor, do nothing. 551 552 If an SSH datasource executor, create a connection to object the remote device. 553 Make a new SSH connection object if there isn't one available. This doesn't 554 actually connect to the device. 555 """ 556 if not self._useSsh: 557 return defer.succeed(None) 558 559 connection = self.pool.get(self._getPoolKey(), None) 560 if connection is None: 561 self.state = SshPerformanceCollectionTask.STATE_CONNECTING 562 log.debug("Creating connection object to %s", self._devId) 563 username = self._device.zCommandUsername 564 password = self._device.zCommandPassword 565 loginTimeout = self._device.zCommandLoginTimeout 566 commandTimeout = self._device.zCommandCommandTimeout 567 keypath = self._device.zKeyPath 568 options = SshOptions(username, password, 569 loginTimeout, commandTimeout, 570 keypath, self._concurrentSessions) 571 572 connection = MySshClient(self._devId, self._manageIp, 573 self._device.zCommandPort, options=options) 574 connection.sendEvent = self._eventService.sendEvent 575 576 self.pool[self._getPoolKey()] = connection 577 578 # Opens SSH connection to device 579 connection.run() 580 581 self._connection = connection 582 self._connection._taskList.add(self) 583 return connection

584

585 - def _close(self):

586 """ 587 If a local datasource executor, do nothing. 588 589 If an SSH datasource executor, relinquish a connection to the remote device. 590 """ 591 if self._connection: 592 self._connection._taskList.discard(self) 593 if not self._connection._taskList: 594 if self._getPoolKey() in self.pool: 595 client = self.pool[self._getPoolKey()] 596 client.clientFinished() 597 client.channelClosed() 598 self._connection = None

599

600 - def connectionFailed(self, msg):

601 """ 602 This method is called by the SSH client when the connection fails. 603 604 @parameter msg: message indicating the cause of the problem 605 @type msg: string 606 """ 607 # Note: Raising an exception and then catching it doesn't work 608 # as it appears that the exception is discarded in PBDaemon.py 609 self.state = TaskStates.STATE_PAUSED 610 log.error("Pausing task %s as %s [%s] connection failure: %s", 611 self.name, self._devId, self._manageIp, msg) 612 self._eventService.sendEvent(STATUS_EVENT, 613 device=self._devId, 614 summary=msg, 615 component=COLLECTOR_NAME, 616 severity=Event.Error) 617 self._commandsToExecute.cancel()

618

619 - def _failure(self, reason):

620 """ 621 Twisted errBack to log the exception for a single device. 622 623 @parameter reason: explanation of the failure 624 @type reason: Twisted error instance 625 """ 626 # Decode the exception 627 if isinstance(reason.value, TimeoutError): 628 cmd, = reason.value.args 629 msg = "Command timed out on device %s: %r" % ( 630 self._devId, cmd.command.split()[0]) 631 log.warning(msg) 632 ev = self._makeCmdEvent(cmd, cmd.severity, msg) 633 self._eventService.sendEvent(ev) 634 635 # Don't log a traceback by not returning a result 636 reason = None 637 638 elif isinstance(reason.value, defer.CancelledError): 639 # The actual issue is logged by connectionFailed 640 # Don't log a traceback by not returning a result 641 msg = "Task %s paused due to connection error" % self.name 642 reason = None 643 644 else: 645 msg = reason.getErrorMessage() 646 if not msg: # Sometimes we get blank error messages 647 msg = reason.__class__ 648 msg = '%s %s' % (self._devId, msg) 649 # Leave 'reason' alone to generate a traceback 650 651 if self._lastErrorMsg != msg: 652 self._lastErrorMsg = msg 653 if msg: 654 log.error(msg) 655 656 if reason: 657 self._eventService.sendEvent(STATUS_EVENT, 658 device=self._devId, 659 summary=msg, 660 severity=Event.Error) 661 662 if self._useSsh: 663 self._delayNextCheck() 664 665 return reason

666

667 - def _connectCallback(self, result):

668 """ 669 Callback called after a successful connect to the remote device. 670 """ 671 if self._useSsh: 672 log.debug("Connected to %s [%s]", self._devId, self._manageIp) 673 else: 674 log.debug("Running command(s) locally") 675 return result

676

677 - def _addDatasource(self, datasource):

678 """ 679 Add a new instantiation of ProcessRunner or SshRunner 680 for every datasource. 681 """ 682 if self._preferences.options.showfullcommand: 683 log.info("Datasource %s command: %s", datasource.name, 684 datasource.command) 685 686 if self._useSsh: 687 runner = SshRunner(self._connection) 688 else: 689 runner = ProcessRunner() 690 691 d = runner.start(datasource) 692 datasource.lastStart = time.time() 693 d.addBoth(datasource.processCompleted) 694 return d

695

696 - def _fetchPerf(self, ignored):

697 """ 698 Get performance data for all the monitored components on a device 699 700 @parameter ignored: required to keep Twisted's callback chain happy 701 @type ignored: result of previous callback 702 """ 703 self.state = SshPerformanceCollectionTask.STATE_FETCH_DATA 704 705 # The keys are the datasource commands, which are by definition unique 706 # to the command run. 707 cacheableDS = {} 708 709 # Bundle up the list of tasks 710 deferredCmds = [] 711 for datasource in self._datasources: 712 datasource.deviceConfig = self._device 713 714 if datasource.command in cacheableDS: 715 cacheableDS[datasource.command].append(datasource) 716 continue 717 cacheableDS[datasource.command] = [] 718 719 task = self._executor.submit(self._addDatasource, datasource) 720 deferredCmds.append(task) 721 722 # Run the tasks 723 dl = defer.DeferredList(deferredCmds, consumeErrors=True) 724 dl.addCallback(self._parseResults, cacheableDS) 725 dl.addCallback(self._storeResults) 726 dl.addCallback(self._updateStatus) 727 728 # Save the list in case we need to cancel the commands 729 self._commandsToExecute = dl 730 return dl

731

732 - def _parseResults(self, resultList, cacheableDS):

733 """ 734 Interpret the results retrieved from the commands and pass on 735 the datapoint values and events. 736 737 @parameter resultList: results of running the commands in a DeferredList 738 @type resultList: array of (boolean, datasource) 739 @parameter cacheableDS: other datasources that can use the same results 740 @type cacheableDS: dictionary of arrays of datasources 741 """ 742 self.state = SshPerformanceCollectionTask.STATE_PARSE_DATA 743 parseableResults = [] 744 for success, datasource in resultList: 745 results = ParsedResults() 746 if not success: 747 # In this case, our datasource is actually a defer.Failure 748 reason = datasource 749 datasource, = reason.value.args 750 msg = "Datasource %s command timed out" % ( 751 datasource.name) 752 ev = self._makeCmdEvent(datasource, msg) 753 results.events.append(ev) 754 755 else: 756 # Re-use our results for any similar datasources 757 cachedDsList = cacheableDS.get(datasource.command) 758 if cachedDsList: 759 for ds in cachedDsList: 760 ds.result = copy(datasource.result) 761 results = ParsedResults() 762 self._processDatasourceResults(ds, results) 763 parseableResults.append( (ds, results) ) 764 results = ParsedResults() 765 766 self._processDatasourceResults(datasource, results) 767 768 parseableResults.append( (datasource, results) ) 769 return parseableResults

770

771 - def _makeParser(self, datasource, eventList):

772 """ 773 Create a parser object to process data 774 775 @parameter datasource: datasource containg information 776 @type datasource: Cmd object 777 @parameter eventList: list of events 778 @type eventList: list of dictionaries 779 """ 780 parser = None 781 try: 782 parser = datasource.parser.create() 783 except Exception, ex: 784 msg = "Error loading parser %s" % datasource.parser 785 log.exception("%s %s %s", self.name, datasource.name, msg) 786 ev = self._makeCmdEvent(datasource, msg) 787 ev['message'] = traceback.format_exc() 788 eventList.append(ev) 789 return parser

790

791 - def _processDatasourceResults(self, datasource, results):

792 """ 793 Process a single datasource's results 794 795 @parameter datasource: datasource containg information 796 @type datasource: Cmd object 797 @parameter results: empty results object 798 @type results: ParsedResults object 799 """ 800 showcommand = self._preferences.options.showfullcommand 801 if not datasource.result.output: 802 msg = "No data returned for command" 803 if showcommand: 804 msg += ": %s" % datasource.command 805 log.warn("%s %s %s", self.name, datasource.name, msg) 806 ev = self._makeCmdEvent(datasource, msg) 807 if showcommand: 808 ev['command'] = datasource.command 809 results.events.append(ev) 810 return 811 812 parser = self._makeParser(datasource, results.events) 813 if not parser: 814 return 815 816 try: 817 parser.preprocessResults(datasource, log) 818 parser.processResults(datasource, results) 819 if not results.events and parser.createDefaultEventUsingExitCode: 820 # Add a failsafe event guessing at the error codes 821 self._addDefaultEvent(datasource, results) 822 if datasource.result.stderr: 823 self._addStderrMsg(datasource.result.stderr, 824 results.events) 825 except Exception, ex: 826 msg = "Error running parser %s" % datasource.parser 827 log.exception("%s %s %s", self.name, datasource.name, msg) 828 ev = self._makeCmdEvent(datasource, msg) 829 ev['message'] = traceback.format_exc() 830 ev['output'] = datasource.result.output 831 results.events.append(ev)

832

833 - def _addDefaultEvent(self, datasource, results):

834 """ 835 If there is no event, send one based on the exit code. 836 """ 837 exitCode = datasource.result.exitCode 838 if exitCode == 0: 839 msg = '' 840 severity = 0 841 else: 842 msg = 'Datasource: %s - Code: %s - Msg: %s' % ( 843 datasource.name, exitCode, getExitMessage(exitCode)) 844 severity = datasource.severity 845 846 ev = self._makeCmdEvent(datasource, msg, severity) 847 results.events.append(ev)

848

849 - def _addStderrMsg(self, stderrMsg, eventList):

850 """ 851 Add the stderr output to error events. 852 853 @parameter stderrMsg: stderr output from the command 854 @type stderrMsg: string 855 @parameter eventList: list of events 856 @type eventList: list of dictionaries 857 """ 858 for event in eventList: 859 if event['severity'] not in ('Clear', 'Info', 'Debug'): 860 event['stderr'] = stderrMsg

861

862 - def _storeResults(self, resultList):

863 """ 864 Store the values in RRD files 865 866 @parameter resultList: results of running the commands 867 @type resultList: array of (datasource, dictionary) 868 """ 869 self.state = SshPerformanceCollectionTask.STATE_STORE_PERF 870 for datasource, results in resultList: 871 for dp, value in results.values: 872 threshData = { 873 'eventKey': datasource.getEventKey(dp), 874 'component': dp.component, 875 } 876 self._dataService.writeRRD( 877 dp.rrdPath, 878 value, 879 dp.rrdType, 880 dp.rrdCreateCommand, 881 datasource.cycleTime, 882 dp.rrdMin, 883 dp.rrdMax, 884 threshData) 885 886 return resultList

887

888 - def _updateStatus(self, resultList):

889 """ 890 Send any accumulated events 891 892 @parameter resultList: results of running the commands 893 @type resultList: array of (datasource, dictionary) 894 """ 895 for datasource, results in resultList: 896 self._clearEvent(datasource, results.events) 897 for ev in results.events: 898 self._eventService.sendEvent(ev, device=self._devId) 899 return resultList

900

901 - def _clearEvent(self, datasource, eventList):

902 """ 903 Ensure that a CLEAR event is sent for any command that 904 successfully completes. 905 """ 906 # If the result is a Failure, no exitCode exists 907 exitCode = getattr(datasource.result, 'exitCode', -1) 908 if exitCode is None or exitCode != 0: 909 return 910 911 clearEvents = [ev for ev in eventList if ev['severity'] == Clear] 912 if not clearEvents: 913 msg = 'Datasource %s command completed successfully' % ( 914 datasource.name) 915 ev = self._makeCmdEvent(datasource, msg, severity=Clear) 916 eventList.append(ev)

917

918 - def _makeCmdEvent(self, datasource, msg, severity=None):

919 """ 920 Create an event using the info in the Cmd object. 921 """ 922 severity = datasource.severity if severity is None else severity 923 ev = dict( 924 device=self._devId, 925 component=datasource.component, 926 eventClass=datasource.eventClass, 927 eventKey=datasource.eventKey, 928 severity=severity, 929 summary=msg 930 ) 931 return ev

932

933 - def _finished(self, result):

934 """ 935 Callback activated when the task is complete 936 937 @parameter result: results of the task 938 @type result: deferred object 939 """ 940 if not isinstance(result, Failure): 941 self._returnToNormalSchedule() 942 943 try: 944 self._close() 945 except Exception, ex: 946 log.warn("Failed to close device %s: error %s" % 947 (self._devId, str(ex))) 948 949 # Return the result so the framework can track success/failure 950 return result

951

952 - def displayStatistics(self):

953 """ 954 Called by the collector framework scheduler, and allows us to 955 see how each task is doing. 956 """ 957 display = "%s useSSH: %s\n" % ( 958 self.name, self._useSsh) 959 if self._lastErrorMsg: 960 display += "%s\n" % self._lastErrorMsg 961 return display

962 963 964 if __name__ == '__main__': 965 # Required for passing classes from zenhub to here 966 from Products.ZenRRD.zencommand import Cmd, DataPointConfig 967 968 myPreferences = SshPerformanceCollectionPreferences() 969 myTaskFactory = SimpleTaskFactory(SshPerformanceCollectionTask) 970 myTaskSplitter = SshPerCycletimeTaskSplitter(myTaskFactory) 971 daemon = CollectorDaemon(myPreferences, myTaskSplitter) 972 daemon.run() 973

Source Code for Module Products.ZenRRD.zencommand