Source Code for Module Products.ZenEvents.EventServer

  1  ############################################################################## 
  2  #  
  3  # Copyright (C) Zenoss, Inc. 2007, 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__='''EventServer 
 12   
 13  Base class for ZenSyslog, ZenTrap and others 
 14   
 15  $Id$ 
 16  ''' 
 17   
 18  __version__ = "$Revision$"[11:-2] 
 19   
 20  import socket 
 21   
 22  import Globals 
 23   
 24  from Products.ZenHub.PBDaemon import PBDaemon, FakeRemote 
 25  from Products.ZenUtils.DaemonStats import DaemonStats 
 26  from Products.ZenUtils.Driver import drive 
 27   
 28  from Products.ZenEvents.ZenEventClasses import App_Start 
 29  from twisted.internet import reactor 
 30   
 31  import time 
 32   
 33 -class Stats: 
 34      totalTime = 0. 
 35      totalEvents = 0 
 36      maxTime = 0. 
 37       
 38 -    def add(self, moreTime): 
 39          self.totalEvents += 1 
 40          self.totalTime += moreTime 
 41          self.maxTime = max(self.maxTime, moreTime) 
 42   
 43 -    def report(self): 
 44          return self.totalTime, self.totalEvents, self.maxTime 
 45   
 46 -class EventServer(PBDaemon): 
 47      'Base class for a daemon whose primary job is to post events' 
 48   
 49      name = 'EventServer' 
 50       
 51 -    def __init__(self): 
 52          PBDaemon.__init__(self, keeproot=True) 
 53          self.stats = Stats() 
 54          self.rrdStats = DaemonStats() 
 55   
 56   
 57 -    def connected(self): 
 58          self.sendEvent(dict(device=self.options.monitor,  
 59                              eventClass=App_Start,  
 60                              summary="%s started" % self.name, 
 61                              severity=0, 
 62                              component=self.name)) 
 63          self.log.info("started") 
 64          self.configure() 
 65   
 66   
 67 -    def model(self): 
 68          return self.services.get('EventService', FakeRemote()) 
 69   
 70   
 71 -    def configure(self): 
 72          def inner(driver): 
 73              self.log.info("fetching default RRDCreateCommand") 
 74              yield self.model().callRemote('getDefaultRRDCreateCommand') 
 75              createCommand = driver.next() 
 76           
 77              self.log.info("getting threshold classes") 
 78              yield self.model().callRemote('getThresholdClasses') 
 79              self.remote_updateThresholdClasses(driver.next()) 
 80           
 81              self.log.info("getting collector thresholds") 
 82              yield self.model().callRemote('getCollectorThresholds') 
 83              self.rrdStats.config(self.options.monitor, self.name, 
 84                                   driver.next(), createCommand) 
 85              self.heartbeat() 
 86              self.reportCycle() 
 87          d = drive(inner) 
 88          def error(result): 
 89              self.log.error("Unexpected error in configure: %s" % result) 
 90          d.addErrback(error) 
 91          return d 
 92   
 93   
 94 -    def sendEvent(self, event, **kw): 
 95          # FIXME: get real event processing stats 
 96          if 'firstTime' in event: 
 97              self.stats.add(min(time.time() - event['firstTime'], 0)) 
 98          PBDaemon.sendEvent(self, event, **kw) 
 99   
100   
101 -    def useUdpFileDescriptor(self, fd): 
102          from twisted.internet import udp 
103          s = socket.fromfd(fd, socket.AF_INET, socket.SOCK_DGRAM) 
104          import os 
105          os.close(fd) 
106          port = s.getsockname()[1] 
107          transport = udp.Port(port, self) 
108          s.setblocking(0) 
109          transport.socket = s 
110          transport.fileno = s.fileno 
111          transport.connected = 1 
112          transport._realPortNumber = port 
113          self.transport = transport 
114          # hack around startListening not being called 
115          self.numPorts = 1 
116          transport.startReading() 
117   
118   
119 -    def useTcpFileDescriptor(self, fd, factory): 
120          import os, socket 
121          for i in range(19800, 19999): 
122              try: 
123                  p = reactor.listenTCP(i, factory) 
124                  os.dup2(fd, p.socket.fileno()) 
125                  p.socket.listen(p.backlog) 
126                  p.socket.setblocking(False) 
127                  p.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 
128                  os.close(fd) 
129                  return p 
130              except socket.error: 
131                  pass 
132          raise socket.error("Unable to find an open socket to listen on") 
133   
134   
135 -    def reportCycle(self): 
136          if self.options.statcycle: 
137              self.report() 
138              reactor.callLater(self.options.statcycle, self.reportCycle) 
139   
140   
141 -    def heartbeat(self): 
142          """Since we don't do anything on a regular basis, just 
143          push heartbeats regularly""" 
144          seconds = self.heartbeatTimeout / 3 
145          reactor.callLater(self.heartbeatTimeout / 3, self.heartbeat) 
146          PBDaemon.heartbeat(self) 
147          totalTime, totalEvents, maxTime = self.stats.report() 
148          for ev in (self.rrdStats.counter('events', 
149                                           seconds, 
150                                           totalEvents) + 
151                     self.rrdStats.counter('totalTime', 
152                                           seconds, 
153                                           int(totalTime * 1000))): 
154              self.sendEvent(ev) 
155   
156           
157 -    def report(self): 
158          'report some simple diagnostics at shutdown' 
159          totalTime, totalEvents, maxTime = self.stats.report() 
160          self.log.info("%d events processed in %.2f seconds", 
161                        totalEvents, 
162                        totalTime) 
163          if totalEvents > 0: 
164              self.log.info("%.5f average seconds per event", 
165                         (totalTime / totalEvents)) 
166              self.log.info("Maximum processing time for one event was %.5f", 
167                            maxTime) 
168   
169   
170 -    def buildOptions(self): 
171          PBDaemon.buildOptions(self) 
172          self.parser.add_option('--statcycle', 
173                                 dest='statcycle', 
174                                 type='int', 
175                                 help='Number of seconds between the writing of statistics', 
176                                 default=0) 
177