Products.ZenRRD.zencommand

1 ########################################################################### 2 # 3 # This program is part of Zenoss Core, an open source monitoring platform. 4 # Copyright (C) 2007, 2009, 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__ = """ZenCommand 15 16 Run Command plugins periodically. 17 18 """ 19 20 import time 21 from pprint import pformat 22 import logging 23 log = logging.getLogger("zen.zencommand") 24 from sets import Set 25 26 from twisted.internet import reactor, defer, error 27 from twisted.internet.protocol import ProcessProtocol 28 from twisted.python import failure 29 from twisted.spread import pb 30 31 import Globals 32 from Products.ZenUtils.Driver import drive, driveLater 33 from Products.ZenUtils.Utils import unused, getExitMessage 34 from Products.ZenRRD.RRDDaemon import RRDDaemon 35 from Products.ZenRRD.RRDUtil import RRDUtil 36 from Products.DataCollector.SshClient import SshClient 37 from Products.ZenEvents.ZenEventClasses import Clear, Error 38 from Products.ZenRRD.CommandParser import ParsedResults 39 40 from Products.DataCollector import Plugins 41 unused(Plugins) 42 MAX_CONNECTIONS = 256 43 44 45

46 -class TimeoutError(Exception):

47 """ 48 Error for a defered call taking too long to complete 49 """ 50

51 - def __init__(self, *args):

52 Exception.__init__(self) 53 self.args = args

54 55

56 -def Timeout(deferred, seconds, obj):

57 "Cause an error on a deferred when it is taking too long to complete" 58 59 def _timeout(deferred, obj): 60 "took too long... call an errback" 61 deferred.errback(failure.Failure(TimeoutError(obj)))

62 63 64 def _cb(arg, timer): 65 "the command finished, possibly by timing out" 66 if not timer.called: 67 timer.cancel() 68 return arg 69 70 71 timer = reactor.callLater(seconds, _timeout, deferred, obj) 72 deferred.addBoth(_cb, timer) 73 return deferred 74 75

76 -class ProcessRunner(ProcessProtocol):

77 """ 78 Provide deferred process execution 79 """ 80 stopped = None 81 exitCode = None 82 output = '' 83

84 - def start(self, cmd):

85 "Kick off the process: run it local" 86 log.debug('running %r' % cmd.command) 87 88 shell = '/bin/sh' 89 self.cmdline = (shell, '-c', 'exec %s' % cmd.command) 90 self.command = ' '.join(self.cmdline) 91 log.debug('cmd line: %r' % self.command) 92 reactor.spawnProcess(self, shell, self.cmdline, env=None) 93 94 d = Timeout(defer.Deferred(), cmd.deviceConfig.commandTimeout, cmd) 95 self.stopped = d 96 self.stopped.addErrback(self.timeout) 97 return d

98

99 - def timeout(self, value):

100 "Kill a process if it takes too long" 101 try: 102 self.transport.signalProcess('KILL') 103 except error.ProcessExitedAlready: 104 log.debug("Command already exited: %s" % self.command) 105 return value

106 107

108 - def outReceived(self, data):

109 "Store up the output as it arrives from the process" 110 self.output += data

111 112

113 - def processEnded(self, reason):

114 "notify the starter that their process is complete" 115 self.exitCode = reason.value.exitCode 116 log.debug('Received exit code: %s' % self.exitCode) 117 log.debug('Command: %r' % self.command) 118 log.debug('Output: %r' % self.output) 119 120 if self.stopped: 121 d, self.stopped = self.stopped, None 122 if not d.called: 123 d.callback(self)

124 125

126 -class MySshClient(SshClient):

127 """ 128 Connection to SSH server at the remote device 129 """ 130

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

132 SshClient.__init__(self, *args, **kw) 133 self.defers = {}

134

135 - def addCommand(self, command):

136 "Run a command against the server" 137 d = defer.Deferred() 138 self.defers[command] = d 139 SshClient.addCommand(self, command) 140 return d

141 142

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

144 "Forward the results of the command execution to the starter" 145 # don't call the CollectorClient.addResult which adds the result to a 146 # member variable for zenmodeler 147 d = self.defers.pop(command, None) 148 if d is None: 149 log.error("Internal error where deferred object not in dictionary." \ 150 " Command = '%s' Data = '%s' Code = '%s'", 151 command, data, code) 152 elif not d.called: 153 d.callback((data, code))

154 155

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

157 "Turn off blocking SshClient.test method" 158 return True

159 160

161 - def clientFinished(self):

162 "We don't need to track commands/results when they complete" 163 SshClient.clientFinished(self) 164 self.cmdmap = {} 165 self._commands = [] 166 self.results = []

167 168

169 -class SshPool:

170 """ 171 Cache all the Ssh connections so they can be managed 172 """ 173

174 - def __init__(self):

175 self.pool = {} 176 self.eventSender = None

177

178 - def get(self, cmd):

179 "Make a new SSH connection if there isn't one available" 180 dc = cmd.deviceConfig 181 result = self.pool.get(dc.device, None) 182 if result is None: 183 log.debug("Creating connection to %s", dc.device) 184 options = Options(dc.username, dc.password, 185 dc.loginTimeout, dc.commandTimeout, 186 dc.keyPath, dc.concurrentSessions) 187 # New param KeyPath 188 result = MySshClient(dc.device, dc.ipAddress, dc.port, 189 options=options) 190 if self.eventSender is not None: 191 result.sendEvent = self.eventSender.sendEvent 192 result.run() 193 self.pool[dc.device] = result 194 return result

195

196 - def reinitializeCollectorClients(self):

197 for collectorClient in self.pool.values(): 198 collectorClient.reinitialize()

199

200 - def _close(self, device):

201 "close the SSH connection to a device, if it exists" 202 c = self.pool.get(device, None) 203 if c: 204 log.debug("Closing connection to %s", device) 205 if c.transport: 206 c.transport.loseConnection() 207 del self.pool[device]

208 209

210 - def close(self, cmd):

211 "symetric close that matches get() method" 212 self._close(cmd.deviceConfig.device)

213 214

215 - def trimConnections(self, schedule):

216 "reduce the number of connections using the schedule for guidance" 217 # compute device list in order of next use 218 devices = [] 219 for c in schedule: 220 device = c.deviceConfig.device 221 if device not in devices: 222 devices.append(device) 223 # close as many devices as needed 224 while devices and len(self.pool) > MAX_CONNECTIONS: 225 self._close(devices.pop())

226 227

228 -class SshRunner:

229 """ 230 Run a single command across a cached SSH connection 231 """ 232 exitCode = None 233 output = None 234

235 - def __init__(self, pool):

236 self.pool = pool

237 238

239 - def start(self, cmd):

240 "Initiate a command on the remote device" 241 self.defer = defer.Deferred() 242 c = self.pool.get(cmd) 243 try: 244 d = Timeout(c.addCommand(cmd.command), 245 cmd.deviceConfig.commandTimeout, 246 cmd) 247 except Exception, ex: 248 log.warning('Error starting command: %s', ex) 249 self.pool.close(cmd) 250 return defer.fail(ex) 251 d.addErrback(self.timeout) 252 d.addBoth(self.processEnded) 253 return d

254 255

256 - def timeout(self, arg):

257 "Deal with slow executing command/connection (close it)" 258 cmd, = arg.value.args 259 # we could send a kill signal, but then we would need to track 260 # the command channel to send it to: just close the connection 261 self.pool.close(cmd) 262 return arg

263 264

265 - def processEnded(self, value):

266 "Deliver ourselves to the starter with the proper attributes" 267 if isinstance(value, failure.Failure): 268 return value 269 self.output, self.exitCode = value 270 return self

271 272

273 -class DeviceConfig(pb.Copyable, pb.RemoteCopy):

274 lastChange = 0. 275 device = '' 276 ipAddress = '' 277 port = 0 278 username = '' 279 password = '' 280 loginTimeout = 0. 281 commandTimeout = 0. 282 keyPath = ''

283 pb.setUnjellyableForClass(DeviceConfig, DeviceConfig) 284 285

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

287 id = '' 288 component = '' 289 rrdPath = '' 290 rrdType = None 291 rrdCreateCommand = '' 292 rrdMin = None 293 rrdMax = None 294

295 - def __init__(self):

296 self.data = {}

297

298 - def __repr__(self):

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

300 301 pb.setUnjellyableForClass(DataPointConfig, DataPointConfig) 302

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

304 """ 305 Holds the config of every command to be run 306 """ 307 command = None 308 useSsh = False 309 cycleTime = None 310 eventClass = None 311 eventKey = None 312 severity = 3 313 lastStart = 0 314 lastStop = 0 315 result = None 316 317

318 - def __init__(self):

319 self.points = []

320

321 - def running(self):

322 return self.lastStop < self.lastStart

323 324

325 - def name(self):

326 cmd, args = (self.command + ' ').split(' ', 1) 327 cmd = cmd.split('/')[-1] 328 return '%s %s' % (cmd, args)

329 330

331 - def nextRun(self):

332 if self.running(): 333 return self.lastStart + self.cycleTime 334 return self.lastStop + self.cycleTime

335 336

337 - def start(self, pool):

338 if self.useSsh: 339 pr = SshRunner(pool) 340 else: 341 pr = ProcessRunner() 342 d = pr.start(self) 343 self.lastStart = time.time() 344 log.debug('Process %s started' % self.name()) 345 d.addBoth(self.processEnded) 346 return d

347 348

349 - def processEnded(self, pr):

350 self.result = pr 351 self.lastStop = time.time() 352 353 # Check for a condition that could cause zencommand to stop cycling. 354 # http://dev.zenoss.org/trac/ticket/4936 355 if self.lastStop < self.lastStart: 356 log.debug('System clock went back?') 357 self.lastStop = self.lastStart 358 359 if not isinstance(pr, failure.Failure): 360 log.debug('Process %s stopped (%s), %.2f seconds elapsed' % ( 361 self.name(), 362 pr.exitCode, 363 self.lastStop - self.lastStart)) 364 return self 365 return pr

366 367

368 - def updateConfig(self, cfg, deviceConfig):

369 self.deviceConfig = deviceConfig 370 self.useSsh = cfg.useSsh 371 self.cycleTime = max(int(cfg.cycleTime), 1) 372 self.eventKey = cfg.eventKey 373 self.eventClass = cfg.eventClass 374 self.severity = cfg.severity 375 self.command = str(cfg.command) 376 self.points = cfg.points 377 if cfg.severity is None: 378 log.warning("severity is None: cfg,%r ; deviceConfig, %r", 379 cfg, deviceConfig) 380 return self

381

382 - def getEventKey(self, point):

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

385

386 - def commandKey(self):

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

390 391 pb.setUnjellyableForClass(Cmd, Cmd) 392

393 -class Options:

394 loginTries=1 395 searchPath='' 396 existenceTest=None 397

398 - def __init__(self, username, password, loginTimeout, commandTimeout, 399 keyPath, concurrentSessions):

400 self.username = username 401 self.password = password 402 self.loginTimeout=loginTimeout 403 self.commandTimeout=commandTimeout 404 self.keyPath = keyPath 405 self.concurrentSessions = concurrentSessions

406 407

408 -class ConfigurationProcessor(object):

409

410 - def __init__(self, config, scheduledCmds, sendEvent):

411 """ 412 config - one of the configurations that was returned by calling 413 getDataSourceCommands on zenhub. 414 scheduledCmds - a dictionary that maps (device, command) to Cmd object 415 for the commands currently on zencommand's schedule. 416 sendEvent - a function that sends events to zenhub using twisted 417 perspective broker. 418 """ 419 self._config = config 420 self._scheduledCmds = scheduledCmds 421 self._sendEvent = sendEvent 422 self._device = config.device 423 self._suppressed = [] # log warning and send event once per device 424 self._summary = 'zCommandUsername is not set'

425

426 - def _sendUsernameEvent(self, severity):

427 "send an event (or clear it) for username not set" 428 self._sendEvent(dict( 429 device=self._device, 430 eventClass='/Cmd/Fail', 431 eventKey='zCommandUsername', 432 severity=severity, 433 component='zencommand', 434 summary=self._summary))

435

436 - def _warnUsernameNotSet(self, command):

437 """ 438 Warn that the username is not set for device and the SSH command cannot be 439 executed. 440 """ 441 if self._device not in self._suppressed: 442 log.warning(self._summary + ' for %s' % self._device) 443 self._sendUsernameEvent(Error) 444 msg = 'username not configured for %s. skipping %s.' 445 log.debug(msg % (self._device, command))

446

447 - def updateCommands(self):

448 """ 449 Go through the Cmd objects in config.commands and update the config on 450 the command with config. If currentDict has the command on it then use 451 that one, otherwise use the command from config. If the device does not have a 452 username set, then don't yield commands that use SSH. 453 """ 454 for cmd in self._config.commands: 455 key = (self._device, cmd.command) 456 if cmd.useSsh: 457 if self._config.username: 458 self._sendUsernameEvent(Clear) 459 else: 460 self._warnUsernameNotSet(cmd.command) 461 if self._device not in self._suppressed: 462 self._suppressed.append(self._device) 463 if key in self._scheduledCmds: 464 del self._scheduledCmds[key] 465 continue 466 if key in self._scheduledCmds: 467 newCmd = self._scheduledCmds.pop(key) 468 else: 469 newCmd = cmd 470 yield newCmd.updateConfig(cmd, self._config) 471 self._suppressed = []

472 473

474 -class zencommand(RRDDaemon):

475 """ 476 Daemon code to schedule commands and run them. 477 """ 478 479 initialServices = RRDDaemon.initialServices + ['CommandConfig'] 480

481 - def __init__(self):

482 RRDDaemon.__init__(self, 'zencommand') 483 self.schedule = [] 484 self.timeout = None 485 self.pool = SshPool() 486 self.pool.eventSender = self 487 self.executed = 0

488

489 - def remote_deleteDevice(self, doomed):

490 self.log.debug("zenhub has asked us to delete device %s" % doomed) 491 self.schedule = [c for c in self.schedule if c.deviceConfig.device != doomed]

492

493 - def remote_updateConfig(self, config):

494 self.log.debug("Configuration update from zenhub") 495 self.log.info("config: %r", config) 496 self.updateConfig([config], [config.device])

497

498 - def remote_updateDeviceList(self, devices):

499 self.log.debug("zenhub sent updated device list %s" % devices) 500 updated = [] 501 lastChanges = dict(devices) # map device name to last change 502 keep = [] 503 for cmd in self.schedule: 504 if cmd.deviceConfig.device in lastChanges: 505 if cmd.lastChange > lastChanges[cmd.device]: 506 updated.append(cmd.deviceConfig.device) 507 keep.append(cmd) 508 else: 509 self.log.info("Removing all commands for %s", cmd.deviceConfig.device) 510 self.schedule = keep 511 if updated: 512 self.log.info("Fetching the config for %s", updated) 513 d = self.model().callRemote('getDataSourceCommands', devices) 514 d.addCallback(self.updateConfig, updated) 515 d.addErrback(self.error)

516

517 - def updateConfig(self, configs, expected):

518 expected = Set(expected) 519 current = {} 520 for c in self.schedule: 521 if c.deviceConfig.device in expected: 522 current[c.deviceConfig.device,c.command] = c 523 # keep all the commands we didn't ask for 524 update = [c for c in self.schedule if c.deviceConfig.device not in expected] 525 for cfg in configs: 526 self.thresholds.updateForDevice(cfg.device, cfg.thresholds) 527 if self.options.device and self.options.device != cfg.device: 528 continue 529 processor = ConfigurationProcessor(cfg, current, self.sendEvent) 530 update.extend(processor.updateCommands()) 531 for device, command in current.keys(): 532 self.log.info("Deleting command %s from %s", device, command) 533 self.schedule = update 534 self.processSchedule()

535

536 - def heartbeatCycle(self, *ignored):

537 "There is no master 'cycle' to send the hearbeat" 538 self.heartbeat() 539 reactor.callLater(self.heartbeatTimeout/3, self.heartbeatCycle) 540 events = [] 541 events += self.rrdStats.gauge('schedule', 542 self.heartbeatTimeout, 543 len(self.schedule)) 544 events += self.rrdStats.counter('commands', 545 self.heartbeatTimeout, 546 self.executed) 547 events += self.rrdStats.counter('dataPoints', 548 self.heartbeatTimeout, 549 self.rrd.dataPoints) 550 events += self.rrdStats.gauge('cyclePoints', 551 self.heartbeatTimeout, 552 self.rrd.endCycle()) 553 self.sendEvents(events)

554 555

556 - def processSchedule(self, *unused):

557 """ 558 Run through the schedule and start anything that needs to be done. 559 Set a timer if we have nothing to do. 560 """ 561 if not self.options.cycle: 562 for cmd in self.schedule: 563 if cmd.running() or cmd.lastStart == 0: 564 break 565 else: 566 self.stop() 567 return 568 try: 569 if self.timeout and not self.timeout.called: 570 self.timeout.cancel() 571 self.timeout = None 572 def compare(x, y): 573 return cmp(x.nextRun(), y.nextRun())

574 self.schedule.sort(compare) 575 self.pool.trimConnections(self.schedule) 576 earliest = None 577 running = 0 578 now = time.time() 579 for c in self.schedule: 580 if c.running(): 581 running += 1 582 583 for c in self.schedule: 584 if running >= self.options.parallel: 585 break 586 if c.nextRun() <= now: 587 c.start(self.pool).addBoth(self.finished) 588 running += 1 589 else: 590 earliest = c.nextRun() - now 591 break 592 593 if earliest is not None: 594 self.pool.reinitializeCollectorClients() 595 self.log.debug("Next command in %d seconds", int(earliest)) 596 self.timeout = reactor.callLater(max(1, earliest), 597 self.processSchedule) 598 except Exception, ex: 599 self.log.exception(ex)

600 601

602 - def sendCmdEvent(self, cmd, severity, summary):

603 """ 604 Send an event using the info in the Cmd object. 605 """ 606 self.sendEvent(dict(device=cmd.deviceConfig.device, 607 component=cmd.component, 608 eventClass=cmd.eventClass, 609 eventKey=cmd.eventKey, 610 severity=severity, 611 summary=summary))

612

613 - def finished(self, cmdOrErr):

614 """ 615 The command has finished. cmdOrErr is either a Cmd instance or a 616 twisted failure. 617 """ 618 self.executed += 1 619 if isinstance(cmdOrErr, failure.Failure): 620 self.error(cmdOrErr) 621 else: 622 cmd = cmdOrErr 623 self._handleExitCode(cmd) 624 self.parseResults(cmd) 625 self.processSchedule()

626

627 - def _handleExitCode(self, cmd):

628 """ 629 zencommand handles sending clears for exit code 0, all other exit codes 630 should be handled by the parser associated with the command 631 """ 632 exitCode = cmd.result.exitCode 633 msg = 'Cmd: %s - Code: %s - Msg: %s' % ( 634 cmd.command, exitCode, getExitMessage(exitCode)) 635 if exitCode == 0: 636 self.sendCmdEvent(cmd, Clear, msg)

637

638 - def error(self, err):

639 """ 640 The finished method indicated that there was a failure. This method 641 is also called by RRDDaemon.errorStop. 642 """ 643 if isinstance(err.value, TimeoutError): 644 cmd, = err.value.args 645 msg = "Command timed out on device %s: %r" % ( 646 cmd.deviceConfig.device, cmd.command) 647 self.log.warning(msg) 648 self.sendCmdEvent(cmd, cmd.severity, msg) 649 else: 650 self.log.exception(err.value)

651

652 - def parseResults(self, cmd):

653 """ 654 Process the results of our command-line, send events 655 and check datapoints. 656 657 @param cmd: command 658 @type: cmd object 659 """ 660 self.log.debug('The result of "%s" was "%r"', cmd.command, cmd.result.output) 661 results = ParsedResults() 662 try: 663 parser = cmd.parser.create() 664 except Exception, ex: 665 self.log.exception("Error loading parser %s" % cmd.parser) 666 import traceback 667 self.sendEvent(dict(device=cmd.deviceConfig.device, 668 summary="Error loading parser %s" % cmd.parser, 669 component="zencommand", 670 message=traceback.format_exc(), 671 agent="zencommand", 672 )) 673 return 674 parser.processResults(cmd, results) 675 676 for ev in results.events: 677 self.sendEvent(ev, device=cmd.deviceConfig.device) 678 679 for dp, value in results.values: 680 self.log.debug("Storing %s = %s into %s" % (dp.id, value, dp.rrdPath)) 681 value = self.rrd.save(dp.rrdPath, 682 value, 683 dp.rrdType, 684 dp.rrdCreateCommand, 685 cmd.cycleTime, 686 dp.rrdMin, 687 dp.rrdMax) 688 self.log.debug("RRD save result: %s" % value) 689 for ev in self.thresholds.check(dp.rrdPath, time.time(), value): 690 eventKey = cmd.getEventKey(dp) 691 if 'eventKey' in ev: 692 ev['eventKey'] = '%s|%s' % (eventKey, ev['eventKey']) 693 else: 694 ev['eventKey'] = eventKey 695 ev['component'] = dp.component 696 self.sendEvent(ev)

697

698 - def fetchConfig(self):

699 def doFetchConfig(driver): 700 try: 701 now = time.time() 702 703 yield self.model().callRemote('propertyItems') 704 self.setPropertyItems(driver.next()) 705 706 yield self.model().callRemote('getDefaultRRDCreateCommand') 707 createCommand = driver.next() 708 709 yield self.model().callRemote('getThresholdClasses') 710 self.remote_updateThresholdClasses(driver.next()) 711 712 713 yield self.model().callRemote('getCollectorThresholds') 714 self.rrdStats.config(self.options.monitor, 715 self.name, 716 driver.next(), 717 createCommand) 718 719 devices = [] 720 if self.options.device: 721 devices = [self.options.device] 722 yield self.model().callRemote('getDataSourceCommands', 723 devices) 724 if not devices: 725 devices = list(Set([c.deviceConfig.device 726 for c in self.schedule])) 727 self.updateConfig(driver.next(), devices) 728 729 self.rrd = RRDUtil(createCommand, 60) 730 731 self.sendEvents( 732 self.rrdStats.gauge('configTime', 733 self.configCycleInterval * 60, 734 time.time() - now)) 735 736 except Exception, ex: 737 self.log.exception(ex) 738 raise

739 740 return drive(doFetchConfig) 741 742

743 - def start(self, driver):

744 """ 745 Fetch the configuration and return a deferred for its completion. 746 Also starts the config cycle 747 """ 748 ex = None 749 try: 750 self.log.debug('Fetching configuration from zenhub') 751 yield self.fetchConfig() 752 driver.next() 753 self.log.debug('Finished config fetch') 754 except Exception, ex: 755 self.log.exception(ex) 756 driveLater(self.configCycleInterval * 60, self.start) 757 if ex: 758 raise ex

759

760 - def buildOptions(self):

761 RRDDaemon.buildOptions(self) 762 763 self.parser.add_option('--parallel', dest='parallel', 764 default=10, type='int', 765 help="Number of devices to collect at one time")

766

767 - def connected(self):

768 d = drive(self.start).addCallbacks(self.processSchedule, self.errorStop) 769 if self.options.cycle: 770 d.addCallback(self.heartbeatCycle)

771 772 773 if __name__ == '__main__': 774 from Products.ZenRRD.zencommand import zencommand 775 z = zencommand() 776 z.run() 777

Source Code for Module Products.ZenRRD.zencommand