Source Code for Module ZenStatus.zenstatus

  1  ########################################################################### 
  2  # 
  3  # This program is part of Zenoss Core, an open source monitoring platform. 
  4  # Copyright (C) 2007, Zenoss Inc. 
  5  # 
  6  # This program is free software; you can redistribute it and/or modify it 
  7  # under the terms of the GNU General Public License version 2 as published by 
  8  # the Free Software Foundation. 
  9  # 
 10  # For complete information please visit: http://www.zenoss.com/oss/ 
 11  # 
 12  ########################################################################### 
 13   
 14  import os 
 15  import time 
 16  import sys 
 17   
 18  from twisted.internet import reactor, defer 
 19   
 20  import Globals # make zope imports work 
 21  from Products.ZenHub.PBDaemon import FakeRemote, PBDaemon as Base 
 22  from Products.ZenUtils.Driver import drive, driveLater 
 23  from Products.ZenStatus.ZenTcpClient import ZenTcpClient 
 24  from Products.ZenEvents.ZenEventClasses import Heartbeat 
 25   
 26  # required for pb.setUnjellyableForClass  
 27  import Products.ZenHub.services.StatusConfig 
 28   
 29  from sets import Set 
 30   
 31 -class Status: 
 32      _running = 0 
 33      _fail = 0 
 34      _success = 0 
 35      _start = 0 
 36      _stop = 0 
 37      _defer = None 
 38   
 39 -    def __init__(self): 
 40          self._remaining = [] 
 41   
 42 -    def start(self, jobs): 
 43          self._success = 0 
 44          self._stop = 0 
 45          self._fail = 0 
 46          self._running = 0 
 47          self._remaining = jobs 
 48          self._start = time.time() 
 49          self._defer = defer.Deferred() 
 50          if not self._remaining: 
 51              self._stop = time.time() 
 52              self._defer.callback(self) 
 53          return self._defer 
 54   
 55 -    def next(self): 
 56          j = self._remaining.pop() 
 57          d = j.start() 
 58          d.addCallbacks(self.success, self.failure) 
 59          self._running += 1 
 60          return d 
 61   
 62 -    def testStop(self, result): 
 63          self._running -= 1 
 64          if self.done(): 
 65              self._stop = time.time() 
 66              self._defer, d = None, self._defer 
 67              d.callback(self) 
 68          return result 
 69   
 70 -    def success(self, result): 
 71          self._success += 1  
 72          return self.testStop(result) 
 73   
 74 -    def failure(self, result): 
 75          self._failure += 1 
 76          return self.testStop(result) 
 77   
 78 -    def done(self): 
 79          return self._running == 0 and not self._remaining 
 80   
 81 -    def stats(self): 
 82          return (len(self._remaining), 
 83                  self._running, 
 84                  self._success, 
 85                  self._fail) 
 86   
 87 -    def duration(self): 
 88          if self.done(): 
 89              return self._stop - self._start 
 90          return time.time() - self._start 
 91   
 92   
 93 -class ZenStatus(Base): 
 94   
 95      name = agent = "zenstatus" 
 96      initialServices = ['EventService', 'StatusConfig'] 
 97      statusCycleInterval = 300 
 98      configCycleInterval = 20 
 99      properties = ('statusCycleInterval', 'configCycleInterval') 
100      reconfigureTimeout = None 
101   
102 -    def __init__(self): 
103          Base.__init__(self, keeproot=True) 
104          self.clients = {} 
105          self.counts = {} 
106          self.status = Status() 
107   
108 -    def configService(self): 
109          return self.services.get('StatusConfig', FakeRemote()) 
110   
111 -    def startScan(self, ignored=None): 
112          d = drive(self.scanCycle) 
113          if not self.options.cycle: 
114              d.addBoth(lambda x: self.stop()) 
115   
116 -    def connected(self): 
117          d = drive(self.configCycle) 
118          d.addCallbacks(self.startScan, self.configError) 
119   
120 -    def configError(self, why): 
121          self.log.error(why.getErrorMessage()) 
122          self.stop() 
123   
124 -    def remote_notifyConfigChanged(self): 
125          self.log.debug("Asynch notification of config change") 
126          if self.reconfigureTimeout and not self.reconfigureTimeout.called: 
127              self.reconfigureTimeout.cancel() 
128          self.reconfigureTimeout = reactor.callLater(5, drive, self.reconfigure) 
129   
130 -    def remote_setPropertyItems(self, items): 
131          self.log.debug("Async update of collection properties") 
132          self.setPropertyItems(items) 
133   
134 -    def setPropertyItems(self, items): 
135          'extract configuration elements used by this server' 
136          table = dict(items) 
137          for name in self.properties: 
138              value = table.get(name, None) 
139              if value is not None: 
140                  if getattr(self, name) != value: 
141                      self.log.debug('Updated %s config to %s' % (name, value)) 
142                  setattr(self, name, value) 
143   
144 -    def remote_deleteDevice(self, device): 
145          self.ipservices = [s for s in self.ipservices if s.cfg.device != device] 
146   
147 -    def configCycle(self, driver): 
148          self.log.info("fetching property items") 
149          yield self.configService().callRemote('propertyItems') 
150          self.setPropertyItems(driver.next()) 
151   
152          d = driveLater(self.configCycleInterval * 60, self.configCycle) 
153          d.addErrback(self.error) 
154   
155          yield drive(self.reconfigure) 
156          driver.next() 
157   
158 -    def reconfigure(self, driver): 
159          self.log.debug("Getting service status") 
160          yield self.configService().callRemote('serviceStatus') 
161          self.counts = {} 
162          for (device, component), count in driver.next(): 
163              self.counts[device, component] = count 
164   
165          self.log.debug("Getting services") 
166          yield self.configService().callRemote('services', 
167                                                self.options.configpath) 
168          self.ipservices = [] 
169          for s in driver.next(): 
170              count = self.counts.get((s.device, s.component), 0) 
171              self.ipservices.append(ZenTcpClient(s, count)) 
172          self.log.debug("ZenStatus configured") 
173           
174   
175 -    def error(self, why): 
176          self.log.error(why.getErrorMessage()) 
177   
178 -    def scanCycle(self, driver): 
179          d = driveLater(self.statusCycleInterval, self.scanCycle) 
180          d.addErrback(self.error) 
181   
182          if not self.status.done(): 
183              duration = self.status.duration() 
184              self.log.warning("Scan cycle not complete in %.2f seconds", 
185                               duration) 
186              if duration < self.statusCycleInterval * 2: 
187                  self.log.warning("Waiting for the cycle to complete") 
188                  return 
189              self.log.warning("Ditching this cycle") 
190   
191          self.log.debug("Getting down devices") 
192          yield self.eventService().callRemote('getDevicePingIssues') 
193          ignored = Set([s[0] for s in driver.next()]) 
194   
195          self.log.debug("Starting scan") 
196          d = self.status.start([i for i in self.ipservices 
197                                 if i.cfg.device not in ignored]) 
198          self.log.debug("Running jobs") 
199          self.runSomeJobs() 
200          yield d  
201          driver.next() 
202          self.log.debug("Scan complete") 
203          self.heartbeat() 
204           
205 -    def heartbeat(self): 
206          _, _, success, fail = self.status.stats() 
207          self.log.info("Finished %d jobs (%d good, %d bad) in %.2f seconds", 
208                        (success + fail), success, fail, self.status.duration()) 
209          if not self.options.cycle: 
210              self.stop 
211              return 
212          from socket import getfqdn 
213          heartbeatevt = dict(eventClass=Heartbeat, 
214                              component='zenstatus', 
215                              device=getfqdn()) 
216          self.sendEvent(heartbeatevt, timeout=self.statusCycleInterval*3) 
217   
218   
219 -    def runSomeJobs(self): 
220          while 1: 
221              left, running, good, bad = self.status.stats() 
222              self.log.debug("Status: left %d running %d good %d bad %d", 
223                             left, running, good, bad) 
224              if not left or running >= self.options.parallel: 
225                  break 
226              d = self.status.next() 
227              d.addCallbacks(self.processTest, self.processError) 
228              self.log.debug("Started job") 
229   
230 -    def processTest(self, job): 
231          self.runSomeJobs() 
232          key = job.cfg.device, job.cfg.component 
233          evt = job.getEvent() 
234          if evt: 
235              self.sendEvent(evt) 
236              self.counts.setdefault(key, 0) 
237              self.counts[key] += 1 
238          else: 
239              if key in self.counts: 
240                  del self.counts[key]  
241           
242 -    def processError(self, error): 
243          self.log.warn(error.getErrorMessage()) 
244   
245 -    def buildOptions(self): 
246          Base.buildOptions(self) 
247          p = self.parser 
248          p.add_option('--configpath', 
249                       dest='configpath', 
250                       default="/Devices/Server", 
251                       help="path to our monitor config ie: /Devices/Server") 
252          p.add_option('--parallel', 
253                       dest='parallel',  
254                       type='int', 
255                       default=50, 
256                       help="number of devices to collect at one time") 
257          p.add_option('--cycletime', 
258                       dest='cycletime', 
259                       type="int", 
260                       default=60, 
261                       help="check events every cycletime seconds") 
262   
263   
264  if __name__=='__main__': 
265      pm = ZenStatus() 
266      pm.run() 
267