Source Code for Module Products.ZenRRD.zenprocess

  1  #! /usr/bin/env python 
  2  ############################################################################## 
  3  #  
  4  # Copyright (C) Zenoss, Inc. 2007, 2009, 2011, all rights reserved. 
  5  #  
  6  # This content is made available according to terms specified in 
  7  # License.zenoss under the directory where your Zenoss product is installed. 
  8  #  
  9  ############################################################################## 
 10   
 11   
 12  __doc__ = """zenprocess 
 13   
 14  Gets SNMP process data from a device's HOST-RESOURCES-MIB 
 15  and store process performance in RRD files. 
 16  """ 
 17   
 18  import logging 
 19  import sys 
 20  import re 
 21  from md5 import md5 
 22  from pprint import pformat 
 23  import os.path 
 24   
 25  import Globals 
 26   
 27  import zope.component 
 28  import zope.interface 
 29   
 30  from Products.ZenCollector.daemon import CollectorDaemon 
 31  from Products.ZenCollector.interfaces import ICollectorPreferences,\ 
 32      IScheduledTask, IEventService, IDataService, IConfigurationListener 
 33  from Products.ZenCollector.tasks import SimpleTaskFactory, SimpleTaskSplitter,\ 
 34      TaskStates 
 35  from Products.ZenEvents import Event 
 36  from Products.ZenEvents.ZenEventClasses import Status_Snmp, Status_OSProcess,\ 
 37      Status_Perf 
 38  from Products.ZenUtils.observable import ObservableMixin 
 39  from Products.ZenUtils.Utils import prepId as globalPrepId 
 40   
 41  # We retrieve our configuration data remotely via a Twisted PerspectiveBroker 
 42  # connection. To do so, we need to import the class that will be used by the 
 43  # configuration service to send the data over, i.e. DeviceProxy. 
 44  from Products.ZenUtils.Utils import unused 
 45  from Products.ZenCollector.services.config import DeviceProxy 
 46   
 47  unused(DeviceProxy) 
 48  from Products.ZenHub.services.ProcessConfig import ProcessProxy 
 49   
 50  unused(ProcessProxy) 
 51  from Products.ZenHub.services.PerformanceConfig import SnmpConnInfo 
 52   
 53  unused(SnmpConnInfo) 
 54   
 55  from twisted.internet import defer, error 
 56  from pynetsnmp.twistedsnmp import Snmpv3Error 
 57   
 58  log = logging.getLogger("zen.zenprocess") 
 59   
 60  # HOST-RESOURCES-MIB OIDs used 
 61  HOSTROOT = '.1.3.6.1.2.1.25' 
 62  RUNROOT = HOSTROOT + '.4' 
 63  NAMETABLE = RUNROOT + '.2.1.2' 
 64  PATHTABLE = RUNROOT + '.2.1.4' 
 65  ARGSTABLE = RUNROOT + '.2.1.5' 
 66  PERFROOT = HOSTROOT + '.5' 
 67  CPU = PERFROOT + '.1.1.1.'        # note trailing dot 
 68  MEM = PERFROOT + '.1.1.2.'        # note trailing dot 
 69   
 70  # Max size for CPU numbers 
 71  WRAP = 0xffffffffL 
 72   
 73  # Regex to see if a string is all hex digits 
 74  IS_MD5 = re.compile('^[a-f0-9]{32}$') 
 75   
 76  PROC_SCAN_ERROR = 'Unable to read processes on device %s' 
 77   
 78 -class HostResourceMIBExecption(Exception): 
 79      pass 
 80   
 81  # Create an implementation of the ICollectorPreferences interface so that the 
 82  # ZenCollector framework can configure itself from our preferences. 
 83 -class ZenProcessPreferences(object): 
 84      zope.interface.implements(ICollectorPreferences) 
 85   
 86 -    def __init__(self): 
 87          """ 
 88          Constructs a new ZenProcessPreferences instance and provide default 
 89          values for needed attributes. 
 90          """ 
 91          self.collectorName = "zenprocess" 
 92          self.defaultRRDCreateCommand = None 
 93          self.configCycleInterval = 20 # minutes 
 94   
 95          #will be updated based on Performance Config property of same name 
 96          self.processCycleInterval = 3 * 60 
 97   
 98          #will be filled in based on buildOptions 
 99          self.options = None 
100   
101          # the configurationService attribute is the fully qualified class-name 
102          # of our configuration service that runs within ZenHub 
103          self.configurationService = 'Products.ZenHub.services.ProcessConfig' 
104   
105      @property 
106 -    def cycleInterval(self): 
107          """ 
108          defined as a property since it is needed by the interface     
109          """ 
110          return self.processCycleInterval 
111   
112 -    def buildOptions(self, parser): 
113          """ 
114          Build a list of command-line options 
115          """ 
116          parser.add_option('--showprocs', 
117                            dest='showprocs', 
118                            action="store_true", 
119                            default=False, 
120                            help="Show the list of processes found."\ 
121                                 "For debugging purposes only.") 
122          parser.add_option('--showrawtables', 
123                            dest='showrawtables', 
124                            action="store_true", 
125                            default=False, 
126                            help="Show the raw SNMP processes data returned "\ 
127                                 "from the device. For debugging purposes only.") 
128          parser.add_option('--captureFilePrefix', dest='captureFilePrefix', 
129                            default='', 
130                            help="Directory and filename to use as a template" 
131                                 " to store SNMP results from device.") 
132   
133 -    def postStartup(self): 
134          pass 
135   
136   
137 -class DeviceStats: 
138 -    def __init__(self, deviceProxy): 
139          self.config = deviceProxy 
140          # map pid number to ProcessStats object 
141          self._pidToProcess = {} 
142          # map ProcessProxy id to ProcessStats object 
143          self._processes = {} 
144          for id, process in deviceProxy.processes.items(): 
145              self._processes[id] = ProcessStats(process) 
146   
147 -    def update(self, deviceProxy): 
148          unused = set(self._processes.keys()) 
149          for id, process in deviceProxy.processes.items(): 
150              unused.discard(id) 
151              if self._processes.get(id): 
152                  self._processes[id].update(process) 
153              else: 
154                  self._processes[id] = ProcessStats(process) 
155   
156          #delete the left overs 
157          for id in unused: 
158              del self._processes[id] 
159              for key, value in self._pidToProcess.items(): 
160                  if value._config.name == id: 
161                      del self._pidToProcess[key] 
162   
163      @property 
164 -    def processStats(self): 
165          """ 
166          The ProcessStats: processes configured to be monitored 
167          """ 
168          return self._processes.values() 
169   
170      @property 
171 -    def pids(self): 
172          """ 
173          returns the pids from being tracked 
174          """ 
175          return self._pidToProcess.keys() 
176   
177      @property 
178 -    def monitoredProcs(self): 
179          """ 
180          returns ProcessStats for which we have a pid 
181          """ 
182          return self._pidToProcess.values() 
183   
184   
185 -class ProcessStats: 
186 -    def __init__(self, processProxy): 
187          self._pids = {} 
188          self._config = processProxy 
189          self.cpu = 0 
190          self.digest = md5('').hexdigest() 
191          if not self._config.ignoreParameters: 
192              # The modeler plugin computes the MD5 hash of the args, 
193              # and then tosses that into the name of the process 
194              result = self._config.name.rsplit(' ', 1) 
195              if len(result) == 2 and result[1] != '': 
196                  self.digest = result[1] 
197   
198 -    def update(self, processProxy): 
199          self._config = processProxy 
200   
201 -    def __str__(self): 
202          """ 
203          Override the Python default to represent ourselves as a string 
204          """ 
205          return str(self._config.name) 
206   
207      __repr__ = __str__ 
208   
209 -    def match(self, name, args, useName=True, useMd5Digest=True): 
210          """ 
211          Perform exact comparisons on the process names. 
212   
213          @parameter name: name of a process to compare 
214          @type name: string 
215          @parameter args: argument list of the process 
216          @type args: string 
217          @parameter useMd5Digest: ignore true result if MD5 doesn't match the process name? 
218          @type useMd5Digest: boolean 
219          @return: does the name match this process's info? 
220          @rtype: Boolean 
221          """ 
222          if self._config.name is None: 
223              return False 
224   
225          # SNMP agents return a 'flexible' number of characters, 
226          # so exact matching isn't always reliable. 
227          processName = ('%s %s' % (name, args or '')).strip() 
228   
229          # Make the comparison 
230          result = re.search(self._config.regex, processName) is not None 
231   
232          # We can a match, but it might not be for us 
233          if result and useMd5Digest: 
234              # Compare this arg list against the digest of this proc 
235              digest = md5(args).hexdigest() 
236              if self.digest and digest != self.digest: 
237                  result = False 
238   
239          if result and useName: 
240              nameOnly = self._config.name.rsplit(' ', 1)[0] 
241              cleanNameOnly = globalPrepId(name) 
242              nameRe = '(.?)' + re.escape(nameOnly) + '$' 
243              nameMatch = re.search(nameRe, cleanNameOnly) 
244              if not nameMatch or nameMatch.group(1) not in ('', '_'): 
245                  log.debug("Discarding match based on name mismatch: %s %s" % (cleanNameOnly, nameOnly)) 
246                  result = False 
247   
248          return result 
249   
250 -    def updateCpu(self, pid, value): 
251          """ 
252          """ 
253          pid = self._pids.setdefault(pid, Pid()) 
254          cpu = pid.updateCpu(value) 
255          if cpu is not None: 
256              self.cpu += cpu 
257              self.cpu %= WRAP 
258   
259 -    def getCpu(self): 
260          """ 
261          """ 
262          return self.cpu 
263   
264 -    def updateMemory(self, pid, value): 
265          """ 
266          """ 
267          self._pids.setdefault(pid, Pid()).memory = value 
268   
269 -    def getMemory(self): 
270          """ 
271          """ 
272          return sum(x.memory for x in self._pids.values() if x.memory is not None) 
273   
274 -    def discardPid(self, pid): 
275          """ 
276          """ 
277          if pid in self._pids: 
278              del self._pids[pid] 
279   
280   
281 -class Pid: 
282      """ 
283      Helper class to track process id information 
284      """ 
285      cpu = None 
286      memory = None 
287   
288 -    def updateCpu(self, n): 
289          """ 
290          """ 
291          if n is not None: 
292              try: 
293                  n = int(n) 
294              except ValueError: 
295                  log.warning("Bad value for CPU: '%s'", n) 
296   
297          if self.cpu is None or n is None: 
298              self.cpu = n 
299              return None 
300          diff = n - self.cpu 
301          if diff < 0: 
302              # don't provide a value when the counter falls backwards 
303              n = None 
304              diff = None 
305          self.cpu = n 
306          return diff 
307   
308 -    def updateMemory(self, n): 
309          """ 
310          """ 
311          self.memory = n 
312   
313 -    def __str__(self): 
314          """ 
315          Override the Python default to represent ourselves as a string 
316          """ 
317          return '<Pid> memory: %s cpu: %s' % (self.memory, self.cpu) 
318   
319      __repr__ = __str__ 
320   
321   
322 -class ConfigListener(object): 
323      zope.interface.implements(IConfigurationListener) 
324   
325 -    def deleted(self, configurationId): 
326          """ 
327          Called when a configuration is deleted from the collector 
328          """ 
329          log.debug('ConfigListener: configuration %s deleted' % configurationId) 
330          ZenProcessTask.DEVICE_STATS.pop(configurationId, None) 
331   
332 -    def added(self, configuration): 
333          """ 
334          Called when a configuration is added to the collector 
335          """ 
336          log.debug('ConfigListener: configuration %s added' % configuration) 
337   
338   
339 -    def updated(self, newConfiguration): 
340          """ 
341          Called when a configuration is updated in collector 
342          """ 
343          log.debug('ConfigListener: configuration %s updated' % newConfiguration) 
344   
345  # Create an implementation of the IScheduledTask interface that will perform 
346  # the actual collection work needed by this collector. 
347 -class ZenProcessTask(ObservableMixin): 
348      """ 
349      A scheduled task that finds instances of configure processes and collects  
350      metrics on the processes  
351      """ 
352      zope.interface.implements(IScheduledTask) 
353   
354      #Keep state about process stats across task updates 
355      DEVICE_STATS = {} 
356   
357      #counter to keep track of total restarted and missing processes 
358      RESTARTED = 0 
359      MISSING = 0 
360   
361      STATE_CONNECTING = 'CONNECTING' 
362      STATE_SCANNING_PROCS = 'SCANNING_PROCESSES' 
363      STATE_FETCH_PERF = 'FETCH_PERF_DATA' 
364      STATE_STORE_PERF = 'STORE_PERF_DATA' 
365      STATE_PARSING_TABLE_DATA = 'PARSING_TABLE_DATA' 
366   
367      statusEvent = {'eventClass': Status_OSProcess, 
368                     'eventGroup': 'Process'} 
369   
370 -    def __init__(self, 
371                   deviceId, 
372                   taskName, 
373                   scheduleIntervalSeconds, 
374                   taskConfig): 
375          super(ZenProcessTask, self).__init__() 
376   
377          #needed for interface 
378          self.name = taskName 
379          self.configId = deviceId 
380          self.interval = scheduleIntervalSeconds 
381          self.state = TaskStates.STATE_IDLE 
382   
383          #the task config corresponds to a DeviceProxy 
384          self._device = taskConfig 
385          self._devId = self._device.name 
386          self._manageIp = self._device.manageIp 
387          self._maxOidsPerRequest = self._device.zMaxOIDPerRequest 
388   
389          self._dataService = zope.component.queryUtility(IDataService) 
390          self._eventService = zope.component.queryUtility(IEventService) 
391          self._preferences = zope.component.queryUtility(ICollectorPreferences, 
392                                                          "zenprocess") 
393          self.snmpProxy = None 
394          self.snmpConnInfo = self._device.snmpConnInfo 
395   
396          self._deviceStats = ZenProcessTask.DEVICE_STATS.get(self._devId) 
397          if self._deviceStats: 
398              self._deviceStats.update(self._device) 
399          else: 
400              self._deviceStats = DeviceStats(self._device) 
401              ZenProcessTask.DEVICE_STATS[self._devId] = self._deviceStats 
402   
403      @defer.inlineCallbacks 
404 -    def doTask(self): 
405          """ 
406          Contact to one device and return a deferred which gathers data from 
407          the device. 
408   
409          @return: job to scan a device 
410          @rtype: Twisted deferred object 
411          """ 
412          try: 
413              # see if we need to connect first before doing any collection 
414              self.openProxy() 
415              log.debug("Opened proxy to %s [%s]", self._devId, self._manageIp) 
416              yield self._collectCallback() 
417          finally: 
418              self._finished() 
419   
420   
421      @defer.inlineCallbacks 
422 -    def _collectCallback(self): 
423          """ 
424          Callback called after a connect or previous collection so that another 
425          collection can take place. 
426          """ 
427          log.debug("Scanning for processes from %s [%s]", 
428                    self._devId, self._manageIp) 
429   
430          self.state = ZenProcessTask.STATE_SCANNING_PROCS 
431          tables = [NAMETABLE, PATHTABLE, ARGSTABLE] 
432          try: 
433              tableResult = yield self._getTables(tables) 
434              summary = 'Process table up for device %s' % self._devId 
435              self._clearSnmpError("%s - timeout cleared" % summary, 'table_scan_timeout') 
436              if self.snmpConnInfo.zSnmpVer == 'v3': 
437                  self._clearSnmpError("%s - v3 error cleared" % summary, 'table_scan_v3_error') 
438              processes = self._parseProcessNames(tableResult) 
439              self._clearSnmpError(summary, 'resource_mib') 
440              self._deviceStats.update(self._device) 
441              processStatuses = self._determineProcessStatus(processes) 
442              self._sendProcessEvents(processStatuses) 
443              self._clearSnmpError(summary) 
444              yield self._fetchPerf() 
445              log.debug("Device %s [%s] scanned successfully", 
446                        self._devId, self._manageIp) 
447          except HostResourceMIBExecption as e: 
448              summary = 'Device %s does not publish HOST-RESOURCES-MIB' %\ 
449                        self._devId 
450              resolution = "Verify with snmpwalk %s %s" %\ 
451                           (self._devId, NAMETABLE ) 
452              log.warn(summary) 
453              self._sendSnmpError(summary, "resource_mib", resolution=resolution) 
454   
455          except error.TimeoutError as e: 
456              log.debug('Timeout fetching tables on device %s' % self._devId) 
457              self._sendSnmpError('%s; Timeout on device' % PROC_SCAN_ERROR % self._devId, 'table_scan_timeout') 
458          except Snmpv3Error as e: 
459              msg = "Cannot connect to SNMP agent on {0._devId}: {1.value}".format(self, str(e)) 
460              log.debug(msg) 
461              self._sendSnmpError('%s; %s' % (PROC_SCAN_ERROR % self._devId, msg), 'table_scan_v3_error') 
462          except Exception as e: 
463              log.exception('Unexpected Error on device %s' % self._devId) 
464              msg = '%s; error: %s' % (PROC_SCAN_ERROR % self._devId, e) 
465              self._sendSnmpError(msg) 
466   
467 -    def _finished(self): 
468          """ 
469          Callback activated when the task is complete 
470          """ 
471          try: 
472              self._close() 
473          except Exception, e: 
474              log.warn("Failed to close device %s: error %s" % 
475                       (self._devId, str(e))) 
476   
477   
478 -    def cleanup(self): 
479          return self._close() 
480   
481 -    def capturePacket(self, hostname, data): 
482          """ 
483          Store SNMP results into files for unit-testing. 
484          """ 
485          # Prep for using capture replay module later 
486          if not hasattr(self, 'captureSerialNum'): 
487              self.captureSerialNum = 0 
488   
489          log.debug("Capturing packet from %s", hostname) 
490          name = "%s-%s-%d" % (self._preferences.options.captureFilePrefix, 
491                               hostname, self.captureSerialNum) 
492   
493          # Don't overwrite previous captures, which will happen if we remodel 
494          # and the serial number gets reset to zero 
495          while os.path.exists(name): 
496              self.captureSerialNum += 1 
497              name = "%s-%s-%d" % (self._preferences.options.captureFilePrefix, 
498                               hostname, self.captureSerialNum) 
499   
500          try: 
501              capFile = open(name, "w") 
502              capFile.write(pformat(data)) 
503              capFile.close() 
504              self.captureSerialNum += 1 
505          except Exception, ex: 
506              log.warn("Couldn't write capture data to '%s' because %s", 
507                       name, str(ex)) 
508   
509 -    def sendRestartEvents(self, afterByConfig, beforeByConfig, restarted): 
510          for procStats, pConfig in restarted.iteritems(): 
511              droppedPids = [] 
512              for pid in beforeByConfig[procStats]: 
513                  if pid not in afterByConfig[procStats]: 
514                      droppedPids.append(pid) 
515              summary = 'Process restarted: %s' % pConfig.originalName 
516              message = '%s\n Using regex \'%s\' Discarded dead pid(s) %s Using new pid(s) %s'\ 
517              % (summary, pConfig.regex, droppedPids, afterByConfig[procStats]) 
518              self._eventService.sendEvent(self.statusEvent, 
519                                           device=self._devId, 
520                                           summary=summary, 
521                                           message=message, 
522                                           component=pConfig.originalName, 
523                                           eventKey=pConfig.processClass, 
524                                           severity=pConfig.severity) 
525              log.info("(%s) %s" % (self._devId, message)) 
526   
527 -    def sendFoundProcsEvents(self, afterByConfig, restarted): 
528          # report alive processes 
529          for processStat in afterByConfig.keys(): 
530              if processStat in restarted: continue 
531              summary = "Process up: %s" % processStat._config.originalName 
532              message = '%s\n Using regex \'%s\' with pid\'s %s '\ 
533              % (summary, processStat._config.regex, afterByConfig[processStat]) 
534              self._eventService.sendEvent(self.statusEvent, 
535                                           device=self._devId, 
536                                           summary=summary, 
537                                           message=message, 
538                                           component=processStat._config.originalName, 
539                                           eventKey=processStat._config.processClass, 
540                                           severity=Event.Clear) 
541              log.debug("(%s) %s" % (self._devId, message)) 
542   
543 -    def _parseProcessNames(self, results): 
544          """ 
545          Parse the process tables and reconstruct the list of processes 
546          that are on the device. 
547   
548          @parameter results: results of SNMP table gets 
549          @type results: dictionary of dictionaries 
550          """ 
551          self.state = ZenProcessTask.STATE_PARSING_TABLE_DATA 
552          if not results or not results[NAMETABLE]: 
553              raise HostResourceMIBExecption() 
554   
555          if self._preferences.options.captureFilePrefix: 
556              self.capturePacket(self._devId, results) 
557   
558          showrawtables = self._preferences.options.showrawtables 
559          args, procs = mapResultsToDicts(showrawtables, results) 
560          if self._preferences.options.showprocs: 
561              self._showProcessList(procs) 
562          return procs 
563   
564 -    def sendMissingProcsEvents(self, missing): 
565          # Look for missing processes 
566          for procConfig in missing: 
567              ZenProcessTask.MISSING += 1 
568              summary = 'Process not running: %s' % procConfig.originalName 
569              message = "%s\n   Using regex \'%s\' \n   All Processes have stopped since the last model occurred. Last Modification time (%s)" \ 
570                          % (summary,procConfig.regex,self._device.lastmodeltime) 
571              self._eventService.sendEvent(self.statusEvent, 
572                                               device=self._devId, 
573                                               summary=summary, 
574                                               message=message, 
575                                               component=procConfig.originalName, 
576                                               eventKey=procConfig.processClass, 
577                                               severity=procConfig.severity) 
578              log.warning("(%s) %s" % (self._devId, message)) 
579   
580 -    def _sendProcessEvents(self, results): 
581          (afterByConfig, afterPidToProcessStats, 
582           beforeByConfig, newPids, restarted, deadPids, missing) = results 
583   
584          self.sendRestartEvents(afterByConfig, beforeByConfig, restarted) 
585          self.sendFoundProcsEvents(afterByConfig, restarted) 
586   
587          for pid in newPids: 
588              log.debug("Found new %s %s pid %d on %s " % ( 
589                  afterPidToProcessStats[pid]._config.originalName, afterPidToProcessStats[pid]._config.name, pid, 
590                  self._devId)) 
591          self._deviceStats._pidToProcess = afterPidToProcessStats 
592          self.sendMissingProcsEvents(missing) 
593   
594          # Store the total number of each process into an RRD 
595          pidCounts = dict((p, 0) for p in self._deviceStats.processStats) 
596   
597          for procStat in self._deviceStats.monitoredProcs: 
598              # monitoredProcs is determined from the current pids in _pidToProcess 
599              # pidCounts is from _deviceStats.processStats which is modeled data 
600              if procStat in pidCounts: 
601                  pidCounts[procStat] += 1 
602              else: 
603                  log.warn('%s monitored proc %s %s not in process stats', self._devId, procStat._config.name, 
604                           procStat._config.originalName) 
605                  log.debug("%s pidcounts is %s", self._devId, pidCounts) 
606          for procName, count in pidCounts.items(): 
607              self._save(procName, 'count_count', count, 'GAUGE') 
608          return "Sent events" 
609   
610 -    def _determineProcessStatus(self, procs): 
611          """ 
612          Determine the up/down/restarted status of processes. 
613   
614          @parameter procs: array of pid, (name, args) info 
615          @type procs: list 
616          """ 
617          beforePids = set(self._deviceStats.pids) 
618          afterPidToProcessStats = {} 
619          pStatsWArgsAndSums, pStatsWoArgs = self._splitPStatMatchers() 
620          for pid, (name, psargs) in procs: 
621              pStats = self._deviceStats._pidToProcess.get(pid) 
622              if pStats: 
623                  # We saw the process before, so there's a good 
624                  # chance that it's the same. 
625                  if pStats.match(name, psargs): 
626                      # Yep, it's the same process 
627                      log.debug("Found process %d on %s, matching %s %s with MD5", 
628                                pid, pStats._config.name, name, psargs) 
629                      log.debug("%s found existing stat %s %s for pid %s - using MD5", self._devId, pStats._config.name, 
630                                pStats._config.originalName, pid) 
631                      afterPidToProcessStats[pid] = pStats 
632                      continue 
633   
634                  elif pStats.match(name, psargs, useMd5Digest=False): 
635                      # In this case, our raw SNMP data from the 
636                      # remote agent got futzed 
637                      # It's the same process. Yay! 
638                      log.debug("%s - Found process %d on %s, matching %s %s without MD5", 
639                                self._devId, pid, pStats._config.name, name, psargs) 
640                      afterPidToProcessStats[pid] = pStats 
641                      continue 
642   
643              # Search for the first match in our list of regexes 
644              # that have arguments AND an MD5-sum argument matching. 
645              # Explicitly *IGNORE* any matchers not modeled by zenmodeler 
646              for pStats in pStatsWArgsAndSums: 
647                  if pStats.match(name, psargs): 
648                      log.debug("%s Found process %d on %s %s", 
649                                self._devId, pid, pStats._config.originalName, pStats._config.name) 
650                      afterPidToProcessStats[pid] = pStats 
651                      break 
652              else: 
653                  # Now look for the first match in our list of regexes 
654                  # that don't have arguments. 
655                  for pStats in pStatsWoArgs: 
656                      if pStats.match(name, psargs, useMd5Digest=False): 
657                          log.debug("Found process %d on %s", 
658                                        pid, pStats._config.name) 
659                          afterPidToProcessStats[pid] = pStats 
660                          break 
661   
662          afterPids = set(afterPidToProcessStats.keys()) 
663          afterByConfig = reverseDict(afterPidToProcessStats) 
664          newPids = afterPids - beforePids 
665          deadPids = beforePids - afterPids 
666   
667          restarted = {} 
668          for pid in deadPids: 
669              procStats = self._deviceStats._pidToProcess[pid] 
670              procStats.discardPid(pid) 
671              if procStats in afterByConfig: 
672                  ZenProcessTask.RESTARTED += 1 
673                  pConfig = procStats._config 
674                  if pConfig.restart: 
675                      restarted[procStats] = pConfig 
676   
677          # Now that we've found all of the stragglers, check to see 
678          # what really is missing or not. 
679          missing = [] 
680          for procStat in self._deviceStats.processStats: 
681              if procStat not in afterByConfig: 
682                  missing.append(procStat._config) 
683   
684          # For historical reasons, return the beforeByConfig 
685          beforeByConfig = reverseDict(self._deviceStats._pidToProcess) 
686   
687          return (afterByConfig, afterPidToProcessStats, 
688                  beforeByConfig, newPids, restarted, deadPids, 
689                  missing) 
690   
691 -    def _splitPStatMatchers(self): 
692          pStatsWArgsAndSums = []; pStatsWoArgs = [] 
693          for pStat in self._deviceStats.processStats: 
694              if pStat._config.ignoreParameters: 
695                  pStatsWoArgs.append(pStat) 
696              else: 
697                  nameList = pStat._config.name.rsplit(' ', 1) 
698                  if len(nameList) < 2: # (name, md5sum) 
699                      nameList = (nameList[0], md5('').hexdigest()) 
700   
701                  possibleMd5Sum = nameList[-1].lower() 
702                  if IS_MD5.match(possibleMd5Sum): 
703                      pStatsWArgsAndSums.append(pStat) 
704          return pStatsWArgsAndSums, pStatsWoArgs 
705   
706      @defer.inlineCallbacks 
707 -    def _fetchPerf(self): 
708          """ 
709          Get performance data for all the monitored processes on a device 
710   
711          @parameter results: results of SNMP table gets 
712          @type results: list of (success, result) tuples 
713          """ 
714          self.state = ZenProcessTask.STATE_FETCH_PERF 
715   
716          oids = [] 
717          for pid in self._deviceStats.pids: 
718              oids.extend([CPU + str(pid), MEM + str(pid)]) 
719          if oids: 
720              singleOids = set() 
721              results = {} 
722              oidsToTest = oids 
723              chunkSize = self._maxOidsPerRequest 
724              while oidsToTest: 
725                  for oidChunk in chunk(oids, chunkSize): 
726                      try: 
727                          log.debug("%s fetching oid(s) %s" % (self._devId, oidChunk)) 
728                          result = yield self._get(oidChunk) 
729                          results.update(result) 
730                      except (error.TimeoutError, Snmpv3Error) as e: 
731                          log.debug("error reading oid(s) %s - %s", oidChunk, e) 
732                          singleOids.update(oidChunk) 
733                  oidsToTest = [] 
734                  if singleOids and chunkSize > 1: 
735                      chunkSize = 1 
736                      log.debug("running oids for %s in single mode %s" % (self._devId, singleOids)) 
737                      oidsToTest = list(singleOids) 
738              self._storePerfStats(results) 
739   
740 -    def _storePerfStats(self, results): 
741          """ 
742          Save the process performance data in RRD files 
743   
744          @parameter results: results of SNMP table gets 
745          @type results: list of {oid:value} dictionaries 
746          @parameter device: proxy object to the remote computer 
747          @type device: Device object 
748          """ 
749          self.state = ZenProcessTask.STATE_STORE_PERF 
750          byConf = reverseDict(self._deviceStats._pidToProcess) 
751          for procStat, pids in byConf.items(): 
752              if len(pids) != 1: 
753                  log.debug("There are %d pids by the name %s - %s", 
754                            len(pids), procStat._config.name, procStat._config.originalName) 
755              procName = procStat._config.name 
756              for pid in pids: 
757                  cpu = results.get(CPU + str(pid), None) 
758                  mem = results.get(MEM + str(pid), None) 
759                  procStat.updateCpu(pid, cpu) 
760                  procStat.updateMemory(pid, mem) 
761              self._save(procName, 'cpu_cpu', procStat.getCpu(), 
762                         'DERIVE', min=0) 
763              self._save(procName, 'mem_mem', 
764                         procStat.getMemory() * 1024, 'GAUGE') 
765          return results 
766   
767 -    def _getTables(self, oids): 
768          """ 
769          Perform SNMP getTable for specified OIDs 
770   
771          @parameter oids: OIDs to gather 
772          @type oids: list of strings 
773          @return: Twisted deferred 
774          @rtype: Twisted deferred 
775          """ 
776          repetitions = self._maxOidsPerRequest / len(oids) 
777          t = self.snmpProxy.getTable(oids, 
778                                      timeout=self.snmpConnInfo.zSnmpTimeout, 
779                                      retryCount=self.snmpConnInfo.zSnmpTries, 
780                                      maxRepetitions=repetitions, 
781                                      limit=sys.maxint) 
782          return t 
783   
784 -    def _get(self, oids): 
785          """ 
786          Perform SNMP get for specified OIDs 
787   
788          @parameter oids: OIDs to gather 
789          @type oids: list of strings 
790          @return: Twisted deferred 
791          @rtype: Twisted deferred 
792          """ 
793          return self.snmpProxy.get(oids, 
794                                    self.snmpConnInfo.zSnmpTimeout, 
795                                    self.snmpConnInfo.zSnmpTries) 
796   
797 -    def openProxy(self): 
798          """ 
799          Create a connection to the remote device 
800          """ 
801          self.state = ZenProcessTask.STATE_CONNECTING 
802          if (self.snmpProxy is None or 
803              self.snmpProxy.snmpConnInfo != self.snmpConnInfo): 
804              self.snmpProxy = self.snmpConnInfo.createSession() 
805              self.snmpProxy.open() 
806   
807 -    def _close(self): 
808          """ 
809          Close down the connection to the remote device 
810          """ 
811          if self.snmpProxy: 
812              self.snmpProxy.close() 
813          self.snmpProxy = None 
814   
815   
816 -    def _showProcessList(self, procs): 
817          """ 
818          Display the processes in a sane manner. 
819   
820          @parameter procs: list of (pid, (name, args)) 
821          @type procs: list of tuples 
822          """ 
823          device_name = self._devId 
824          proc_list = ['%s %s %s' % (pid, name, args) for pid, (name, args) in sorted(procs)] 
825          proc_list.append('') 
826          log.info("#===== Processes on %s:\n%s", device_name, '\n'.join(proc_list)) 
827   
828   
829 -    def _sendSnmpError(self, message, eventKey=None, **kwargs): 
830          event = self.statusEvent.copy() 
831          event.update(kwargs) 
832          self._eventService.sendEvent(event, 
833                                       eventClass=Status_Snmp, 
834                                       device=self._devId, 
835                                       severity=Event.Error, 
836                                       eventKey=eventKey, 
837                                       summary=message) 
838   
839 -    def _clearSnmpError(self, message, eventKey=None): 
840          """ 
841          Send an event to clear other events. 
842   
843          @parameter message: clear text 
844          @type message: string 
845          """ 
846          self._eventService.sendEvent(self.statusEvent, 
847                                       eventClass=Status_Snmp, 
848                                       device=self._devId, 
849                                       summary=message, 
850                                       agent='zenprocess', 
851                                       eventKey=eventKey, 
852                                       severity=Event.Clear) 
853   
854 -    def _save(self, pidName, statName, value, rrdType, min='U'): 
855          """ 
856          Save a value into an RRD file 
857   
858          @param pidName: process id of the monitored process 
859          @type pidName: string 
860          @param statName: metric name 
861          @type statName: string 
862          @param value: data to be stored 
863          @type value: number 
864          @param rrdType: RRD data type (eg ABSOLUTE, DERIVE, COUNTER) 
865          @type rrdType: string 
866          """ 
867          deviceName = self._devId 
868          path = 'Devices/%s/os/processes/%s/%s' % (deviceName, pidName, statName) 
869          try: 
870              self._dataService.writeRRD(path, value, rrdType, min=min) 
871          except Exception, ex: 
872              summary = "Unable to save data for process-monitor RRD %s" %\ 
873                        path 
874              log.critical(summary) 
875   
876              message = "Data was value= %s, type=%s" %\ 
877                        ( value, rrdType ) 
878              log.critical(message) 
879              log.exception(ex) 
880   
881              import traceback 
882   
883              trace_info = traceback.format_exc() 
884   
885              self._eventService.sendEvent(dict( 
886                  dedupid="%s|%s" % (self._preferences.options.monitor, 
887                                     'RRD write failure'), 
888                  severity=Event.Critical, 
889                  device=self._preferences.options.monitor, 
890                  eventClass=Status_Perf, 
891                  component="RRD", 
892                  pidName=pidName, 
893                  statName=statName, 
894                  path=path, 
895                  message=message, 
896                  traceback=trace_info, 
897                  summary=summary)) 
898   
899   
900 -def mapResultsToDicts(showrawtables, results): 
901      """ 
902      Parse the process tables and reconstruct the list of processes 
903      that are on the device. 
904   
905      @parameter showrawtables: log the raw table info? 
906      @type showrawtables: boolean 
907      @parameter results: results of SNMP table gets ie (OID + pid, value) 
908      @type results: dictionary of dictionaries 
909      @return: maps relating names and pids to each other 
910      @rtype: dictionary, list of tuples 
911      """ 
912   
913      def extract(dictionary, oid, value): 
914          """ 
915          Helper function to extract SNMP table data. 
916          """ 
917          pid = int(oid.split('.')[-1]) 
918          dictionary[pid] = value.strip() 
919   
920      names, paths, args = {}, {}, {} 
921      if showrawtables: 
922          log.info("NAMETABLE = %r", results[NAMETABLE]) 
923      for row in results[NAMETABLE].items(): 
924          extract(names, *row) 
925   
926      if showrawtables: 
927          log.info("PATHTABLE = %r", results[PATHTABLE]) 
928      for row in results[PATHTABLE].items(): 
929          extract(paths, *row) 
930   
931      if showrawtables: 
932          log.info("ARGSTABLE = %r", results[ARGSTABLE]) 
933      for row in results[ARGSTABLE].items(): 
934          extract(args, *row) 
935   
936      procs = [] 
937      for pid, name in names.items(): 
938          path = paths.get(pid, '') 
939          if path and path.find('\\') == -1: 
940              name = path 
941          arg = args.get(pid, '') 
942          procs.append((pid, (name, arg) )) 
943   
944      return args, procs 
945   
946   
947 -def reverseDict(d): 
948      """ 
949      Return a dictionary with keys and values swapped: 
950      all values are lists to handle the different keys mapping to the same value 
951      """ 
952      result = {} 
953      for a, v in d.items(): 
954          result.setdefault(v, []).append(a) 
955      return result 
956   
957   
958 -def chunk(lst, n): 
959      """ 
960      Break lst into n-sized chunks 
961      """ 
962      return [lst[i:i + n] for i in range(0, len(lst), n)] 
963   
964  # Collector Daemon Main entry point 
965  # 
966  if __name__ == '__main__': 
967      myPreferences = ZenProcessPreferences() 
968   
969      myTaskFactory = SimpleTaskFactory(ZenProcessTask) 
970      myTaskSplitter = SimpleTaskSplitter(myTaskFactory) 
971      daemon = CollectorDaemon(myPreferences, myTaskSplitter, ConfigListener()) 
972      daemon.run() 
973