Source Code for Module Products.ZenUtils.PBUtil

  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  # taken from r1.10 of buildbot.sf.net/buildbot/pbutil.py 
 12   
 13  # flumotion has nearly the same code 
 14   
 15  __doc__ = """PBUtil 
 16  Base classes handy for use with PB clients. 
 17  """ 
 18   
 19  import logging 
 20  zenlog = logging.getLogger("zen.pbclientfactory") 
 21   
 22  from twisted.spread import pb 
 23   
 24  from twisted.spread.pb import PBClientFactory 
 25  from twisted.internet import protocol, reactor, defer, task 
 26  from twisted.internet.error import ConnectionClosed 
 27   
 28 -class ReconnectingPBClientFactory(PBClientFactory, 
 29                                    protocol.ReconnectingClientFactory): 
 30      """Reconnecting client factory for PB brokers. 
 31   
 32      Like PBClientFactory, but if the connection fails or is lost, the factory 
 33      will attempt to reconnect. 
 34   
 35      Instead of using f.getRootObject (which gives a Deferred that can only 
 36      be fired once), override the gotRootObject method. 
 37   
 38      Instead of using the newcred f.login (which is also one-shot), call 
 39      f.startLogin() with the credentials and client, and override the 
 40      gotPerspective method. 
 41   
 42      Instead of using the oldcred f.getPerspective (also one-shot), call 
 43      f.startGettingPerspective() with the same arguments, and override 
 44      gotPerspective. 
 45   
 46      gotRootObject and gotPerspective will be called each time the object is 
 47      received (once per successful connection attempt). You will probably want 
 48      to use obj.notifyOnDisconnect to find out when the connection is lost. 
 49   
 50      If an authorization error occurs, failedToGetPerspective() will be 
 51      invoked. 
 52   
 53      To use me, subclass, then hand an instance to a connector (like 
 54      TCPClient). 
 55      """ 
 56      __pychecker__='no-override' 
 57   
 58      # maxDelay(secs) set to 5 minute maximum delay before attempting to 
 59      # reconnect 
 60      maxDelay = 300 
 61   
 62 -    def __init__(self, connectTimeout=30, pingPerspective=False, pingInterval=30, pingtimeout=120): 
 63          PBClientFactory.__init__(self) 
 64          self._doingLogin = False 
 65          self._doingGetPerspective = False 
 66          self._scheduledConnectTimeout = None 
 67          self._connectTimeout = connectTimeout 
 68          # should the perspective be pinged. Perspective must have a ping method 
 69          self._shouldPingPerspective = pingPerspective 
 70          # how often to ping 
 71          self._pingInterval = pingInterval 
 72          # how long to wait for a ping before closing connection 
 73          self._pingTimeoutTime = pingtimeout 
 74          # ref to the scheduled ping timeout call 
 75          self._pingTimeout = None 
 76          # looping call doing the ping 
 77          self._pingCheck = None 
 78   
 79          self._perspective = None 
 80   
 81 -    def connectTCP(self, host, port): 
 82          factory = self 
 83          self.connector = reactor.connectTCP(host, port, factory) 
 84          return self.connector 
 85   
 86 -    def clientConnectionFailed(self, connector, reason): 
 87          zenlog.debug("Failed to create connection to %s:%s - %s", 
 88                       connector.host, connector.port, reason) 
 89          self._perspective = None 
 90          self._cancelConnectTimeout() 
 91          PBClientFactory.clientConnectionFailed(self, connector, reason) 
 92          # Twisted-1.3 erroneously abandons the connection on non-UserErrors. 
 93          # To avoid this bug, don't upcall, and implement the correct version 
 94          # of the method here. 
 95          if self.continueTrying: 
 96              self.connector = connector 
 97              self.retry() 
 98   
 99 -    def clientConnectionLost(self, connector, reason, reconnecting=1): 
100          zenlog.debug("Lost connection to %s:%s - %s", connector.host, 
101                       connector.port, reason) 
102          self._perspective = None 
103          self._cancelConnectTimeout() 
104          PBClientFactory.clientConnectionLost(self, connector, reason, 
105                                               reconnecting=reconnecting) 
106          RCF = protocol.ReconnectingClientFactory 
107          RCF.clientConnectionLost(self, connector, reason) 
108   
109 -    def clientConnectionMade(self, broker): 
110          zenlog.debug("Connected") 
111          self._cancelConnectTimeout() 
112          self.resetDelay() 
113          PBClientFactory.clientConnectionMade(self, broker) 
114          if self._doingLogin: 
115              self._startConnectTimeout("Login") 
116              self.doLogin(self._root) 
117          if self._doingGetPerspective: 
118              self.doGetPerspective(self._root) 
119          self.gotRootObject(self._root) 
120   
121 -    def startedConnecting(self, connector): 
122          zenlog.debug("Starting connection...") 
123          self._startConnectTimeout("Initial connect") 
124          self.connecting() 
125   
126 -    def __getstate__(self): 
127          # this should get folded into ReconnectingClientFactory 
128          d = self.__dict__.copy() 
129          d['connector'] = None 
130          d['_callID'] = None 
131          return d 
132   
133      # oldcred methods 
134   
135 -    def getPerspective(self, *args): 
136          raise RuntimeError( "getPerspective is one-shot: use startGettingPerspective instead" ) 
137   
138 -    def startGettingPerspective(self, username, password, serviceName, 
139                                  perspectiveName=None, client=None): 
140          self._doingGetPerspective = True 
141          if perspectiveName == None: 
142              perspectiveName = username 
143          self._oldcredArgs = (username, password, serviceName, 
144                               perspectiveName, client) 
145   
146 -    def doGetPerspective(self, root): 
147          # oldcred getPerspective() 
148          (username, password, 
149           serviceName, perspectiveName, client) = self._oldcredArgs 
150          d = self._cbAuthIdentity(root, username, password) 
151          d.addCallback(self._cbGetPerspective, 
152                        serviceName, perspectiveName, client) 
153          d.addCallbacks(self._gotPerspective, self.failedToGetPerspective) 
154   
155   
156      # newcred methods 
157   
158 -    def login(self, credentials, client=None): 
159          from Products.ZenUtils.Utils import unused 
160          unused(credentials, client) 
161          raise RuntimeError( "Login is one-shot: use startLogin instead" ) 
162   
163 -    def startLogin(self, credentials, client=None): 
164          self._credentials = credentials 
165          self._client = client 
166          self._doingLogin = True 
167   
168 -    def doLogin(self, root): 
169          # newcred login() 
170          zenlog.debug("Sending credentials") 
171          d = self._cbSendUsername(root, self._credentials.username, 
172                                   self._credentials.password, self._client) 
173          d.addCallbacks(self._gotPerspective, self.failedToGetPerspective) 
174          return d 
175   
176 -    def _gotPerspective(self, perspective): 
177          self._cancelConnectTimeout() 
178          self._cancelPingTimeout() 
179          self._perspective = perspective 
180          if self._shouldPingPerspective: 
181              reactor.callLater(0, self._startPingCycle) 
182          self.gotPerspective(perspective) 
183   
184   
185 -    def _disconnect(self): 
186          if self._broker: 
187              self.disconnect() 
188          elif self.connector: 
189              try: 
190                  self.connector.disconnect() 
191              except Exception: 
192                  zenlog.exception('Could not disconnect') 
193          else: 
194              zenlog.debug('No connector or broker to disconnect') 
195   
196      # methods for connecting and login timeout 
197 -    def _startConnectTimeout(self, msg): 
198          self._cancelConnectTimeout() 
199          self._scheduledConnectTimeout = reactor.callLater(self._connectTimeout, self._timeoutConnect, msg) 
200   
201 -    def _timeoutConnect(self, msg): 
202          zenlog.info("%s timed out after %s seconds", msg, self._connectTimeout) 
203          self._disconnect() 
204   
205 -    def _cancelConnectTimeout(self): 
206          self._scheduledConnectTimeout, timeout = None, self._scheduledConnectTimeout 
207          if timeout and timeout.active(): 
208              zenlog.debug("Cancelling connect timeout") 
209              timeout.cancel() 
210   
211      # methods to check connection is active 
212 -    def _startPingTimeout(self): 
213          if not self._pingTimeout: 
214              self._pingTimeout = reactor.callLater(self._pingTimeoutTime, 
215                  self._doPingTimeout) 
216   
217 -    def _cancelPingTimeout(self): 
218          self._pingTimeout, timeout = None, self._pingTimeout 
219          if timeout and timeout.active(): 
220              zenlog.debug("Cancelling ping timeout") 
221              timeout.cancel() 
222   
223 -    def _doPingTimeout(self): 
224          if self._perspective: 
225              zenlog.warn("Perspective ping timed out after %s seconds", self._pingTimeoutTime) 
226              self._disconnect() 
227   
228      @defer.inlineCallbacks 
229 -    def _startPingCycle(self): 
230          if not self._pingCheck: 
231              pingCheck = task.LoopingCall(self._pingPerspective) 
232              self._pingCheck = pingCheck 
233              try: 
234                  yield pingCheck.start(self._pingInterval) 
235              except Exception: 
236                  zenlog.exception("perspective ping loop died") 
237              finally: 
238                  # should only happen at shutdown 
239                  zenlog.info("perspective ping loop ended") 
240   
241      @defer.inlineCallbacks 
242 -    def _pingPerspective(self): 
243          try: 
244              if self._perspective: 
245                  zenlog.debug('pinging perspective') 
246                  self._startPingTimeout() 
247                  response = yield self._perspective.callRemote('ping') 
248                  zenlog.debug("perspective %sed", response) 
249              else: 
250                  zenlog.debug('skipping perspective ping') 
251              self._cancelPingTimeout() 
252          except ConnectionClosed: 
253              zenlog.info("Connection was closed") 
254              self._cancelPingTimeout() 
255          except Exception: 
256              zenlog.exception("ping perspective exception") 
257   
258      # methods to override 
259   
260 -    def connecting(self): 
261          """ 
262          Called when a connection is about to be attempted. Can be the initial 
263          connect or a retry/reconnect 
264          """ 
265          pass 
266   
267 -    def gotPerspective(self, perspective): 
268          """The remote avatar or perspective (obtained each time this factory 
269          connects) is now available.""" 
270          pass 
271   
272 -    def gotRootObject(self, root): 
273          """The remote root object (obtained each time this factory connects) 
274          is now available. This method will be called each time the connection 
275          is established and the object reference is retrieved.""" 
276          pass 
277   
278 -    def failedToGetPerspective(self, why): 
279          """The login process failed, most likely because of an authorization 
280          failure (bad password), but it is also possible that we lost the new 
281          connection before we managed to send our credentials. 
282          """ 
283          self._cancelConnectTimeout() 
284          zenlog.debug("ReconnectingPBClientFactory.failedToGetPerspective") 
285          if why.check(pb.PBConnectionLost): 
286              zenlog.debug("we lost the brand-new connection") 
287              # retrying might help here, let clientConnectionLost decide 
288              return 
289   
290          zenlog.warning("Cancelling attempts to connect") 
291          self.stopTrying() # logging in harder won't help 
292          if why.type == 'twisted.cred.error.UnauthorizedLogin': 
293              zenlog.critical("zenhub username/password combination is incorrect!") 
294              # Don't exit as Enterprise caches info and can survive 
295          else: 
296              zenlog.critical("Unknown connection problem to zenhub %s", why.type) 
297