Source Code for Module Products.ZenUtils.Watchdog

  1  #! /usr/bin/env python  
  2  ############################################################################## 
  3  #  
  4  # Copyright (C) Zenoss, Inc. 2007, 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__='''watchdog for zenoss daemons 
 13   
 14  Run a program that is expected to run forever.  If the program stops, 
 15  restart it. 
 16   
 17  ''' 
 18   
 19  import Globals 
 20  from Products.ZenUtils.Utils import zenPath 
 21  import logging 
 22   
 23  import socket as s 
 24  import os, sys, time, signal, select 
 25   
 26 -class TimeoutError(Exception): pass 
 27 -class UnexpectedFailure(Exception): pass 
 28   
 29  log = logging.getLogger('watchdog')     
 30   
 31  # time to spend waiting around for a child to die after we kill it 
 32  DEATH_WATCH_TIME = 10 
 33   
 34  # time to wait between tests of a childs imminent death 
 35  BUSY_WAIT_SLEEP = 0.5 
 36   
 37 -def _sleep(secs): 
 38      "Sleep, but don't raise an exception if interrupted" 
 39      try: 
 40          time.sleep(secs) 
 41      except: 
 42          pass 
 43   
 44 -class ExitStatus: 
 45      "Model a child's exit status" 
 46 -    def __init__(self, status): 
 47          self.status = status 
 48   
 49 -    def __str__(self): 
 50          if self.signaled(): 
 51              return 'Killed with signal %d' % self.signal() 
 52          return 'Exited with code %d' % self.exitCode() 
 53   
 54 -    def __repr__(self): 
 55          return '<ExitStatus %d (%s)>' % (self.status, self) 
 56   
 57 -    def signaled(self): 
 58          return os.WIFSIGNALED(self.status) 
 59   
 60 -    def exitCode(self): 
 61          if self.signaled(): 
 62              raise ValueError(str(self)) 
 63          return os.WEXITSTATUS(self.status) 
 64   
 65 -    def signal(self): 
 66          if not self.signaled(): 
 67              raise ValueError(str(self)) 
 68          return os.WTERMSIG(self.status) 
 69           
 70   
 71 -class Watcher: 
 72      """Run the given command, and expect periodic input on a shared 
 73      UNIX-domain socket.  If the command does not connect to the socket 
 74      in startTimeout seconds, or it does not report every cycleTimeout 
 75      seconds, then the process is restarted.  The Watchdog will 
 76      increase the time-between restarts until maxTime is achieved""" 
 77   
 78 -    def __init__(self, 
 79                   socketPath, 
 80                   cmd, 
 81                   startTimeout = None, 
 82                   cycleTimeout = 1, 
 83                   maxTime = 30): 
 84          if startTimeout == None: 
 85              startTimeout = 120 
 86          self.socketPath = socketPath 
 87          self.cmd = cmd 
 88          self.startTimeout = startTimeout 
 89          self.cycleTimeout = cycleTimeout 
 90          self.maxTime = maxTime 
 91          self.stop = False 
 92          self.childPid = -1 
 93   
 94 -    def _kill(self): 
 95          """Send a signal to a process and wait for it to stop.  Use 
 96          progressively more serious signals to get it to stop. 
 97          """ 
 98          if self.childPid <= 0: 
 99              return 
100          signals = signal.SIGINT, signal.SIGTERM, signal.SIGKILL 
101          for sig in signals: 
102              log.debug("Killing %d with %d", self.childPid, sig) 
103              os.kill(self.childPid, sig) 
104              stopTime = time.time() + DEATH_WATCH_TIME / len(signals) 
105              while time.time() < stopTime: 
106                  try: 
107                      pid, status = os.waitpid(self.childPid, os.WNOHANG) 
108                      if pid: 
109                          return ExitStatus(status) 
110                  except os.error: 
111                      pass 
112                  _sleep(BUSY_WAIT_SLEEP) 
113   
114 -    def _readWait(self, sock, timeout): 
115          "Wait for a file descriptor to become readable" 
116          endTime = time.time() + timeout 
117          # Loop because signals can cause select stop early 
118          while not self.stop and time.time() < endTime: 
119              diff = endTime - time.time() 
120              try: 
121                  log.debug("waiting %f seconds" % diff) 
122                  rd, wr, ex = select.select([sock], [], [], diff) 
123              except Exception: 
124                  continue 
125              if rd: 
126                  return sock 
127          return None 
128   
129 -    def _runOnce(self): 
130          try: 
131              if os.path.exists(self.socketPath): 
132                  os.unlink(self.socketPath) 
133          except OSError: 
134              log.exception("Problem removing old socket %s" % self.socketPath) 
135          cmd = self.cmd + ['--watchdogPath', self.socketPath] 
136          cmd.insert(0, sys.executable) 
137          sock = s.socket(s.AF_UNIX, s.SOCK_STREAM) 
138          sock.bind(self.socketPath) 
139          self.childPid = os.fork() 
140          if self.childPid < 0: 
141              log.error("Unable to fork") 
142              return 
143          if self.childPid == 0: 
144              # child 
145              try: 
146                  log.debug('Running %r' % (cmd,)) 
147                  os.execlp(cmd[0], *cmd) 
148              except: 
149                  log.exception("Exec failed!") 
150                  sys.exit(0) 
151          try: 
152              sock.setblocking(False) 
153              sock.listen(1) 
154              if not self._readWait(sock, self.startTimeout): 
155                  if not self.stop: 
156                      raise TimeoutError("getting initial connection from process") 
157              log.debug('Waiting for command to connect %r' % (cmd,)) 
158              conn, addr = sock.accept() 
159              conn.setblocking(False) 
160              try: 
161                  buf = '' 
162                  while not self.stop: 
163                      # get input from the child 
164                      if not self._readWait(conn, self.cycleTimeout * 2): 
165                          if not self.stop: 
166                              raise TimeoutError("getting status from process") 
167                      try: 
168                          bytes = conn.recv(1024) 
169                      except Exception: 
170                          continue 
171                      if bytes == '':          # EOF 
172                          pid, status = os.waitpid(self.childPid, os.WNOHANG) 
173                          if pid == self.childPid: 
174                              status = ExitStatus(status) 
175                              self.childPid = -1 
176                              if status.signaled(): 
177                                  raise UnexpectedFailure(status) 
178                              if status.exitCode() != 0: 
179                                  log.error("Child exited with status %d" % 
180                                            status.exitCode()) 
181                                  raise UnexpectedFailure(status) 
182                              return 
183                          else: 
184                              _sleep(0.1) 
185                              continue 
186                      # interpret the data as an updated cycleTime 
187                      buf += bytes 
188                      lines = buf.split('\n') 
189                      if lines: 
190                          buf = lines[-1] 
191                          line = lines[0] 
192                          if line: 
193                              log.debug("Child sent %s" % line) 
194                              try: 
195                                  self.cycleTimeout = max(int(line), 1) 
196                                  log.debug("Watchdog cycleTimeout is %d", 
197                                            self.cycleTimeout) 
198                              except ValueError: 
199                                  log.exception("Unable to convert cycleTime") 
200              finally: 
201                  conn.close() 
202          finally: 
203              os.unlink(self.socketPath) 
204              self._kill() 
205   
206 -    def _stop(self, *unused): 
207          self.stop = True 
208   
209 -    def run(self): 
210          sleepTime = 1 
211          signal.signal(signal.SIGINT, self._stop) 
212          while not self.stop: 
213              try: 
214                  self._runOnce() 
215                  return 
216              except TimeoutError, ex: 
217                  log.error("Timeout: %s" % ex.args) 
218              except UnexpectedFailure, ex: 
219                  status = ex.args[0] 
220                  log.error("Child died: %s" % status) 
221              except Exception, ex: 
222                  log.exception(ex) 
223              if not self.stop: 
224                  log.debug("Waiting %.2f seconds before restarting", sleepTime) 
225                  _sleep(sleepTime) 
226                  prog = self.cmd[0].split('/')[-1].split('.')[0] 
227                  log.error("Restarting %s" % prog) 
228              sleepTime = min(1.5 * sleepTime, self.maxTime) 
229   
230 -class Reporter: 
231 -    def __init__(self, path): 
232          self.sock = s.socket(s.AF_UNIX, s.SOCK_STREAM) 
233          self.sock.connect(path) 
234          self.sock.setblocking(False) 
235   
236 -    def niceDoggie(self, cycleTime): 
237          cycleTime = max(1, int(cycleTime)) 
238          try: 
239              try: 
240                  self.sock.recv(1) 
241                  log.error("Received input on report socket: probably EOF") 
242                  sys.exit(1) 
243              except: 
244                  pass 
245              self.sock.send('%d\n' % cycleTime) 
246          except Exception: 
247              log.exception("Unable to report to the watchdog.") 
248              sys.exit(1) 
249   
250 -    def close(self): 
251          self.sock.close() 
252   
253 -def main(): 
254      '''Little test for the Watchdog. 
255      Usage: 
256            python Watchdog.py -- python Watchdog.py -e 1 
257   
258      This will repeatedly run a child that exits with exit code 1. 
259      The child does periodic reports over the watchdog socket. 
260      ''' 
261      import getopt 
262      global log 
263      opts, cmd = getopt.getopt(sys.argv[1:], 'p:m:d:e:c:', 'watchdogPath=') 
264      socketPath = 'watchdog.%d' % os.getpid() 
265      maxTime = 30 
266      cycleTime = 1 
267      level = 20 
268      child = None 
269      exitCode = 0 
270      for opt, arg in opts: 
271          if opt == '-p': 
272              socketPath = arg 
273          if opt == '-e': 
274              exitCode = int(arg) 
275          if opt == '-m': 
276              maxTime = float(arg) 
277          if opt == '-d': 
278              level = int(arg) 
279          if opt == '-c': 
280              cycleTime = int(arg) 
281          if opt == '--watchdogPath': 
282              socketPath = arg 
283              child = True 
284      if not child: 
285          socketPath = zenPath('var', socketPath) 
286   
287      logging.basicConfig(level=level) 
288      log = logging.getLogger('watchdog') 
289   
290      if child: 
291          r = Reporter(socketPath) 
292          print 'Connected' 
293          for i in range(3): 
294              time.sleep(1) 
295              r.niceDoggie(1) 
296              sys.stdout.write('*') 
297              sys.stdout.flush() 
298          time.sleep(2) 
299          r.close() 
300          print 'Closed' 
301          sys.exit(exitCode) 
302      else: 
303          w = Watcher(socketPath, cmd, cycleTimeout=cycleTime, maxTime=maxTime) 
304          w.run() 
305   
306  if __name__ == '__main__': 
307      main() 
308   
309  __all__ = ['Watcher', 'Reporter'] 
310