twisted/runner/procmon.py

   1 # -*- test-case-name: twisted.runner.test.test_procmon -*-
   2 # Copyright (c) Twisted Matrix Laboratories.
   3 # See LICENSE for details.
   4
   5 """
   6 Support for starting, monitoring, and restarting child process.
   7 """
   8 import warnings
   9
  10 from twisted.python import log
  11 from twisted.internet import error, protocol, reactor as _reactor
  12 from twisted.application import service
  13 from twisted.protocols import basic
  14
  15 class DummyTransport:
  16
  17     disconnecting = 0
  18
  19 transport = DummyTransport()
  20
  21 class LineLogger(basic.LineReceiver):
  22
  23     tag = None
  24     delimiter = '\n'
  25
  26     def lineReceived(self, line):
  27         log.msg('[%s] %s' % (self.tag, line))
  28
  29
  30 class LoggingProtocol(protocol.ProcessProtocol):
  31
  32     service = None
  33     name = None
  34     empty = 1
  35
  36     def connectionMade(self):
  37         self.output = LineLogger()
  38         self.output.tag = self.name
  39         self.output.makeConnection(transport)
  40
  41
  42     def outReceived(self, data):
  43         self.output.dataReceived(data)
  44         self.empty = data[-1] == '\n'
  45
  46     errReceived = outReceived
  47
  48
  49     def processEnded(self, reason):
  50         if not self.empty:
  51             self.output.dataReceived('\n')
  52         self.service.connectionLost(self.name)
  53
  54
  55 class ProcessMonitor(service.Service):
  56     """
  57     ProcessMonitor runs processes, monitors their progress, and restarts
  58     them when they die.
  59
  60     The ProcessMonitor will not attempt to restart a process that appears to
  61     die instantly -- with each "instant" death (less than 1 second, by
  62     default), it will delay approximately twice as long before restarting
  63     it.  A successful run will reset the counter.
  64
  65     The primary interface is L{addProcess} and L{removeProcess}. When the
  66     service is running (that is, when the application it is attached to is
  67     running), adding a process automatically starts it.
  68
  69     Each process has a name. This name string must uniquely identify the
  70     process.  In particular, attempting to add two processes with the same
  71     name will result in a C{KeyError}.
  72
  73     @type threshold: C{float}
  74     @ivar threshold: How long a process has to live before the death is
  75         considered instant, in seconds.  The default value is 1 second.
  76
  77     @type killTime: C{float}
  78     @ivar killTime: How long a process being killed has to get its affairs
  79         in order before it gets killed with an unmaskable signal.  The
  80         default value is 5 seconds.
  81
  82     @type minRestartDelay: C{float}
  83     @ivar minRestartDelay: The minimum time (in seconds) to wait before
  84         attempting to restart a process.  Default 1s.
  85
  86     @type maxRestartDelay: C{float}
  87     @ivar maxRestartDelay: The maximum time (in seconds) to wait before
  88         attempting to restart a process.  Default 3600s (1h).
  89
  90     @type _reactor: L{IReactorProcess} provider
  91     @ivar _reactor: A provider of L{IReactorProcess} and L{IReactorTime}
  92         which will be used to spawn processes and register delayed calls.
  93
  94     """
  95     threshold = 1
  96     killTime = 5
  97     minRestartDelay = 1
  98     maxRestartDelay = 3600
  99
 100
 101     def __init__(self, reactor=_reactor):
 102         self._reactor = reactor
 103
 104         self.processes = {}
 105         self.protocols = {}
 106         self.delay = {}
 107         self.timeStarted = {}
 108         self.murder = {}
 109         self.restart = {}
 110
 111
 112     def __getstate__(self):
 113         dct = service.Service.__getstate__(self)
 114         del dct['_reactor']
 115         dct['protocols'] = {}
 116         dct['delay'] = {}
 117         dct['timeStarted'] = {}
 118         dct['murder'] = {}
 119         dct['restart'] = {}
 120         return dct
 121
 122
 123     def addProcess(self, name, args, uid=None, gid=None, env={}):
 124         """
 125         Add a new monitored process and start it immediately if the
 126         L{ProcessMonitor} service is running.
 127
 128         Note that args are passed to the system call, not to the shell. If
 129         running the shell is desired, the common idiom is to use
 130         C{ProcessMonitor.addProcess("name", ['/bin/sh', '-c', shell_script])}
 131
 132         @param name: A name for this process.  This value must be
 133             unique across all processes added to this monitor.
 134         @type name: C{str}
 135         @param args: The argv sequence for the process to launch.
 136         @param uid: The user ID to use to run the process.  If C{None},
 137             the current UID is used.
 138         @type uid: C{int}
 139         @param gid: The group ID to use to run the process.  If C{None},
 140             the current GID is used.
 141         @type uid: C{int}
 142         @param env: The environment to give to the launched process. See
 143             L{IReactorProcess.spawnProcess}'s C{env} parameter.
 144         @type env: C{dict}
 145         @raises: C{KeyError} if a process with the given name already
 146             exists
 147         """
 148         if name in self.processes:
 149             raise KeyError("remove %s first" % (name,))
 150         self.processes[name] = args, uid, gid, env
 151         self.delay[name] = self.minRestartDelay
 152         if self.running:
 153             self.startProcess(name)
 154
 155
 156     def removeProcess(self, name):
 157         """
 158         Stop the named process and remove it from the list of monitored
 159         processes.
 160
 161         @type name: C{str}
 162         @param name: A string that uniquely identifies the process.
 163         """
 164         self.stopProcess(name)
 165         del self.processes[name]
 166
 167
 168     def startService(self):
 169         """
 170         Start all monitored processes.
 171         """
 172         service.Service.startService(self)
 173         for name in self.processes:
 174             self.startProcess(name)
 175
 176
 177     def stopService(self):
 178         """
 179         Stop all monitored processes and cancel all scheduled process restarts.
 180         """
 181         service.Service.stopService(self)
 182
 183         # Cancel any outstanding restarts
 184         for name, delayedCall in self.restart.items():
 185             if delayedCall.active():
 186                 delayedCall.cancel()
 187
 188         for name in self.processes:
 189             self.stopProcess(name)
 190
 191
 192     def connectionLost(self, name):
 193         """
 194         Called when a monitored processes exits. If
 195         L{ProcessMonitor.running} is C{True} (ie the service is started), the
 196         process will be restarted.
 197         If the process had been running for more than
 198         L{ProcessMonitor.threshold} seconds it will be restarted immediately.
 199         If the process had been running for less than
 200         L{ProcessMonitor.threshold} seconds, the restart will be delayed and
 201         each time the process dies before the configured threshold, the restart
 202         delay will be doubled - up to a maximum delay of maxRestartDelay sec.
 203
 204         @type name: C{str}
 205         @param name: A string that uniquely identifies the process
 206             which exited.
 207         """
 208         # Cancel the scheduled _forceStopProcess function if the process
 209         # dies naturally
 210         if name in self.murder:
 211             if self.murder[name].active():
 212                 self.murder[name].cancel()
 213             del self.murder[name]
 214
 215         del self.protocols[name]
 216
 217         if self._reactor.seconds() - self.timeStarted[name] < self.threshold:
 218             # The process died too fast - backoff
 219             nextDelay = self.delay[name]
 220             self.delay[name] = min(self.delay[name] * 2, self.maxRestartDelay)
 221
 222         else:
 223             # Process had been running for a significant amount of time
 224             # restart immediately
 225             nextDelay = 0
 226             self.delay[name] = self.minRestartDelay
 227
 228         # Schedule a process restart if the service is running
 229         if self.running and name in self.processes:
 230             self.restart[name] = self._reactor.callLater(nextDelay,
 231                                                          self.startProcess,
 232                                                          name)
 233
 234
 235     def startProcess(self, name):
 236         """
 237         @param name: The name of the process to be started
 238         """
 239         # If a protocol instance already exists, it means the process is
 240         # already running
 241         if name in self.protocols:
 242             return
 243
 244         args, uid, gid, env = self.processes[name]
 245
 246         proto = LoggingProtocol()
 247         proto.service = self
 248         proto.name = name
 249         self.protocols[name] = proto
 250         self.timeStarted[name] = self._reactor.seconds()
 251         self._reactor.spawnProcess(proto, args[0], args, uid=uid,
 252                                           gid=gid, env=env)
 253
 254
 255     def _forceStopProcess(self, proc):
 256         """
 257         @param proc: An L{IProcessTransport} provider
 258         """
 259         try:
 260             proc.signalProcess('KILL')
 261         except error.ProcessExitedAlready:
 262             pass
 263
 264
 265     def stopProcess(self, name):
 266         """
 267         @param name: The name of the process to be stopped
 268         """
 269         if name not in self.processes:
 270             raise KeyError('Unrecognized process name: %s' % (name,))
 271
 272         proto = self.protocols.get(name, None)
 273         if proto is not None:
 274             proc = proto.transport
 275             try:
 276                 proc.signalProcess('TERM')
 277             except error.ProcessExitedAlready:
 278                 pass
 279             else:
 280                 self.murder[name] = self._reactor.callLater(
 281                                             self.killTime,
 282                                             self._forceStopProcess, proc)
 283
 284
 285     def restartAll(self):
 286         """
 287         Restart all processes. This is useful for third party management
 288         services to allow a user to restart servers because of an outside change
 289         in circumstances -- for example, a new version of a library is
 290         installed.
 291         """
 292         for name in self.processes:
 293             self.stopProcess(name)
 294
 295
 296     def __repr__(self):
 297         l = []
 298         for name, proc in self.processes.items():
 299             uidgid = ''
 300             if proc[1] is not None:
 301                 uidgid = str(proc[1])
 302             if proc[2] is not None:
 303                 uidgid += ':'+str(proc[2])
 304
 305             if uidgid:
 306                 uidgid = '(' + uidgid + ')'
 307             l.append('%r%s: %r' % (name, uidgid, proc[0]))
 308         return ('<' + self.__class__.__name__ + ' '
 309                 + ' '.join(l)
 310                 + '>')