Products.ZenRRD.zencommand

1 ############################################################################## 2 # 3 # Copyright (C) Zenoss, Inc. 2007, 2009, 2010, all rights reserved. 4 # 5 # This content is made available according to terms specified in 6 # License.zenoss under the directory where your Zenoss product is installed. 7 # 8 ############################################################################## 9 10 11 __doc__ = """ZenCommand 12 13 Run Command plugins periodically. 14 15 """ 16 17 import time 18 from pprint import pformat 19 import logging 20 log = logging.getLogger("zen.zencommand") 21 import traceback 22 from copy import copy 23 24 from twisted.internet import reactor, defer, error 25 from twisted.internet.protocol import ProcessProtocol 26 from twisted.python.failure import Failure 27 28 from twisted.spread import pb 29 30 import Globals 31 import zope.interface 32 33 from Products.ZenUtils.Utils import unused, getExitMessage 34 from Products.DataCollector.SshClient import SshClient 35 from Products.ZenEvents.ZenEventClasses import Clear, Cmd_Fail 36 from Products.ZenRRD.CommandParser import ParsedResults 37 38 from Products.ZenCollector.daemon import CollectorDaemon 39 from Products.ZenCollector.interfaces import ICollectorPreferences,\ 40 IDataService,\ 41 IEventService,\ 42 IScheduledTask 43 from Products.ZenCollector.tasks import SimpleTaskFactory,\ 44 SubConfigurationTaskSplitter,\ 45 TaskStates, \ 46 BaseTask 47 from Products.ZenCollector.pools import getPool 48 from Products.ZenEvents import Event 49 from Products.ZenUtils.Executor import TwistedExecutor 50 51 from Products.DataCollector import Plugins 52 unused(Plugins) 53 54 MAX_CONNECTIONS = 250 55 MAX_BACK_OFF_MINUTES = 20 56 57 # We retrieve our configuration data remotely via a Twisted PerspectiveBroker 58 # connection. To do so, we need to import the class that will be used by the 59 # configuration service to send the data over, i.e. DeviceProxy. 60 from Products.ZenCollector.services.config import DeviceProxy 61 unused(DeviceProxy) 62 63 COLLECTOR_NAME = "zencommand" 64 POOL_NAME = 'SshConfigs' 65

66 -class SshPerformanceCollectionPreferences(object):

67 zope.interface.implements(ICollectorPreferences) 68

69 - def __init__(self):

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

88

89 - def buildOptions(self, parser):

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

109

110 - def postStartup(self):

111 pass

112 113

114 -class SshPerCycletimeTaskSplitter(SubConfigurationTaskSplitter):

115 subconfigName = 'datasources' 116

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

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

119 120

121 -class TimeoutError(Exception):

122 """ 123 Error for a defered call taking too long to complete 124 """

125 - def __init__(self, *args):

126 Exception.__init__(self) 127 self.args = args

128 129

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

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

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

149 -class ProcessRunner(ProcessProtocol):

150 """ 151 Provide deferred process execution for a *single* command 152 """ 153 stopped = None 154 exitCode = None 155 output = '' 156 stderr = '' 157

158 - def start(self, cmd):

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

175

176 - def timeout(self, value):

177 """ 178 Kill a process gracefully if it takes too long 179 """ 180 try: 181 self.transport.signalProcess('INT') 182 reactor.callLater(2, self._reap) 183 except error.ProcessExitedAlready: 184 log.debug("Command already exited: %s", self.command.split()[0]) 185 return value

186

187 - def _reap(self):

188 """ 189 Kill a process forcefully if it takes too long 190 """ 191 try: 192 self.transport.signalProcess('KILL') 193 except Exception: 194 pass

195

196 - def outReceived(self, data):

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

201

202 - def errReceived(self, data):

203 """ 204 Store up the output as it arrives from the process 205 """ 206 self.stderr += data

207

208 - def processEnded(self, reason):

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

225 226

227 -class MySshClient(SshClient):

228 """ 229 Connection to SSH server at the remote device 230 """

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

232 SshClient.__init__(self, *args, **kw) 233 self.defers = {} 234 self._taskList = set()

235

236 - def addCommand(self, command):

237 """ 238 Run a command against the server 239 """ 240 d = defer.Deferred() 241 self.defers[command] = d 242 SshClient.addCommand(self, command) 243 return d

244

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

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

258

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

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

269

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

271 """ 272 Turn off blocking SshClient.test method 273 """ 274 return True

275

276 - def clientFinished(self):

277 """ 278 We don't need to track commands/results when they complete 279 """ 280 SshClient.clientFinished(self) 281 self.cmdmap = {} 282 self._commands = [] 283 self.results = []

284

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

286 """ 287 If we didn't connect let the modeler know 288 289 @param connector: connector associated with this failure 290 @type connector: object 291 @param reason: failure object 292 @type reason: object 293 """ 294 self.clientFinished() 295 message= reason.getErrorMessage() 296 for task in list(self._taskList): 297 task.connectionFailed(message)

298 299

300 -class SshOptions:

301 loginTries=1 302 searchPath='' 303 existenceTest=None 304

305 - def __init__(self, username, password, loginTimeout, commandTimeout, 306 keyPath, concurrentSessions):

307 self.username = username 308 self.password = password 309 self.loginTimeout=loginTimeout 310 self.commandTimeout=commandTimeout 311 self.keyPath = keyPath 312 self.concurrentSessions = concurrentSessions

313 314

315 -class SshRunner(object):

316 """ 317 Run a single command across a cached SSH connection 318 """ 319

320 - def __init__(self, connection):

321 self._connection = connection 322 self.exitCode = None 323 self.output = None 324 self.stderr = None

325

326 - def start(self, cmd):

327 "Initiate a command on the remote device" 328 self.defer = defer.Deferred(canceller=self._canceller) 329 try: 330 d = timeoutCommand(self._connection.addCommand(cmd.command), 331 self._connection.commandTimeout, 332 cmd) 333 except Exception, ex: 334 log.warning('Error starting command: %s', ex) 335 return defer.fail(ex) 336 d.addErrback(self.timeout) 337 d.addBoth(self.processEnded) 338 return d

339

340 - def _canceller(self, deferToCancel):

341 if not deferToCancel.mytimer.called: 342 deferToCancel.mytimer.cancel() 343 return None

344

345 - def timeout(self, arg):

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

350

351 - def processEnded(self, value):

352 "Deliver ourselves to the starter with the proper attributes" 353 if isinstance(value, Failure): 354 return value 355 self.output, self.exitCode = value 356 return self

357 358

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

360 id = '' 361 component = '' 362 rrdPath = '' 363 rrdType = None 364 rrdCreateCommand = '' 365 rrdMin = None 366 rrdMax = None 367

368 - def __init__(self):

369 self.data = {}

370

371 - def __repr__(self):

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

373 374 pb.setUnjellyableForClass(DataPointConfig, DataPointConfig) 375

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

377 """ 378 Holds the config of every command to be run 379 """ 380 device = '' 381 command = None 382 ds = '' 383 useSsh = False 384 cycleTime = None 385 eventClass = None 386 eventKey = None 387 severity = 3 388 lastStart = 0 389 lastStop = 0 390 result = None 391 env = None 392

393 - def __init__(self):

394 self.points = []

395

396 - def processCompleted(self, pr):

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

418

419 - def getEventKey(self, point):

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

422

423 - def commandKey(self):

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

427 - def __str__(self):

428 return ' '.join(map(str, [ 429 self.ds, 430 'useSSH=%s' % self.useSsh, 431 self.cycleTime, 432 ]))

433 434 pb.setUnjellyableForClass(Cmd, Cmd) 435 436 437 STATUS_EVENT = { 'eventClass' : Cmd_Fail, 438 'component' : 'command', 439 } 440

441 -class SshPerformanceCollectionTask(BaseTask):

442 """ 443 A task that performs periodic performance collection for devices providing 444 data via SSH connections. 445 """ 446 zope.interface.implements(IScheduledTask) 447 448 STATE_CONNECTING = 'CONNECTING' 449 STATE_FETCH_DATA = 'FETCH_DATA' 450 STATE_PARSE_DATA = 'PARSING_DATA' 451 STATE_STORE_PERF = 'STORE_PERF_DATA' 452

453 - def __init__(self, 454 taskName, 455 configId, 456 scheduleIntervalSeconds, 457 taskConfig):

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

502

503 - def __str__(self):

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

506

507 - def cleanup(self):

508 self._cleanUpPool() 509 self._close()

510

511 - def _getPoolKey(self):

512 """ 513 Get the key under which the client should be stored in the pool. 514 """ 515 username = self._device.zCommandUsername 516 password = self._device.zCommandPassword 517 ip = self._manageIp 518 port = self._device.zCommandPort 519 return hash((username, password, ip, port))

520

521 - def _cleanUpPool(self):

522 """ 523 Close the connection currently associated with this task. 524 """ 525 poolkey = self._getPoolKey() 526 if poolkey in self.pool: 527 client = self.pool[poolkey] 528 tasklist = client._taskList 529 if not tasklist: 530 # No other tasks, so safe to clean up 531 transport = client.transport 532 if transport: 533 transport.loseConnection() 534 del self.pool[poolkey]

535

536 - def doTask(self):

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

554

555 - def _connect(self):

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

591

592 - def _close(self):

593 """ 594 If a local datasource executor, do nothing. 595 596 If an SSH datasource executor, relinquish a connection to the remote device. 597 """ 598 if self._connection: 599 self._connection._taskList.discard(self) 600 if not self._connection._taskList: 601 if self._getPoolKey() in self.pool: 602 client = self.pool[self._getPoolKey()] 603 client.clientFinished() 604 client.channelClosed() 605 self._connection = None

606

607 - def connectionFailed(self, msg):

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

625

626 - def _failure(self, reason):

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

673

674 - def _connectCallback(self, result):

675 """ 676 Callback called after a successful connect to the remote device. 677 """ 678 if self._useSsh: 679 log.debug("Connected to %s [%s]", self._devId, self._manageIp) 680 else: 681 log.debug("Running command(s) locally") 682 return result

683

684 - def _addDatasource(self, datasource):

685 """ 686 Add a new instantiation of ProcessRunner or SshRunner 687 for every datasource. 688 """ 689 if self._preferences.options.showfullcommand: 690 log.info("Datasource %s command: %s", datasource.name, 691 datasource.command) 692 693 if self._useSsh: 694 runner = SshRunner(self._connection) 695 else: 696 runner = ProcessRunner() 697 698 d = runner.start(datasource) 699 datasource.lastStart = time.time() 700 d.addBoth(datasource.processCompleted) 701 return d

702

703 - def _fetchPerf(self, ignored):

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

738

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

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

777

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

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

797

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

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

840

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

842 """ 843 If there is no event, send one based on the exit code. 844 """ 845 exitCode = datasource.result.exitCode 846 if exitCode == 0: 847 msg = '' 848 severity = 0 849 else: 850 msg = 'Datasource: %s - Code: %s - Msg: %s' % ( 851 datasource.name, exitCode, getExitMessage(exitCode)) 852 severity = datasource.severity 853 854 ev = self._makeCmdEvent(datasource, msg, severity) 855 results.events.append(ev)

856

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

858 """ 859 Add the stderr output to error events. 860 861 @parameter stderrMsg: stderr output from the command 862 @type stderrMsg: string 863 @parameter eventList: list of events 864 @type eventList: list of dictionaries 865 """ 866 for event in eventList: 867 if event['severity'] not in ('Clear', 'Info', 'Debug'): 868 event['stderr'] = stderrMsg

869

870 - def _storeResults(self, resultList):

871 """ 872 Store the values in RRD files 873 874 @parameter resultList: results of running the commands 875 @type resultList: array of (datasource, dictionary) 876 """ 877 self.state = SshPerformanceCollectionTask.STATE_STORE_PERF 878 for datasource, results in resultList: 879 for dp, value in results.values: 880 threshData = { 881 'eventKey': datasource.getEventKey(dp), 882 'component': dp.component, 883 } 884 self._dataService.writeRRD( 885 dp.rrdPath, 886 value, 887 dp.rrdType, 888 dp.rrdCreateCommand, 889 datasource.cycleTime, 890 dp.rrdMin, 891 dp.rrdMax, 892 threshData) 893 894 return resultList

895

896 - def _updateStatus(self, resultList):

897 """ 898 Send any accumulated events 899 900 @parameter resultList: results of running the commands 901 @type resultList: array of (datasource, dictionary) 902 """ 903 for datasource, results in resultList: 904 self._clearEvent(datasource, results.events) 905 for ev in results.events: 906 self._eventService.sendEvent(ev, device=self._devId) 907 return resultList

908

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

910 """ 911 Ensure that a CLEAR event is sent for any command that 912 successfully completes. 913 """ 914 # If the result is a Failure, no exitCode exists 915 exitCode = getattr(datasource.result, 'exitCode', -1) 916 # Don't send if non-zero exit code or no output returned 917 if exitCode is None or exitCode != 0 or not datasource.result.output.strip(): 918 return 919 920 clearEvents = [ev for ev in eventList if ev['severity'] == Clear] 921 if not clearEvents: 922 msg = 'Datasource %s command completed successfully' % ( 923 datasource.name) 924 ev = self._makeCmdEvent(datasource, msg, severity=Clear) 925 eventList.append(ev)

926

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

928 """ 929 Create an event using the info in the Cmd object. 930 """ 931 severity = datasource.severity if severity is None else severity 932 ev = dict( 933 device=self._devId, 934 component=datasource.component, 935 eventClass=datasource.eventClass, 936 eventKey=datasource.eventKey, 937 severity=severity, 938 summary=msg 939 ) 940 return ev

941

942 - def _finished(self, result):

943 """ 944 Callback activated when the task is complete 945 946 @parameter result: results of the task 947 @type result: deferred object 948 """ 949 if not isinstance(result, Failure): 950 self._returnToNormalSchedule() 951 952 try: 953 self._close() 954 except Exception, ex: 955 log.warn("Failed to close device %s: error %s" % 956 (self._devId, str(ex))) 957 958 # Return the result so the framework can track success/failure 959 return result

960

961 - def displayStatistics(self):

962 """ 963 Called by the collector framework scheduler, and allows us to 964 see how each task is doing. 965 """ 966 display = "%s useSSH: %s\n" % ( 967 self.name, self._useSsh) 968 if self._lastErrorMsg: 969 display += "%s\n" % self._lastErrorMsg 970 return display

971 972 973 if __name__ == '__main__': 974 # Required for passing classes from zenhub to here 975 from Products.ZenRRD.zencommand import Cmd, DataPointConfig 976 977 myPreferences = SshPerformanceCollectionPreferences() 978 myTaskFactory = SimpleTaskFactory(SshPerformanceCollectionTask) 979 myTaskSplitter = SshPerCycletimeTaskSplitter(myTaskFactory) 980 daemon = CollectorDaemon(myPreferences, myTaskSplitter) 981 daemon.run() 982

Source Code for Module Products.ZenRRD.zencommand