ServiceCallGPCS.py (31972B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\script\net\ServiceCallGPCS.py 2 import copy 3 import random 4 import sys 5 import weakref 6 from collections import defaultdict 7 import blue 8 import bluepy 9 import log 10 import uthread 11 import time 12 from carbon.common.lib import const 13 from carbon.common.script.net import machobase, machoNet, machoNetAddress, machoNetPacket, objectCaching 14 from carbon.common.script.net.machoNetExceptions import ProxyRedirect, UberMachoException, UnMachoDestination, WrongMachoNode 15 from carbon.common.script.net.moniker import UpdateMoniker 16 from carbon.common.script.sys import basesession 17 from carbon.common.script.util.timerstuff import ClockThisWithoutTheStars 18 from cluster import MACHONETMSG_TYPE_SESSIONCHANGENOTIFICATION, MACHONETMSG_TYPE_SESSIONINITIALSTATENOTIFICATION 19 from eveexceptions import UserError 20 from eveprefs import prefs, boot 21 from carbon.common.lib.markers import PopMark, PushMark 22 from stringutil import strx 23 from machonet_tracing import EgressTracer, IngressTracer 24 25 class CoreServiceCall(): 26 __guid__ = 'gpcs.CoreServiceCall' 27 __notifyevents__ = ['OnUpdateServiceMapping'] 28 __recvServiceCallCount__ = defaultdict(lambda : 0) 29 __sendServiceCallCount__ = defaultdict(lambda : 0) 30 __recvServiceCallSourceCount__ = defaultdict(lambda : defaultdict(lambda : 0)) 31 __sendNotifyCount__ = defaultdict(lambda : 0) 32 33 def __init__(self): 34 self.services = {} 35 sm.RegisterNotify(self) 36 self.logClientCall = prefs.GetValue('logClientCall', 0) or prefs.GetValue('clusterMode', '') == 'LOCAL' 37 38 def UpdateRecvServiceCallStats(self, service, method, source): 39 key = str(service) + '.' + str(method) 40 if source.addressType == const.ADDRESS_TYPE_NODE: 41 sourceId = source.nodeID 42 elif source.addressType == const.ADDRESS_TYPE_CLIENT: 43 sourceId = 'Client' 44 else: 45 sourceId = 'Unknown' 46 CoreServiceCall.__recvServiceCallCount__[key] += 1 47 CoreServiceCall.__recvServiceCallSourceCount__[key][sourceId] += 1 48 49 def UpdateSendNotifyStats(self, method): 50 CoreServiceCall.__sendNotifyCount__[method] += 1 51 52 def UpdateSendServiceCallStats(self, service, method): 53 key = str(service) + '.' + str(method) 54 CoreServiceCall.__sendServiceCallCount__[key] += 1 55 56 def OnUpdateServiceMapping(self, svc, nodeID): 57 self.services[svc] = nodeID 58 59 def ResetAutoResolveCache(self): 60 self.services.clear() 61 62 @bluepy.TimedFunction('ServiceCallGPCS::CallUp') 63 def CallUp(self, packet): 64 if not packet.payload[0]: 65 packet.payload = packet.payload[1] 66 return self.ForwardCallUp(packet) 67 destinationService = getattr(packet.destination, 'service', None) 68 if destinationService is None: 69 return self.ForwardCallUp(packet) 70 destinationMethod = packet.payload[1] 71 if destinationMethod in packet.service.__dict__.get('__restrictedcalls__', {}) or destinationMethod.startswith('_'): 72 raise RuntimeError('This call can only be the result of a hack attack', destinationMethod) 73 if destinationMethod == 'MachoBindObject': 74 bind_start = time.time() 75 callWithArgs = packet.payload[2] 76 call = callWithArgs[-1] 77 args = callWithArgs[:-1] 78 bindParams = args[0] 79 nodeID = packet.service.MachoResolveObject(bindParams) 80 if nodeID and nodeID != self.machoNet.GetNodeID(): 81 self.machoNet.LogError('MachoBindObject received on wrong node. Should have gone to ', nodeID, ', bindParams=', bindParams, ', packet=', packet) 82 raise WrongMachoNode(nodeID) 83 boundObject = packet.service.MachoGetObjectBoundToSession(bindParams, session) 84 if self.logClientCall and packet.source.addressType == const.ADDRESS_TYPE_CLIENT: 85 kwargs = packet.payload[3] if len(packet.payload) >= 4 else None 86 self.machoNet.LogClientCall(session, destinationService, destinationMethod, callWithArgs, kwargs) 87 bind_end = time.time() 88 if call is not None: 89 if isinstance(boundObject, basesession.ObjectConnection): 90 objclass = boundObject.GetObjectConnectionLogClass() 91 else: 92 objclass = machobase.AssignLogName(object) 93 method, args, keywords = call 94 IngressTracer.start_macho_bind_trace(packet, objclass, method, bind_start, bind_end) 95 if method in boundObject.__dict__.get('__restrictedcalls__', {}) or method.startswith('_'): 96 raise RuntimeError('This call can only be the result of a hack attack', method) 97 callTimer = basesession.CallTimer('%s::%s (UnBoundObjectCall\\Server)' % (objclass, method)) 98 try: 99 func = getattr(boundObject, method) 100 self.UpdateRecvServiceCallStats(objclass, method, packet.source) 101 with IngressTracer.record_process_call_ms(): 102 if not keywords: 103 retval = ClockThisWithoutTheStars('Remote Unbound Object Call::%s::%s' % (objclass, method), func, args) 104 else: 105 retval = ClockThisWithoutTheStars('Remote Unbound Object Call::%s::%s' % (objclass, method), func, args, keywords) 106 payload = (boundObject, retval) 107 finally: 108 callTimer.Done() 109 110 else: 111 payload = (boundObject, None) 112 else: 113 IngressTracer.start_trace(packet, destinationService, destinationMethod) 114 args = packet.payload[2] 115 method = getattr(packet.service, destinationMethod) 116 callTimer = basesession.CallTimer('%s::%s (ServiceCall\\Server)' % (destinationService, destinationMethod)) 117 if self.logClientCall and packet.source.addressType == const.ADDRESS_TYPE_CLIENT and destinationMethod != 'MachoBindObject': 118 kwargs = packet.payload[3] if len(packet.payload) >= 4 else None 119 self.machoNet.LogClientCall(session, destinationService, destinationMethod, args, kwargs) 120 try: 121 self.UpdateRecvServiceCallStats(destinationService, method.__method__, packet.source) 122 with IngressTracer.record_process_call_ms(): 123 if len(packet.payload) == 3: 124 payload = ClockThisWithoutTheStars('Remote Service Call::%s::%s' % (destinationService, destinationMethod), method, args) 125 else: 126 payload = ClockThisWithoutTheStars('Remote Service Call::%s::%s' % (destinationService, destinationMethod), method, args, packet.payload[3]) 127 finally: 128 callTimer.Done() 129 130 if session.contextOnly: 131 packet.srcTransport.InstallSessionIfRequired(destinationService, destinationMethod) 132 packet.srcTransport = None 133 return packet.Response(payload=payload) 134 135 def CallDown(self, packet): 136 packet.payload = (0, packet.payload) 137 return self.ForwardCallDown(packet) 138 139 def NotifyUp(self, packet): 140 if packet.payload[0]: 141 if getattr(packet.destination, 'service', None) is not None: 142 if session.contextOnly and packet.command != MACHONETMSG_TYPE_SESSIONCHANGENOTIFICATION and packet.command != MACHONETMSG_TYPE_SESSIONINITIALSTATENOTIFICATION: 143 packet.srcTransport.InstallSessionIfRequired(packet.destination.service, packet.payload[1]) 144 packet.srcTransport = None 145 method = getattr(packet.service, packet.payload[1]) 146 callTimer = basesession.CallTimer('%s::%s (ServiceNotify\\Server)' % (packet.destination.service, packet.payload[1])) 147 try: 148 if len(packet.payload) == 3: 149 ClockThisWithoutTheStars('Remote Service Notify::%s::%s' % (packet.destination.service, packet.payload[1]), method, packet.payload[2]) 150 else: 151 ClockThisWithoutTheStars('Remote Service Notify::%s::%s' % (packet.destination.service, packet.payload[1]), method, packet.payload[2], packet.payload[3]) 152 finally: 153 callTimer.Done() 154 155 return 156 else: 157 packet.payload = packet.payload[1] 158 return self.ForwardNotifyUp(packet) 159 160 def NotifyDown(self, packet): 161 if packet.destination.broadcastID: 162 self.UpdateSendNotifyStats(packet.destination.broadcastID) 163 packet.payload = (0, packet.payload) 164 return self.ForwardNotifyDown(packet) 165 166 def RemoteServiceCall(self, sess, nodeID, service, method, *args, **keywords): 167 return self.RemoteServiceCallWithoutTheStars(sess, nodeID, service, method, args, keywords) 168 169 def RemoteServiceCallWithoutTheStars(self, sess, nodeID, service, method, args, keywords): 170 while 1: 171 try: 172 if sess is None: 173 self.machoNet.LogError('Session is None during RemoteServiceCall') 174 log.LogTraceback() 175 sessionVars1 = self._GetRemoteServiceCallVariables(sess) 176 mask = sess.Masquerade() 177 try: 178 cachable, versionCheck, throwOK, cachedResultRecord, cacheKey, cacheDetails = sm.GetService('objectCaching').PerformCachedMethodCall(None, service, method, args) 179 if cachable: 180 if not versionCheck: 181 if throwOK: 182 self.LogMethodCall(objectCaching.CacheOK(), service, method + '[Cached]', args, keywords) 183 raise objectCaching.CacheOK() 184 elif cachedResultRecord: 185 self.LogMethodCall(cachedResultRecord['lret'], service, method + '[Cached]', args, keywords) 186 return cachedResultRecord['lret'] 187 if nodeID is None and machoNet.mode == 'client': 188 nodeID = self.services.get(service, None) 189 if nodeID is None and service in self.machoNet.serviceInfo: 190 where = self.machoNet.serviceInfo[service] 191 if where: 192 for k, v in self.machoNet.serviceInfo.iteritems(): 193 if k != service and v and (v.startswith(where) or where.startswith(v)): 194 nodeID = self.services.get(k, None) 195 break 196 197 if nodeID is None and self.services and service not in ('DB2',): 198 self.machoNet.LogInfo("I haven't got a clue where ", service, ' should run, although I could guess any of these: ', str(self.services.keys())) 199 if nodeID is None: 200 if machoNet.mode == 'server': 201 proxies = self.machoNet.GetConnectedProxyNodes() 202 if not proxies: 203 raise UnMachoDestination('No proxy available') 204 destination = machoNetAddress.MachoAddress(nodeID=random.choice(proxies), service=service) 205 else: 206 destination = machoNetAddress.MachoAddress(service=service) 207 elif isinstance(nodeID, machoNetAddress.MachoAddress): 208 destination = nodeID 209 else: 210 destination = machoNetAddress.MachoAddress(nodeID=nodeID, service=service) 211 if keywords: 212 kw2 = copy.copy(keywords) 213 else: 214 kw2 = {} 215 if kw2.get('machoVersion', None) is None: 216 if cachedResultRecord is not None and cachedResultRecord['version'] is not None: 217 kw2['machoVersion'] = cachedResultRecord['version'] 218 else: 219 kw2['machoVersion'] = 1 220 if currentcall: 221 machoTimeout = kw2.get('machoTimeout', currentcall.packet.oob.get('machoTimeout', None)) 222 else: 223 machoTimeout = kw2.get('machoTimeout', None) 224 if 'machoTimeout' in kw2: 225 del kw2['machoTimeout'] 226 payload = (1, 227 method, 228 args, 229 kw2) 230 packet = machoNetPacket.CallReq(destination=destination, payload=payload) 231 if machoTimeout is not None: 232 packet.oob['machoTimeout'] = machoTimeout 233 with EgressTracer(packet, service, method) as trace: 234 callTimer = basesession.CallTimer('%s::%s (ServiceCall\\Client)' % (service, method)) 235 try: 236 response = self.ForwardCallDown(packet) 237 ret = response.payload 238 self.UpdateSendServiceCallStats(service, method) 239 self.LogMethodCall(ret, service, method, args, keywords) 240 except RuntimeError: 241 raise 242 except (UpdateMoniker, 243 UserError, 244 ProxyRedirect, 245 WrongMachoNode): 246 raise 247 except objectCaching.CacheOK: 248 if cachedResultRecord is None: 249 raise 250 sys.exc_clear() 251 self.LogMethodCall(cachedResultRecord['lret'], service, method, args, keywords) 252 sm.GetService('objectCaching').UpdateVersionCheckPeriod(cacheKey) 253 return cachedResultRecord['lret'] 254 finally: 255 callTimer.Done() 256 257 if isinstance(ret, objectCaching.CachedMethodCallResult): 258 sm.GetService('objectCaching').CacheMethodCall(service, method, args, ret) 259 ret = ret.GetResult() 260 return ret 261 finally: 262 mask.UnMask() 263 sessionVars2 = self._GetRemoteServiceCallVariables(sess) 264 265 except WrongMachoNode as e: 266 sys.exc_clear() 267 self._HandleServiceCallWrongNode(e, service, sessionVars1, sessionVars2) 268 self.services[service] = e.payload 269 if nodeID == e.payload: 270 self.machoNet.LogError('Redirected to same node', nodeID) 271 raise UnMachoDestination('Failed to redirect call, because we were being redirected to the same place we were trying to call in the first place (%s)' % nodeID) 272 else: 273 self.machoNet.LogInfo('Redirecting call to node ', e.payload, ' that was originally intended for node ', nodeID) 274 nodeID = e.payload 275 except UserError as e: 276 if e.msg in ('UnMachoDestination', 'GPSTransportClosed'): 277 self.services.clear() 278 raise 279 except Exception: 280 if service in self.services: 281 otherNodeID = self.services[service] 282 for s, n in self.services.items()[:]: 283 if n and n in (otherNodeID, nodeID): 284 del self.services[s] 285 286 raise 287 288 def _GetRemoteServiceCallVariables(self, session): 289 return () 290 291 def _HandleServiceCallWrongNode(self, e, service, sessionVars1, sessionVars2): 292 pass 293 294 def RemoteServiceNotify(self, sess, nodeID, service, method, *args, **keywords): 295 return self.RemoteServiceNotifyWithoutTheStars(sess, nodeID, service, method, args, keywords) 296 297 def RemoteServiceNotifyWithoutTheStars(self, sess, nodeID, service, method, args, keywords): 298 while 1: 299 try: 300 if sess is None: 301 self.machoNet.LogError('Session is None during RemoteServiceNotify') 302 log.LogTraceback() 303 if nodeID is None and machoNet.mode == 'client': 304 nodeID = self.services.get(service, None) 305 if nodeID is None and service in self.machoNet.serviceInfo: 306 where = self.machoNet.serviceInfo[service] 307 for k, v in self.machoNet.serviceInfo.iteritems(): 308 if k != service and v == where: 309 nodeID = self.services.get(k, None) 310 break 311 312 if nodeID is None and self.machoNet.serviceInfo.get(service, None) is not None: 313 self.machoNet.LogInfo("I haven't got a clue where ", service, ' should run, although I could guess any of these: ', str(self.services.keys())) 314 if nodeID is None: 315 if machoNet.mode == 'server': 316 proxies = self.machoNet.GetConnectedProxyNodes() 317 if not proxies: 318 raise UnMachoDestination('No proxy available') 319 destination = machoNetAddress.MachoAddress(nodeID=random.choice(proxies), service=service) 320 else: 321 destination = machoNetAddress.MachoAddress(service=service) 322 elif isinstance(nodeID, machoNetAddress.MachoAddress): 323 destination = nodeID 324 else: 325 destination = machoNetAddress.MachoAddress(nodeID=nodeID, service=service) 326 if keywords: 327 kw2 = copy.copy(keywords) 328 else: 329 kw2 = {} 330 if currentcall: 331 machoTimeout = kw2.get('machoTimeout', currentcall.packet.oob.get('machoTimeout', None)) 332 else: 333 machoTimeout = kw2.get('machoTimeout', None) 334 if 'machoTimeout' in kw2: 335 del kw2['machoTimeout'] 336 if kw2: 337 payload = (1, 338 method, 339 args, 340 kw2) 341 else: 342 payload = (1, method, args) 343 packet = machoNetPacket.Notification(destination=destination, payload=payload) 344 if machoTimeout is not None: 345 packet.oob['machoTimeout'] = machoTimeout 346 mask = sess.Masquerade() 347 try: 348 callTimer = basesession.CallTimer('%s::%s (ServiceNotify\\Client)' % (service, method)) 349 try: 350 self.ForwardNotifyDown(packet) 351 finally: 352 callTimer.Done() 353 354 finally: 355 mask.UnMask() 356 357 except WrongMachoNode as e: 358 sys.exc_clear() 359 self.services[service] = e.payload 360 if nodeID == e.payload: 361 self.machoNet.LogError('Redirected to same node', nodeID) 362 raise UnMachoDestination('Failed to redirect call, because we were being redirected to the same place we were trying to call in the first place (%s)' % nodeID) 363 else: 364 self.machoNet.LogInfo('Redirecting call to node ', e.payload, ' that was originally intended for node ', nodeID) 365 nodeID = e.payload 366 continue 367 except UserError as e: 368 if e.msg in ('UnMachoDestination', 'GPSTransportClosed'): 369 self.services.clear() 370 raise 371 except Exception: 372 self.services.clear() 373 raise 374 375 return 376 377 def LogMethodCall(self, result, logname, method, args, keywords): 378 logChannel = log.methodcalls 379 if logChannel.IsOpen(log.LGINFO): 380 if isinstance(result, Exception): 381 eorr = ', EXCEPTION=' 382 else: 383 eorr = ', retval=' 384 if keywords: 385 logwhat = [logname, 386 '::', 387 method, 388 ' args=', 389 args, 390 ', keywords={', 391 keywords, 392 '}', 393 eorr, 394 result] 395 else: 396 logwhat = [logname, 397 '::', 398 method, 399 ' args=', 400 args, 401 eorr, 402 result] 403 timer = PushMark(logname + '::LogPseudoMethodCall') 404 try: 405 s = 'failed to convert' 406 for getter in [str, unicode, repr]: 407 try: 408 s = ''.join(map(getter, logwhat)) 409 break 410 except: 411 sys.exc_clear() 412 413 if len(s) > 2500: 414 s = s[:2500] 415 logChannel.Log(s, log.LGINFO, 1) 416 finally: 417 PopMark(timer) 418 419 def ConnectToRemoteService(self, servicename, nodeID = None, sess = None): 420 if sess is None: 421 sess = session.GetActualSession() 422 return MachoServiceConnection(self, sess, servicename, nodeID) 423 424 def UberMachoRemoteServiceCall(self, nodeGroup, batchInterval, sess, service, method, *args, **keywords): 425 if batchInterval: 426 queue = sess.GetSessionVariable(('batchedCallQueue', batchInterval), ({}, {})) 427 if not queue[0]: 428 uthread.worker('GPCS::BroadcastStuff::BatchedUberMachoRemoteServiceCall', self.__BatchedUberMachoRemoteServiceCall, nodeGroup, sess, batchInterval) 429 i = len(queue[0]) 430 queue[0][i] = (service, 431 method, 432 args, 433 keywords) 434 queue[1][i] = uthread.Channel() 435 return queue[1][i].receive() 436 else: 437 return self.__UberMachoRemoteServiceCall(sess, service, method, self.__GetNodeGroup(nodeGroup), *args, **keywords) 438 439 def __GetNodeGroup(self, nodeGroup): 440 if nodeGroup == 0: 441 if machoNet.mode == 'proxy': 442 return self.machoNet.GetConnectedSolNodes() 443 else: 444 return self.machoNet.GetConnectedProxyNodes() 445 elif nodeGroup == 1: 446 if machoNet.mode == 'proxy': 447 return self.machoNet.GetConnectedProxyNodes() + [self.machoNet.GetNodeID()] 448 else: 449 return self.machoNet.GetConnectedSolNodes() + [self.machoNet.GetNodeID()] 450 else: 451 return self.machoNet.GetConnectedNodes() + [self.machoNet.GetNodeID()] 452 453 def __BatchedUberMachoRemoteServiceCall(self, nodeGroup, sess, batchInterval): 454 blue.pyos.synchro.SleepWallclock(batchInterval / const.MSEC) 455 queue = sess.GetSessionVariable(('batchedCallQueue', batchInterval)) 456 sess.SetSessionVariable(('batchedCallQueue', batchInterval), ({}, {})) 457 scatteredRetvals = self.__UberMachoRemoteServiceCall(sess, 'sessionMgr', 'BatchedRemoteCall', self.__GetNodeGroup(nodeGroup), queue[0]) 458 gatheredRetvals = {} 459 for isexception, nodeID, retval in scatteredRetvals: 460 if isexception: 461 for callID in queue[0].iterkeys(): 462 if callID not in gatheredRetvals: 463 gatheredRetvals[callID] = [] 464 gatheredRetvals[callID].append((isexception, nodeID, retval)) 465 466 else: 467 for isexception2, callID, retval2 in retval: 468 if callID not in gatheredRetvals: 469 gatheredRetvals[callID] = [] 470 gatheredRetvals[callID].append((isexception2, nodeID, retval2)) 471 472 for callID, channel in queue[1].iteritems(): 473 channel.send(gatheredRetvals[callID]) 474 475 def __UberMachoRemoteServiceCall(self, sess, service, method, nodes, *args, **keywords): 476 retvals = [] 477 dude = UberDude(len(nodes), service, method) 478 if not len(nodes): 479 raise UnMachoDestination('No target nodes available for UberMachoRemoteServiceCall') 480 uberMachoRaise = keywords.get('uberMachoRaise', False) 481 if 'uberMachoRaise' in keywords: 482 keywords = copy.copy(keywords) 483 del keywords['uberMachoRaise'] 484 if 'machoTimeout' in keywords: 485 keywords = copy.copy(keywords) 486 else: 487 keywords['machoTimeout'] = 60 488 machoTimeout = keywords.get('machoTimeout', 60) 489 for each in nodes: 490 uthread.worker(blue.pyos.taskletTimer.GetCurrent() + '::RPC', UnterMachoRemoteServiceCall, self, dude, retvals, sess, each, service, method, args, keywords) 491 492 if dude.Await(machoTimeout) == 2: 493 log.LogTraceback('Local Timeout') 494 n = copy.copy(nodes) 495 for r in retvals: 496 n.remove(r[1]) 497 498 for nodeID in n: 499 self.machoNet.LogError('Node ', nodeID, " has not completed it's uber-macho call") 500 retvals.append((1, nodeID, RuntimeError('Local Timeout'))) 501 502 if uberMachoRaise: 503 for isexception, nodeID, ret in retvals: 504 if isexception: 505 raise UberMachoException(retvals) 506 507 return retvals 508 509 def ConnectToAllNeighboringServices(self, servicename, sess, batchInterval): 510 return UberMachoServiceConnection(self, sess, servicename, 0, batchInterval) 511 512 def ConnectToAllSiblingServices(self, servicename, sess, batchInterval): 513 return UberMachoServiceConnection(self, sess, servicename, 1, batchInterval) 514 515 def ConnectToAllServices(self, servicename, sess, batchInterval): 516 return UberMachoServiceConnection(self, sess, servicename, 2, batchInterval) 517 518 519 class UberDude(): 520 521 def __init__(self, ubercount, service, method): 522 self.ubercount = ubercount 523 if ubercount: 524 self.channel = uthread.Channel(('UberMachoRemoteServiceCall', service + '::' + method)) 525 else: 526 self.channel = None 527 self.tasklet = None 528 529 def Done(self): 530 self.ubercount -= 1 531 if self.ubercount <= 0: 532 if self.channel is not None and self.channel.queue: 533 self.channel.send(1) 534 self.channel = None 535 if self.tasklet: 536 self.tasklet.kill() 537 self.tasklet = None 538 539 def Await(self, timeoutInterval): 540 self.tasklet = uthread.new(self.__Timeout, timeoutInterval) 541 if self.channel: 542 return self.channel.receive() 543 return 0 544 545 def __Timeout(self, timeoutInterval): 546 try: 547 blue.pyos.synchro.SleepWallclock(1000 * timeoutInterval) 548 self.tasklet = None 549 if self.channel: 550 self.ubercount = 0 551 self.channel.send(2) 552 except TaskletExit: 553 pass 554 555 556 def UnterMachoRemoteServiceCall(self, dude, retvals, sess, nodeID, service, method, args, keywords): 557 try: 558 if nodeID == sm.services['machoNet'].GetNodeID(): 559 if 'machoTimeout' in keywords: 560 keywords = copy.copy(keywords) 561 del keywords['machoTimeout'] 562 retvals.append((0, nodeID, getattr(sess.ConnectToService(service), method)(*args, **keywords))) 563 else: 564 retvals.append((0, nodeID, self.RemoteServiceCallWithoutTheStars(sess, nodeID, service, method, args, keywords))) 565 except Exception as e: 566 log.LogException() 567 retvals.append((1, nodeID, e)) 568 sys.exc_clear() 569 except Exception as e: 570 log.LogTraceback() 571 retvals.append((1, nodeID, e)) 572 raise 573 finally: 574 dude.Done() 575 576 577 class UberMachoServiceConnection(): 578 579 def __init__(self, gpcs, sess, servicename, nodeGroup, batchInterval): 580 try: 581 self.__dict__['session'] = weakref.proxy(sess) 582 except Exception: 583 self.__dict__['session'] = sess 584 sys.exc_clear() 585 586 self.__dict__['servicename'] = servicename 587 self.__dict__['gpcs'] = weakref.proxy(gpcs) 588 self.__dict__['nodeGroup'] = nodeGroup 589 self.__dict__['batchInterval'] = batchInterval 590 591 def __str__(self): 592 return '<UberMachoRemoteService:' + strx(self.__dict__['servicename']) + '>' 593 594 def __nonzero__(self): 595 return 1 596 597 def __repr__(self): 598 return str(self) 599 600 def __getattr__(self, attribute): 601 if attribute in self.__dict__: 602 return self.__dict__[attribute] 603 if attribute[:2] == '__': 604 raise AttributeError(attribute) 605 return UberMachoServiceCallWrapper(self.__dict__['gpcs'], self.__dict__['session'], self.__dict__['servicename'], attribute, self.__dict__['nodeGroup'], self.__dict__['batchInterval']) 606 607 608 class UberMachoServiceCallWrapper(): 609 610 def __init__(self, gpcs, sess, service, method, nodeGroup, batchInterval): 611 self.session = sess 612 self.gpcs = gpcs 613 self.service = service 614 self.method = method 615 self.nodeGroup = nodeGroup 616 self.batchInterval = batchInterval 617 618 def __call__(self, *args, **keywords): 619 mask = self.session.Masquerade() 620 try: 621 callTimer = basesession.CallTimer('%s::%s (UberMachoRemoteServiceCall)' % (self.service, self.method)) 622 try: 623 return self.gpcs.UberMachoRemoteServiceCall(self.nodeGroup, self.batchInterval, self.session, self.service, self.method, *args, **keywords) 624 finally: 625 callTimer.Done() 626 627 finally: 628 mask.UnMask() 629 self.__dict__.clear() 630 631 632 class MachoServiceConnection(): 633 blocking = True 634 635 def __init__(self, gpcs, sess, servicename, nodeID = None): 636 try: 637 self.session = weakref.proxy(sess) 638 except TypeError: 639 self.session = sess 640 641 self.servicename = servicename 642 try: 643 self.gpcs = weakref.proxy(gpcs) 644 except TypeError: 645 self.gpcs = gpcs 646 647 self.nodeID = nodeID 648 649 def __repr__(self): 650 return '<RemoteService: %s>' % self.servicename 651 652 def __getattr__(self, attribute): 653 if attribute.startswith('__'): 654 raise AttributeError(attribute) 655 if attribute == 'NoBlock': 656 noBlock = MachoServiceConnection(self.gpcs, self.session, self.servicename, self.nodeID) 657 noBlock.blocking = False 658 return noBlock 659 660 def callable(*args, **kwargs): 661 if self.blocking: 662 call = self.gpcs.RemoteServiceCallWithoutTheStars 663 else: 664 call = self.gpcs.RemoteServiceNotifyWithoutTheStars 665 666 def doCall(): 667 method = attribute 668 mask = self.session.Masquerade() 669 try: 670 callTimer = basesession.CallTimer('%s::%s (RemoteServiceCall)' % (self.servicename, method)) 671 try: 672 return call(self.session, self.nodeID, self.servicename, method, args, kwargs) 673 finally: 674 callTimer.Done() 675 676 finally: 677 mask.UnMask() 678 679 if boot.role == 'client': 680 if kwargs.get('noCallThrottling', None) is None: 681 key = (self.servicename, 682 attribute, 683 str(args), 684 str(kwargs)) 685 return machobase.ThrottledCall(key, doCall) 686 del kwargs['noCallThrottling'] 687 return doCall() 688 689 return callable