machoNet.py (231673B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\script\net\machoNet.py 2 from __future__ import print_function 3 from future.utils import raise_ 4 import collections 5 import copy 6 import json 7 import numbers 8 import os 9 import random 10 import socket 11 import struct 12 import weakref 13 import sys 14 import blue 15 import bluepy 16 import locks 17 import log 18 import stackless 19 import telemetry 20 import uthread 21 import uthread2 22 import carbon.common.script.net.machobase as macho 23 import carbon.common.script.net.ExceptionMappingGPCS as gpcs 24 import cluster 25 import monolithconfig 26 import monolithmetrics 27 import monolithsentry 28 import monolithdisconnectlogger 29 from carbon.common.script.net.addressCache import MachoAddressCache 30 from carbon.common.script.net.cachedObject import CachedObject as utilCachedObject 31 from carbon.common.script.net.GPSExceptions import GPSAddressOccupied, GPSBadAddress, GPSTransportClosed 32 from carbon.common.script.net.machoNetAddress import MachoAddress 33 from carbon.common.script.net.machoNetExceptions import MachoException, MachoWrappedException, SessionUnavailable, UnMachoChannel, UnMachoDestination, UserRejectedByVIP 34 from carbon.common.script.net.machoNetPacket import Notification, PingReq, IdentificationReq 35 from carbon.common.script.net.machoNetTransport import MachoTransport 36 from carbon.common.script.net.machoRunTimeStats import MachoRunTimeStats 37 from carbon.common.script.net.machobase import DumpsSanitized 38 from carbon.common.script.sys import basesession 39 from carbon.common.script.sys import crowset, service 40 from carbon.common.script.sys.buildversion import GetBuildVersionAsInt 41 from carbon.common.script.util import format as utilFormat 42 from carbon.common.script.util.weight import ChooseWeighted as weightedChoice 43 from carbon.common.script.sys.service import ROLE_SERVICE, ROLE_ADMIN, ROLE_ANY, ROLE_VIPLOGIN, SERVICE_RUNNING, SERVICE_START_PENDING, ROLE_GML, ROLE_PLAYER 44 from eve.common.lib import appConst as const 45 from eve.common.script.sys.idCheckers import IsEveUser, IsDustUser 46 from evecrypto.crypto import CryptoHash 47 from eveexceptions import ConnectionError, UserError 48 from eveprefs import prefs, boot 49 from monolithmetrics import _prometheus 50 from uthread2 import StartTasklet 51 from utillib import strx, KeyVal 52 from machonet_tracing import IngressTracer 53 import launchdarkly 54 import logging 55 try: 56 from carbon.server.script.net.onlinedata import OnlineData 57 except ImportError: 58 from carbon.common.script.net.onlinedata import OnlineData 59 60 if macho.mode != 'client': 61 from featuremanagement.server.flag import flag_bool 62 else: 63 from featuremanagement.client.flag import flag_bool 64 METRIC_SESSIONS = 'machonet.sessions' 65 METRIC_SESSIONS_TOTAL = METRIC_SESSIONS + '.total' 66 LD_FLAG_TRANSPORT_METRICS = 'machonet_transport_metrics_enabled' 67 Enter = blue.pyos.taskletTimer.EnterTasklet 68 Leave = blue.pyos.taskletTimer.ReturnFromTasklet 69 OOB_SESSIONNOTIFICATION = 1 70 CLUSTER_ID = 1 71 DEFAULT_SUBNETWORK = getattr(boot, 'preferredSubnet', '10.') 72 PREFERRED_SUBNETWORKS = {'server': getattr(boot, 'preferredServerSubnet', DEFAULT_SUBNETWORK), 73 'proxy': getattr(boot, 'preferredProxySubnet', DEFAULT_SUBNETWORK)} 74 offsetMap = {'tcp:packet:client': 0, 75 'tcp:packet:machoNet': 1, 76 'tcp:raw:http': 2, 77 'tcp:raw:http2': 3, 78 'tcp:http:cow': 4, 79 'tcp:raw:status': 5} 80 offsetMap = {'client': offsetMap, 81 'server': offsetMap, 82 'proxy': offsetMap} 83 offsetStep = {'proxy': 1000, 84 'server': 10, 85 'client': 10} 86 _cached_gps_map = None 87 88 def get_gps_map(): 89 global _cached_gps_map 90 if _cached_gps_map is None: 91 from carbon.common.script.net.SocketGPS import SecureSocketPacketTransportFactory 92 from carbon.common.script.net.SocketGPS import SocketPacketTransportFactory 93 from carbon.common.script.net.SocketGPS import SocketTransportFactory 94 _cached_gps_map = {'client': {'tcp:raw:http': SocketTransportFactory, 95 'tcp:packet:server': SecureSocketPacketTransportFactory}, 96 'proxy': {'tcp:packet:client': SecureSocketPacketTransportFactory, 97 'tcp:packet:machoNet': SocketPacketTransportFactory, 98 'tcp:raw:http': SocketTransportFactory}, 99 'server': {'tcp:packet:machoNet': SocketPacketTransportFactory, 100 'tcp:raw:http': SocketTransportFactory}} 101 import iocp 102 if iocp.UsingSSL(): 103 from carbon.common.script.net.SocketGPS import SSLSocketPacketTransportFactory 104 log.general.Log('Encryption: SSL', log.LGINFO) 105 _cached_gps_map['client']['tcp:packet:server'] = SSLSocketPacketTransportFactory 106 _cached_gps_map['proxy']['tcp:packet:client'] = SSLSocketPacketTransportFactory 107 else: 108 log.general.Log('Encryption: MachoNet', log.LGINFO) 109 if iocp.UsingHTTPS(): 110 from carbon.common.script.net.SocketGPS import SSLSocketTransportFactory 111 _cached_gps_map['proxy']['tcp:raw:http'] = SSLSocketTransportFactory 112 _cached_gps_map['server']['tcp:raw:http'] = SSLSocketTransportFactory 113 log.general.Log('Using HTTPS', log.LGINFO) 114 return _cached_gps_map 115 116 117 _cached_gpcs_map = None 118 119 def get_gpcs_map(): 120 global _cached_gpcs_map 121 if _cached_gpcs_map is None: 122 from carbon.common.script.net.ExceptionWrapperGPCS import ExceptionWrapper 123 from carbon.common.script.net.ExceptionMappingGPCS import ExceptionMappingGPCS 124 from carbon.common.script.net.ResolveGPCS import ResolveGPCS 125 from carbon.common.script.net.ObjectCallGPCS import ObjectCall 126 from eve.common.script.net.eveServiceCallGPCS import ServiceCall 127 from eve.common.script.net.eveBroadcastStuffGPCS import BroadcastStuff 128 from carbon.common.script.net.SessionChangeGPCS import SessionChange 129 _cached_gpcs_map = {'server': {None: [ExceptionWrapper, 130 ExceptionMappingGPCS, 131 ResolveGPCS, 132 ObjectCall, 133 ServiceCall, 134 BroadcastStuff], 135 'sessionchange': [SessionChange]}, 136 'proxy': {None: [ExceptionWrapper, 137 ExceptionMappingGPCS, 138 ObjectCall, 139 ServiceCall, 140 BroadcastStuff], 141 'sessionchange': [SessionChange]}, 142 'client': {None: [ExceptionWrapper, 143 ExceptionMappingGPCS, 144 ObjectCall, 145 ServiceCall, 146 BroadcastStuff], 147 'sessionchange': [SessionChange]}} 148 return _cached_gpcs_map 149 150 151 class MachoNetService(service.Service): 152 __guid__ = 'svc.machoNet' 153 __displayname__ = 'MachoNet Service' 154 __startupdependencies__ = [] 155 if boot.role == 'server': 156 __startupdependencies__ = ['DB2', 'dbLog'] 157 __dependencies__ = [] 158 __configvalues__ = {'acceptThreadCount': 20, 159 'cleanupInterval': 180, 160 'defaultClientPortOffset': 15000, 161 'defaultServerPortOffset': 10000, 162 'defaultProxyPortOffset': 26000, 163 'defaultTunnelPortOffset': 50000, 164 'bannedIPs': [], 165 'vipMode': 0, 166 'reducedTimeouts': 0, 167 'callTimeOutInterval': 480 if boot.role != 'client' else 3600, 168 'clientKeepAliveTimerInterval': 240, 169 'serverKeepAliveTimerInterval': 60, 170 'clientLogonQueuePollTime': 60, 171 'minimumSolCount': 0, 172 'minimumProxyCount': 0, 173 'nodeRefreshInterval': 30, 174 'proxyStatPollInterval': 180, 175 'proxyStatSmoothie': 10, 176 'nodeLoadPush': 300, 177 'disconnectUnauthorizedUsersPollInterval': 3600, 178 'disconnectUnauthorizedUsersDelayInterval': 15, 179 'clientSessionTimeoutGranularity': 60, 180 'connectionLimit': 1000, 181 'maxLoginsPerMinute': 60, 182 'acceptDelay': 10, 183 'autoLogoffAuthenticatedTransportInterval': 180, 184 'debugHeartBeat': 0, 185 'largeResponseWarningThreshold': 6000000, 186 'largeResponseErrorThreshold': 8000000, 187 'callCounterEnabled': 0} 188 __exportedcalls__ = {'CheckACL': [ROLE_SERVICE | ROLE_ADMIN], 189 'IsThisUserCoolForLogin': [ROLE_SERVICE], 190 'GetTransport': [ROLE_SERVICE], 191 'GetTransportOfNode': [ROLE_SERVICE], 192 'GetNodeFromAddress': [ROLE_SERVICE | ROLE_ADMIN], 193 'GetNodeIDsFromServiceMask': [ROLE_SERVICE | ROLE_ADMIN], 194 'CheckAddressCache': [ROLE_SERVICE | ROLE_ADMIN], 195 'LookupAddressCache': [ROLE_SERVICE | ROLE_ADMIN], 196 'ClearNodeOfAddresses': [ROLE_SERVICE | ROLE_ADMIN], 197 'GetFullAddressCache': [ROLE_SERVICE | ROLE_ADMIN], 198 'GetFullServiceMaskAddressCache': [ROLE_SERVICE | ROLE_ADMIN], 199 'GetServiceMaskByServiceID': [ROLE_SERVICE | ROLE_ADMIN], 200 'GetNodeID': [ROLE_SERVICE | ROLE_ADMIN], 201 'GetServerStatus': [ROLE_ANY], 202 'GetGPCS': [ROLE_SERVICE], 203 'GetConnectionProperties': [ROLE_ANY], 204 'SetConnectionProperty': [ROLE_ADMIN], 205 'RefreshConnectivity': [ROLE_ADMIN], 206 'GetAddressAndStaticTagsForPrometheus': [ROLE_SERVICE | ROLE_ADMIN], 207 'ConnectToServer': [ROLE_ANY], 208 'ConnectToAddress': [ROLE_SERVICE], 209 'StartTransportReader': [ROLE_SERVICE], 210 'DisconnectFromServer': [ROLE_ANY], 211 'IsConnected': [ROLE_ANY], 212 'GetBasePortNumber': [ROLE_ANY], 213 'GetClientSessionID': [ROLE_ANY], 214 'GetServiceInfo': [ROLE_ANY], 215 'GetTime': [ROLE_ANY], 216 'Shutdown': [ROLE_SERVICE], 217 'GracefulShutdown': [ROLE_SERVICE], 218 'GracefulShutdownStarted': [ROLE_SERVICE], 219 'GracefulShutdownComplete': [ROLE_SERVICE], 220 'IsClusterShuttingDown': [ROLE_SERVICE], 221 'TerminateUnconnectedNodes': [ROLE_SERVICE], 222 'TerminateClient': [ROLE_SERVICE], 223 'TerminateAllClients': [ROLE_SERVICE], 224 'ConnectivityTest': [ROLE_SERVICE], 225 'GetCPULoad': [ROLE_SERVICE], 226 'GetNetworkStats': [ROLE_SERVICE], 227 'GetBlockingCallStats': [ROLE_SERVICE], 228 'GetMetrics': [ROLE_SERVICE], 229 'GetConnectedSolNodes': [ROLE_SERVICE], 230 'GetConnectedProxyNodes': [ROLE_SERVICE], 231 'GetConnectedNodes': [ROLE_SERVICE], 232 'GetClusterGameStatistics': [ROLE_SERVICE], 233 'GetClusterGameStatisticsForClient': [ROLE_PLAYER], 234 'GetSessionCounts': [ROLE_SERVICE], 235 'SetClusterSessionCounts': [ROLE_SERVICE], 236 'GetClusterSessionCounts': [ROLE_SERVICE], 237 'SendProvisionalResponse': [ROLE_SERVICE], 238 'GetIDOfAddress': [ROLE_SERVICE], 239 'ForwardProxyBroadcast': [ROLE_SERVICE], 240 'ForwardSinglecastByCharID': [ROLE_SERVICE], 241 'PrimeAddressCache': [ROLE_SERVICE], 242 'AreTheseServicesRunning': [ROLE_SERVICE], 243 'GetValidClientCodeHash': [ROLE_SERVICE], 244 'ReloadClientCodeHash': [ROLE_SERVICE], 245 'GetLogonQueuePosition': [ROLE_SERVICE], 246 'GetLogonQueueStats': [ROLE_SERVICE], 247 'GetPolarisExternalTunnelingAddress': [ROLE_SERVICE], 248 'IsResurrectedNode': [ROLE_SERVICE], 249 'CalculatePortNumber': [ROLE_SERVICE], 250 'GetPortOffsetStep': [ROLE_SERVICE], 251 'GetGlobalConfig': [ROLE_SERVICE], 252 'SetGlobalConfigValue': [ROLE_SERVICE | ROLE_ADMIN], 253 'GetGlobalConfigValue': [ROLE_SERVICE | ROLE_ADMIN], 254 'ForwardNotificationToNode': [ROLE_SERVICE], 255 'GetBaseTunnelPortOffset': [ROLE_SERVICE], 256 'PerformanceTest': [ROLE_SERVICE], 257 'LogClientCall': [ROLE_SERVICE], 258 'ForwardCharacterNotification': [ROLE_SERVICE], 259 'DisconnectUsers': [ROLE_SERVICE], 260 'SendMessageToPlayers': [ROLE_SERVICE | ROLE_ADMIN | ROLE_GML], 261 'GetRuntimeDetails': [ROLE_SERVICE], 262 'GetConnectedClusterName': [ROLE_ANY], 263 'GetConnectedClusterMode': [ROLE_ANY]} 264 __counters__ = {'dataSent': 'traffic', 265 'dataReceived': 'traffic', 266 'blockingCallTimes': 'traffic', 267 'compressedBytes': 'traffic', 268 'decompressedBytes': 'traffic', 269 'broadcastsResolved': 'traffic', 270 'broadcastsMissed': 'normal'} 271 __notifyevents__ = {'ProcessSessionChange', 272 'OnNewNode', 273 'OnNodeDeath', 274 'OnMachoObjectDisconnect', 275 'OnObjectPublicAttributesUpdated', 276 'OnClusterStarting', 277 'OnRefreshConnectivity', 278 'OnNodeLoadPush', 279 'OnPeopleWhoShouldntBeLoggedInNotification', 280 'OnVIPListChanged', 281 'OnTimeReset', 282 'OnClientCodeUpdated', 283 'OnClearFetchedData', 284 'OnGlobalConfigUpdated', 285 'OnClusterDesegmentation', 286 'OnCacheServiceReady', 287 'OnCacheServiceGroupChange'} 288 __gpcsmethodnames__ = {'Objectcast', 289 'ObjectcastWithoutTheStars', 290 'Queuedcast', 291 'QueuedcastWithoutTheStars', 292 'Scattercast', 293 'ScattercastWithoutTheStars', 294 'Broadcast', 295 'ClientBroadcast', 296 'ClusterBroadcast', 297 'ServerBroadcast', 298 'ProxyBroadcast', 299 'NodeBroadcast', 300 'NarrowcastByClientIDs', 301 'NarrowcastByClientIDsWithoutTheStars', 302 'NarrowcastByCharIDs', 303 'NarrowcastByUserIDs', 304 'NarrowcastByNodeIDs', 305 'SinglecastByServiceMask', 306 'SinglecastByClientID', 307 'SinglecastByCharID', 308 'SinglecastByNodeID', 309 'SinglecastByUserID', 310 'ConnectToRemoteService', 311 'ConnectToAllServices', 312 'ConnectToAllNeighboringServices', 313 'ConnectToAllSiblingServices', 314 'RemoteServiceCallWithoutTheStars', 315 'RemoteServiceNotifyWithoutTheStars', 316 'ResetAutoResolveCache', 317 'RemoteServiceCall', 318 'RemoteServiceNotify', 319 'OnObjectPublicAttributesUpdated', 320 'ReliableSinglecastByCharID', 321 'ReliableSinglecastByUserID'} 322 __pingrequestorresponse__ = {cluster.MACHONETMSG_TYPE_PING_RSP, cluster.MACHONETMSG_TYPE_PING_REQ} 323 __sessioninitorchangenotification__ = {cluster.MACHONETMSG_TYPE_SESSIONINITIALSTATENOTIFICATION, cluster.MACHONETMSG_TYPE_SESSIONCHANGENOTIFICATION} 324 __clientallowedcommands__ = {cluster.MACHONETMSG_TYPE_ERRORRESPONSE, 325 cluster.MACHONETMSG_TYPE_NOTIFICATION, 326 cluster.MACHONETMSG_TYPE_CALL_RSP, 327 cluster.MACHONETMSG_TYPE_CALL_REQ, 328 cluster.MACHONETMSG_TYPE_RESOLVE_REQ, 329 cluster.MACHONETMSG_TYPE_PING_REQ, 330 cluster.MACHONETMSG_TYPE_PING_RSP} 331 __notificationtypes__ = {cluster.MACHONETMSG_TYPE_NOTIFICATION, cluster.MACHONETMSG_TYPE_SESSIONCHANGENOTIFICATION, cluster.MACHONETMSG_TYPE_SESSIONINITIALSTATENOTIFICATION} 332 __calltypes__ = {cluster.MACHONETMSG_TYPE_CALL_REQ, cluster.MACHONETMSG_TYPE_RESOLVE_REQ} 333 __responsetypes__ = {cluster.MACHONETMSG_TYPE_ERRORRESPONSE, cluster.MACHONETMSG_TYPE_CALL_RSP, cluster.MACHONETMSG_TYPE_RESOLVE_RSP} 334 __server_scattercast_session_variables__ = {'userid', 'objectID', 'charid'} 335 metricsMap = {'CARBON:MachoChar': const.zmetricCounter_MachoChar, 336 'CARBON:MachoUser': const.zmetricCounter_MachoUser} 337 message_bus = None 338 339 def __init__(self): 340 self.shutdown = None 341 self.fetchedServiceInfo = None 342 self.clockReset = False 343 self.handlingClientAuthentication = 0 344 self.scheduleCount = 0 345 if macho.mode == 'server': 346 self.__dependencies__.append('DB2') 347 self.clusterSessionStatistics = {} 348 self.clusterSessionStatsHistory = [] 349 self.nodeCPULoadValue = {} 350 self.serverNames = {} 351 service.Service.__init__(self) 352 if macho.mode == 'server' and ('/compileconstants' in blue.pyos.GetArg() or '/compile' in blue.pyos.GetArg()): 353 self.connectToCluster = 0 354 else: 355 self.connectToCluster = 1 356 self.acceptStart = None 357 self.clientLogonQueue = KeyVal(address='', timestamp=None, position=None, history=None, reportedQueued=False) 358 self.serverLogonQueue = [] 359 self.clientHalfBakedTransports = {} 360 self.namedtransports = {} 361 self.transportsByID = {} 362 self.transportIDbyClientID = {} 363 self.transportIDbySessionID = {} 364 self.transportIDbyProxyNodeID = {} 365 self.transportIDbySolNodeID = {} 366 self.transportIDbyAppNodeID = {} 367 self.externalAddressesByNodeID = {} 368 self.channelHandlersUp = {} 369 self.channelHandlersDown = {} 370 self.polarisExternalTunneling = None 371 self.basePortNumber = self.defaultProxyPortOffset 372 self.baseTunnelPortOffset = self.defaultTunnelPortOffset 373 self.nodeID = None 374 self.machineID = None 375 self.nodeIndex = None 376 self.expectedLoadValue = {} 377 self.clientIDOffset = 1 378 self.stop = 1 379 self.transportID = 1 380 self.calls = {} 381 self.timeoutCalls = {} 382 self.callsCountMax = 0 383 self.callsCountMin = 0 384 self.callID = 1 385 self._addressCache = MachoAddressCache() 386 self.myProxyNodeID = None 387 self.dontLayTracksTo = {} 388 if macho.mode == 'client': 389 self.authenticating = 0 390 self.gettingServerStatus = 0 391 self.clientSessionID = 0 392 self.seq = 0 393 self.seqwait = {} 394 self.notifySequenceIDByNodeID = collections.defaultdict(int) 395 self.serviceInfo = {} 396 self.locks = {} 397 self.serviceMappings = {} 398 self.serviceMaskByServiceID = {} 399 self.serviceNameByServiceID = {} 400 self.factories = {} 401 self.spam = set() 402 self.kickedTick = 0 403 self.startedWhen = blue.os.GetWallclockTime() 404 self.paon = None 405 self.paonLock = uthread.Semaphore('MachoNet:paonLock') 406 self.vipLock = uthread.Semaphore('MachoNet:vipLock') 407 self.fetchedDataLock = uthread.Semaphore('MachoNet:fetchedDataLock') 408 self.clusterStartLock = uthread.Semaphore('MachoNet:clusterStartLock') 409 self.callCounter = collections.defaultdict(int) 410 for svc in sm.services.itervalues(): 411 if hasattr(svc, '__machoresolve__'): 412 self.serviceInfo[svc.__logname__] = svc.__machoresolve__ 413 414 self.globalConfig = None 415 self.shutdownInProgress = False 416 self.serviceMask = cluster.SERVICE_NONE 417 self.getNodeFromAddressWarnsDone = set() 418 self.statusData = {} 419 self.statusCallbacks = {} 420 self.onlineData = OnlineData(self) 421 self.heartbeat_lagging = False 422 self.metricsEnabled = flag_bool(LD_FLAG_TRANSPORT_METRICS, False, self._configure_metrics_flag) 423 self.logger = logging.getLogger('machoNet') 424 425 def SendMessageToPlayers(self, message, includeEve, includeDust): 426 recipients = [] 427 for sid, sess in basesession.allSessionsBySID.iteritems(): 428 userID = getattr(sess, 'userid', None) 429 clientID = getattr(sess, 'clientID', None) 430 if userID and clientID: 431 if IsDustUser(userID) and includeDust or IsEveUser(userID) and includeEve: 432 recipients.append(clientID) 433 434 self.NarrowcastByClientIDs(recipients, 'OnServerMessage', message) 435 436 def ForwardNotificationToNode(self, source, destination, userID, payload): 437 orginalClientID = source.clientID 438 localTransport = self.transportsByID[self.transportIDbyClientID[orginalClientID]] 439 packet = Notification(source=source, destination=destination, userID=userID, payload=payload) 440 self.HandleMessage(localTransport, packet) 441 self.SinglecastByClientID(orginalClientID, 'OnUpdateServiceMapping', destination.service, destination.nodeID) 442 443 def GetBaseTunnelPortOffset(self): 444 return self.baseTunnelPortOffset 445 446 def IsResurrectedNode(self): 447 return self.resurrectedNode 448 449 def GetPolarisExternalTunnelingAddress(self): 450 if macho.mode == 'server' and self.polarisExternalTunneling is not None and self.polarisExternalTunneling.proxyNodeID in self.transportIDbyProxyNodeID: 451 return self.polarisExternalTunneling.address 452 else: 453 lastNodeID = None 454 last = None 455 for nodeID, transportID in self.transportIDbyProxyNodeID.iteritems(): 456 machoTransport = self.transportsByID.get(transportID, None) 457 if machoTransport is not None: 458 host2, port2 = machoTransport.transport.GetAddress().split(':') 459 tmp2 = host2.split('.') 460 if last is None or int(tmp2[3]) < last: 461 last = int(tmp2[3]) 462 lastNodeID = nodeID 463 464 if macho.mode == 'server': 465 self.polarisExternalTunneling = KeyVal(proxyNodeID=lastNodeID, address=self.session.ConnectToProxyServerService('machoNet', lastNodeID).GetPolarisExternalTunnelingAddress()) 466 return self.polarisExternalTunneling.address 467 address = self.namedtransports['tcp:raw:http'].GetESPAddress() 468 host, port = address.split(':') 469 tmp = host.split('.') 470 last = int(host.split('.')[3]) 471 host = '.'.join(tmp[:3]) + '.' + str(last) 472 return host + ':%d' % (self.defaultTunnelPortOffset + 1) 473 474 def SessionsFromClientIDs(self, clientIDs): 475 sessions = [] 476 for clientID in clientIDs: 477 transportID = self.transportIDbyClientID.get(clientID, None) 478 if transportID is not None: 479 transport = self.transportsByID.get(transportID, None) 480 if transport is not None: 481 sessions.append(transport._SessionFromClientID(clientID)) 482 483 return sessions 484 485 def GetFactory(self, factoryClass, args): 486 k = args[1] 487 if k not in self.factories: 488 f = CreateInstance(factoryClass, args) 489 if 'machoNet' in k: 490 f.MaxPacketSize = 1073741824 491 self.factories[k] = (factoryClass, args, f) 492 else: 493 n, a, f = self.factories[k] 494 if n != factoryClass or a != args: 495 log.LogTraceback() 496 raise RuntimeError('Thou shall not GetFactory with differing factoryName or args') 497 return f 498 499 def Run(self, memStream = None): 500 self.vipkeys = None 501 self.vipUsers = None 502 self.peopleWhoShouldntBeLoggedIn = {} 503 self.heartbeatInterval = 60 504 self.state = SERVICE_START_PENDING 505 service.Service.Run(self, memStream) 506 self.LogInfo('Running macho net') 507 self.stop = 0 508 self.clusterStartupPhase = False 509 self.clusterIsDesegmented = False 510 self.availableLoginSlots = 0 511 if macho.mode in ('server', 'proxy'): 512 machineName = blue.pyos.GetEnv().get('COMPUTERNAME', 'LOCALHOST').lower() 513 monolithmetrics.set_macho_machine_name(machineName) 514 if macho.mode == 'client' and not int(prefs.GetValue('http', 0)): 515 try: 516 del offsetMap[macho.mode]['tcp:raw:http'] 517 except Exception: 518 sys.exc_clear() 519 520 if macho.mode == 'client': 521 self.basePortNumber = self.defaultClientPortOffset 522 elif macho.mode == 'server': 523 self.basePortNumber = self.defaultServerPortOffset 524 elif macho.mode == 'proxy': 525 self.basePortNumber = self.defaultProxyPortOffset 526 else: 527 return 528 blue.net.SetMode(macho.mode) 529 self.LogInfo('Opening listen ports') 530 self.resurrectedNode = False 531 fixedhop = None 532 nodeIndex = 0 533 self.LogInfo('MachoNet is starting up using the NEW cluster configuration tables') 534 for arg in blue.pyos.GetArg(): 535 if arg.lower().startswith('/machohop='): 536 fixedhop = int(arg[len('/machohop='):]) 537 if arg.lower().startswith('/machoresurrection'): 538 self.resurrectedNode = True 539 540 gpsMap = get_gps_map() 541 offsets = offsetMap[macho.mode].items() 542 offsets = [ (k, v) for k, v in offsets if k in gpsMap[macho.mode] ] 543 offsets.sort(key=lambda t: t[1]) 544 transports = {} 545 if fixedhop is not None: 546 self.basePortNumber += fixedhop * offsetStep[macho.mode] 547 self.baseTunnelPortOffset += fixedhop * offsetStep[macho.mode] 548 nodeIndex = fixedhop 549 while True: 550 try: 551 for each, offset in offsets: 552 port = self.basePortNumber + offset 553 self.LogInfo('Open listen transport for ', each, ' on port ', port) 554 factory = self.GetFactory(gpsMap[macho.mode][each], (macho.mode == 'proxy' and each.endswith(':client'), each)) 555 transports[each] = factory.Listen(port) 556 557 except GPSAddressOccupied as e: 558 self.LogInfo('port ', port, ' occupied') 559 for each in transports.itervalues(): 560 each.Close('Desired address occupied in Run') 561 562 transports.clear() 563 sys.exc_clear() 564 else: 565 self.nodeIndex = nodeIndex 566 monolithmetrics.set_macho_node_index(self.nodeIndex) 567 monolithsentry.set_macho_node_index(self.nodeIndex) 568 break 569 570 if fixedhop is not None: 571 errreason = 'Failed to acquire the required port range running in fixed hop (=%d) mode. Terminating process' % (fixedhop,) 572 self.LogError(errreason) 573 log.Quit(errreason) 574 self.basePortNumber += offsetStep[macho.mode] 575 nodeIndex += 1 576 self.baseTunnelPortOffset += offsetStep[macho.mode] 577 if self.basePortNumber > self.defaultTunnelPortOffset: 578 raise RuntimeError("Something is seriously wrong. I've tried plenty of port number ranges, and still no luck.") 579 self.LogInfo('looping, base=', self.basePortNumber) 580 581 self.LogInfo('All transports acquired: ', transports, ', ', offsetMap[macho.mode]) 582 self.namedtransports = transports 583 self.SetStatusKeyValuePair('clusterStatus', -800) 584 self.SetStatusKeyValuePair('clusterStatusText', 'Sockets acqired') 585 if macho.mode == 'server' or macho.mode == 'proxy': 586 uthread.new(self.StatusSocketLoop) 587 if macho.mode == 'proxy': 588 self.nodeID = self.GetIDOfAddress(self.GetTransport('tcp:packet:client').GetInternalAddress(), clientMode=False) 589 print('nodeID=', self.nodeID) 590 elif macho.mode == 'server': 591 self.LogInfo('Registering node') 592 self.dbzcluster = self.DB2.GetSchema('zcluster') 593 self.dbcluster = self.DB2.GetSchema('cluster') 594 self.dbzuser = self.DB2.GetSchema('zuser') 595 retry = False 596 while True: 597 self.dbzcluster.Nodes_TrashLimbos(self.cleanupInterval) 598 host, port = self.GetTransport('tcp:packet:machoNet').GetExternalAddress().split(':') 599 internalAddress = self.GetTransport('tcp:packet:machoNet').GetInternalAddress().split(':')[0] 600 self.LogInfo('Registering to the cluster database. ClusterID:', CLUSTER_ID, 'machineID:', None, 'machineName:', machineName, 'nodeIndex:', nodeIndex) 601 ret = self.dbzcluster.Nodes_Register2(CLUSTER_ID, None, machineName, self.nodeIndex, 'SOL', macho.version, boot.version, GetBuildVersionAsInt(), blue.os.pid, self.resurrectedNode, internalAddress) 602 errreason = None 603 if ret == -1: 604 errreason = "zcluster.Nodes_Register refused to accept the registration. The cluster has already passed it's connectivity tests." 605 if not retry: 606 lastHeartBeat = self.DB2.SQL('SELECT lastHeartBeat= MAX(heartbeat) FROM zcluster.nodes')[0].lastHeartBeat 607 countDateTime = self.DB2.SQL("SELECT countDateTime = CONVERT(datetime, [value], 121) FROM zsystem.settings WHERE [group] = 'zcluster' AND [key] = 'TrashLimboNodes-LastCall'")[0].countDateTime 608 if lastHeartBeat is not None: 609 delay = 3000 + (self.cleanupInterval * const.SEC + lastHeartBeat - blue.os.GetWallclockTime()) / const.MSEC 610 if countDateTime is not None: 611 delay = max(delay, 3000 + (countDateTime + self.cleanupInterval * const.SEC - blue.os.GetWallclockTime()) / const.MSEC) 612 if delay > 0: 613 print(errreason) 614 self.LogWarn(errreason) 615 print('The previous run might however be dead, delaying and retrying after ', delay, ' ms...') 616 self.LogWarn('The previous run might however be dead, delaying and retrying after ', delay, ' ms...') 617 blue.pyos.synchro.SleepWallclock(delay) 618 retry = True 619 continue 620 elif ret == -2: 621 errreason = 'An app lock timeout occurred in zcluster.Nodes_Register2' 622 elif ret == -3: 623 errreason = ('The machine ', machineName, ' is not a member of the specified cluster. Configure zcluster.proxies and zcluster.machines correctly.') 624 elif ret == -4: 625 localip = GetPreferredHostByName(machineName) 626 if machineName != localip: 627 self.LogError('Did not find machine configured for', machineName, 'Retrying with the ip address:', localip) 628 machineName = localip 629 continue 630 else: 631 errreason = 'Could not find a machineName configured to match: %s. Configure zcluster.machines correctly.' % machineName 632 elif isinstance(ret, numbers.Integral): 633 errreason = 'An unspecified error occurred while registering the node: %d' % ret 634 if errreason: 635 self.LogError(errreason) 636 log.Quit(errreason) 637 break 638 639 self.nodeID = ret[0][0].nodeID 640 self.serviceMask = ret[0][0].serviceMask 641 self.clusterGroup = ret[0][0].clusterGroup 642 monolithmetrics.set_macho_cluster_group(self.clusterGroup) 643 monolithsentry.set_macho_cluster_group(self.clusterGroup) 644 self.machineID = ret[0][0].machineID 645 monolithsentry.set_macho_machine_id(self.machineID) 646 if self.serviceMask is None: 647 self.serviceMask = cluster.SERVICE_ALL 648 monolithmetrics.set_macho_service_mask(self.serviceMask) 649 monolithsentry.set_macho_service_mask(self.serviceMask) 650 self.premapped = prefs.clusterMode in ('TEST', 'LIVE') 651 for each in ret[1]: 652 self.serviceMappings[each.serviceName] = [each.serviceID, each.serviceMask] 653 self.serviceMaskByServiceID[each.serviceID] = each.serviceMask 654 self.serviceNameByServiceID[each.serviceID] = each.serviceName 655 656 for each in ret[2]: 657 self.serverNames[each.nodeID] = getattr(each, 'machineName', None) 658 659 if self.connectToCluster and not self.premapped: 660 polarisNodeID = self.GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) 661 self.LogInfo('Server Registration Complete. NodeID:', self.nodeID, ' serviceMask:', self.serviceMask, ', machineID:', self.machineID, ' nodeIndex:', self.nodeIndex) 662 else: 663 self.nodeID = None 664 monolithsentry.set_macho_node_id(self.nodeID) 665 if macho.mode != 'client': 666 monolithmetrics.event(title='Eve Node Started') 667 monolithmetrics.increment(metric='node.started') 668 if monolithconfig.on_server(): 669 launchdarkly.get_client().init_machonet_values(machine_id=self.machineID, node_id=self.nodeID, node_index=self.nodeIndex, service_mask=self.serviceMask, cluster_group=self.clusterGroup, cluster_name=monolithconfig.get_value('clusterName'), cluster_mode=monolithconfig.get_value('clusterMode'), role=mode) 670 self.SetStatusKeyValuePair('clusterStatus', -700) 671 self.SetStatusKeyValuePair('clusterStatusText', 'Node registered') 672 self.SetStatusCallBack('nodeID', lambda : self.nodeID) 673 if self.nodeID: 674 blue.net.SetMyNodeID(self.nodeID) 675 if macho.mode == 'server': 676 self.LogInfo('Establishing timers') 677 uthread.worker('machoNet::OnHeartBeat', self.OnHeartBeat) 678 uthread.worker('machoNet::OnCleanUp', self.OnCleanUp) 679 uthread.worker('machoNet::GracefulShutdown', self.__GracefulShutdownWorker) 680 self.acceptThreadCount = 1 681 self.LogInfo('Monkeypatching acceptThreadCount. Value now ' + str(self.acceptThreadCount)) 682 machoListeners = [ k for k in gpsMap[macho.mode].keys() if k.find(':packet:') >= 0 ] 683 machoListeners = [ k for k in machoListeners if k in offsetMap[macho.mode] ] 684 for eachTransport in machoListeners: 685 threadCount = self.acceptThreadCount 686 if 'machoNet' in eachTransport: 687 threadCount = 1 688 self.LogInfo('Launching ', threadCount, ' listen threads for transport ', eachTransport) 689 for eachThread in range(0, threadCount): 690 uthread.new(self.AcceptLoop, eachTransport).context = 'machoNet::AcceptLoop' 691 692 self.acceptThreadCount = 20 693 self.LogInfo('Restoring value of acceptThreadCount') 694 if macho.mode == 'server' and self.connectToCluster: 695 self.LogInfo('Connectivity[Startup]: Establishing communications with proxy servers') 696 layTracksTo = self._GetLayTracksTo() 697 if not layTracksTo: 698 self.LayTracksIfNeeded('127.0.0.1:%d' % (self.defaultProxyPortOffset + offsetMap['proxy']['tcp:packet:machoNet']), None, 'machoNet::Run') 699 else: 700 layTracksTo = list(layTracksTo) 701 random.shuffle(layTracksTo) 702 for each in layTracksTo: 703 self.LayTracksIfNeeded('%s:%d' % (each.ipAddress, each.port), each.nodeID, 'machoNet::Run') 704 705 self.SetStatusKeyValuePair('clusterStatus', -600) 706 self.SetStatusKeyValuePair('clusterStatusText', 'LayTracks done') 707 self.LogInfo('Creating handlers') 708 machoNet = weakref.proxy(self) 709 gpcsMap = get_gpcs_map() 710 for each in gpcsMap[macho.mode]: 711 nodes = [] 712 for d in gpcsMap[macho.mode][each]: 713 nodes.append(CreateInstance(d)) 714 715 for i in range(0, len(nodes)): 716 prior = machoNet 717 next = None 718 if i > 0: 719 prior = nodes[i - 1] 720 if i + 1 < len(nodes): 721 next = nodes[i + 1] 722 nodes[i].next = next 723 nodes[i].prior = prior 724 nodes[i].machoNet = machoNet 725 726 for i in range(0, len(nodes)): 727 current = nodes[i] 728 while current is not None and current != machoNet: 729 if not hasattr(nodes[i], 'ForwardNotifyDown') and current.prior is not None and hasattr(current.prior, 'NotifyDown'): 730 nodes[i].ForwardNotifyDown = current.prior.NotifyDown 731 if not hasattr(nodes[i], 'ForwardCallDown') and current.prior is not None and hasattr(current.prior, 'CallDown'): 732 nodes[i].ForwardCallDown = current.prior.CallDown 733 current = current.prior 734 735 current = nodes[i] 736 while current is not None and current != machoNet: 737 if not hasattr(nodes[i], 'ForwardNotifyUp') and current.next is not None and hasattr(current.next, 'NotifyUp'): 738 nodes[i].ForwardNotifyUp = current.next.NotifyUp 739 if not hasattr(nodes[i], 'ForwardCallUp') and current.next is not None and hasattr(current.next, 'CallUp'): 740 nodes[i].ForwardCallUp = current.next.CallUp 741 current = current.next 742 743 self.channelHandlersUp[each] = nodes[0] 744 if not hasattr(self.channelHandlersUp[each], 'CallUp') and hasattr(self.channelHandlersUp[each], 'ForwardCallUp'): 745 self.channelHandlersUp[each].CallUp = self.channelHandlersUp[each].ForwardCallUp 746 if not hasattr(self.channelHandlersUp[each], 'NotifyUp') and hasattr(self.channelHandlersUp[each], 'ForwardNotifyUp'): 747 self.channelHandlersUp[each].NotifyUp = self.channelHandlersUp[each].ForwardNotifyUp 748 self.channelHandlersDown[each] = nodes[len(nodes) - 1] 749 if not hasattr(self.channelHandlersDown[each], 'CallDown') and hasattr(self.channelHandlersDown[each], 'ForwardCallDown'): 750 self.channelHandlersDown[each].CallDown = self.channelHandlersDown[each].ForwardCallDown 751 if not hasattr(self.channelHandlersDown[each], 'NotifyDown') and hasattr(self.channelHandlersDown[each], 'ForwardNotifyDown'): 752 self.channelHandlersDown[each].NotifyDown = self.channelHandlersDown[each].ForwardNotifyDown 753 754 if macho.mode == 'server' and self.connectToCluster: 755 self.dbzmetric = self.DB2.GetSchema('zmetric') 756 uthread.new(self.__NodeWatcher).context = 'machoNet::NodeWatcher' 757 elif macho.mode == 'proxy': 758 uthread.new(self.__ClientSessionMaxTimePoll).context = 'machoNet::ClientSessionMaxTimePoll' 759 uthread.new(self.__TimeoutCallLoop).context = 'machoNet::TimeoutCallLoop' 760 uthread.new(self.__KATLoop).context = 'machoNet::KATLoop' 761 self.acceptStart = blue.os.GetWallclockTime() 762 self.SetStatusKeyValuePair('clusterStatus', -500) 763 self.SetStatusKeyValuePair('clusterStatusText', 'Machonet accept start') 764 self.stats = MachoRunTimeStats(self.transportsByID) 765 self.stats.Enable() 766 self.LogNotice('MachoNet started with acceptDelay of %d seconds. I will accept connections at %s' % (self.acceptDelay, utilFormat.FmtDateEng(self.acceptStart + self.acceptDelay * const.SEC))) 767 self.state = SERVICE_RUNNING 768 for each in self.__gpcsmethodnames__: 769 current = self.GetGPCS() 770 while 1: 771 if hasattr(current, each): 772 setattr(self, each, getattr(current, each)) 773 break 774 if not hasattr(current, 'prior') or current.prior is None: 775 self.LogError("MachoNet GPCS doesn't contain ", each, 'current=', current) 776 break 777 current = current.prior 778 779 if macho.mode == 'server' and self.connectToCluster: 780 self.dbzclient = self.DB2.GetSchema('zclient') 781 if macho.mode in ('server', 'proxy'): 782 self.LogNotice('Starting driftTimeMetrics on', macho.mode, machineName, nodeIndex) 783 from eve.server.script.net import timedriftmetrics 784 self._timeDriftMetrics = timedriftmetrics.TimeDriftMetrics(nodeIndex) 785 self._timeDriftMetrics.start_worker() 786 else: 787 self._timeDriftMetrics = None 788 sm.FavourMe(self.OnNodeDeath) 789 790 def OnCacheServiceReady(self): 791 monolithconfig.set_service_manager(sm) 792 793 def OnCacheServiceGroupChange(self, group): 794 monolithconfig.flush(group=group) 795 796 def _begin_publishing_status(self): 797 proxy_node_id = self.GetNodeID() 798 proxy_name = self.GetLocalHostName() 799 proxy_ipv4_address = self.namedtransports['tcp:packet:machoNet'].GetInternalAddress().split(':')[0] 800 proxy_ipv4 = struct.unpack('!I', socket.inet_aton(proxy_ipv4_address))[0] 801 esp_ipv4 = self.namedtransports['tcp:raw:http'].GetESPAddress() 802 metrics_port = _prometheus.prometheus_port 803 start_time = blue.sysinfo.processStartTime 804 while True: 805 try: 806 self.session.ConnectToRemoteService('externalQueueMgr').PublishProxyEvent(proxy_node_id, proxy_name, proxy_ipv4, esp_ipv4, metrics_port, start_time, os.getpid()) 807 except Exception as e: 808 log.LogException('Proxy node publish attempt failed') 809 810 blue.synchro.SleepWallclock(5000) 811 812 def IsClusterStarted(self): 813 return self.clusterStartupPhase 814 815 def OnClusterStarting(self, polarisNodeID): 816 with self.clusterStartLock: 817 if self.clusterStartupPhase: 818 return 819 self.SetStatusKeyValuePair('clusterStatus', -80) 820 self.SetStatusKeyValuePair('clusterStatusText', 'OnClusterStarting') 821 if macho.mode == 'proxy': 822 StartTasklet(self._begin_publishing_status) 823 uthread.new(self.__LoginQueueUpdater).context = 'machoNet::LoginQueueUpdater' 824 if macho.mode == 'server' and (self.GetNodeID() == self.GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) or self.IsResurrectedNode()): 825 config = self.dbzcluster.Cluster_DowntimeInfo() 826 if len(config): 827 when = config[0].shutdownWhen * const.MIN + blue.os.GetWallclockTime() / const.DAY * const.DAY 828 if when < blue.os.GetWallclockTime(): 829 when += const.DAY 830 if self.IsResurrectedNode(): 831 self.GracefulShutdown(config[0].shutdownUserID, when, 0, config[0].shutdownReason) 832 else: 833 self.session.ConnectToAllSolServerServices('machoNet').GracefulShutdown(config[0].shutdownUserID, when, 0, config[0].shutdownReason) 834 for transportID, transport in self.transportsByID.iteritems(): 835 nodeID = getattr(transport, 'nodeID', None) 836 if nodeID is None: 837 log.LogTraceback('Transport with nodeID == None connected before OnClusterStarting') 838 else: 839 isProxy = nodeID in self.transportIDbyProxyNodeID 840 isPolaris = nodeID == polarisNodeID 841 address, serviceMask = self.externalAddressesByNodeID[nodeID] 842 sm.ScatterEvent('OnNewNode', nodeID, address, isProxy, isPolaris, serviceMask) 843 844 if macho.mode == 'proxy': 845 cfg.LoadCfgData() 846 if macho.mode == 'server' and self.connectToCluster: 847 uthread.new(self.__SessionStatWatcher).context = 'machoNet::SessionStatWatcher' 848 uthread.new(self.__NodeLoadPush).context = 'machoNet::NodeLoadPush' 849 uthread.new(self.__DisconnectUnauthorizedUsers).context = 'machoNet::DisconnectUnauthorizedUsers' 850 sm.ScatterEvent('OnClusterStartup') 851 if macho.mode == 'proxy': 852 self.SetStatusKeyValuePair('clusterStatus', 0) 853 self.SetStatusKeyValuePair('clusterStatusText', 'Ready') 854 self.GetGlobalConfig() 855 self.clusterStartupPhase = True 856 857 def OnNewNode(self, nodeID, nodeAddress, isProxy, isPolaris, serviceMask): 858 if isProxy: 859 self._addressCache.UnmarkDeadNode(nodeID) 860 self.LogNotice('machoNet::OnNewNode: Calling LoadNodeToServiceMaskMapping(', ','.join([str(nodeID), str(serviceMask)]), ')') 861 self._addressCache.LoadNodeToServiceMaskMapping(nodeID, serviceMask) 862 if macho.mode == 'server': 863 clientListener = self.GetTransport('tcp:packet:client') 864 self.namedtransports['tcp:packet:client'] = None 865 if clientListener is not None: 866 clientListener.Close('The server has accepted a proxy connection, so it is closing all direct client connections.') 867 hasSpewed = False 868 for transportID in self.transportIDbyClientID.values(): 869 transportIDs = self.transportIDbyProxyNodeID.values() + self.transportIDbySolNodeID.values() 870 if transportID not in transportIDs: 871 if transportIDs: 872 if not hasSpewed: 873 hasSpewed = True 874 self.LogError('Disconnecting a node->node connection!?! Man do I smell deep shit going on here...') 875 self.LogError('Got a new node, and started iterating over transports to close them.') 876 self.LogError("Found a deadite that isn'it a known node->node transport.") 877 self.LogError('However, that is total crud, and we know it, because this is a sol server with proxy or sol connections.') 878 self.LogError("That means there are NO direct client connections, and we've got a situation here.") 879 self.LogError("Here's a stacktrace:") 880 log.LogTraceback() 881 self.LogError("And here's the new info we got:") 882 self.LogError('nodeID=', nodeID) 883 self.LogError('address=', nodeAddress) 884 self.LogError('isPolaris=', isPolaris) 885 self.LogError('I think I need an Alkaseltzer...') 886 self.LogError('self.transportsByID[transportID]=', self.transportsByID[transportID]) 887 else: 888 self.transportsByID[transportID].Close('The server has established communications with a proxy server. All direct communications are banned.') 889 890 def OnTimeReset(self, oldTime, newTime): 891 if self.acceptStart: 892 delta = newTime - oldTime 893 self.LogInfo('MachoNet (%s) acceptTime adjusted for DB time change' % macho.mode, self.acceptStart, delta, self.acceptStart + delta) 894 self.acceptStart += delta 895 else: 896 self.LogInfo('MachoNet (%s) acceptTime not adjusted for DB time change (not used yet?)' % macho.mode) 897 898 def ConnectivityTest(self, offset, proxyNodeCount, serverNodeCount): 899 if offset is not None: 900 if offset and not self.clockReset and macho.mode == 'proxy': 901 now = blue.os.GetWallclockTimeNow() 902 newnow = now + offset 903 print('Correcting clock... advancing ', float(offset) / float(const.SEC), ' secs') 904 blue.pyos.synchro.ResetClock(newnow) 905 sm.ScatterEvent('OnTimeReset', now, newnow) 906 self.clockReset = True 907 if len(self.transportIDbySolNodeID) != serverNodeCount: 908 self.LogInfo('ConnectivityTest Failed: expected ', serverNodeCount, ' server nodes, found ', len(self.transportIDbySolNodeID)) 909 if len(self.transportIDbyProxyNodeID) != proxyNodeCount: 910 self.LogInfo('ConnectivityTest Failed: expected ', proxyNodeCount, ' proxy nodes, found ', len(self.transportIDbyProxyNodeID)) 911 return len(self.transportIDbySolNodeID) == serverNodeCount and len(self.transportIDbyProxyNodeID) == proxyNodeCount 912 913 def OnObjectPublicAttributesUpdated(self, *args): 914 pass 915 916 def Stop(self, memStream = None, reason = 'The server is stopping'): 917 self.LogInfo('Stopping macho net node ', self.nodeID) 918 self.stop = 1 919 self.acceptStart = None 920 if macho.mode == 'server': 921 if self.nodeID is not None and sm.IsServiceRunning('DB2'): 922 self.DB2.SetAllowSynchronousCalls(1) 923 self.dbzcluster.Nodes_Trash(self.nodeID, reason, 1) 924 self.DB2.SetAllowSynchronousCalls(0) 925 while len(self.namedtransports): 926 ls = self.namedtransports.values() 927 self.namedtransports = {} 928 for each in ls: 929 if each is not None: 930 try: 931 each.disconnectsilently = getattr(each, 'disconnectsilently', 2) 932 each.Close('The machoNet service is stopping') 933 except Exception: 934 log.LogException() 935 self.LogInfo('Exception during Close ignored.') 936 sys.exc_clear() 937 938 while len(self.transportsByID): 939 ls = self.transportsByID.values() 940 self.transportsByID = {} 941 for each in ls: 942 try: 943 each.disconnectsilently = getattr(each, 'disconnectsilently', 2) 944 each.Close('The machoNet service is stopping') 945 except Exception: 946 log.LogException() 947 self.LogInfo('Exception during Close ignored.') 948 sys.exc_clear() 949 950 self.transportIDbySolNodeID = {} 951 self.transportIDbyAppNodeID = {} 952 self.transportIDbyProxyNodeID = {} 953 blue.net.ClearProxyNodes() 954 self.transportIDbyClientID = {} 955 self.transportIDbySessionID = {} 956 service.Service.Stop(self, memStream) 957 958 def GetHtmlStateDetails(self, k, v, detailed): 959 from carbon.common.script.util import htmlwriter 960 wr = htmlwriter.HtmlWriter() 961 if k == 'basePortNumber': 962 return ('Base Port', v) 963 if k == 'defaultTunnelPortOffset': 964 return ('Default Tunneling Port', v) 965 if k == '_addressCache': 966 if detailed: 967 nodeMap = self._addressCache.GetCachedServiceAddressMap() 968 hd = ['Node', 'Service', 'Addresses'] 969 li = [] 970 for node in sorted(nodeMap.keys()): 971 services = {} 972 for service, address in nodeMap[node]: 973 if service in services: 974 services[service].append(address) 975 else: 976 services[service] = [address] 977 978 for service in sorted(services.keys()): 979 addresses = sorted(services[service]) 980 li.append([node, service, ', '.join((str(addr) for addr in addresses))]) 981 982 return ('Address Cache', wr.GetTable(hd, li)) 983 else: 984 return ('Address Cache', self._addressCache.Report()) 985 elif k == 'namedtransports': 986 if detailed: 987 hd = ['Name', 'GPS Transport'] 988 li = [] 989 for each in v.iterkeys(): 990 li.append([each, v[each].__class__.__name__]) 991 992 return ('Named Transports', wr.GetTable(hd, li)) 993 else: 994 r = '' 995 comma = '' 996 for each in v.iterkeys(): 997 r = r + comma + each + '=' + v[each].__class__.__name__ 998 comma = ', ' 999 1000 return ('Named Transports', r) 1001 else: 1002 if k == 'dontLayTracksTo': 1003 r = '' 1004 comma = '' 1005 for each, at in v: 1006 r = r + comma + '%s at %s' % (each, at) 1007 comma = ', ' 1008 1009 return ('Currently Laying Tracks to, or have already laid tracks to ', r) 1010 if k == 'transportsByID': 1011 if detailed: 1012 hd = ['TRID', 1013 'Type', 1014 'Address', 1015 'NodeID/<br>ClientID', 1016 'Dependants', 1017 'Sessions', 1018 'Calls', 1019 'RTT<br>(ms)', 1020 'TDiff<br>(ms)'] 1021 li = [] 1022 for each in v.iterkeys(): 1023 if v[each].nodeID: 1024 theID = v[each].nodeID 1025 else: 1026 theID = v[each].clientID 1027 r = '' 1028 comma = '' 1029 for sess in v[each].sessions.iterkeys(): 1030 r = r + comma + strx(sess) + '=' + strx(v[each].sessions[sess].sid) 1031 comma = ', ' 1032 1033 if abs(v[each].estimatedRTT / const.MSEC) >= 100: 1034 rtt = '%d' % (v[each].estimatedRTT / const.MSEC) 1035 else: 1036 rtt = '%-2.1f' % (v[each].estimatedRTT / const.MSEC) 1037 if abs(v[each].timeDiff / const.MSEC) >= 100: 1038 tdiff = '%d' % (v[each].timeDiff / const.MSEC) 1039 else: 1040 tdiff = '%-2.1f' % (v[each].timeDiff / const.MSEC) 1041 li.append([each, 1042 v[each].transportName, 1043 v[each].transport.address, 1044 theID, 1045 strx(v[each].dependants), 1046 htmlwriter.Swing(r), 1047 strx(v[each].calls), 1048 rtt, 1049 tdiff]) 1050 li.sort() 1051 1052 return ('Transports by ID', wr.GetTable(hd, li)) 1053 else: 1054 r = '' 1055 comma = '' 1056 for each in v.iterkeys(): 1057 r = r + comma + strx(each) + '=' + strx(v[each].transport.address) 1058 comma = ', ' 1059 1060 return ('Transports by ID', r) 1061 elif k == 'transportIDbyClientID': 1062 if detailed: 1063 hd = ['ClientID', 'TRID'] 1064 li = [] 1065 for each in v.iterkeys(): 1066 li.append([each, v[each]]) 1067 1068 return ('TransportID by ClientID', wr.GetTable(hd, li)) 1069 else: 1070 r = '' 1071 comma = '' 1072 for each in v.iterkeys(): 1073 r = r + comma + strx(each) + '=>' + strx(v[each]) 1074 comma = ', ' 1075 1076 return ('TransportID by ClientID', r) 1077 elif k in ('channelHandlersUp', 'channelHandlersDown'): 1078 inout = 'Inbound' 1079 if k == 'channelHandlersDown': 1080 inout = 'Outbound' 1081 if detailed: 1082 hd = ['Channel', 'Handler'] 1083 li = [] 1084 for each in v.iterkeys(): 1085 li.append([each, v[each].__class__.__name__]) 1086 1087 return ('%s Channel Handlers' % inout, wr.GetTable(hd, li)) 1088 else: 1089 r = '' 1090 comma = '' 1091 for each in v.iterkeys(): 1092 r = r + comma + strx(each) + '=>' + v[each].__class__.__name__ 1093 comma = ', ' 1094 1095 return ('%s Channel Handlers' % inout, r) 1096 elif k == 'transportIDbySolNodeID': 1097 if detailed: 1098 hd = ['Sol NodeID', 'TRID'] 1099 li = [] 1100 for each in v.iterkeys(): 1101 li.append([each, v[each]]) 1102 1103 return ('TransportID by Sol NodeID', wr.GetTable(hd, li)) 1104 else: 1105 r = '' 1106 comma = '' 1107 for each in v.iterkeys(): 1108 r = r + comma + strx(each) + '=>' + strx(v[each]) 1109 comma = ', ' 1110 1111 return ('TransportID by Sol NodeID', r) 1112 elif k == 'transportIDbyProxy NodeID': 1113 if detailed: 1114 hd = ['Proxy NodeID', 'TRID'] 1115 li = [] 1116 for each in v.iterkeys(): 1117 li.append([each, v[each]]) 1118 1119 return ('TransportID by Proxy NodeID', wr.GetTable(hd, li)) 1120 else: 1121 r = '' 1122 comma = '' 1123 for each in v.iterkeys(): 1124 r = r + comma + strx(each) + '=>' + strx(v[each]) 1125 comma = ', ' 1126 1127 return ('TransportID by Proxy NodeID', r) 1128 else: 1129 if k == 'connectionLimit': 1130 return ('Max Connections', v) 1131 if k == 'callID': 1132 return ('Call ID', 'A running number used to identify blocking calls made on this server. The current value is %d, so %d blocking calls have been made since the service was started.' % (v, v - 1)) 1133 if k == 'transportID': 1134 return ('Transport ID', 'A running number used to identify transports connected to this server. The current value is %d, so %d transports have been used by the service so far.' % (v, v - 1)) 1135 if k == 'nodeID': 1136 return ('Node ID', 'For proxies and sol servers, a unique address that can be used for addressing packets to this server. Value=%d' % v) 1137 if k == 'mode': 1138 return ('Macho Mode', 'Whether this is a client, server, or proxy. Value=%s' % v) 1139 if k == 'stop': 1140 if v: 1141 v = 'The service is stopping, so it will be doing so.' 1142 else: 1143 v = 'The service is running, so it will leave them alone for now.' 1144 return ('Stop?', "Whether or not the service should shut down it's transports. %s" % v) 1145 1146 def add_session_transport_lookup(self, session_id, transport_id): 1147 self.transportIDbySessionID[session_id] = transport_id 1148 monolithmetrics.increment(metric=METRIC_SESSIONS) 1149 monolithmetrics.gauge(metric=METRIC_SESSIONS_TOTAL, value=len(self.transportIDbySessionID)) 1150 1151 def remove_session_transport_lookup(self, session_id): 1152 del self.transportIDbySessionID[session_id] 1153 monolithmetrics.decrement(metric=METRIC_SESSIONS) 1154 monolithmetrics.gauge(metric=METRIC_SESSIONS_TOTAL, value=len(self.transportIDbySessionID)) 1155 1156 def LogClientCall(self, session, objectName, method, args): 1157 pass 1158 1159 def PerformanceTest(self, *args): 1160 if 'dbdip' in args: 1161 self.dbzcluster.Cluster_DowntimeInfo() 1162 return ['dbdip', args] 1163 return ['server', args] 1164 1165 def ForwardCharacterNotification(self, charid, method, *args): 1166 callTimer = basesession.CallTimer('ForwardCharacterNotification::%s (Broadcast\\Client)' % method) 1167 try: 1168 clientIDs = [ sess.clientID for sess in basesession.FindSessions('charid', [charid]) ] 1169 if len(clientIDs): 1170 self.NarrowcastByClientIDsWithoutTheStars(clientIDs, method, args) 1171 else: 1172 machoNet = self.session.ConnectToSolServerService('machoNet') 1173 nodeID = machoNet.GetNodeFromAddress(cluster.SERVICE_CHARACTER, charid % const.CHARNODE_MOD) 1174 if nodeID == self.GetNodeID(): 1175 self.ReliableSinglecastByCharID(charid, method, *args) 1176 else: 1177 machoNet = self.session.ConnectToRemoteService('machoNet', nodeID) 1178 machoNet.ForwardCharacterNotification(charid, method, *args) 1179 finally: 1180 callTimer.Done() 1181 1182 def __str__(self): 1183 return '<MachoNet Service>' 1184 1185 def __repr__(self): 1186 return str(self) 1187 1188 def IsConnected(self): 1189 if self.GetTransport('ip:packet:server') and not self.authenticating: 1190 return 1 1191 else: 1192 return 0 1193 1194 def DisconnectFromServer(self): 1195 if macho.mode != 'client': 1196 raise AttributeError('The DisconnectFromServer method should only be called on the client') 1197 if 'ip:packet:server' in self.namedtransports: 1198 self.LogInfo('Closing connection to server') 1199 transport = self.namedtransports['ip:packet:server'] 1200 transport.disconnectsilently = getattr(transport, 'disconnectsilently', 1) 1201 transport.Close('User disconnected', 'CLIENTDISCONNECTED') 1202 1203 def GetServerStatus(self, address, forceQueueCheck = False): 1204 if self.authenticating or self.gettingServerStatus or not forceQueueCheck and address.upper() == self.clientLogonQueue.address and blue.os.GetWallclockTime() - (self.clientLogonQueue.timestamp or blue.os.GetWallclockTime()) < self.clientLogonQueuePollTime * const.SEC: 1205 raise UserError('AlreadyConnecting') 1206 self.gettingServerStatus = 1 1207 try: 1208 rawTransport = None 1209 rawTransportCached = False 1210 try: 1211 if address in self.clientHalfBakedTransports: 1212 rawTransport = self.clientHalfBakedTransports[address] 1213 del self.clientHalfBakedTransports[address] 1214 if rawTransport.IsClosed(): 1215 rawTransport = None 1216 if rawTransport is None: 1217 factory = self.GetFactory(get_gps_map()[macho.mode]['tcp:packet:server'], (0, 'tcp:packet:server')) 1218 rawTransport = factory.Connect(address) 1219 serverInfo = rawTransport.Authenticate(None, None, None) 1220 self.clientHalfBakedTransports[address] = rawTransport 1221 rawTransportCached = True 1222 return (lambda since = blue.os.GetWallclockTime(): sm.GetService('machoNet').GetOKMessage(since), serverInfo) 1223 except GPSBadAddress as e: 1224 if not rawTransportCached and rawTransport is not None: 1225 rawTransport.Close('GPSBadAddress', exception=e) 1226 sys.exc_clear() 1227 return (('/Carbon/MachoNet/ServerStatus/BadAddress', {}), {}) 1228 except GPSTransportClosed as e: 1229 sys.exc_clear() 1230 if not rawTransportCached and rawTransport is not None: 1231 rawTransport.Close('GPSTransportClosed', exception=e) 1232 if str(e.codename) != str(boot.codename) or e.reasonCode == 'HANDSHAKE_INCOMPATIBLERELEASE': 1233 reason = ('/Carbon/MachoNet/ServerStatus/IncompatibleRelease', {}) 1234 elif e.version != boot.version or e.reasonCode == 'HANDSHAKE_INCOMPATIBLEVERSION': 1235 reason = ('/Carbon/MachoNet/ServerStatus/IncompatibleVersion', {}) 1236 elif e.machoVersion != macho.version or e.reasonCode == 'HANDSHAKE_INCOMPATIBLEPROTOCOL': 1237 reason = ('/Carbon/MachoNet/ServerStatus/IncompatibleProtocol', {}) 1238 elif e.build != boot.build or e.reasonCode == 'HANDSHAKE_INCOMPATIBLEBUILD': 1239 reason = ('/Carbon/MachoNet/ServerStatus/IncompatibleBuild', {}) 1240 elif str(e.region) != str(boot.region) or e.reasonCode == 'HANDSHAKE_INCOMPATIBLEREGION': 1241 reason = ('/Carbon/MachoNet/ServerStatus/IncompatibleRegion', {}) 1242 elif e.reasonCode == 'ACL_SHUTTINGDOWN': 1243 reason = ('/Carbon/MachoNet/ServerStatus/ShuttingDown', {}) 1244 when = blue.os.GetWallclockTime() + random.randrange(15, 60) * const.SEC 1245 reason = lambda when = when, reason = reason: (reason if blue.os.GetWallclockTime() < when else None) 1246 elif e.reasonCode == 'ACL_NOTACCEPTING': 1247 reason = ('/Carbon/MachoNet/ServerStatus/NotAcceptingConnections', {}) 1248 when = blue.os.GetWallclockTime() + random.randrange(60, 300) * const.SEC 1249 reason = lambda when = when, reason = reason: (reason if blue.os.GetWallclockTime() < when else None) 1250 elif e.reasonCode == 'ACL_ACCEPTDELAY': 1251 reasonMessage = '/Carbon/MachoNet/ServerStatus/StartingUp' 1252 when = blue.os.GetWallclockTime() + (5 + e.reasonArgs['seconds']) * const.SEC 1253 reason = lambda when = when: ((reasonMessage, {'progress': when - blue.os.GetWallclockTime()}) if blue.os.GetWallclockTime() < when else None) 1254 elif e.reasonCode == 'ACL_PROXYFULL': 1255 if e.reasonArgs: 1256 reason = ('/Carbon/MachoNet/ServerStatus/ProxyFullWithLimit', {'limit': e.reasonArgs['limit']}) 1257 else: 1258 reason = ('/Carbon/MachoNet/ServerStatus/ProxyFull', {}) 1259 when = blue.os.GetWallclockTime() + random.randrange(30, 120) * const.SEC 1260 reason = lambda when = when, reason = reason: (reason if blue.os.GetWallclockTime() < when else None) 1261 elif e.reasonCode == 'ACL_PROXYNOTCONNECTED': 1262 reason = ('/Carbon/MachoNet/ServerStatus/ProxyNotConnected', {}) 1263 when = blue.os.GetWallclockTime() + random.randrange(30, 120) * const.SEC 1264 reason = lambda when = when, reason = reason: (reason if blue.os.GetWallclockTime() < when else None) 1265 elif e.reasonCode == 'ACL_IPADDRESSBAN': 1266 reason = ('/Carbon/MachoNet/ServerStatus/IPBanned', {}) 1267 elif e.origin == 'client': 1268 reason = ('/Carbon/MachoNet/ServerStatus/Unknown', {}) 1269 when = blue.os.GetWallclockTime() + random.randrange(60, 300) * const.SEC 1270 reason = lambda when = when, reason = reason: (reason if blue.os.GetWallclockTime() < when else None) 1271 else: 1272 reason = lambda since = blue.os.GetWallclockTime(): sm.GetService('machoNet').GetOKMessage(since) 1273 if e.origin == 'client': 1274 return (reason, {}) 1275 else: 1276 return (reason, {'boot_version': getattr(e, 'version', None), 1277 'boot_build': getattr(e, 'build', None), 1278 'boot_codename': str(getattr(e, 'codename', None)), 1279 'boot_region': str(getattr(e, 'region', None)), 1280 'cluster_usercount': getattr(e, 'loggedOnUserCount', None), 1281 'macho_version': getattr(e, 'machoVersion', None)}) 1282 except Exception as e: 1283 if not rawTransportCached and rawTransport is not None: 1284 rawTransport.Close('Disconnected', exception=e) 1285 raise e 1286 1287 finally: 1288 self.gettingServerStatus = 0 1289 1290 def GetValidClientCodeHash(self): 1291 if not hasattr(self, 'clientHash'): 1292 self.OnClientCodeUpdated() 1293 return getattr(self, 'clientHash') 1294 1295 def ReloadClientCodeHash(self): 1296 clientUpdate = self.dbzclient.CodeHashes_Select(None, GetBuildVersionAsInt()) 1297 if len(clientUpdate) > 1: 1298 self.LogError("Can't have more than one valid client code update") 1299 return 1300 if len(clientUpdate) == 0: 1301 return 1302 row = clientUpdate[0] 1303 return KeyVal(hash=row.hash, fileurl=row.fileUrl, build=row.build) 1304 1305 def OnClientCodeUpdated(self): 1306 hs = self.session.ConnectToRemoteService('machoNet').ReloadClientCodeHash() 1307 setattr(self, 'clientHash', hs) 1308 1309 def GetOKMessage(self, since): 1310 if self.clientLogonQueue.position == 1 and self.clientLogonQueue.reportedQueued: 1311 return ('/Carbon/MachoNet/ServerStatus/HeadOfQueue', {}) 1312 elif self.clientLogonQueue.position > 1: 1313 if not self.gettingServerStatus and blue.os.GetWallclockTime() - max(self.clientLogonQueue.timestamp or since, since) > self.clientLogonQueuePollTime * const.SEC: 1314 return None 1315 self.clientLogonQueue.reportedQueued = True 1316 return ('/Carbon/MachoNet/ServerStatus/InQueue', {'position': self.clientLogonQueue.position}) 1317 else: 1318 return ('/Carbon/MachoNet/ServerStatus/OK', {}) 1319 1320 def SetLogonQueuePosition(self, position): 1321 self.clientLogonQueue.position = position 1322 self.clientLogonQueue.timestamp = blue.os.GetWallclockTime() 1323 if position is None: 1324 self.clientLogonQueue.history = None 1325 elif self.clientLogonQueue.history is None: 1326 self.clientLogonQueue.history = KeyVal(initial=KeyVal(time=blue.os.GetWallclockTime(), pos=position), prev=KeyVal(time=blue.os.GetWallclockTime(), pos=position), last=KeyVal(time=blue.os.GetWallclockTime(), pos=position)) 1327 else: 1328 self.clientLogonQueue.history.prev = self.clientLogonQueue.history.last 1329 self.clientLogonQueue.history.last = KeyVal(time=blue.os.GetWallclockTime(), pos=position) 1330 1331 def GetPortOffsetStep(self, portTypeName): 1332 return offsetStep[portTypeName] 1333 1334 def GetGlobalConfig(self): 1335 if macho.mode != 'client' and self.globalConfig is None: 1336 with locks.TempLock('Getting the values, hold your horses', locks.RLock): 1337 if self.globalConfig is not None: 1338 return self.globalConfig 1339 if macho.mode == 'proxy': 1340 self.globalConfig = self.session.ConnectToSolServerService('machoNet').GetGlobalConfig() 1341 else: 1342 self.globalConfig = {k.strip():v for k, v in self.DB2.GetSchema('zsystem').GlobalConfig_Select()} 1343 self.LogInfo('Got the following global config:', self.globalConfig) 1344 sm.ScatterEvent('OnGlobalConfigChanged', self.globalConfig) 1345 elif macho.mode == 'client' and self.globalConfig is None: 1346 return {} 1347 return self.globalConfig 1348 1349 def InitClientGlobalConfig(self, globalConfig): 1350 if not monolithconfig.on_client(): 1351 self.LogWarn('InitClientGlobalConfig called on {}'.format(monolithconfig._get_boot_role())) 1352 return 1353 self.globalConfig = globalConfig 1354 self.LogNotice('Got global config values:', globalConfig) 1355 monolithconfig.defer_global_config_updates() 1356 sm.ScatterEvent('OnGlobalConfigChanged', globalConfig) 1357 1358 def SetGlobalConfigValue(self, key, value, isDelete = False, isUpdate = True, isUpdateClients = True): 1359 dbzsystem = self.DB2.GetSchema('zsystem') 1360 if isDelete: 1361 dbzsystem.GlobalConfig_Delete(key) 1362 self.LogNotice('Global config key deleted', key) 1363 value = None 1364 else: 1365 rs = dbzsystem.GlobalConfig_Select(key) 1366 if len(rs) == 0: 1367 dbzsystem.GlobalConfig_Insert(key, value) 1368 self.LogNotice('Global config added', key, value) 1369 else: 1370 dbzsystem.GlobalConfig_Update(key, value) 1371 self.LogNotice('Global config updated', key, value) 1372 if isUpdate: 1373 self.ClusterBroadcast('OnGlobalConfigUpdated', key, value) 1374 self.LogNotice('Global config updated cluster', key, value) 1375 if isUpdateClients: 1376 self.Broadcast('OnGlobalConfigUpdated', key, value) 1377 self.LogNotice('Global config updated connected clients', key, value) 1378 1379 def GetGlobalConfigValue(self, key): 1380 if key: 1381 row = self.DB2.GetSchema('zsystem').GlobalConfig_Select(key)[0] 1382 return (row.key, row.value) 1383 return ('', '') 1384 1385 def CalculatePortNumber(self, mapname, portTypeName): 1386 attributeName = 'default' + portTypeName[0].upper() + portTypeName[1:] + 'PortOffset' 1387 basePort = getattr(self, attributeName, None) 1388 if basePort: 1389 return basePort + offsetMap[portTypeName][mapname] 1390 raise Exception('Unsupported role') 1391 1392 def ConnectToAddress(self, address, mapname, destMachoRole, destinationNodeID = None, named = False, withReader = True, forcedPortNumber = None): 1393 port = forcedPortNumber or self.CalculatePortNumber(mapname, destMachoRole) 1394 address += ':' + str(port) 1395 transportID = self.transportID 1396 self.transportID = self.transportID + 1 1397 discard, destTypeName = mapname.split(':', 1) 1398 ipDestTypeName = 'ip:' + destTypeName 1399 factory = self.GetFactory(get_gps_map()[macho.mode][mapname], (0, mapname)) 1400 rawTransport = factory.Connect(address) 1401 transport = None 1402 try: 1403 transport = MachoTransport(transportID, rawTransport, ipDestTypeName, self) 1404 self.transportsByID[transportID] = transport 1405 if named: 1406 self.namedtransports[ipDestTypeName] = transport 1407 rawTransport = None 1408 transport._AssociateWithSession(forceNodeID=self.GetNodeID()) 1409 blue.net.EnumerateTransport(transport.transport.socket.getSocketDescriptor(), transport.transportID, transport.transportName, 0, 0, self.GetNodeID()) 1410 if withReader: 1411 self.StartTransportReader(transport) 1412 if destinationNodeID: 1413 self.transportIDbyAppNodeID[destinationNodeID] = transportID 1414 return transport 1415 except Exception as e: 1416 if transport is not None: 1417 transport.Close('Disconnected', exception=e) 1418 if rawTransport is not None: 1419 rawTransport.Close('Disconnected', exception=e) 1420 raise e 1421 1422 def StartTransportReader(self, transport): 1423 transport.currentReaders += 1 1424 uthread.new(self.TransportReader, transport).context = 'machoNet::Transport Reader::ConnectToAddress' 1425 1426 @telemetry.ZONE_METHOD 1427 def ConnectToServer(self, address, userName, password, sockettype = 'tcp', ssoToken = None): 1428 if macho.mode != 'client': 1429 raise AttributeError('The ConnectToServer method should only be called on the client') 1430 if self.authenticating: 1431 raise UserError('AlreadyConnecting') 1432 if address.upper() == self.clientLogonQueue.address and self.clientLogonQueue.position > 1: 1433 raise UserError('AlreadyConnecting') 1434 loginProgressSteps = 3 1435 self.authenticating = 2 1436 try: 1437 self.clientSessionID += 1 1438 try: 1439 dirty = 0 1440 for each in session.__persistvars__: 1441 if getattr(session, each) != session.GetDefaultValueOfAttribute(each): 1442 self.LogError('Connecting with a dirty session. ', each, ' was ', getattr(session, each), ' but should have been ', session.GetDefaultValueOfAttribute(each)) 1443 dirty = 1 1444 1445 if dirty: 1446 log.LogTraceback('Connecting with a dirty session') 1447 raise UserError('DirtySession') 1448 self.LogInfo('Connecting to server ', address) 1449 if 'ip:packet:server' in self.namedtransports: 1450 transport = self.namedtransports['ip:packet:server'] 1451 transport.disconnectsilently = getattr(transport, 'disconnectsilently', 1) 1452 transport.Close('Reconnecting') 1453 rawTransport = None 1454 transport = None 1455 try: 1456 mapname = '%s:packet:server' % sockettype 1457 sm.ScatterEvent('OnProcessLoginProgress', 'loginprogress::connecting', '', 1, loginProgressSteps) 1458 if address in self.clientHalfBakedTransports: 1459 rawTransport = self.clientHalfBakedTransports[address] 1460 del self.clientHalfBakedTransports[address] 1461 if rawTransport.IsClosed(): 1462 rawTransport = None 1463 if rawTransport is None: 1464 factory = self.GetFactory(get_gps_map()[macho.mode]['%s:packet:server' % sockettype], (0, mapname)) 1465 rawTransport = factory.Connect(address) 1466 sm.ScatterEvent('OnProcessLoginProgress', 'loginprogress::authenticating', '', 2, loginProgressSteps) 1467 self.LogInfo('Authenticating user ', userName) 1468 self.clientLogonQueue.address = address.upper() 1469 self.clientLogonQueue.timestamp = None 1470 self.clientLogonQueue.position = None 1471 self.clientLogonQueue.reportedQueued = False 1472 response = rawTransport.Authenticate(userName, password, ssoToken) 1473 sm.GetService('objectCaching').LoadCache(rawTransport.address) 1474 transportID = self.transportID 1475 self.transportID = self.transportID + 1 1476 transport = MachoTransport(transportID, rawTransport, 'ip:packet:server', self) 1477 self.transportsByID[transportID] = transport 1478 rawTransport = None 1479 self.namedtransports['ip:packet:server'] = transport 1480 transport.currentReaders += 1 1481 uthread.new(self.TransportReader, self.transportsByID[transportID]).context = 'machoNet::TransportReader::ConnectToServer' 1482 transport.userID = response['session_init']['userid'] 1483 self.myProxyNodeID = response['proxy_nodeid'] 1484 transport._AssociateWithSession() 1485 self.LogInfo('ConnectToServer established transport ', transport) 1486 blue.net.EnumerateTransport(transport.transport.socket.getSocketDescriptor(), transport.transportID, transport.transportName, transport.userID, 0, self.myProxyNodeID) 1487 blue.net.AddProxyNode(transport.transportID, self.myProxyNodeID) 1488 except UnMachoDestination as e: 1489 if transport is not None: 1490 transport.Close('UnMachoDestination', exception=e) 1491 if rawTransport is not None: 1492 rawTransport.Close('UnMachoDestination', exception=e) 1493 sm.ScatterEvent('OnConnectionRefused') 1494 exctype, exc, tb = sys.exc_info() 1495 try: 1496 raise_(UserError('OnConnectionRefused', {'what': e.payload}), None, tb) 1497 finally: 1498 exctype = None 1499 exc = None 1500 tb = None 1501 1502 except UserError as e: 1503 rawTransportCached = False 1504 if e.msg == 'AutLogonFailureTotalUsers': 1505 if rawTransport is not None and not rawTransport.IsClosed(): 1506 self.clientHalfBakedTransports[address] = rawTransport 1507 rawTransportCached = True 1508 self.SetLogonQueuePosition(e.dict.get('position', None)) 1509 if not rawTransportCached: 1510 if transport is not None: 1511 transport.Close('UserError', exception=e) 1512 if rawTransport is not None: 1513 rawTransport.Close('UserError', exception=e) 1514 sm.ScatterEvent('OnConnectionRefused') 1515 if e.msg != 'AuthenticationFailure': 1516 log.LogTraceback() 1517 raise 1518 except GPSBadAddress as e: 1519 if transport is not None: 1520 transport.Close('GPSBadAddress', exception=e) 1521 if rawTransport is not None: 1522 rawTransport.Close('GPSBadAddress', exception=e) 1523 sm.ScatterEvent('OnConnectionRefused') 1524 exctype, exc, tb = sys.exc_info() 1525 try: 1526 raise_(UserError('OnConnectionBadAddress', {'what': e.reason}), None, tb) 1527 finally: 1528 exctype = None 1529 exc = None 1530 tb = None 1531 1532 except GPSTransportClosed as e: 1533 exctype, exc, tb = sys.exc_info() 1534 rawTransportCached = False 1535 try: 1536 sm.ScatterEvent('OnConnectionRefused') 1537 if e.reason == 'Connected transport closed': 1538 raise_(UserError('OnConnectionFailed', {'what': 'The server was either not running or inaccessible'}), None, tb) 1539 else: 1540 if e.reasonCode in gpcs.MESSAGES: 1541 reason = gpcs.MESSAGES[e.reasonCode] 1542 reasonParameters = {'what': (const.UE_LOC, reason, e.reasonArgs)} 1543 err = UserError('OnConnectionFailed', reasonParameters) 1544 elif e.reasonCode not in cfg.messages: 1545 reasonParameters = {'what': e.reason} 1546 err = UserError('OnConnectionFailed', reasonParameters) 1547 else: 1548 err = UserError(e.reasonCode, e.reasonArgs) 1549 if e.reason == 'AutLogonFailureTotalUsers': 1550 if rawTransport is not None and not rawTransport.IsClosed(): 1551 self.clientHalfBakedTransports[address] = rawTransport 1552 rawTransportCached = True 1553 self.SetLogonQueuePosition(e.reasonArgs.get('position', None)) 1554 raise_(err, None, tb) 1555 finally: 1556 if not rawTransportCached: 1557 if transport is not None: 1558 transport.Close('GPSTransportClosed', exception=e) 1559 if rawTransport is not None: 1560 rawTransport.Close('GPSTransportClosed', exception=e) 1561 exctype = None 1562 exc = None 1563 tb = None 1564 1565 except Exception as e: 1566 if transport is not None: 1567 transport.Close('Disconnected', exception=e) 1568 if rawTransport is not None: 1569 rawTransport.Close('Disconnected', exception=e) 1570 self.LogInfo('ConnectToServer blew up big time') 1571 log.LogException() 1572 raise 1573 1574 self.serviceInfo = self.RemoteServiceCall(session, self.myProxyNodeID, 'machoNet', 'GetServiceInfo') 1575 sm.ChainEvent('ProcessLoginProgressDone') 1576 sm.ScatterEvent('OnProcessLoginProgress', 'loginprogress::done', '', 3, loginProgressSteps, response) 1577 finally: 1578 self.authenticating = 1 1579 1580 global_config = response.get('config_vals', {}) 1581 self.InitClientGlobalConfig(global_config) 1582 return response 1583 finally: 1584 self.authenticating = 0 1585 1586 def GetBasePortNumber(self): 1587 return self.basePortNumber 1588 1589 def GetClusterGameStatistics(self, key, default): 1590 return default 1591 1592 def GetClusterGameStatisticsForClient(self, key, default): 1593 return default 1594 1595 def GetConnectionProperties(self): 1596 retval = {} 1597 if macho.mode == 'proxy': 1598 retval['vipMode'] = self.vipMode 1599 retval['limit'] = self.connectionLimit 1600 retval['maxLoginsPerMinute'] = self.maxLoginsPerMinute 1601 retval['pendingConnections'] = len(self.serverLogonQueue) 1602 retval['availableLoginSlots'] = self.availableLoginSlots 1603 retval['clients'] = len(self.transportIDbyClientID) 1604 retval['proxies'] = len(self.transportIDbyProxyNodeID) 1605 retval['servers'] = len(self.transportIDbySolNodeID) 1606 retval['solNodeIDs'] = self.transportIDbySolNodeID.keys() 1607 retval['proxyNodeIDs'] = self.transportIDbyProxyNodeID.keys() 1608 retval['espaddress'] = self.namedtransports['tcp:raw:http'].GetESPAddress() 1609 retval['externaladdress'] = self.namedtransports['tcp:packet:client'].GetExternalAddress() 1610 retval['internaladdress'] = self.namedtransports['tcp:packet:machoNet'].GetInternalAddress() 1611 retval['bannedIPs'] = self.bannedIPs 1612 retval['acceptDelay'] = self.acceptDelay 1613 retval['acceptStart'] = self.acceptStart 1614 elif macho.mode == 'client': 1615 retval['bytes_out'] = self.dataSent.Current() 1616 retval['bytes_in'] = self.dataReceived.Current() 1617 retval['packets_out'] = self.dataSent.Count() 1618 retval['packets_in'] = self.dataReceived.Count() 1619 retval['blocking_calls'] = self.blockingCallTimes.Count() 1620 retval['blocking_call_times'] = self.blockingCallTimes.Current() 1621 retval['calls_outstanding'] = len(self.calls) 1622 else: 1623 retval['limit'] = self.connectionLimit 1624 retval['maxLoginsPerMinute'] = self.maxLoginsPerMinute 1625 retval['proxiedclients'] = len(self.transportIDbyClientID) 1626 retval['proxies'] = len(self.transportIDbyProxyNodeID) 1627 retval['servers'] = len(self.transportIDbySolNodeID) 1628 retval['solNodeIDs'] = self.transportIDbySolNodeID.keys() 1629 retval['proxyNodeIDs'] = self.transportIDbyProxyNodeID.keys() 1630 retval['address'] = self.namedtransports['tcp:packet:machoNet'].GetExternalAddress() 1631 retval['internaladdress'] = self.namedtransports['tcp:packet:machoNet'].GetInternalAddress() 1632 retval['acceptDelay'] = self.acceptDelay 1633 if macho.mode != 'client': 1634 retval['blue.os.pid'] = blue.os.pid 1635 created = blue.sysinfo.processStartTime 1636 kernel = blue.sysinfo.GetProcessTimes().systemTime 1637 user = blue.sysinfo.GetProcessTimes().userTime 1638 cputime = kernel + user 1639 uptime = blue.os.GetWallclockTime() - created 1640 memory = blue.sysinfo.GetMemory() 1641 retval['blue.os.started'] = utilFormat.FmtDateEng(created) 1642 retval['blue.os.uptime'] = utilFormat.FmtTimeEng(uptime) 1643 retval['blue.os.cpuload'] = self.GetCPULoad() 1644 retval['blue.os.cputime'] = utilFormat.FmtTimeEng(cputime) 1645 retval['blue.os.kerneltime'] = utilFormat.FmtTimeEng(kernel) 1646 retval['blue.os.usertime'] = utilFormat.FmtTimeEng(user) 1647 retval['blue.os.ramReal'] = memory.totalPhysical - memory.availablePhysical 1648 retval['blue.w32.PagefileUsage'] = memory.pageFile 1649 retval['blue.os.simDilation'] = blue.os.simDilation 1650 return retval 1651 1652 def GetAddressAndStaticTagsForPrometheus(self): 1653 if macho.mode == 'client': 1654 return {} 1655 if sm.IsServiceRunning('tcpRawProxyService'): 1656 address = self.namedtransports['tcp:raw:http'].GetESPAddress() 1657 external_address = self.namedtransports['tcp:packet:client'].GetExternalAddress() 1658 internal_address = self.namedtransports['tcp:packet:machoNet'].GetInternalAddress() 1659 else: 1660 first_proxy = min(self.GetConnectedProxyNodes()) 1661 tcp_proxy = self.session.ConnectToRemoteService('tcpRawProxyService', first_proxy) 1662 host, ports = tcp_proxy.GetESPTunnelingAddressByNodeID() 1663 port = ports.get(self.nodeID) 1664 address = '%s:%s' % (host, port) 1665 external_address = self.namedtransports['tcp:packet:machoNet'].GetExternalAddress() 1666 internal_address = self.namedtransports['tcp:packet:machoNet'].GetInternalAddress() 1667 hostname = self.GetLocalHostName() 1668 info = {'node_id': self.nodeID, 1669 'esp_tunnel_address': address, 1670 'role': macho.mode, 1671 'hostname': hostname, 1672 'machine_id': self.machineID, 1673 'metrics_port': _prometheus.prometheus_port, 1674 'node_index': self.nodeIndex, 1675 'service_mask': self.serviceMask, 1676 'external_address': external_address.split(':')[0], 1677 'internal_address': internal_address.split(':')[0]} 1678 if macho.mode == 'server': 1679 info['cluster_group'] = self.clusterGroup 1680 else: 1681 info['cluster_group'] = 0 1682 return info 1683 1684 def GetCPULoad(self): 1685 now = blue.os.GetWallclockTime() 1686 then = now - const.MIN * 5 1687 total = 0 1688 lastTime = now 1689 i = len(blue.pyos.cpuUsage) - 1 1690 while i > 0 and blue.pyos.cpuUsage[i].timestamp >= then: 1691 total += blue.pyos.cpuUsage[i].userProcessCpuUsage 1692 lastTime = blue.pyos.cpuUsage[i].timestamp 1693 i -= 1 1694 1695 if lastTime == now: 1696 return 0.0 1697 return min(100.0 * (float(total) / (float(now) - float(lastTime))), 100.0) 1698 1699 def GetNetworkStats(self, socketServers): 1700 import _socket 1701 if not socketServers: 1702 stackless.set_schedule_callback(None) 1703 self.scheduleCount = 0 1704 return 1705 if self.scheduleCount == 0: 1706 1707 def ScheduleCounter(prev, next): 1708 self.scheduleCount += 1 1709 1710 stackless.set_schedule_callback(ScheduleCounter) 1711 stats = _socket.getstats() 1712 headings = stats.keys() 1713 line = [ stats[k] for k in headings ] 1714 1715 def replace(a, b): 1716 headings[headings.index(a)] = b 1717 1718 replace('BytesReceived', 'bytesRecvd') 1719 replace('BytesSent', 'bytesSent') 1720 replace('PacketsSent', 'countWrites') 1721 replace('PacketsReceived', 'countReads') 1722 stats = {'global socket stats': [headings, line]} 1723 return KeyVal(stats=stats, scheduleCount=self.scheduleCount, taskletCount=len(bluepy.tasklets)) 1724 1725 def GetBlockingCallStats(self): 1726 min = self.callsCountMin 1727 max = self.callsCountMax 1728 self.callsCountMin = self.callsCountMax = len(self.calls) 1729 return (min, max) 1730 1731 def GetMetrics(self): 1732 from carbon.common.script.net.BroadcastStuffGPCS import CoreBroadcastStuff 1733 from carbon.common.script.net.ServiceCallGPCS import CoreServiceCall 1734 kv = KeyVal({'CallsReceived': {}, 1735 'CallsSent': {}, 1736 'NotificationsReceived': {}, 1737 'NotificationsSent': {}}) 1738 for functionName, receivedCount in CoreServiceCall.__recvServiceCallCount__.iteritems(): 1739 kv.CallsReceived[functionName] = receivedCount 1740 1741 for functionName, sentCount in CoreServiceCall.__sendServiceCallCount__.iteritems(): 1742 kv.CallsSent[functionName] = sentCount 1743 1744 for functionName, receivedCount in CoreBroadcastStuff.__notifyRecvCount__.iteritems(): 1745 kv.NotificationsReceived[functionName] = receivedCount 1746 1747 for functionName, sentCount in CoreServiceCall.__sendNotifyCount__.iteritems(): 1748 kv.NotificationsSent[functionName] = sentCount 1749 1750 return kv 1751 1752 def SetConnectionProperty(self, k, v, doPersist = True): 1753 if k == 'vipMode': 1754 monolithmetrics.event('Eve VIP Mode Changed', 'VIP Mode set to: %s' % v) 1755 if k == 'limit': 1756 self.connectionLimit = v 1757 elif k in ('maxLoginsPerMinute', 'vipMode'): 1758 if doPersist: 1759 setattr(self, k, v) 1760 else: 1761 self.__dict__[k] = v 1762 if k == 'maxLoginsPerMinute': 1763 self.availableLoginSlots = min(self.availableLoginSlots, self.maxLoginsPerMinute) 1764 elif k == 'maxTrialUsers': 1765 self.maxTrialUsers = v 1766 elif k in ('bannedIPs', 'allowedIPs', 'useIPACL', 'acceptDelay'): 1767 if doPersist: 1768 setattr(self, k, v) 1769 else: 1770 self.__dict__[k] = v 1771 elif k == 'bannedIPs': 1772 self.bannedIPs = v 1773 bannz0red = [] 1774 for clientID, transportID in self.transportIDbyClientID.items(): 1775 if transportID not in self.transportsByID: 1776 continue 1777 transport = self.transportsByID[transportID] 1778 for bannedIP in v: 1779 if transport.transport.address.startswith(bannedIP): 1780 bannz0red.append(transport) 1781 break 1782 1783 for transport in bannz0red: 1784 transport.Close('Your IP address has been banned', 'ACL_IPADDRESSBAN', {'address': transport.transport.address.split(':')[0]}) 1785 1786 elif k == 'acceptDelay': 1787 self.acceptDelay = v 1788 1789 def CheckACL(self, address, espCheck = False): 1790 if not espCheck and self.shutdown is not None: 1791 if self.shutdown.when < blue.os.GetWallclockTime(): 1792 self.LogInfo('Rejecting connection from ', address, " because we're shutting down") 1793 return ('The cluster is shutting down', 'ACL_SHUTTINGDOWN') 1794 if self.shutdown.when - blue.os.GetWallclockTime() < 3 * const.MIN: 1795 self.LogInfo('Rejecting connection from ', address, " because we're shutting down in less than 3 minutes") 1796 return ('The cluster is shutting down in a few moments', 'ACL_SHUTTINGDOWN', {'when': (17, self.shutdown.when)}) 1797 if not espCheck and self.acceptDelay and blue.os.GetWallclockTime() - self.acceptStart < const.SEC * self.acceptDelay: 1798 self.LogInfo('Rejecting connection from ', address, " because we're still starting and our accept delay hasn't passed") 1799 return ('Starting up...(%d sec.)' % (self.acceptDelay - (blue.os.GetWallclockTime() - self.acceptStart) / const.SEC), 'ACL_ACCEPTDELAY', {'seconds': self.acceptDelay - (blue.os.GetWallclockTime() - self.acceptStart) / const.SEC}) 1800 if not self.clusterStartupPhase: 1801 if prefs.clusterMode in ('LIVE', 'TEST'): 1802 self.LogWarn('Accept Delay too low, the accept delay should represent the time from startup until the first user can connect') 1803 self.LogInfo('Rejecting connection from ', address, " because we're still starting") 1804 return ('Starting up...', 'ACL_ACCEPTDELAY', {'seconds': 0}) 1805 if not self.transportIDbySolNodeID: 1806 self.LogInfo('Rejecting connection from ', address, ' because we do not have a sol server connection') 1807 return ('Proxy not connected', 'ACL_PROXYNOTCONNECTED') 1808 if espCheck: 1809 return 1810 if self.connectionLimit == 0 or self.maxLoginsPerMinute == 0: 1811 self.LogInfo('Rejecting connection from ', address, " because we're not accepting any connections") 1812 return ('Not accepting', 'ACL_NOTACCEPTING') 1813 for each in self.bannedIPs: 1814 if address.startswith(each): 1815 self.LogInfo('Rejecting connection from ', address, ' because the IP address is banned by the ', each, ' rule') 1816 return ('Your IP address has been banned', 'ACL_IPADDRESSBAN', {'address': address.split(':')[0]}) 1817 1818 def GetSessionCounts(self, attr = None, default = None): 1819 if default is None: 1820 default = [0, {}] 1821 sessionMgr = sm.services.get('sessionMgr', None) 1822 if sessionMgr is None: 1823 if attr is not None: 1824 return default 1825 return {} 1826 else: 1827 stats = sessionMgr.GetSessionStatistics() 1828 if attr is None: 1829 return stats 1830 return stats.get(attr, default) 1831 1832 def SetClusterSessionCounts(self, clusterSessionStatistics): 1833 now = blue.os.GetWallclockTime() 1834 self.clusterSessionStatsHistory.append((now, clusterSessionStatistics)) 1835 if len(self.clusterSessionStatsHistory) > self.proxyStatSmoothie: 1836 self.clusterSessionStatsHistory = self.clusterSessionStatsHistory[1:] 1837 1838 def GetClusterSessionCounts(self, attr = None, default = None): 1839 if default is None: 1840 default = [0, {}] 1841 if len(self.clusterSessionStatsHistory) == 0: 1842 if attr is not None: 1843 return default 1844 return {} 1845 else: 1846 stats = self.clusterSessionStatsHistory[-1][1] 1847 if attr is None: 1848 return stats 1849 return stats.get(attr, default) 1850 1851 def __SessionStatWatcher(self): 1852 allProxies = None 1853 while not getattr(self, 'stop', True): 1854 blue.pyos.synchro.SleepWallclock(1000 * self.proxyStatPollInterval) 1855 if self.GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) == self.GetNodeID(): 1856 self.LogInfo('SessionStatWatcher collecting session statistics from all proxies') 1857 try: 1858 if allProxies is None: 1859 allProxies = self.session.ConnectToAllProxyServerServices('machoNet') 1860 self.__SessionStatWatcher2(allProxies) 1861 except Exception: 1862 self.LogError('Exception during session stat stuff') 1863 log.LogException() 1864 sys.exc_clear() 1865 1866 def __SessionStatWatcher2(self, allProxies): 1867 if self.transportIDbyProxyNodeID: 1868 allStats = allProxies.GetSessionCounts() 1869 summ = {} 1870 for isError, nodeID, stats in allStats: 1871 if not isError: 1872 for attribute, countsTuple in stats.iteritems(): 1873 if attribute not in summ: 1874 summ[attribute] = [0, {}] 1875 valueCounts = countsTuple[1] 1876 for attrValue, count in valueCounts.iteritems(): 1877 summ[attribute][0] += count 1878 if attrValue not in summ[attribute][1]: 1879 summ[attribute][1][attrValue] = count 1880 else: 1881 summ[attribute][1][attrValue] += count 1882 1883 else: 1884 log.LogTraceback('Exception during proxy stat poll for proxy %s' % str(nodeID)) 1885 sys.exc_clear() 1886 return 1887 1888 self.clusterSessionStatistics = summ 1889 self._StoreMetricsToDB(self.clusterSessionStatistics) 1890 allProxies.SetClusterSessionCounts(self.clusterSessionStatistics) 1891 1892 def GetNumberOfEVEUsersConnected(self): 1893 return self.clusterSessionStatistics.get('EVE:Online', [None, None])[0] 1894 1895 def _StoreMetricsToDB(self, metrics): 1896 try: 1897 for virtualAttr, zmetric in self.metricsMap.iteritems(): 1898 count = metrics[virtualAttr][0] if virtualAttr in metrics else 0 1899 if count > 0: 1900 try: 1901 self.dbzmetric.SimpleCounters_Insert(zmetric, count) 1902 except Exception: 1903 log.LogException('Error writing metric %s (%s) to zmetric.simpleCounters' % (zmetric, virtualAttr)) 1904 sys.exc_clear() 1905 1906 except Exception: 1907 log.LogException() 1908 except: 1909 log.LogException() 1910 raise 1911 1912 def _GetLayTracksTo(self): 1913 with self.paonLock: 1914 if self.paon is None or blue.os.GetWallclockTime() - self.paon[1] > 5 * const.SEC: 1915 self.LogInfo('Connectivity[PAON]: Fetching node/proxy info') 1916 rowDescriptor = blue.DBRowDescriptor((('nodeID', const.DBTYPE_I4), ('ipAddress', const.DBTYPE_STR), ('port', const.DBTYPE_I4))) 1917 layTracksTo = crowset.CRowset(rowDescriptor, []) 1918 proxies = self.dbzcluster.Proxies_Select() 1919 if not proxies and macho.mode == 'server': 1920 ip = self.namedtransports['tcp:packet:machoNet'].GetExternalAddress().split(':')[0] 1921 clientport = self.defaultProxyPortOffset + offsetMap[macho.mode]['tcp:packet:client'] 1922 serverport = self.defaultProxyPortOffset + offsetMap[macho.mode]['tcp:packet:machoNet'] 1923 proxyNodeID = self.GetIDOfAddress(ip + ':%d' % clientport, clientMode=False) 1924 layTracksTo.InsertNew([proxyNodeID, ip, serverport]) 1925 self.LogWarn('Connectivity[PAON]: Warning: No proxy servers registered in DB, defaulting to localhost') 1926 for proxy in proxies: 1927 if proxy.enabled == 1: 1928 for i in xrange(proxy.nodeCount): 1929 proxyIpAddress = GetPreferredHostByName(proxy.ipAddress) 1930 clientport = self.defaultProxyPortOffset + offsetMap[macho.mode]['tcp:packet:client'] + 1000 * i 1931 serverport = self.defaultProxyPortOffset + offsetMap[macho.mode]['tcp:packet:machoNet'] + 1000 * i 1932 proxyNodeID = self.GetIDOfAddress('%s:%d' % (proxyIpAddress, clientport), clientMode=False) 1933 if self.GetNodeID() != proxyNodeID: 1934 self.LogInfo('Connectivity[PAON]: Laying tracks to self=', self.GetNodeID(), ' proxy=', proxyNodeID) 1935 layTracksTo.InsertNew([proxyNodeID, proxyIpAddress, serverport]) 1936 1937 for sol in self.dbzcluster.Nodes_SelectOthers2(self.nodeID): 1938 port = self.defaultServerPortOffset + offsetStep[macho.mode] * sol.nodeIndex 1939 port += offsetMap[macho.mode]['tcp:packet:machoNet'] 1940 self.LogInfo('Connectivity[PAON]: Laying tracks to self=', self.GetNodeID(), ' sol=', sol.nodeID, ' at:', sol.machineName, ':', port) 1941 layTracksTo.InsertNew([sol.nodeID, sol.machineName, port]) 1942 1943 self.paon = (layTracksTo, blue.os.GetWallclockTime()) 1944 else: 1945 self.LogInfo('Connectivity[PAON]: Node/proxy info is up to date') 1946 self.LogInfo('Connectivity[PAON]: returning ', len(self.paon[0]), ' proxies and servers') 1947 return self.paon[0] 1948 1949 def OnRefreshConnectivity(self): 1950 self.RefreshConnectivity() 1951 1952 def RefreshConnectivity(self): 1953 if macho.mode == 'server': 1954 self.LogInfo('Connectivity[RefreshConnectivity]: Refreshing communications with proxy servers') 1955 layTracksTo = self._GetLayTracksTo() 1956 if not layTracksTo: 1957 addresses = [('127.0.0.1:%d' % (self.defaultProxyPortOffset + offsetMap['proxy']['tcp:packet:machoNet']), None)] 1958 else: 1959 addresses = [] 1960 for each in layTracksTo: 1961 addresses.append(('%s:%d' % (each.ipAddress, each.port), each.nodeID)) 1962 1963 current = [] 1964 for each in self.transportsByID.itervalues(): 1965 current.append(each) 1966 1967 for node, addr in self.LayingTracksTo(): 1968 self.LogInfo('Connectivity[RefreshConnectivity]: Already laying tracks to node ', node, ' at ', addr) 1969 1970 random.shuffle(addresses) 1971 for address, nodeID in addresses: 1972 self.LayTracksIfNeeded(address, nodeID, 'RefreshConnectivity') 1973 1974 self.LogInfo('Connectivity[RefreshConnectivity]: End') 1975 return len(layTracksTo) 1976 1977 def GracefulShutdown(self, userID, when, duration, explanationLabel): 1978 if when: 1979 self.LogNotice('Graceful Shutdown requested by ', userID, ' to occur at ', utilFormat.FmtDateEng(when), ', duration ', duration or 0, ' minutes, explanation: ', explanationLabel) 1980 if when < blue.os.GetWallclockTime() + const.HOUR: 1981 notify = blue.os.GetWallclockTime() 1982 else: 1983 notify = when - const.HOUR 1984 self.shutdown = KeyVal(notify=notify, userID=userID, when=when, duration=duration, explanationLabel=explanationLabel) 1985 else: 1986 self.LogNotice('Graceful Shutdown cancelled by ', userID, ', explanation: ', explanationLabel) 1987 self.shutdown = None 1988 if self.GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) == self.GetNodeID(): 1989 self.Broadcast('OnClusterShutdownCancelled', explanationLabel) 1990 sm.ScatterEvent('OnClusterShutdownCancelled', explanationLabel) 1991 self.dbzcluster.Broadcasts_Insert(userID, 'LOCALIZED:' + explanationLabel) 1992 1993 def __GracefulShutdownWorker(self): 1994 pollingInterval = 10 1995 while not getattr(self, 'stop', True): 1996 try: 1997 if self.shutdown is not None and self.GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) == self.GetNodeID(): 1998 self.LogInfo('GracefulShutdownWorker: Probing') 1999 now = blue.os.GetWallclockTime() 2000 if self.shutdown.notify is not None: 2001 self.__BroadcastShutdownInformation(now, pollingInterval) 2002 elif self.shutdown.when < now: 2003 self.__PerformGracefulShutdown() 2004 except Exception: 2005 log.LogException('Exception in shutdown worker') 2006 sys.exc_clear() 2007 2008 blue.pyos.synchro.SleepWallclock(pollingInterval * 1000) 2009 2010 def __BroadcastShutdownInformation(self, now, pollingInterval): 2011 self.LogInfo('GracefulShutdownWorker: Checking Notification, now=', now, ', notify=', self.shutdown.notify) 2012 sortOfNow = now + pollingInterval * 2 * const.SEC 2013 if sortOfNow <= self.shutdown.notify: 2014 return 2015 self.LogInfo('GracefulShutdownWorker: Sending Notification') 2016 if self.shutdown.when - sortOfNow > 15 * const.MIN: 2017 self.LogInfo('GracefulShutdownWorker: Prepping 15 minute Notification') 2018 self.shutdown.notify = self.shutdown.when - 15 * const.MIN 2019 elif self.shutdown.when - sortOfNow > 5 * const.MIN: 2020 self.LogInfo('GracefulShutdownWorker: Prepping 5 minute Notification') 2021 self.shutdown.notify = self.shutdown.when - 5 * const.MIN 2022 else: 2023 self.LogInfo('GracefulShutdownWorker: Prepping actual shutdown') 2024 self.shutdown.notify = None 2025 args = (self.shutdown.explanationLabel, self.shutdown.when, self.shutdown.duration) 2026 self.Broadcast('OnClusterShutdownInitiated', *args) 2027 sm.ScatterEvent('OnClusterShutdownInitiated', *args) 2028 self.dbzcluster.Broadcasts_Insert(self.shutdown.userID, self.shutdown.explanationLabel) 2029 2030 def __PerformGracefulShutdown(self): 2031 self.LogNotice('GracefulShutdownWorker: Shutdown Initiated') 2032 eventID = None 2033 nodeCount = proxyCount = -1 2034 exceptionCount = 0 2035 try: 2036 counterDate = self.DB2.SaveMetricsDate() 2037 if counterDate: 2038 eventID = self.dbLog.LogProcessClusterShutdownStarted('machoNet') 2039 dbzmetric = self.DB2.GetSchema('zmetric') 2040 alles = self.session.ConnectToAllSolServerServices('machoNet').GetMetrics() 2041 nodeCount = len(alles) 2042 summaryReceived = {} 2043 summarySent = {} 2044 for isException, nodeID, kv in alles: 2045 blue.pyos.BeNice() 2046 if isException: 2047 exceptionCount += 1 2048 continue 2049 for functionName, receivedCount in kv.CallsReceived.iteritems(): 2050 if receivedCount > 0: 2051 if functionName in summaryReceived: 2052 summaryReceived[functionName] += receivedCount 2053 else: 2054 summaryReceived[functionName] = receivedCount 2055 2056 for functionName, receivedCount in kv.NotificationsReceived.iteritems(): 2057 if receivedCount > 0: 2058 if functionName in summaryReceived: 2059 summaryReceived[functionName] += receivedCount 2060 else: 2061 summaryReceived[functionName] = receivedCount 2062 2063 for functionName, sentCount in kv.CallsSent.iteritems(): 2064 if sentCount > 0: 2065 if functionName in summarySent: 2066 summarySent[functionName] += sentCount 2067 else: 2068 summarySent[functionName] = sentCount 2069 2070 for functionName, sentCount in kv.NotificationsSent.iteritems(): 2071 if sentCount > 0: 2072 if functionName in summarySent: 2073 summarySent[functionName] += sentCount 2074 else: 2075 summarySent[functionName] = sentCount 2076 2077 for functionName, receivedCount in summaryReceived.iteritems(): 2078 dbzmetric.KeyCounters_UpdateMulti(30021, 'D', counterDate, 2000000004, None, unicode(functionName), receivedCount) 2079 2080 for functionName, sentCount in summarySent.iteritems(): 2081 dbzmetric.KeyCounters_UpdateMulti(30022, 'D', counterDate, 2000000004, None, unicode(functionName), sentCount) 2082 2083 alles = self.session.ConnectToAllProxyServerServices('machoNet').GetMetrics() 2084 proxyCount = len(alles) 2085 summaryReceived = {} 2086 summarySent = {} 2087 for isException, nodeID, kv in alles: 2088 blue.pyos.BeNice() 2089 if isException: 2090 exceptionCount += 1 2091 continue 2092 for functionName, receivedCount in kv.CallsReceived.iteritems(): 2093 if receivedCount > 0: 2094 if functionName in summaryReceived: 2095 summaryReceived[functionName] += receivedCount 2096 else: 2097 summaryReceived[functionName] = receivedCount 2098 2099 for functionName, receivedCount in kv.NotificationsReceived.iteritems(): 2100 if receivedCount > 0: 2101 if functionName in summaryReceived: 2102 summaryReceived[functionName] += receivedCount 2103 else: 2104 summaryReceived[functionName] = receivedCount 2105 2106 for functionName, sentCount in kv.CallsSent.iteritems(): 2107 if sentCount > 0: 2108 if functionName in summarySent: 2109 summarySent[functionName] += sentCount 2110 else: 2111 summarySent[functionName] = sentCount 2112 2113 for functionName, sentCount in kv.NotificationsSent.iteritems(): 2114 if sentCount > 0: 2115 if functionName in summarySent: 2116 summarySent[functionName] += sentCount 2117 else: 2118 summarySent[functionName] = sentCount 2119 2120 for functionName, receivedCount in summaryReceived.iteritems(): 2121 dbzmetric.KeyCounters_UpdateMulti(30023, 'D', counterDate, 2000000004, None, unicode(functionName), receivedCount) 2122 2123 for functionName, sentCount in summarySent.iteritems(): 2124 dbzmetric.KeyCounters_UpdateMulti(30024, 'D', counterDate, 2000000004, None, unicode(functionName), sentCount) 2125 2126 except Exception: 2127 log.LogException('Exception in shutdown worker - Metrics phase') 2128 sys.exc_clear() 2129 2130 if eventID: 2131 self.dbLog.LogProcessClusterShutdownCompleted(eventID, int_1=nodeCount, int_2=proxyCount, int_3=exceptionCount) 2132 log.Quit = None 2133 self.LogNotice('Metrics phase completed. Notifying all nodes of immediate shutdown') 2134 eventID = self.dbLog.LogProcessClusterShutdownStarted('gracefulClusterShutdown') 2135 if self.message_bus is None: 2136 from carbon.common.script.net.messagebus.machoNetMessenger import MachoNetMessenger 2137 self.message_bus = MachoNetMessenger(sm.GetService('externalQueueMgr')) 2138 self.message_bus.shutdown_requested() 2139 try: 2140 self.session.ConnectToAllServices('machoNet').GracefulShutdownStarted() 2141 except Exception: 2142 log.LogException('Exception in shutdown worker - GracefulShutdownStarted phase') 2143 sys.exc_clear() 2144 2145 self.LogNotice('Terminating connected clients') 2146 massLogoutEvent = None 2147 try: 2148 self.session.ConnectToAllProxyServerServices('machoNet').TerminateAllClients('The cluster is shutting down') 2149 massLogoutEvent = uthread2.Event() 2150 sm.ChainEvent('ProcessClientTerminatedBecauseOfShutDown', massLogoutEvent) 2151 except Exception: 2152 log.LogException('Exception in shutdown worker - client disconnection phase') 2153 sys.exc_clear() 2154 2155 self.LogNotice('Calling ProcessClusterShutdown') 2156 try: 2157 sm.ChainEvent('ProcessClusterShutdown') 2158 except Exception: 2159 log.LogException('Exception in shutdown worker - ChainEvent ProcessClusterShutdown phase') 2160 sys.exc_clear() 2161 2162 self.LogNotice('Calling GracefulShutdownComplete') 2163 try: 2164 self.session.ConnectToAllServices('machoNet').GracefulShutdownComplete(self.GetNodeID()) 2165 except Exception: 2166 log.LogException('Exception in shutdown worker - GracefulShutdownComplete phase') 2167 sys.exc_clear() 2168 2169 self.LogNotice('Polaris node Completing GracefulShutdown') 2170 self.dbLog.LogProcessClusterShutdownCompleted(eventID) 2171 if massLogoutEvent is not None: 2172 startTime = blue.os.GetWallclockTimeNow() 2173 massLogoutEvent.wait(10.0) 2174 self.LogNotice('Waited', (blue.os.GetWallclockTimeNow() - startTime) / const.MSEC, 'milliseconds for mass logout event') 2175 self.GracefulShutdownComplete(None) 2176 2177 def IsClusterShuttingDown(self): 2178 return self.shutdownInProgress 2179 2180 def GracefulShutdownStarted(self): 2181 self.shutdownInProgress = True 2182 self.LogNotice('Cluster GracefulShutdownStarted on node', self.nodeID, ' - Disabling ObjectKillah') 2183 try: 2184 basesession.objectKillaThread.kill() 2185 except Exception: 2186 self.LogException() 2187 2188 if macho.mode == 'proxy': 2189 self.LogNotice('GracefulShutdownStarted on node', self.nodeID, ' - Setting VIP mode') 2190 self.SetConnectionProperty('vipMode', True, doPersist=False) 2191 sm.ChainEvent('ProcessGracefulShutdownStarted') 2192 self.LogNotice('GracefulShutdownStarted done on node %s' % self.nodeID) 2193 2194 def GracefulShutdownComplete(self, callerNodeID): 2195 if callerNodeID == self.nodeID and macho.mode == 'server': 2196 return 2197 sm.ChainEvent('ProcessGracefulShutdownComplete') 2198 nodeString = 'Master' if callerNodeID is None else 'Slave' 2199 self.LogNotice('GracefulShutdownComplete running on %s %s node %s; called from %s' % (nodeString, 2200 macho.mode, 2201 self.nodeID, 2202 callerNodeID)) 2203 if macho.mode == 'server': 2204 try: 2205 self.LogNotice('GracefulShutdownComplete (%s node %s) - Trashing node... ' % (nodeString, self.nodeID)) 2206 self.dbzcluster.Nodes_Trash(self.nodeID, 'Graceful Cluster Shutdown - %s' % nodeString, 1) 2207 except Exception: 2208 self.LogException('Exception while trashing %s node %s' % (nodeString, self.nodeID)) 2209 2210 self.LogNotice('Threading out Terminate in GracefulShutdownComplete (%s) stackless.runcount = %s' % (self.nodeID, stackless.getruncount())) 2211 uthread.worker('Shutdown thread', bluepy.Terminate, 0, 'Graceful Cluster Shutdown (%s - %s)' % (self.nodeID, nodeString)) 2212 2213 def Shutdown(self, who): 2214 self.LogError('Shutdown requested by ', who, '. Shutting down.') 2215 if macho.mode == 'server': 2216 self.dbzcluster.Nodes_Trash(self.nodeID, 'Shutdown called by %s' % who, 1) 2217 bluepy.Terminate(0, 'Shutdown called by %s' % who) 2218 2219 def TerminateUnconnectedNodes(self): 2220 if macho.mode == 'proxy': 2221 for each in self.nodeCPULoadValue.iterkeys(): 2222 if each not in self.transportIDbySolNodeID: 2223 self.LogNotice('Terminating unconnected node', each) 2224 uthread.worker('machoNet::ServerBroadcast::OnNodeDeath', self.ServerBroadcast, 'OnNodeDeath', each, 1, "It's Terminate Unconnected Nodes' fault") 2225 2226 def TerminateClient(self, reason, clientID): 2227 if macho.mode != 'proxy': 2228 return 2229 transportID = self.transportIDbyClientID.get(clientID, None) 2230 if transportID is not None: 2231 transport = self.transportsByID.get(transportID, None) 2232 if transport is not None: 2233 self.LogInfo('Closing transport', transportID, transport) 2234 transport.Close(reason) 2235 2236 def TerminateAllClients(self, reason): 2237 if macho.mode != 'proxy': 2238 return 2239 while self.transportIDbyClientID: 2240 for clientID in self.transportIDbyClientID.keys(): 2241 transport = self.transportsByID.get(self.transportIDbyClientID[clientID], None) 2242 if transport.transportName == 'tcp:packet:client': 2243 monolithdisconnectlogger.log_server_shutdown_disconnect(transport.transport.address) 2244 self.TerminateClient(reason, clientID) 2245 2246 monolithmetrics.flush() 2247 2248 self.LogNotice('All clients terminated on', macho.mode, 'node', self.nodeID) 2249 2250 def __NodeWatcher(self): 2251 connectionCount = -1 2252 while not self.IsClusterShuttingDown() and not self.stop: 2253 self.LogInfo('NodeWatcher: Probing') 2254 try: 2255 if len(self.transportIDbyProxyNodeID) + len(self.transportIDbySolNodeID) != connectionCount or macho.mode == 'server' and not len(self.transportIDbyProxyNodeID) or macho.mode == 'proxy' and not len(self.transportIDbySolNodeID): 2256 self.LogWarn("I think I've got a connectivity problem. Attempting autorefresh.") 2257 connectionCount = self.RefreshConnectivity() 2258 self.LogInfo('NodeWatcher: Sleeping for ', self.nodeRefreshInterval, ' seconds') 2259 except Exception: 2260 log.LogException('Exception during NodeWatcher') 2261 sys.exc_clear() 2262 2263 blue.pyos.synchro.SleepWallclock(1000 * self.nodeRefreshInterval) 2264 2265 if self.stop: 2266 self.LogNotice('NodeWatcher: Shutting down due to node shutdown') 2267 else: 2268 self.LogNotice('NodeWatcher: Shutting down due to cluster shutdown') 2269 2270 def __LoginQueueUpdater(self): 2271 while not getattr(self, 'stop', True): 2272 self.availableLoginSlots = min(len(self.serverLogonQueue), self.availableLoginSlots) 2273 maxConnections = max(0, self.connectionLimit - len(self.transportIDbyClientID)) 2274 self.availableLoginSlots = min(self.availableLoginSlots + self.maxLoginsPerMinute, maxConnections) 2275 if len(self.serverLogonQueue) > self.availableLoginSlots: 2276 self.LogWarn('Some clients are waiting in a login queue. Clients at login screen:', len(self.serverLogonQueue), 'Available slots:', self.availableLoginSlots) 2277 blue.pyos.synchro.SleepWallclock(const.MIN / const.MSEC) 2278 2279 def __ClientSessionMaxTimePoll(self): 2280 while not getattr(self, 'stop', True): 2281 self.LogInfo('ClientSessionMaxTimePoll: Sleeping for ', self.clientSessionTimeoutGranularity, ' seconds') 2282 blue.pyos.synchro.SleepWallclock(1000 * self.clientSessionTimeoutGranularity) 2283 try: 2284 for transportID in self.transportIDbyClientID.values(): 2285 if transportID in self.transportsByID: 2286 transport = self.transportsByID[transportID] 2287 clientID = transport.clientID 2288 if clientID in transport.sessions: 2289 s = transport.sessions[clientID] 2290 if s.maxSessionTime and s.maxSessionTime < blue.os.GetWallclockTime(): 2291 transport.Close('Game Time Expired', 'GAMETIMEEXPIRED') 2292 2293 except Exception: 2294 log.LogException('Exception during __ClientSessionMaxTimePoll') 2295 sys.exc_clear() 2296 2297 def __NodeLoadPush(self): 2298 while not getattr(self, 'stop', True): 2299 try: 2300 blue.pyos.synchro.SleepWallclock(1000 * self.nodeLoadPush) 2301 if macho.mode == 'server' and self.GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) == self.GetNodeID(): 2302 cluGetNodes = self.dbzcluster.Nodes_Select2() 2303 self.nodeCPULoadValue = {} 2304 for r in cluGetNodes: 2305 self.nodeCPULoadValue[r.nodeID] = r.currentCpu or 0.0 2306 self.serverNames[r.nodeID] = r.machineName or None 2307 2308 self.ClusterBroadcast('OnNodeLoadPush', self.nodeCPULoadValue, self.serverNames) 2309 except UnMachoDestination as e: 2310 self.LogWarn('UnMachoDestination (', e, ') while attempting to acquire load stats') 2311 sys.exc_clear() 2312 except Exception: 2313 self.LogError('Exception while attempting to acquire load stats') 2314 log.LogException() 2315 sys.exc_clear() 2316 2317 def __DisconnectUnauthorizedUsers(self): 2318 while not getattr(self, 'stop', True): 2319 try: 2320 blue.pyos.synchro.SleepWallclock(1000 * self.disconnectUnauthorizedUsersPollInterval) 2321 if self.GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) == self.GetNodeID(): 2322 peopleWhoShouldntBeLoggedIn = self.dbzcluster.Sessions_SelectInactive() 2323 if peopleWhoShouldntBeLoggedIn: 2324 self.ProxyBroadcast('OnPeopleWhoShouldntBeLoggedInNotification', peopleWhoShouldntBeLoggedIn) 2325 except Exception: 2326 self.LogError('Exception while attempting to disconnect unauthorized users') 2327 log.LogException() 2328 sys.exc_clear() 2329 2330 def DisconnectUsers(self, userIDs): 2331 clientIDs = [ s.clientID for s in basesession.FindSessions('userid', userIDs) ] 2332 for cid in clientIDs: 2333 if cid in self.transportIDbyClientID: 2334 tid = self.transportIDbyClientID[cid] 2335 t = self.transportsByID[tid] 2336 t.Close('Client disconnected', 'CLIENTDISCONNECTED') 2337 2338 def OnPeopleWhoShouldntBeLoggedInNotification(self, peopleWhoShouldntBeLoggedIn): 2339 lastPeopleWhoShouldntBeLoggedIn = self.peopleWhoShouldntBeLoggedIn 2340 self.peopleWhoShouldntBeLoggedIn = {} 2341 for sessionID, status in peopleWhoShouldntBeLoggedIn: 2342 if sessionID not in self.transportIDbySessionID: 2343 continue 2344 transportID = self.transportIDbySessionID.get(sessionID, None) 2345 if transportID is not None: 2346 transport = self.transportsByID[transportID] 2347 if status in (5, 12, 13): 2348 uthread.worker('machoNet::DelayedClose', self.__DelayedClose, transport, 'ACCOUNTBANNED') 2349 elif sessionID in lastPeopleWhoShouldntBeLoggedIn: 2350 self.LogOffSession(sessionID) 2351 else: 2352 self.peopleWhoShouldntBeLoggedIn[sessionID] = True 2353 2354 def __DelayedClose(self, transport, message): 2355 clientID = transport.clientID 2356 if message == 'ACCOUNTBANNED': 2357 self.SinglecastByClientID(clientID, 'OnServerMessage', ('/Carbon/MachoNet/DelayedCloseNotification/AccountBanned', {})) 2358 blue.pyos.synchro.SleepWallclock(self.disconnectUnauthorizedUsersDelayInterval * 1000) 2359 transport.Close(message, message) 2360 2361 def OnNodeLoadPush(self, nodeCPULoadValue, serverNames): 2362 self.nodeCPULoadValue = nodeCPULoadValue 2363 self.serverNames = serverNames 2364 2365 def GetLocalHostName(self): 2366 if macho.mode == 'client': 2367 return None 2368 return blue.pyos.GetEnv().get('COMPUTERNAME', 'UNKNOWNHOST') 2369 2370 class Sequencer(uthread.TaskletSequencer): 2371 2372 def __init__(self, machoNet): 2373 uthread.TaskletSequencer.__init__(self) 2374 self.machoNet = machoNet 2375 2376 def OnSleep(self, sequence): 2377 self.machoNet.LogInfo('Sequencer: Going to sleep, I am %d, expected %d, queue:%r' % (sequence, self.expected, self.queue)) 2378 2379 def OnWakeUp(self, sequence): 2380 self.machoNet.LogInfo('Sequencer: Waking up, I am %d, expected %d, queue:%r' % (sequence, self.expected, self.queue)) 2381 2382 def OnWakingUp(self, sequence, target): 2383 self.machoNet.LogInfo('Sequencer: Waking dude up, I am %d, dude %d, queue:%r' % (sequence, target, self.queue)) 2384 2385 def WaitForSequenceNumber(self, address, seq): 2386 if not seq or macho.mode != 'client': 2387 return 2388 if address.addressType == const.ADDRESS_TYPE_NODE and address.nodeID: 2389 try: 2390 sequencer = self.seqwait[address.nodeID] 2391 except KeyError: 2392 sequencer = self.Sequencer(self) 2393 self.seqwait[address.nodeID] = sequencer 2394 2395 if sequencer.IsClosed(): 2396 self.LogError('Sequence Tasklet is closed in WaitForSequenceNumber') 2397 sequencer.Pass(seq) 2398 else: 2399 self.LogWarn('Unexpected source address type in wait for sequence number, address=', address) 2400 2401 def OnNodeDeath(self, nodeID, confirmed, reason = None): 2402 self.LogNotice('machoNet::OnNodeDeath(', ','.join([str(nodeID), str(confirmed), str(reason)]), ') running on node', self.nodeID) 2403 if self._addressCache.IsDeadNode(nodeID): 2404 self.LogNotice('OnNodeDeath - Already handled the death of', nodeID) 2405 return 2406 if not self._addressCache.IsKnownNode(nodeID): 2407 self.LogNotice('machoNet::OnNodeDeath: Node', nodeID, 'is totally alien to this node...') 2408 return 2409 if not confirmed: 2410 return 2411 self._addressCache.MarkDeadNode(nodeID) 2412 if macho.mode == 'server': 2413 if self.nodeID == int(nodeID): 2414 self.dbzcluster.Nodes_Trash(self.nodeID, 'Some (proxy) sent me an OnNodeDeath for my own nodeID', 1) 2415 log.Quit('Some (proxy) sent me an OnNodeDeath for my own nodeID') 2416 else: 2417 self.dbzcluster.Nodes_Trash(int(nodeID), 'Killed by node %s, because: %s' % (self.nodeID, reason)) 2418 transport = self.GetTransportOfNode(nodeID) 2419 if transport is not None: 2420 transport.Close(reason=reason, noSend=True) 2421 2422 def LayTracksIfNeeded(self, address, nodeID, whencemsg): 2423 if self.DontLayTracksTo(nodeID): 2424 return 2425 self.LogInfo('Connectivity[%s]: ' % whencemsg, self.nodeID, ' laying tracks to ', nodeID, ' at ', address) 2426 uthread.worker('machoNet::LayTracksIfNeeded', self.LayTracks, address, nodeID, whencemsg) 2427 2428 def DontLayTracksTo(self, nodeID): 2429 if nodeID == self.nodeID: 2430 return True 2431 if nodeID in self.transportIDbySolNodeID or nodeID in self.transportIDbyProxyNodeID: 2432 return True 2433 if nodeID in self.dontLayTracksTo: 2434 return True 2435 return False 2436 2437 def LayingTracksTo(self): 2438 return self.dontLayTracksTo.items() 2439 2440 def LayTracks(self, address, nodeID, whence): 2441 if self.DontLayTracksTo(nodeID): 2442 return 2443 logname = 'Connectivity[LayTracks tid = %s, %s]:' % (id(stackless.getcurrent()), whence) 2444 self.LogInfo(logname, 'start', repr(address), 'nodeID=', nodeID) 2445 self.dontLayTracksTo[nodeID] = address 2446 try: 2447 if nodeID == self.GetNodeID(): 2448 self.LogError(logname, 'Trying to lay tracks to self!') 2449 log.LogTraceback() 2450 return 2451 if self.nodeID is None: 2452 self.LogError(logname, 'Cannot lay tracks before we have a node ID. Sleeping...') 2453 while self.nodeID is None: 2454 blue.pyos.synchro.SleepWallclock(100) 2455 2456 self.LogInfo(logname, 'Woke up, have nodeID', self.nodeID) 2457 if False: 2458 nodeType = self.GetNodeTypeFromID(nodeID) 2459 if nodeType == macho.mode and nodeID <= self.nodeID: 2460 self.LogInfo(logname, 'Not laying tracks to a higher ranking node (%d to %d), expecting incoming connection.' % (self.nodeID, nodeID)) 2461 return 2462 if nodeType == 'server' and macho.mode == 'proxy': 2463 self.LogInfo(logname, 'Not laying tracks from proxy to server, expecting incoming connection.') 2464 return 2465 try: 2466 factory = self.GetFactory(get_gps_map()[macho.mode]['tcp:packet:machoNet'], (0, 'tcp:packet:machoNet')) 2467 self.LogInfo(logname, 'Attempting connect to ', repr(address)) 2468 rawTransport = factory.Connect(address) 2469 transport = None 2470 self.LogInfo(logname, 'Connect succeeded! [%r->%r]' % (rawTransport.localaddress, rawTransport.address)) 2471 try: 2472 transportID = self.transportID 2473 self.transportID = self.transportID + 1 2474 transport = MachoTransport(transportID, rawTransport, 'tcp:packet:machoNet', self) 2475 others = [] 2476 if macho.mode == 'server': 2477 layTracksTo = self._GetLayTracksTo() 2478 for each in layTracksTo: 2479 others.append(('%s:%d' % (each.ipAddress, each.port), each.nodeID)) 2480 2481 myaddress = self.GetTransport('tcp:packet:machoNet').GetExternalAddress() 2482 else: 2483 myaddress = self.GetTransport('tcp:packet:machoNet').GetInternalAddress() 2484 self.LogInfo(logname, 'sending idrq to node', nodeID, 'at', address) 2485 transport.Write(IdentificationReq(nodeID=self.nodeID, source=MachoAddress(nodeID=self.nodeID), myaddress=myaddress, others=others, isProxy=macho.mode == 'proxy', isApp=False, serviceMask=self.serviceMask)) 2486 response = transport.Read() 2487 self.LogInfo(logname, 'got response to idrq from', nodeID, 'at', address) 2488 if response.command != cluster.MACHONETMSG_TYPE_IDENTIFICATION_RSP: 2489 raise RuntimeError('UnexpectedResponse', 'Got a bad response to IdentificationReq') 2490 if nodeID and response.nodeID != nodeID: 2491 self.LogError(logname, 'I thought I was connecting to ', nodeID, ', but got a response from ', response.nodeID) 2492 log.LogTraceback('Got a misguided response to IdentificationReq (expected: %s, actual: %s)' % (nodeID, response.nodeID)) 2493 raise RuntimeError('UnexpectedResponse', 'Got a misguided response to IdentificationReq') 2494 accepted, reason = response.accepted 2495 self.transportsByID[transportID] = transport 2496 rawTransport = None 2497 if not accepted: 2498 self.LogInfo(logname, response.nodeID, ' rejected connection: ', repr(reason)) 2499 transport.Close('%s rejected connection: %r.' % (response.nodeID, reason), noSend=True) 2500 elif response.nodeID in self.transportIDbyProxyNodeID or response.nodeID in self.transportIDbySolNodeID: 2501 self.LogError(logname, "I've already got a connection established to ", response.nodeID, ". Two's a crowd in this case. Ciao.") 2502 transport.Close("I've already got a connection established to %s." % response.nodeID) 2503 else: 2504 self.LogInfo(logname, 'Succesfully established a connection to ', response.nodeID, ' reason: ', repr(reason)) 2505 transport.Finalize(response.nodeID) 2506 blue.net.EnumerateTransport(transport.transport.socket.getSocketDescriptor(), transport.transportID, transport.transportName, 0, 0, transport.nodeID) 2507 if response.isProxy: 2508 self.transportIDbyProxyNodeID[transport.nodeID] = transportID 2509 blue.net.AddProxyNode(transportID, transport.nodeID) 2510 else: 2511 self.transportIDbySolNodeID[transport.nodeID] = transportID 2512 transport.currentReaders += 1 2513 uthread.worker('machoNet::TransportReader::LayTracks', self.TransportReader, self.transportsByID[transportID]) 2514 self.externalAddressesByNodeID[response.nodeID] = (address, response.serviceMask) 2515 if self.clusterStartupPhase: 2516 if macho.mode == 'proxy': 2517 polarisID = self.session.ConnectToSolServerService('machoNet').GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) 2518 else: 2519 polarisID = self.GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) 2520 isPolaris = response.nodeID == polarisID 2521 sm.ScatterEvent('OnNewNode', response.nodeID, address, response.isProxy, isPolaris, response.serviceMask) 2522 others = response.others 2523 for otherAddress, otherNodeID in others: 2524 self.LayTracksIfNeeded(otherAddress, otherNodeID, 'LayTracks::response.others') 2525 2526 except GPSTransportClosed as e: 2527 if transport is not None: 2528 transport.Close('GPSTransportClosed', exception=e) 2529 if rawTransport is not None: 2530 rawTransport.Close('GPSTransportClosed', exception=e) 2531 raise 2532 except Exception as e: 2533 if transport is not None: 2534 transport.Close('LayTracks blew up', exception=e) 2535 if rawTransport is not None: 2536 rawTransport.Close('LayTracks blew up', exception=e) 2537 log.LogException() 2538 2539 except GPSTransportClosed as e: 2540 ps = 'proxy' if macho.mode != 'proxy' else 'server' 2541 self.LogWarn(logname, 'Failed to connect to ', ps, ' on ', address, repr(e)) 2542 except Exception: 2543 log.LogException() 2544 2545 finally: 2546 del self.dontLayTracksTo[nodeID] 2547 self.LogInfo(logname, 'End') 2548 2549 def RegisterXmlRpc(self): 2550 2551 def GracefulShutdownWrapper(whenMinutes, duration, explaination): 2552 userID = 0 2553 when = whenMinutes * const.MIN + blue.os.GetWallclockTime() 2554 self.session.ConnectToAllSolServerServices('machoNet').GracefulShutdown(userID, when, duration, explaination) 2555 return True 2556 2557 sm.services['xmlrpcService'].Register(ROLE_ADMIN, GracefulShutdownWrapper, 'machoNet', 'GracefulShutdown') 2558 2559 def IsReady(): 2560 return self.clusterStartupPhase 2561 2562 sm.services['xmlrpcService'].Register(ROLE_ANY, IsReady, 'machoNet') 2563 2564 def SetVIPMode(onoff): 2565 s = self.session.ConnectToAllProxyServerServices('machoNet') 2566 s.SetConnectionProperty('vipMode', onoff) 2567 return self.vipMode 2568 2569 sm.services['xmlrpcService'].Register(ROLE_ADMIN, SetVIPMode, 'machoNet') 2570 2571 def SetProxyConnectionLimit(limit): 2572 s = self.session.ConnectToAllProxyServerServices('machoNet') 2573 s.SetConnectionProperty('limit', limit) 2574 return self.connectionLimit 2575 2576 sm.services['xmlrpcService'].Register(ROLE_ADMIN, SetProxyConnectionLimit, 'machoNet') 2577 2578 def _MapServiceNameToServiceID(self, serviceName): 2579 if serviceName in self.serviceMappings: 2580 serviceID = self.serviceMappings[serviceName][0] 2581 return serviceID 2582 return serviceName 2583 2584 def SetNodeOfAddress(self, service, address, nodeID): 2585 if nodeID is None: 2586 self._addressCache.RemoveServiceAddress(service, address) 2587 else: 2588 self._addressCache.SetServiceAddress(service, address, nodeID) 2589 2590 def CheckAddressCache(self, service, address, lazyGetIfNotFound = False): 2591 suggestedNodeID, service, address = self._GetNodeFromAddressAdjustments(service, address) 2592 if isinstance(service, str): 2593 if macho.mode != 'client': 2594 calledBy = log.WhoCalledMe() 2595 if calledBy not in self.getNodeFromAddressWarnsDone: 2596 self.LogWarn('Passing a string(', service, ') serviceID into CheckAddressCache, will be mapped to a int internally. Please update the usage at:', calledBy) 2597 self.getNodeFromAddressWarnsDone.add(calledBy) 2598 serviceID = self._MapServiceNameToServiceID(service) 2599 if serviceID: 2600 service = serviceID 2601 nodeID = self._addressCache.GetByServiceAddress(service, address) 2602 if not nodeID and lazyGetIfNotFound: 2603 uthread.new(self.GetNodeFromAddress, service, address) 2604 return nodeID 2605 2606 def LookupAddressCache(self, serviceName, address): 2607 return self._addressCache.GetByServiceAddress(serviceName, address) 2608 2609 def BlockGetNodeFromAddress(self): 2610 self.isGetNodeFromAddressBlocked = True 2611 2612 def UnblockGetNodeFromAddress(self): 2613 self.isGetNodeFromAddressBlocked = False 2614 2615 @bluepy.TimedFunction('machoNet::GetNodeFromAddress') 2616 def GetNodeFromAddress(self, serviceID, address): 2617 if getattr(self, 'isGetNodeFromAddressBlocked', False): 2618 if (serviceID, address) != (cluster.SERVICE_POLARIS, 0): 2619 raise RuntimeError('somebody called GetNodeFromAddress while blocked (%s, %s)' % (serviceID, address)) 2620 suggestedNodeID, serviceID, address = self._GetNodeFromAddressAdjustments(serviceID, address) 2621 if isinstance(serviceID, str): 2622 calledBy = log.WhoCalledMe() 2623 if calledBy not in self.getNodeFromAddressWarnsDone: 2624 stack = '' 2625 for i in xrange(10): 2626 stack += '%s\n' % log.WhoCalledMe(i) 2627 2628 self.LogWarn('Passing a string(', serviceID, ') serviceID into GetNodeFromAddress, will be mapped to a int internally. Please update the usage at:', stack) 2629 self.getNodeFromAddressWarnsDone.add(calledBy) 2630 serviceID = self._MapServiceNameToServiceID(serviceID) 2631 if not serviceID: 2632 log.LogTraceback(extraText='Received a servicename into GetNodeFromAddress which did no map to a valid servicemask:' + str(serviceID)) 2633 if address is None: 2634 raise RuntimeError('GetNodeFromAddress only works if you specify an address...') 2635 key = (serviceID, address) 2636 while 1: 2637 nodeID = self._addressCache.GetByServiceAddress(serviceID, address) 2638 if nodeID is None: 2639 myNodeID = self.nodeID 2640 if not self.connectToCluster: 2641 myNodeID = -1 2642 uthread.Lock(self, serviceID, address) 2643 try: 2644 nodeID = self._addressCache.GetByServiceAddress(serviceID, address) 2645 if nodeID is None: 2646 borkPrevention = 10 2647 serviceMask = self.serviceMaskByServiceID[serviceID] 2648 for i in xrange(100): 2649 if serviceMask not in self.expectedLoadValue: 2650 uthread.Lock(self, 'CluGetExpectedLoadValue') 2651 try: 2652 self.expectedLoadValue[serviceID] = self.dbzcluster.LoadStats_ExpectedLoadValue(serviceID) 2653 finally: 2654 uthread.UnLock(self, 'CluGetExpectedLoadValue') 2655 2656 nodeID = self._GetNodeFromAddressFromDB(myNodeID, serviceID, int(address), suggestedNodeID, self.expectedLoadValue[serviceID], serviceMask) 2657 if nodeID <= 0: 2658 if nodeID == -1: 2659 self.LogError('zcluster.Cluster_NodeFromAddress informed us that an app lock timeout occurred') 2660 self.LogError('Sleeping and retrying. This is very bad.') 2661 blue.pyos.synchro.SleepWallclock(15000) 2662 continue 2663 elif nodeID == -2: 2664 self.LogError("zcluster.Cluster_NodeFromAddress informed us that we're dead") 2665 log.LogTraceback() 2666 blue.pyos.synchro.SleepWallclock(60000) 2667 self.dbzcluster.Nodes_Trash(self.nodeID, "zcluster.Cluster_NodeFromAddress informed us that we're dead", 1) 2668 log.Quit("zcluster.Cluster_NodeFromAddress informed us that we're dead") 2669 elif nodeID == -3: 2670 if borkPrevention > 0: 2671 self.LogWarn('Cluster Segmentation Initial Fragmentation Prevention for ', key, ': Sleeping for ', 1 + self.heartbeatInterval / 10, ' seconds to allow other segments to pass the safety time.') 2672 blue.pyos.synchro.SleepWallclock(1000 * (1 + self.heartbeatInterval / 10)) 2673 borkPrevention = borkPrevention - 1 2674 continue 2675 if borkPrevention == 0: 2676 log.LogTraceback("Couldn't allocate address on any server. Cluster borked beyond recognition. Performing emergency cluster desegmentation.") 2677 self.dbzcluster.Cluster_Desegmentation() 2678 self.ClusterBroadcast('OnClusterDesegmentation') 2679 continue 2680 else: 2681 log.LogTraceback("Couldn't allocate address on any server after desegmentification. Cluster borked beyond recognition.") 2682 raise RuntimeError('The address you asked for could not be allocated on any server. This cluster must be seriously borked, to say the least...') 2683 elif nodeID == -4: 2684 self.LogWarn('zcluster.Cluster_NodeFromAddress informed us that the address we requested (', serviceID, ',', address, ") was recently mapped on a node that has not yet acknowledged it's death") 2685 self.LogWarn('Sleeping and retrying.') 2686 blue.pyos.synchro.SleepWallclock(15000) 2687 continue 2688 else: 2689 self.LogError('zcluster.Cluster_NodeFromAddress went haywire') 2690 log.LogTraceback('Bork Bork in zcluster.Cluster_NodeFromAddress') 2691 self.dbzcluster.Nodes_Trash(self.nodeID, 'zcluster.Cluster_NodeFromAddress went haywire and returned an illegal value', 1) 2692 log.Quit('zcluster.Cluster_NodeFromAddress went haywire and returned an illegal value') 2693 if nodeID == 0: 2694 log.LogTraceback('zcluster.Cluster_NodeFromAddress returned 0???') 2695 nodeID = None 2696 if nodeID is not None: 2697 self._addressCache.SetServiceAddress(serviceID, address, nodeID) 2698 break 2699 else: 2700 log.LogTraceback('zcluster.Cluster_NodeFromAddress returned 0 or None??? Entering bork-prevention loop') 2701 blue.pyos.synchro.SleepWallclock(5000) 2702 2703 if nodeID is None: 2704 log.LogTraceback('Failing to acquire a nodeID for address (%s). Raising.' % strx(key)) 2705 raise RuntimeError('GetNodeFromAddress failed to acquire a nodeID for the address (1)') 2706 finally: 2707 uthread.UnLock(self, serviceID, address) 2708 2709 if nodeID is None: 2710 raise RuntimeError("Couldn't resolve %s:%s to a node" % (serviceID, address)) 2711 if nodeID is None: 2712 log.LogTraceback('Failed to acquire a nodeID for address (%s). Raising.' % strx(key)) 2713 raise RuntimeError('GetNodeFromAddress failed to acquire a nodeID for the address (2)') 2714 if self._addressCache.IsDeadNode(nodeID): 2715 self.LogError('Tried to return a dead mapping: Dead Node ', nodeID, ' has been registered in the address cache for ', key) 2716 self._addressCache.RemoveAllForNode(nodeID) 2717 blue.pyos.synchro.SleepWallclock(3000) 2718 else: 2719 break 2720 2721 return nodeID 2722 2723 def ClearNodeOfAddresses(self, addresses): 2724 totalCleared = 0 2725 for service, address in addresses: 2726 try: 2727 self.LogWarn('ClearNodeOfAddresses: clearing (', service, ', ', address, ')') 2728 uthread.Lock(self, service, address) 2729 self.SetNodeOfAddress(service, address, None) 2730 totalCleared += 1 2731 finally: 2732 uthread.UnLock(self, service, address) 2733 2734 return totalCleared 2735 2736 def GetNodeIDsFromServiceMask(self, serviceMask): 2737 return self._addressCache.GetByServiceMask(serviceMask) 2738 2739 def GetFullAddressCache(self): 2740 return (dict(self._addressCache.GetCachedServiceAddressMap()), dict(self._addressCache.GetCachedServiceMaskMap())) 2741 2742 def GetFullServiceMaskAddressCache(self): 2743 return self._addressCache.GetCachedServiceMaskMap() 2744 2745 def GetServiceMaskByServiceID(self, serviceID): 2746 return self.serviceMaskByServiceID.get(serviceID, None) 2747 2748 def OnClusterDesegmentation(self): 2749 self.LogWarn('Cluster desegmentation') 2750 self.clusterIsDesegmented = True 2751 for nodeID in self.transportIDbySolNodeID.iterkeys(): 2752 self._addressCache.LoadNodeToServiceMaskMapping(nodeID, cluster.SERVICE_ALL) 2753 2754 if macho.mode == 'server': 2755 self.serviceMask = cluster.SERVICE_ALL 2756 2757 def _GetNodeFromAddressFromDB(self, myNodeID, serviceMapping, address, suggestedNodeID, expectedLoadValue, serviceMask): 2758 return self.dbzcluster.Cluster_NodeFromAddress(myNodeID, serviceMapping, address, suggestedNodeID, expectedLoadValue, serviceMask) 2759 2760 def GuessNodeIDFromAddress(self, serviceName, address): 2761 return None 2762 2763 def _GetNodeFromAddressAdjustments(self, service, address): 2764 return (None, service, address) 2765 2766 def PrimeAddressCache(self, allMappings): 2767 mapped = 0 2768 for each in allMappings: 2769 if self._addressCache.IsDeadNode(each.nodeID): 2770 continue 2771 serviceID = self._MapServiceNameToServiceID(each.serviceName) 2772 if self._addressCache.SetServiceAddress(serviceID or each.serviceName, each.addressID, each.nodeID): 2773 mapped += 1 2774 2775 if mapped: 2776 self.LogInfo('PrimeAddressCache mapped %d new addresses' % mapped) 2777 2778 def ClearAddressCache(self): 2779 self._addressCache.Clear() 2780 2781 def RemoveAllForNode(self, nodeId): 2782 self._addressCache.RemoveAllForNode(nodeId) 2783 2784 def HandleUnMachoDestination(self, nodeID, packet): 2785 if self.IsClusterShuttingDown(): 2786 return 2787 self._addressCache.RemoveAllForNode(nodeID) 2788 self.LogWarn('UnMachoDestination received. Performing provisional node death handling for packet=', packet) 2789 if self._addressCache.IsKnownNode(nodeID): 2790 self.ServerBroadcast('OnNodeDeath', nodeID, 0) 2791 2792 def AreTheseServicesRunning(self, requiredServices): 2793 for required in requiredServices: 2794 if required not in sm.services: 2795 self.LogInfo('AreTheseServicesRunning: Waiting for %s to initiate startup' % required) 2796 return required 2797 2798 for s in sm.services.itervalues(): 2799 if s.state == service.SERVICE_FAILED: 2800 self.LogInfo('AreTheseServicesRunning: %s was SERVICE_FAILED' % s.__guid__) 2801 if s.__error__: 2802 raise_(s.__error__[1], None, s.__error__[2]) 2803 else: 2804 raise_(RuntimeError, 'Service %s failed' % s.__guid__) 2805 if s.state != service.SERVICE_RUNNING: 2806 self.LogInfo('AreTheseServicesRunning: Waiting for %s' % s.__guid__) 2807 return s.__guid__.split('.')[1] 2808 for dependant in s.__dependencies__: 2809 if dependant not in sm.services: 2810 self.LogInfo('AreTheseServicesRunning: Waiting for %s, which is a dependant of %s, to initiate startup' % (dependant, s.__guid__.split('.')[1])) 2811 return dependant 2812 2813 if 'config' in requiredServices: 2814 import __builtin__ 2815 if 'const' not in dir(__builtin__): 2816 self.LogInfo('AreTheseServicesRunning: const not in builtin') 2817 return 'config::const' 2818 if 'cfg' not in dir(__builtin__): 2819 self.LogInfo('AreTheseServicesRunning: cfg not in builtin') 2820 return 'config::cfg' 2821 2822 def GetNodeID(self): 2823 if not hasattr(self, 'nodeID'): 2824 return 2825 if macho.mode == 'server': 2826 while self.nodeID is None: 2827 self.LogError("Some dude is trying to get our nodeID before we're up and running... Can't have that, now can we? Let's sleeping on it.") 2828 log.LogTraceback() 2829 blue.pyos.synchro.SleepWallclock(1000) 2830 2831 return self.nodeID 2832 2833 def GetAcceptDelay(self): 2834 return self.acceptDelay 2835 2836 def GetNodeTypeFromID(self, nodeID): 2837 if nodeID >= const.maxNodeID: 2838 return 'proxy' 2839 else: 2840 return 'server' 2841 2842 def GetClientSessionID(self): 2843 if macho.mode == 'client': 2844 return (self.clientSessionID, session.userid) 2845 return 0 2846 2847 def GetMachoTransportByClientID(self, clientID): 2848 return self.transportsByID.get(self.transportIDbyClientID.get(clientID, None), None) 2849 2850 def GetSessionByClientID(self, clientID): 2851 transport = self.GetMachoTransportByClientID(clientID) 2852 if transport is None: 2853 return 2854 else: 2855 return transport.sessions.get(clientID, None) 2856 2857 def GetConnectedProxyNodes(self): 2858 return self.transportIDbyProxyNodeID.keys() 2859 2860 def GetConnectedSolNodes(self): 2861 return self.transportIDbySolNodeID.keys() 2862 2863 def GetConnectedNodes(self): 2864 return self.GetConnectedProxyNodes() + self.GetConnectedSolNodes() 2865 2866 def GetTransportOfNode(self, nodeID): 2867 transportID = self.transportIDbyProxyNodeID.get(nodeID, None) or self.transportIDbySolNodeID.get(nodeID, None) or self.transportIDbyAppNodeID.get(nodeID, None) 2868 return self.transportsByID.get(transportID, None) 2869 2870 def GetMachoTransportsByTransportIDs(self, ids): 2871 retval = [] 2872 for each in ids: 2873 if each in self.transportsByID: 2874 retval.append(self.transportsByID[each]) 2875 2876 return retval 2877 2878 def GetTransport(self, transportName): 2879 return self.namedtransports.get(transportName, None) 2880 2881 def OnHeartBeat(self): 2882 while not getattr(self, 'stop', True): 2883 blue.pyos.synchro.SleepWallclock(int(self.heartbeatInterval * 1000)) 2884 if self.nodeID is not None and (self.shutdown is None or self.shutdown.when > blue.os.GetWallclockTime()): 2885 try: 2886 secs = self.dbzcluster.Nodes_HeartBeat(self.nodeID) 2887 if self.debugHeartBeat: 2888 self.LogNotice('Node', self.nodeID, 'Done sending Heartbeat @ ', utilFormat.FmtDateEng(blue.os.GetWallclockTime()), ' result ', repr(secs)) 2889 except ConnectionError as e: 2890 self.LogError('!! ---------------------------------------------------------- !!') 2891 self.LogError('ConnectionError exception:[%s]. If SHUTDOWN, then possibly due to a too long/unrecoverable DB Cluster fail over.' % e) 2892 self.LogError('!! ---------------------------------------------------------- !!') 2893 log.LogException() 2894 self._OnHeartBeatLost('ConnectionError exception:[%s]. If SHUTDOWN, then possibly due to a too long/unrecoverable DB Cluster fail over.') 2895 except Exception: 2896 self.LogError('!! ---------------------------------------------------------- !!') 2897 self.LogError('Oh my God! An exception during zcluster.Nodes_HeartBeat!!!') 2898 self.LogError('!! ---------------------------------------------------------- !!') 2899 log.LogException() 2900 self._OnHeartBeatLost('Exception during zcluster.Nodes_HeartBeat') 2901 2902 if not __debug__: 2903 if secs > self.cleanupInterval: 2904 self._OnHeartBeatLagging() 2905 if not hasattr(self, 'slowdeath'): 2906 self.slowdeath = 1 2907 self.LogError('Node ', self.nodeID, ' is dead. The server is configured with a ', self.cleanupInterval, " second cleanup interval, but it's been ", secs, ' seconds since') 2908 self.LogError("he last performed a heartbeat. That means every other server on the network have assumed that he's dead") 2909 self.LogError('and will remove his entries from the database, effectively turning him into a mummy that wanders around,') 2910 self.LogError('borking up everything.') 2911 self.LogError("The server is overloaded and dead as far as everybody is concerned. This may NEVER happen on a live server. It's heartbeat is supposed to be every ", self.heartbeatInterval, " secs, but it's down to ", secs, '. Braindeath occurred at ', self.cleanupInterval) 2912 self.LogError('A reboot is probably in order.') 2913 elif secs > self.cleanupInterval - self.heartbeatInterval: 2914 self.LogError("The server is extremely loaded, and basically taking the highway to doom. It's heartbeat is supposed to be every ", self.heartbeatInterval, " secs, but it's down to ", secs, '. Braindeath occurs at ', self.cleanupInterval) 2915 elif secs > self.heartbeatInterval * 4 / 3: 2916 self.LogWarn("The server is very loaded, beware... It's heartbeat is supposed to be every ", self.heartbeatInterval, " secs, but it's down to ", secs, '. Braindeath occurs at ', self.cleanupInterval) 2917 elif secs < 0: 2918 self.LogError('!! ---------------------------------------------------------- !!') 2919 self.LogError("Oh my God! I've been removed from zcluster.nodes!!! That means I'm already dead, I just didn't know about it!!!!") 2920 self.LogError("How embarrassing. I must be causing all sorts of trouble. I think I'll just lie down and die of shame.") 2921 self.LogError('!! ---------------------------------------------------------- !!') 2922 self._OnHeartBeatLost("zcluster.Nodes_HeartBeat indicated that I've been removed from zcluster.nodes") 2923 else: 2924 self._OnHeartBeatNormal() 2925 2926 def _OnHeartBeatLagging(self): 2927 if not self.heartbeat_lagging: 2928 self.heartbeat_lagging = True 2929 sm.ScatterEvent('OnNodeHeartbeatLagging') 2930 2931 def _OnHeartBeatNormal(self): 2932 if self.heartbeat_lagging: 2933 self.heartbeat_lagging = False 2934 sm.ScatterEvent('OnNodeHeartbeatBackToNormal') 2935 2936 def _OnHeartBeatLost(self, reason = 'zcluster.Nodes_HeartBeat problem'): 2937 cleanup_timeout_in_seconds = 10 2938 sm.ScatterEvent('OnNodeHeartbeatLost', reason=reason, cleanup_timeout_in_seconds=cleanup_timeout_in_seconds) 2939 self.Stop(reason=reason) 2940 uthread2.sleep(cleanup_timeout_in_seconds + 1) 2941 log.Quit(reason) 2942 2943 def OnCleanUp(self): 2944 if not __debug__: 2945 while not getattr(self, 'stop', True): 2946 try: 2947 blue.pyos.synchro.SleepWallclock(int(self.cleanupInterval * 1000)) 2948 self.dbzcluster.Nodes_TrashLimbos(self.cleanupInterval) 2949 except Exception: 2950 log.LogException('Exception during OnCleanUp. Ignoring.') 2951 sys.exc_clear() 2952 2953 def GetGPCS(self, channel = None): 2954 if macho.mode == 'client': 2955 while self.authenticating == 2: 2956 blue.pyos.synchro.SleepWallclock(500) 2957 2958 while not self.state == SERVICE_RUNNING: 2959 self.LogError("GetGPCS shouldn't be called on machoNet before it's running...") 2960 log.LogTraceback() 2961 blue.pyos.synchro.SleepWallclock(500) 2962 2963 return self.channelHandlersDown.get(channel, None) 2964 2965 def Ping(self, pingCount = 5, nodeID = None, silent = False): 2966 if nodeID is None: 2967 address = MachoAddress() 2968 daNode = 'random node' 2969 else: 2970 address = MachoAddress(nodeID=nodeID) 2971 daNode = 'node %s' % nodeID 2972 2973 def prepared_print(prepared): 2974 lengths = [0] * len(prepared[0]) 2975 for i in range(len(prepared[0])): 2976 for each in prepared: 2977 each[i] = strx(each[i]) 2978 if len(each[i]) > lengths[i]: 2979 lengths[i] = len(each[i]) + 1 2980 2981 for each in prepared: 2982 each[i] += ' ' * (lengths[i] - len(each[i])) 2983 2984 for i in range(0, len(prepared)): 2985 each = prepared[i] 2986 if i < 2: 2987 each[1] = each[1][:-lengths[3] + 1] 2988 each[2] = each[2][:-lengths[3] + 1] 2989 else: 2990 each[1] = each[1][-lengths[3]:] 2991 each[2] = each[2][-lengths[3]:] 2992 2993 for each in prepared: 2994 x = '' 2995 for other in each: 2996 x += other 2997 2998 if not silent: 2999 print(x) 3000 3001 summaries = [] 3002 for i in range(pingCount): 3003 if not silent: 3004 print('---------------------------------------------------------------------------------------------') 3005 print(' MachoNet Ping statistics for %s' % daNode) 3006 print('---------------------------------------------------------------------------------------------') 3007 prepared = [['Where', 3008 'TickID', 3009 'Time', 3010 'Time-Tick', 3011 'curr-start', 3012 'curr-start', 3013 'Time-Last Time', 3014 'Time-Last Time'], ['', 3015 '', 3016 '', 3017 '', 3018 '(blues)', 3019 '(millisecs)', 3020 '(blues)', 3021 '(millisecs)']] 3022 retval = self._BlockingCall(PingReq(destination=address, times=[(blue.os.GetWallclockTime(), blue.os.GetWallclockTimeNow(), macho.mode + '::start')])) 3023 retval.times.append((blue.os.GetWallclockTime(), blue.os.GetWallclockTimeNow(), macho.mode + '::stop')) 3024 for n in range(len(retval.times)): 3025 each = retval.times[n] 3026 stopminusstartb = '' 3027 stopminusstartm = '' 3028 m = 0 3029 if each[2] in ('proxy::after_tasklet', 'server::after_tasklet', 'client::after_tasklet'): 3030 m = 1 3031 elif each[2] in ('proxy::writing', 'server::turnaround', 'client::stop'): 3032 m = 2 3033 if m: 3034 other = retval.times[n - m] 3035 stopminusstartb = long(round(float(each[1] - other[1]))) 3036 stopminusstartm = long(round(float(each[1] - other[1]) * 10.0 / float(const.MSEC))) 3037 if stopminusstartm: 3038 stopminusstartm = str(stopminusstartm / 10) + '.' + str(stopminusstartm % 10) 3039 else: 3040 stopminusstartm = '' 3041 tmltb = '' 3042 tmltm = '' 3043 if n > 2: 3044 tmltb = long(each[1]) - long(retval.times[n - 1][1]) 3045 tmltm = long(round(float(tmltb) * 10.0 / float(const.MSEC))) 3046 if tmltm: 3047 tmltm = str(tmltm / 10) + '.' + str(tmltm % 10) 3048 else: 3049 tmltm = '' 3050 prepared.append([each[2], 3051 each[0], 3052 each[1], 3053 each[1] - each[0], 3054 stopminusstartb, 3055 stopminusstartm, 3056 tmltb, 3057 tmltm]) 3058 3059 prepared_print(prepared) 3060 clientFirst = len(prepared) 3061 proxyFirst = len(prepared) 3062 serverFirst = len(prepared) 3063 for i in reversed(range(len(prepared))): 3064 if prepared[i][0].startswith('client::'): 3065 clientFirst = i 3066 if prepared[i][0].startswith('proxy::'): 3067 proxyFirst = i 3068 if prepared[i][0].startswith('server::'): 3069 serverFirst = i 3070 3071 clientLast = -1 3072 proxyLast = -1 3073 serverLast = -1 3074 for i in range(len(prepared)): 3075 if prepared[i][0].startswith('client::'): 3076 clientLast = i 3077 if prepared[i][0].startswith('proxy::'): 3078 proxyLast = i 3079 if prepared[i][0].startswith('server::'): 3080 serverLast = i 3081 3082 cr = 10000000000.0 3083 if clientFirst < clientLast: 3084 cr = (float(prepared[clientLast][2].strip()) - float(prepared[clientFirst][2].strip())) / const.MSEC 3085 if not silent: 3086 print('client roundtrip: ', cr, ' msec') 3087 pr = 10000000000.0 3088 if proxyFirst < proxyLast: 3089 pr = (float(prepared[proxyLast][2].strip()) - float(prepared[proxyFirst][2].strip())) / const.MSEC 3090 if not silent: 3091 print('proxy roundtrip: ', cr, ' msec') 3092 server1 = [ p[0].strip() for p in prepared ].index('proxy::writing') 3093 server2 = [ p[0].strip() for p in prepared ].index('proxy::handle_message', server1) 3094 diff = float(prepared[server2][2]) - float(prepared[server1][2]) 3095 diff /= const.MSEC 3096 pspr = diff 3097 if not silent: 3098 print('proxy-server-proxy rountrip: %sms' % pspr) 3099 tickids = [] 3100 for each in prepared[2:]: 3101 if each[1] not in tickids: 3102 tickids.append(each[1]) 3103 3104 if not silent: 3105 print('Roundtrip tick count=', len(tickids)) 3106 summaries.append((cr, 3107 pr, 3108 pspr, 3109 len(tickids))) 3110 3111 average = [ sum([ s[i] for s in summaries ]) / len(summaries) for i in range(len(summaries[0])) ] 3112 minerage = [ min([ s[i] for s in summaries ]) for i in range(len(summaries[0])) ] 3113 if not silent: 3114 print('Totals for: Client rountrip, proxy roundtrip, proxy-server-proxy roundtrip, tickcount') 3115 print('Average: ' + repr(average)) 3116 print('Minimum: ' + repr(minerage)) 3117 return retval.times 3118 3119 def GetServiceInfo(self): 3120 with self.fetchedDataLock: 3121 if self.fetchedServiceInfo is None: 3122 if macho.mode == 'proxy': 3123 self.fetchedServiceInfo = self.session.ConnectToRemoteService('machoNet').GetServiceInfo() 3124 self.fetchedServiceInfo.GetCachedObject() 3125 else: 3126 serviceInfo = {} 3127 for svc in sm.services.itervalues(): 3128 if hasattr(svc, '__machoresolve__'): 3129 serviceInfo[svc.__logname__] = svc.__machoresolve__ 3130 3131 self.fetchedServiceInfo = utilCachedObject(1, 'machoNet.serviceInfo', serviceInfo) 3132 return self.fetchedServiceInfo 3133 3134 def ClearFetchedData(self): 3135 with self.fetchedDataLock: 3136 self.fetchedServiceInfo = None 3137 if macho.mode == 'server': 3138 self.ProxyBroadcast('OnClearFetchedData') 3139 3140 def OnClearFetchedData(self): 3141 if macho.mode == 'proxy': 3142 self.ClearFetchedData() 3143 3144 def OnGlobalConfigUpdated(self, key, value): 3145 self.LogNotice('OnGlobalConfigUpdated', key, value) 3146 if self.globalConfig is None: 3147 self.globalConfig = self.GetGlobalConfig() or {} 3148 if value is None: 3149 if key in self.globalConfig: 3150 del self.globalConfig[key] 3151 else: 3152 self.globalConfig[key] = value 3153 monolithconfig.trigger_global_config_updates() 3154 sm.ScatterEvent('OnGlobalConfigChanged', self.globalConfig) 3155 3156 @bluepy.TimedFunction('machoNet::__IsVIP') 3157 def __IsVIP(self, vipKey): 3158 if vipKey is None: 3159 return False 3160 with self.vipLock: 3161 if self.vipkeys is None: 3162 self.__PrimeVIPList() 3163 return vipKey in self.vipkeys 3164 3165 @bluepy.TimedFunction('machoNet::__IsVIPbyUser') 3166 def __IsVIPbyUser(self, userID): 3167 if userID is None: 3168 return False 3169 with self.vipLock: 3170 if self.vipUsers is None: 3171 self.__PrimeVIPList() 3172 return userID in self.vipUsers 3173 3174 def __PrimeVIPList(self): 3175 self.vipkeys = set() 3176 self.vipUsers = set() 3177 for r in self.session.ConnectToSolServerService('DB2').CallProc('zuser.Users_SelectByRole', ROLE_VIPLOGIN): 3178 self.vipkeys.add(CryptoHash(utilFormat.CaseFold(r.userName))) 3179 self.vipUsers.add(r.userID) 3180 3181 def OnVIPListChanged(self): 3182 with self.vipLock: 3183 self.vipkeys = None 3184 self.vipUsers = None 3185 3186 def IsThisUserCoolForLogin(self, userID): 3187 if self.IsInVIP(): 3188 return self.__IsVIPbyUser(userID) 3189 return True 3190 3191 def IsInVIP(self): 3192 return bool(self.vipMode) 3193 3194 @bluepy.TimedFunction('machoNet::GetLogonQueuePosition') 3195 def GetLogonQueuePosition(self, transportID, vipKey): 3196 if self.__IsVIP(vipKey): 3197 return 1 3198 try: 3199 position = self.serverLogonQueue.index(transportID) + 2 3200 except ValueError: 3201 log.LogException() 3202 position = 2 3203 3204 position -= self.availableLoginSlots 3205 return max(1, position) 3206 3207 def PutTransportInLogonQueue(self, transportID): 3208 self.serverLogonQueue.append(transportID) 3209 3210 def RemoveTransportFromLogonQueue(self, transportID): 3211 self.serverLogonQueue.remove(transportID) 3212 3213 def GetLogonQueueStats(self): 3214 loggedInConnections = len(self.transportIDbyClientID) 3215 pendingConnections = len(self.serverLogonQueue) 3216 return (loggedInConnections, pendingConnections, self.availableLoginSlots) 3217 3218 def Authenticate(self, transport, request, sessionID): 3219 clientID = self.GetIDOfAddress(transport.address, clientMode=True) 3220 ssoToken = request.get('user_sso_token', None) 3221 authSvc = self.session.ConnectToRemoteService('authentication') 3222 loginInfo, access_token, computer_hash = authSvc.Login(sessionID, request['user_name'], request['user_password'], request['user_password_hash'], const.userConnectTypeClient, transport.address, clientID, request, ssoToken=ssoToken, clientVersion=request['boot_build']) 3223 ret = {'sessionID': sessionID, 3224 'user_clientid': clientID, 3225 'session_init': loginInfo, 3226 'access_token': access_token, 3227 'computer_hash': computer_hash} 3228 user_id = loginInfo['userid'] 3229 if not self.IsThisUserCoolForLogin(user_id): 3230 ipaddress = transport.address.split(':')[0] 3231 authSvc.Logout(sessionID, ipaddress=ipaddress) 3232 raise UserRejectedByVIP() 3233 transport.sessionID = sessionID 3234 uthread.worker('machoNet::AutoLogOffClient', self.__AutoLogOffClient, transport, sessionID) 3235 return ret 3236 3237 def __AutoLogOffClient(self, transport, sessionID): 3238 blue.pyos.synchro.SleepWallclock(self.autoLogoffAuthenticatedTransportInterval * 1000) 3239 if sessionID not in self.transportIDbySessionID and not transport.IsClosed(): 3240 transport.Close('Authentication Timeout', 'HANDSHAKE_TIMEOUT_AUTHENTICATED') 3241 ipaddress = transport.address.split(':')[0] 3242 self.session.ConnectToRemoteService('authentication').Logout(sessionID, ipaddress=ipaddress) 3243 3244 def LogOffSession(self, sessionID, ipaddress = None): 3245 if sessionID and not self.IsClusterShuttingDown(): 3246 uthread.worker('machoNet::LogOffSession', self.session.ConnectToRemoteService('authentication').Logout, sessionID, ipaddress=ipaddress) 3247 3248 def GetTime(self): 3249 return blue.os.GetWallclockTimeNow() 3250 3251 def SendProvisionalResponse(self, method, *args, **keywords): 3252 if currentcall: 3253 timeout = keywords.get('machoTimeout', self.callTimeOutInterval) 3254 response = currentcall.packet.Response(payload=()) 3255 response.oob['provisional'] = (timeout, method, args) 3256 response.trace_id = None 3257 response.parent_span_id = None 3258 self.LogInfo('Sending provisional response and updating timeout interval', response) 3259 self._NonBlockingCall(response) 3260 else: 3261 self.LogWarn('Ignoring provisional response request, currentcall=', currentcall) 3262 3263 def __TimeoutCall(self, channel): 3264 channel.send_exception(RuntimeError, ('OnMachoTimeout', {'what': 'A low-level timeout occurred during a remote service request'})) 3265 3266 def __TimeoutCallLoop(self): 3267 while not getattr(self, 'stop', True): 3268 try: 3269 timeout = [] 3270 blue.pyos.synchro.SleepWallclock(100 * self.callTimeOutInterval) 3271 if self.reducedTimeouts: 3272 timeoutCalls = self.timeoutCalls 3273 else: 3274 timeoutCalls = self.calls 3275 for k in timeoutCalls.keys(): 3276 if blue.os.GetWallclockTime() > self.calls[k][1]: 3277 v = self.calls[k] 3278 timeout.append((self.__TimeoutCall, (v[0],))) 3279 del self.calls[k] 3280 if k in self.timeoutCalls: 3281 del self.timeoutCalls[k] 3282 self.callsCountMin = min(self.callsCountMin, len(self.calls)) 3283 if k in v[2]: 3284 del v[2][k] 3285 3286 if timeout: 3287 uthread.parallel(timeout) 3288 except Exception: 3289 log.LogException('Exception during Timeout Call Loop') 3290 sys.exc_clear() 3291 3292 def __KATLoop(self): 3293 if macho.mode == 'client': 3294 keepAliveTimerInterval = self.clientKeepAliveTimerInterval 3295 else: 3296 keepAliveTimerInterval = self.serverKeepAliveTimerInterval 3297 while not getattr(self, 'stop', True): 3298 try: 3299 blue.pyos.synchro.SleepWallclock(keepAliveTimerInterval * 1000 / 4) 3300 if not self.clockReset: 3301 continue 3302 time = blue.os.GetWallclockTime() 3303 clientCheck = time - self.clientKeepAliveTimerInterval * const.SEC 3304 clientDead = time - 3 * self.clientKeepAliveTimerInterval * const.SEC 3305 serverCheck = time - self.serverKeepAliveTimerInterval * const.SEC 3306 serverDead = time - 3 * self.serverKeepAliveTimerInterval * const.SEC 3307 for transport in self.transportsByID.itervalues(): 3308 if not transport.lastPing: 3309 continue 3310 if transport.clientID: 3311 check, dead = clientCheck, clientDead 3312 else: 3313 check, dead = serverCheck, serverDead 3314 req, resp = transport.lastPing 3315 if resp >= check: 3316 continue 3317 if hasattr(transport, 'done_broadcasting_close'): 3318 continue 3319 if resp < dead and req > resp: 3320 if transport.clientID: 3321 transport.Close('Client Transport keep-alive timer has expired', 'KEEPALIVEEXPIRED') 3322 else: 3323 transport.Close('Proxy/Server Transport keep-alive timer was determined expired by %s #%s' % (macho.mode, self.nodeID)) 3324 elif not transport.pinging: 3325 uthread.worker('machoNet::KATPing', self.__KATPing, transport) 3326 3327 except Exception: 3328 log.LogException('Exception during KAT Loop') 3329 sys.exc_clear() 3330 3331 def __KATPing(self, transport): 3332 try: 3333 transport.pinging = True 3334 try: 3335 t0 = blue.os.GetWallclockTimeNow() 3336 if not transport.lastPing: 3337 transport.lastPing = [0, 0] 3338 transport.lastPing[0] = t0 3339 response = self._BlockingCall(PingReq(destination=transport.GetMachoAddressOfOtherSide(), times=[])) 3340 t1 = blue.os.GetWallclockTimeNow() 3341 tResponse = response.times[0][1] - transport.estimatedRTT / 2 3342 tDiff = t1 - t0 3343 transport.estimatedRTT = transport.estimatedRTT * 0.85 + (t1 - t0) * 0.15 3344 transport.timeDiff = transport.timeDiff * 0.85 + (t0 + (t1 - t0) / 2 - tResponse) * 0.15 3345 transport.lastPing[1] = t1 3346 finally: 3347 transport.pinging = False 3348 3349 except UnMachoDestination as e: 3350 self.LogWarn('UnMachoDestination encountered while pinging ', transport.GetMachoAddressOfOtherSide(), '. Probably irrelevant') 3351 sys.exc_clear() 3352 except GPSTransportClosed as e: 3353 self.LogInfo("KATPing detected a closed transport. That's what it's supposed to do...") 3354 if transport.transportID in self.transportsByID: 3355 self.transportsByID[transport.transportID].Close('KATPing detected that the socket was closed', 'KEEPALIVEEXPIRED') 3356 sys.exc_clear() 3357 except Exception: 3358 self.LogError("Unhandled exception in KATPing. Shouldn't be a problem, but I wanna see it") 3359 self.LogError('Worst case, transport keepalive timers should be failing to do their job') 3360 log.LogException() 3361 sys.exc_clear() 3362 3363 def _BlockingCall(self, packet, destTransport = None): 3364 if destTransport is None: 3365 transports = self._GetTransports(packet.destination) 3366 if len(transports) > 1: 3367 raise UnMachoDestination('A blocking call cannot be performed using a forking address. Use ConnectToAllNeighboringServices instead.') 3368 destTransport = transports[0] 3369 callID = self.callID 3370 self.callID += 1 3371 if macho.mode == 'client': 3372 source = MachoAddress(clientID=0, callID=callID) 3373 else: 3374 source = MachoAddress(nodeID=self.nodeID, callID=callID) 3375 packet.source = source 3376 self.calls[callID] = [uthread.Channel('machoNet::BlockingCall'), 3377 blue.os.GetWallclockTime() + packet.oob.get('machoTimeout', self.callTimeOutInterval) * const.SEC, 3378 destTransport.calls, 3379 callID] 3380 packetTimeout = packet.oob.get('machoTimeout', None) 3381 if packetTimeout: 3382 self.timeoutCalls[callID] = blue.os.GetWallclockTime() + packetTimeout * const.SEC 3383 self.callsCountMax = max(self.callsCountMax, len(self.calls)) 3384 try: 3385 destTransport.calls[callID] = self.calls[callID][0] 3386 before = blue.os.GetWallclockTime() 3387 try: 3388 destTransport.Write(packet) 3389 while 1: 3390 retval = self.calls[callID][0].receive() 3391 if macho.mode == 'client': 3392 destTransport.TagPacketSizes(packet, retval) 3393 provisional = retval.oob.get('provisional', None) 3394 if provisional is None: 3395 break 3396 if macho.mode == 'client': 3397 self.LogInfo('Received Provisional Response, provisional=', provisional) 3398 self.calls[callID][1] = blue.os.GetWallclockTime() + provisional[0] * const.SEC 3399 if provisional[1].startswith('Process'): 3400 sm.ChainEventWithoutTheStars(provisional[1], provisional[2]) 3401 elif provisional[1].startswith('Do'): 3402 sm.SendEventWithoutTheStars(provisional[1], provisional[2]) 3403 else: 3404 sm.ScatterEventWithoutTheStars(provisional[1], provisional[2]) 3405 3406 finally: 3407 if callID in destTransport.calls: 3408 del destTransport.calls[callID] 3409 after = blue.os.GetWallclockTime() 3410 diff = after - before 3411 self.blockingCallTimes.Add(diff) 3412 3413 except ReferenceError: 3414 exctype, exc, tb = sys.exc_info() 3415 try: 3416 raise_(RuntimeError('ReferenceError'), None, tb) 3417 finally: 3418 exctype = None 3419 xc = None 3420 tb = None 3421 3422 finally: 3423 if callID in self.timeoutCalls: 3424 del self.timeoutCalls[callID] 3425 if callID in self.calls: 3426 del self.calls[callID] 3427 self.callsCountMin = min(self.callsCountMin, len(self.calls)) 3428 if callID in destTransport.calls: 3429 del destTransport.calls[callID] 3430 3431 if retval.command == cluster.MACHONETMSG_TYPE_ERRORRESPONSE: 3432 if retval.code == cluster.MACHONETERR_UNMACHODESTINATION: 3433 raise UnMachoDestination(retval.payload) 3434 elif retval.code == cluster.MACHONETERR_UNMACHOCHANNEL: 3435 raise UnMachoChannel(retval.payload) 3436 elif retval.code == cluster.MACHONETERR_WRAPPEDEXCEPTION: 3437 raise MachoWrappedException(retval.payload) 3438 else: 3439 raise MachoException(retval.payload) 3440 return retval 3441 3442 CallDown = _BlockingCall 3443 3444 def _NonBlockingCall(self, packet): 3445 try: 3446 transports = self._GetTransports(packet.destination) 3447 for each in transports: 3448 each.Write(packet) 3449 each.TagPacketSizes(packet) 3450 3451 except UnMachoDestination as e: 3452 self.LogInfo("_NonBlockingCall failed for an UnMachoDestination. The reason given was '", e.payload, "'. Generally speaking, this is probably just fine and dandy.") 3453 self.LogInfo('Packet=', packet) 3454 sys.exc_clear() 3455 except GPSTransportClosed as e: 3456 self.LogInfo("_NonBlockingCall failed for a closed transport. The reason given was '", e.reason, "'. Generally speaking, this is probably just fine and dandy.") 3457 self.LogInfo('Packet=', packet) 3458 sys.exc_clear() 3459 3460 NotifyDown = _NonBlockingCall 3461 3462 def __TransportChoice(self, transportIDs): 3463 if len(transportIDs) == 1: 3464 return transportIDs[0] 3465 options = [] 3466 for transportID in transportIDs: 3467 try: 3468 nodeID = self.transportsByID.get(transportID).nodeID 3469 cpu = self.nodeCPULoadValue.get(nodeID, 1.0) 3470 weight = 1.0 / float(max(0.01, cpu)) 3471 except: 3472 weight = 1.0 3473 3474 options.append([weight, transportID]) 3475 3476 if len(options) > 1: 3477 options.sort(lambda x, y: cmp(y[0], x[0])) 3478 options.pop() 3479 return weightedChoice(options) 3480 3481 @bluepy.TimedFunction('machoNet::_GetTransports') 3482 def _GetTransports(self, destAddress, srcTransport = None): 3483 if macho.mode == 'client': 3484 srcTransport = self.namedtransports.get('tcp:packet:machoNet', None) 3485 if srcTransport and destAddress.addressType == const.ADDRESS_TYPE_BROADCAST and len(destAddress.narrowcast) == 1: 3486 appNodeID = destAddress.narrowcast[0] 3487 if appNodeID in self.transportIDbyAppNodeID: 3488 machoTransport = self.transportsByID[self.transportIDbyAppNodeID[appNodeID]] 3489 return [machoTransport] 3490 if 'ip:packet:server' not in self.namedtransports: 3491 raise GPSTransportClosed('The transport has not yet been connected, or authentication was not successful.') 3492 return [self.namedtransports['ip:packet:server']] 3493 else: 3494 transportID = None 3495 if destAddress.addressType == const.ADDRESS_TYPE_ANY: 3496 if srcTransport is not None and srcTransport.dependants: 3497 mk, mv = [], 0 3498 for k, v in srcTransport.dependants.iteritems(): 3499 if v > mv: 3500 mk = [k] 3501 mv = v 3502 elif v == mv: 3503 mk.append(k) 3504 3505 if mk: 3506 transportID = self.__TransportChoice(mk) 3507 else: 3508 transportID = self.__TransportChoice(srcTransport.dependants.keys()) 3509 elif macho.mode == 'server': 3510 if self.transportIDbyProxyNodeID: 3511 transportID = random.choice(self.transportIDbyProxyNodeID.values()) 3512 else: 3513 raise UnMachoDestination('Could not connect to any proxy server') 3514 elif self.transportIDbySolNodeID: 3515 transportID = self.__TransportChoice(self.transportIDbySolNodeID.values()) 3516 else: 3517 raise UnMachoDestination('Could not connect to any sol server') 3518 elif destAddress.addressType == const.ADDRESS_TYPE_NODE: 3519 if self._addressCache.IsDeadNode(destAddress.nodeID): 3520 raise UnMachoDestination('The specified proxy or server node (%s) is dead' % destAddress.nodeID) 3521 if destAddress.nodeID in self.transportIDbySolNodeID: 3522 transportID = self.transportIDbySolNodeID[destAddress.nodeID] 3523 elif destAddress.nodeID in self.transportIDbyProxyNodeID: 3524 transportID = self.transportIDbyProxyNodeID[destAddress.nodeID] 3525 elif destAddress.nodeID in self.transportIDbyAppNodeID: 3526 transportID = self.transportIDbyAppNodeID[destAddress.nodeID] 3527 elif macho.mode == 'server': 3528 if len(self.transportIDbyProxyNodeID): 3529 transportID = random.choice(self.transportIDbyProxyNodeID.values()) 3530 else: 3531 if destAddress.nodeID == self.nodeID: 3532 self.LogError('This node (%s) is attempting to resolve a MachoNet Transport to itself' % self.nodeID) 3533 raise UnMachoDestination('This node (%s) is attempting to resolve a MachoNet Transport to itself' % self.nodeID) 3534 if not self._addressCache.IsDeadNode(destAddress.nodeID): 3535 self._addressCache.MarkDeadNode(destAddress.nodeID) 3536 reason = "The node wasn't registered in the proxy's transportIDbySolNodeID or transportIDbyProxyNodeID maps" 3537 self.LogNotice('Sending OnNodeDeath for', destAddress.nodeID, 'because', reason) 3538 self.ServerBroadcast('OnNodeDeath', destAddress.nodeID, 1, reason) 3539 raise UnMachoDestination('The specified proxy or server node (%s) could not be reached' % destAddress.nodeID) 3540 elif destAddress.addressType == const.ADDRESS_TYPE_CLIENT: 3541 if destAddress.clientID in self.transportIDbyClientID: 3542 transportID = self.transportIDbyClientID[destAddress.clientID] 3543 else: 3544 raise UnMachoDestination('The client (%s) could not be reached' % destAddress.clientID) 3545 else: 3546 if destAddress.addressType == const.ADDRESS_TYPE_BROADCAST: 3547 retval = [] 3548 transportIDs = self.GetTransportIDsFromBroadcastAddress(destAddress) 3549 for each in transportIDs: 3550 retval.append(self.transportsByID[each]) 3551 3552 if not len(retval): 3553 self.LogInfo('None of the specified destination addresses could be reached for ', destAddress) 3554 return retval 3555 raise UnMachoDestination('The specified address type is invalid') 3556 if transportID is None: 3557 self.LogError('Some dude is sending a weird address (', destAddress, ")that doesn't resolve to a transport ID") 3558 log.LogTraceback() 3559 raise UnMachoDestination('The transportID is invalid') 3560 if transportID not in self.transportsByID: 3561 raise UnMachoDestination('The transportID does not identify a known transport') 3562 return [self.transportsByID[transportID]] 3563 3564 def GetTransportIDsFromBroadcastAddress(self, destAddress): 3565 return self._GetTransportIDsFromBroadcastAddress(destAddress) 3566 3567 @bluepy.TimedFunction('machoNet::_GetTransportIDsFromBroadcastAddress') 3568 def _GetTransportIDsFromBroadcastAddress(self, destAddress): 3569 if destAddress.idtype is None: 3570 idtype = None 3571 scattered = 0 3572 elif destAddress.idtype[0] in ('*', '+'): 3573 idtype = destAddress.idtype[1:] 3574 scattered = 1 3575 else: 3576 idtype = destAddress.idtype 3577 scattered = 0 3578 if len(destAddress.narrowcast): 3579 clientIDs = [] 3580 nodeIDs = [] 3581 done = 0 3582 if idtype == 'clientID': 3583 if macho.mode == 'server' and len(destAddress.narrowcast) >= 2 * len(self.transportIDbyProxyNodeID): 3584 nodeIDs = self.transportIDbyProxyNodeID.iterkeys() 3585 else: 3586 clientIDs = destAddress.narrowcast 3587 done = 1 3588 elif idtype == 'nodeID': 3589 nodeIDs = destAddress.narrowcast 3590 done = 1 3591 elif idtype == 'serviceMask': 3592 nodeIDs = self.ResolveServiceMaskToNodeIDs(destAddress.narrowcast[0]) 3593 done = 1 3594 elif macho.mode == 'server': 3595 if idtype in self.__server_scattercast_session_variables__: 3596 if scattered: 3597 nodeIDs = self.transportIDbyProxyNodeID.iterkeys() 3598 done = 1 3599 elif len(destAddress.narrowcast) == 1: 3600 done, nodeIDs = self._GetServerBroadcastNodesSingle(idtype, destAddress, scattered) 3601 if done is None: 3602 self.LogWarn('Sending a packet by some funky address type (', idtype, '). Resorting to scattercast') 3603 nodeIDs = self.transportIDbyProxyNodeID.iterkeys() 3604 done = 1 3605 else: 3606 done, nodeIDs = self._GetServerBroadcastNodesMultiple(idtype, destAddress, scattered) 3607 if done is None: 3608 if not scattered: 3609 self.LogWarn('Sending a packet via a non-scattered complex address that resorts to a scattercast. address: ', destAddress) 3610 else: 3611 self.LogInfo('Sending a packet via a scattered complex address. address: ', destAddress) 3612 nodeIDs = self.transportIDbyProxyNodeID.iterkeys() 3613 done = 1 3614 if not done: 3615 notfound = [] 3616 if len(destAddress.narrowcast) == 1 and (idtype, destAddress.narrowcast[0]) in self.spam: 3617 spam = 1 3618 elif macho.mode == 'server': 3619 if self.transportIDbyProxyNodeID: 3620 spam, clientIDs, notfound = basesession.FindClientsAndHoles(idtype, destAddress.narrowcast, len(self.transportIDbyProxyNodeID) * 2) 3621 else: 3622 spam, clientIDs, notfound = basesession.FindClientsAndHoles(idtype, destAddress.narrowcast, 20) 3623 else: 3624 spam, clientIDs, notfound = basesession.FindClientsAndHoles(idtype, destAddress.narrowcast, None) 3625 if len(destAddress.narrowcast) == 1 and spam: 3626 self.LogInfo('Interpreting ', destAddress.narrowcast, ' as a persistant spam address.') 3627 if len(self.spam) > 1000: 3628 self.spam.clear() 3629 self.spam.add((idtype, destAddress.narrowcast[0])) 3630 if spam or macho.mode == 'server' and destAddress.idtype[0] == '*' and len(notfound): 3631 clientIDs = [] 3632 if macho.mode == 'server': 3633 nodeIDs = self.transportIDbyProxyNodeID.iterkeys() 3634 else: 3635 nodeIDs = self.transportIDbySolNodeID.iterkeys() 3636 transportIDs = set() 3637 if macho.mode == 'server' and self.transportIDbyProxyNodeID: 3638 for clientID in clientIDs: 3639 try: 3640 nodeID = self.GetProxyNodeIDFromClientID(clientID) 3641 except Exception as e: 3642 self.LogWarn('Unable to map clientID', repr(clientID), 'to proxyID. Dropping message Exception type=', type(e), ' args=', e.args) 3643 continue 3644 3645 if nodeID in self.transportIDbyProxyNodeID: 3646 transportIDs.add(self.transportIDbyProxyNodeID[nodeID]) 3647 if len(transportIDs) == len(self.transportIDbyProxyNodeID): 3648 if len(transportIDs) > 2: 3649 self.LogWarn('All proxies targeted in a clientID based routing decision. If this happens frequently, the caller should resort to a better casting method.') 3650 break 3651 3652 else: 3653 for each in clientIDs: 3654 if each in self.transportIDbyClientID and self.transportIDbyClientID[each] in self.transportsByID: 3655 transportIDs.add(self.transportIDbyClientID[each]) 3656 else: 3657 self.LogInfo('Transport for client ', each, ' not found while sending narrowcast.') 3658 3659 for each in nodeIDs: 3660 if macho.mode == 'server' and each in self.transportIDbyProxyNodeID and self.transportIDbyProxyNodeID[each] in self.transportsByID: 3661 transportIDs.add(self.transportIDbyProxyNodeID[each]) 3662 elif macho.mode == 'proxy' and each in self.transportIDbySolNodeID and self.transportIDbySolNodeID[each] in self.transportsByID: 3663 transportIDs.add(self.transportIDbySolNodeID[each]) 3664 elif each in self.transportIDbyAppNodeID and self.transportIDbyAppNodeID[each] in self.transportsByID: 3665 transportIDs.add(self.transportIDbyAppNodeID[each]) 3666 elif each in self.transportIDbySolNodeID and self.transportIDbySolNodeID[each] in self.transportsByID: 3667 transportIDs.add(self.transportIDbySolNodeID[each]) 3668 else: 3669 self.LogInfo('Transport for node', each, 'not found while sending narrowcast.') 3670 3671 transportIDs = list(transportIDs) 3672 elif idtype == 'nodeID': 3673 if destAddress.idtype[0] == '+': 3674 transportIDs = self.transportIDbyProxyNodeID.values() 3675 else: 3676 transportIDs = self.transportIDbySolNodeID.values() 3677 elif idtype == 'clientID' and macho.mode == 'server': 3678 transportIDs = self.transportIDbyProxyNodeID.values() 3679 elif macho.mode == 'server': 3680 transportIDs = self.transportsByID.keys() 3681 else: 3682 transportIDs = self.transportIDbyClientID.values() 3683 if transportIDs: 3684 self.broadcastsResolved.Add(len(transportIDs)) 3685 else: 3686 self.broadcastsMissed.Add() 3687 return transportIDs 3688 3689 def __GetServerBroadcastNodesStub(self, idtype, destAddress, scattered): 3690 done = None 3691 nodeIDs = None 3692 return (done, nodeIDs) 3693 3694 _GetServerBroadcastNodesSingle = __GetServerBroadcastNodesStub 3695 _GetServerBroadcastNodesMultiple = __GetServerBroadcastNodesStub 3696 3697 def ResolveServiceMaskToNodeIDs(self, serviceMask): 3698 return self._addressCache.ResolveServiceMask(serviceMask) 3699 3700 def IsServiceMapped(self, serviceID): 3701 return bool(1 << serviceID - 1 & self.serviceMask) 3702 3703 def OnMachoObjectDisconnect(self, objectID, clientID, refID): 3704 sess = None 3705 if macho.mode == 'client': 3706 sess = session 3707 elif clientID in self.transportIDbyClientID: 3708 transportID = self.transportIDbyClientID[clientID] 3709 if transportID in self.transportsByID: 3710 transport = self.transportsByID[transportID] 3711 sess = transport.sessions.get(clientID, None) 3712 if sess is not None: 3713 sess.UnregisterMachoObject(objectID, refID) 3714 3715 def ProcessSessionChange(self, isremote, sess, change): 3716 if 'charid' in change and change['charid'][1] is not None and hasattr(sess, 'clientID'): 3717 blue.net.MapCharToClient(change['charid'][1], sess.clientID) 3718 if 'userid' in change and change['userid'][1] is None or 'charid' in change and change['charid'][0] is not None: 3719 clientID = getattr(sess, 'clientID', 0) 3720 if not clientID and 'clientID' in change: 3721 clientID = change['clientID'][0] 3722 if clientID: 3723 if clientID in self.transportIDbyClientID and self.transportIDbyClientID[clientID] not in self.transportIDbyProxyNodeID.itervalues() and self.transportIDbyClientID[clientID] not in self.transportIDbySolNodeID.itervalues(): 3724 transportID = self.transportIDbyClientID[clientID] 3725 if 'userid' in change and change['userid'][1] is None: 3726 reason = "The user's connection has been usurped on the " + macho.mode 3727 else: 3728 reason = 'Hot-swapping characters is not permitted' 3729 if macho.mode == 'server' and (len(self.transportIDbyProxyNodeID) or len(self.transportIDbySolNodeID)): 3730 self.LogError('Usurping node->node connection!?! Man do I smell deep shit going on here...') 3731 self.LogError('Received a session change notification where userID or charID was disappearing.') 3732 self.LogError('The session has a clientID(', clientID, "), so we should disconnect it's transport (", transportID, ').') 3733 self.LogError("This should be safe, because the transport isn'it a known node->node transport.") 3734 self.LogError('However, that is total crud, and we know it, because this is a sol server with proxy connections.') 3735 self.LogError("That means there are NO direct client connections, and we've got a situation here.") 3736 self.LogError("Here's a stacktrace:") 3737 log.LogTraceback() 3738 self.LogError("And here's a session dump:") 3739 sess.LogSessionError('USURPSERVER') 3740 self.LogError("And here's the change we got:") 3741 self.LogError('isremote=', isremote) 3742 self.LogError('session=', sess) 3743 self.LogError('change=', change) 3744 self.LogError("and here's some important info") 3745 self.LogError('self.transportIDbyClientID=', self.transportIDbyClientID) 3746 self.LogError('self.transportIDbyProxyNodeID=', self.transportIDbyProxyNodeID) 3747 self.LogError('self.transportIDbySolNodeID=', self.transportIDbySolNodeID) 3748 self.LogError('self.transportsByID[transportID]=', self.transportsByID[transportID]) 3749 self.LogError('Odds are that this server is going down anyway, and an explanation will be found elsewhere') 3750 else: 3751 uthread.worker('machoNet::DelayedTransportCloseFromProcessSessionChange', self.transportsByID[transportID].Close, reason) 3752 elif not isremote and 'charid' in change and change['charid'][1] is not None and self.shutdown is not None and self.shutdown.notify is not None and self.shutdown.when - const.HOUR <= blue.os.GetWallclockTime(): 3753 clientID = getattr(sess, 'clientID', 0) 3754 if not clientID and 'clientID' in change: 3755 clientID = change['clientID'][1] 3756 if clientID: 3757 self.SinglecastByClientID(clientID, 'OnClusterShutdownInitiated', self.shutdown.explanationLabel, self.shutdown.when, self.shutdown.duration) 3758 3759 def HandleAccept(self, transname, transport, llv): 3760 transportID = self.transportID 3761 counter = 0 3762 try: 3763 self.transportID = self.transportID + 1 3764 while not transport.IsClosed(): 3765 if counter > 0: 3766 blue.pyos.synchro.SleepWallclock(3000) 3767 machoTransport = None 3768 try: 3769 sessionID = basesession.GetNewSid() 3770 if llv: 3771 eveInVIP = self.IsInVIP() 3772 llvresponse = transport.HandleClientAuthentication(self.onlineData.get_logon_count(), transportID, counter, sessionID, eveInVIP) 3773 counter += 1 3774 if not llvresponse or type(llvresponse) is str and llvresponse != 'OK': 3775 self.LogError('Low-level version check failure while accepting from ', transport.address, '. Undoubtedly a handshake request.') 3776 continue 3777 elif type(llvresponse) is str: 3778 self.LogInfo('Finished dealing with accept from ', transport.address, ', probably a queue request') 3779 continue 3780 if self.stop: 3781 return 3782 if llv: 3783 machoTransport = MachoTransport(transportID, transport, transname, self) 3784 self.transportsByID[transportID] = machoTransport 3785 transport = None 3786 machoTransport.transport.SetKeepalive(20, 20) 3787 machoTransport.clientID = machoTransport.transport.handShake['user_clientid'] 3788 machoTransport.userID = machoTransport.transport.handShake['session_init']['userid'] 3789 machoTransport.sessionID = sessionID 3790 blue.net.EnumerateTransport(machoTransport.transport.socket.getSocketDescriptor(), machoTransport.transportID, machoTransport.transportName, machoTransport.userID, machoTransport.clientID, 0) 3791 blue.net.AddClient(machoTransport.clientID, machoTransport.transportID) 3792 s = basesession.CreateSession(sessionID, const.session.SESSION_TYPE_GAME) 3793 if macho.mode == 'proxy': 3794 self.session.ConnectToSolServerService('userSvc').ProcessEVESessionCreated(sessionID, machoTransport.transport.address) 3795 elif macho.mode == 'server': 3796 sm.ChainEvent('ProcessEVESessionCreated', sessionID, machoTransport.transport.address) 3797 s.__dict__['clientID'] = machoTransport.clientID 3798 llvresponse['session_init']['role'] |= s.role 3799 machoTransport.sessions[machoTransport.clientID] = s 3800 self.transportIDbyClientID[machoTransport.clientID] = machoTransport.transportID 3801 self.add_session_transport_lookup(machoTransport.sessionID, machoTransport.transportID) 3802 s.LogSessionHistory('Authenticating user %s via machoNet from address %s' % (llvresponse['user_name'], machoTransport.transport.address)) 3803 s.SetAttributes(llvresponse['session_init']) 3804 s.LogSessionHistory('Authenticated user %s via machoNet from address %s' % (llvresponse['user_name'], machoTransport.transport.address)) 3805 if not s.role & ROLE_VIPLOGIN: 3806 self.LogInfo('non-vip transport', machoTransport, 'has connected, with session', s, 'There are', self.availableLoginSlots, 'login slots remaining') 3807 self.availableLoginSlots -= 1 3808 else: 3809 machoTransport = MachoTransport(transportID, transport, transname, self) 3810 self.transportsByID[transportID] = machoTransport 3811 transport = None 3812 blue.net.EnumerateTransport(machoTransport.transport.socket.getSocketDescriptor(), machoTransport.transportID, machoTransport.transportName, 0, 0, 0) 3813 machoTransport.currentReaders += 1 3814 uthread.worker('machoNet::TransportReader', self.TransportReader, self.transportsByID[transportID]) 3815 except GPSTransportClosed as e: 3816 if machoTransport is not None: 3817 machoTransport.Close('GPSTransportClosed', exception=e) 3818 if transport is not None: 3819 transport.Close('GPSTransportClosed', exception=e) 3820 sys.exc_clear() 3821 except Exception as e: 3822 if machoTransport is not None: 3823 machoTransport.Close('Disconnected', exception=e) 3824 if transport is not None: 3825 transport.Close('Disconnected', exception=e) 3826 log.LogException() 3827 self.LogWarn('Caught %s in HandleAccept' % repr(e)) 3828 sys.exc_clear() 3829 3830 return 3831 3832 finally: 3833 if llv: 3834 try: 3835 self.RemoveTransportFromLogonQueue(transportID) 3836 except AttributeError: 3837 pass 3838 except ValueError: 3839 pass 3840 3841 def AcceptLoop(self, transname): 3842 try: 3843 listenTransport = self.GetTransport(transname) 3844 while not getattr(self, 'stop', True): 3845 try: 3846 self.LogInfo('Blocking on accept for ', transname) 3847 transport = None 3848 transport = listenTransport.Accept() 3849 self.LogInfo('Accepted a transport from ', transport.address) 3850 uthread.worker('machoNet::HandleAccept', self.HandleAccept, transname, transport, transname != 'tcp:packet:machoNet') 3851 except GPSTransportClosed: 3852 sys.exc_clear() 3853 self.LogWarn('ListenTransport closed') 3854 return 3855 3856 except Exception: 3857 log.LogException() 3858 self.LogError('AcceptLoop bombed') 3859 sys.exc_clear() 3860 3861 def TransportReader(self, srcTransport): 3862 srcTransport.currentReaders -= 1 3863 try: 3864 while 1: 3865 if srcTransport.currentReaders > srcTransport.desiredReaders: 3866 return 3867 sys.exc_clear() 3868 try: 3869 theMessage = None 3870 msgSession = None 3871 theMessage = srcTransport.Read() 3872 if srcTransport.currentReaders < srcTransport.desiredReaders: 3873 srcTransport.currentReaders += 1 3874 uthread.worker('machoNet::TransportReader', self.TransportReader, srcTransport) 3875 if theMessage.command == cluster.MACHONETMSG_TYPE_SESSIONCHANGENOTIFICATION and macho.mode == 'proxy': 3876 destTransports = self._GetTransports(theMessage.destination, srcTransport) 3877 if len(destTransports) == 1: 3878 destTransport = destTransports[0] 3879 if srcTransport.transportName == 'tcp:packet:machoNet': 3880 if srcTransport.transportID in destTransport.dependants: 3881 destTransport.dependants[srcTransport.transportID] += 1 3882 theMessage.destination = theMessage.source 3883 theMessage.source = MachoAddress(clientID=destTransport.clientID) 3884 msgSession = destTransport._SessionAndChannelAndIDFromPacket(theMessage)[0] 3885 theMessage.sessionVersion = msgSession.receivedVersion 3886 msgSession.receivedVersion += 1 3887 self.LogInfo('Proxy incremented receivedVersion of session(', msgSession.sid, ') to ', msgSession.receivedVersion) 3888 try: 3889 self.HandleMessage(destTransport, theMessage, 1) 3890 except Exception: 3891 self.LogError('MAJOR BORKUPP!!! This is baaaaddddddd') 3892 log.LogTraceback() 3893 msgSession.LogSessionError('Session state change bombed') 3894 raise 3895 3896 msgSession.version += 1 3897 self.LogInfo('Proxy incrementing actual session(', msgSession.sid, ")'s version number to ", msgSession.version) 3898 else: 3899 self.LogError('Funky Session Change packet received, ', theMessage) 3900 elif len(destTransports): 3901 self.LogError('Received a session change notification destined for ', len(destTransports), ' transports. packet=', theMessage) 3902 else: 3903 self.LogWarn('Ignoring non-routable session change notification. Hopefully a disconnect event? packet=', theMessage) 3904 elif theMessage.command % 2: 3905 self.HandleMessage(srcTransport, theMessage) 3906 else: 3907 msgSession = srcTransport._SessionAndChannelAndIDFromPacket(theMessage)[0] 3908 theMessage.sessionVersion = msgSession.receivedVersion 3909 if theMessage.command in self.__sessioninitorchangenotification__: 3910 msgSession.receivedVersion += 1 3911 self.LogInfo('Node incremented receivedVersion of session(', msgSession.sid, ') to ', msgSession.receivedVersion) 3912 self.HandleMessage(srcTransport, theMessage) 3913 msgSession.version += 1 3914 self.LogInfo('Node incrementing actual session(', msgSession.sid, ")'s version number to ", msgSession.version) 3915 else: 3916 self.HandleMessage(srcTransport, theMessage) 3917 except SessionUnavailable as e: 3918 self.LogError('Message abandoned in TransportReader due to unavailable user session') 3919 srcTransport.Write(theMessage.ErrorResponse(cluster.MACHONETERR_WRAPPEDEXCEPTION, (DumpsSanitized(RuntimeError(e.payload)),))) 3920 sys.exc_clear() 3921 except UnMachoDestination as e: 3922 if theMessage.destination.addressType != const.ADDRESS_TYPE_CLIENT or theMessage.command == cluster.MACHONETMSG_TYPE_CALL_REQ: 3923 srcTransport.Write(theMessage.ErrorResponse(cluster.MACHONETERR_UNMACHODESTINATION, e.payload)) 3924 else: 3925 self.LogWarn('Message abandoned in TransportReader due to unroutable destination address, reason=', e.payload) 3926 self.LogWarn('Packet=', theMessage) 3927 sys.exc_clear() 3928 except GPSTransportClosed: 3929 raise 3930 3931 except GPSTransportClosed as e: 3932 srcTransport.Close(**e.GetCloseArgs()) 3933 sys.exc_clear() 3934 except TaskletExit: 3935 self.LogInfo('TransportReader is shutting down - tasklet exit command received.') 3936 sys.exc_clear() 3937 except Exception: 3938 log.LogException() 3939 srcTransport.Close('TransportReader is shutting down - an unhandled exception has occurred.', 'UNHANDLEDEXCEPTION') 3940 sys.exc_clear() 3941 except: 3942 log.LogException() 3943 srcTransport.Close('TransportReader shutting down - a bizzarre low level exception has occurred.', 'UNHANDLEDEXCEPTION') 3944 sys.exc_clear() 3945 3946 def IsMessageSane(self, srcTransport, theMessage): 3947 if srcTransport.transportName == 'tcp:packet:client': 3948 if theMessage.destination.addressType in self.__borc__: 3949 srcTransport.Close('Client is trying to send C2C') 3950 return False 3951 if theMessage.command not in self.__clientallowedcommands__: 3952 srcTransport.Close('Client is trying to send disallowed commands') 3953 return False 3954 if getattr(srcTransport, 'userID', None) is None: 3955 srcTransport.Close('Client is trying to send over unauthenticated wire, message=%s' % strx(theMessage)) 3956 return False 3957 if hasattr(srcTransport, 'userID') and srcTransport.userID != theMessage.userID: 3958 srcTransport.Close('User is trying to spoof another user') 3959 return False 3960 elif macho.mode == 'proxy' or macho.mode == 'server': 3961 if srcTransport.nodeID is None and theMessage.command != cluster.MACHONETMSG_TYPE_IDENTIFICATION_REQ: 3962 self.LogError('Unidentified or fubarred wire?') 3963 log.LogTraceback() 3964 self.LogError('Extra info, tn=', srcTransport.transportName, ', t.node=', srcTransport.nodeID, ', transport=', srcTransport) 3965 self.LogError('Msg=', theMessage) 3966 srcTransport.Close('Server or Proxy is trying to send over unidentified wire, message=%s' % strx(theMessage)) 3967 return False 3968 if macho.mode == 'proxy': 3969 if srcTransport.nodeID is not None and theMessage.source.addressType == const.ADDRESS_TYPE_NODE and theMessage.source.nodeID != srcTransport.nodeID: 3970 if self.ShouldAcceptForwardedMessage(srcTransport, theMessage): 3971 pass 3972 elif theMessage.command != cluster.MACHONETMSG_TYPE_RESOLVE_RSP: 3973 raise RuntimeError('Server is trying to pull a fast one. source=%s, transport=%s' % (theMessage.source.nodeID, srcTransport.nodeID)) 3974 return True 3975 3976 __borc__ = (const.ADDRESS_TYPE_BROADCAST, const.ADDRESS_TYPE_CLIENT) 3977 __aorb__ = (const.ADDRESS_TYPE_ANY, const.ADDRESS_TYPE_BROADCAST) 3978 3979 @bluepy.TimedFunction('machoNet::HandleMessage') 3980 def HandleMessage(self, srcTransport, theMessage, skipSanityCheck = 0): 3981 with IngressTracer(theMessage): 3982 if theMessage.command in self.__pingrequestorresponse__: 3983 theMessage.times.append((blue.os.GetWallclockTime(), blue.os.GetWallclockTimeNow(), macho.mode + '::handle_message')) 3984 theMessage.Changed() 3985 try: 3986 if srcTransport.transportID not in self.transportsByID: 3987 self.LogWarn("Packet received on a transport that wasn't in transportsByID. transport=", srcTransport) 3988 self.LogWarn('Packet=', theMessage) 3989 srcTransport.Close("Packet received on a transport that wasn't in transportsByID") 3990 return 3991 if self.stop: 3992 srcTransport.Close('The service is stopping') 3993 return 3994 if not skipSanityCheck: 3995 if not self.IsMessageSane(srcTransport, theMessage): 3996 return 3997 if self.callCounterEnabled: 3998 try: 3999 self.callCounter[theMessage.source.addressType, theMessage.destination.addressType] += 1 4000 except Exception as e: 4001 self.LogError('Error adding to callCounter', e) 4002 sys.exc_clear() 4003 4004 if theMessage.source.addressType == const.ADDRESS_TYPE_CLIENT and theMessage.source.clientID not in self.transportIDbyClientID: 4005 self.transportIDbyClientID[theMessage.source.clientID] = srcTransport.transportID 4006 blue.net.AddClient(theMessage.source.clientID, srcTransport.transportID) 4007 if theMessage.command % 2: 4008 if theMessage.destination.addressType in self.__aorb__: 4009 srcTransport.Close('A response must be sent to a specific address, not a forking one') 4010 elif theMessage.source.addressType in self.__aorb__: 4011 srcTransport.Close('A request must be sent from a specific address, not a forking one') 4012 forward = 0 4013 routesTo = theMessage.RoutesTo(MachoAddress(nodeID=self.nodeID)) 4014 if macho.mode == 'client' or routesTo: 4015 if theMessage.command in self.__notificationtypes__: 4016 if macho.mode == 'proxy': 4017 if theMessage.command == cluster.MACHONETMSG_TYPE_SESSIONCHANGENOTIFICATION: 4018 try: 4019 srcTransport.SessionNotification(theMessage) 4020 except UnMachoDestination as e: 4021 self.LogInfo('UnMachoDestination encountered while forwarding session change notification: ', e) 4022 sys.exc_clear() 4023 4024 elif theMessage.destination.addressType == const.ADDRESS_TYPE_NODE and theMessage.destination.nodeID == self.nodeID: 4025 try: 4026 srcTransport.SessionNotification(theMessage) 4027 except UnMachoDestination as e: 4028 self.LogInfo('UnMachoDestination encountered while handling a proxy notification: ', e) 4029 sys.exc_clear() 4030 4031 elif theMessage.destination.addressType == const.ADDRESS_TYPE_BROADCAST and theMessage.destination.idtype == '+nodeID': 4032 try: 4033 srcTransport.SessionNotification(theMessage) 4034 except UnMachoDestination as e: 4035 self.LogInfo('UnMachoDestination encountered while handling a proxy broadcast: ', e) 4036 sys.exc_clear() 4037 4038 else: 4039 if routesTo == 2: 4040 theMessage2 = copy.deepcopy(theMessage) 4041 srcTransport.SessionNotification(theMessage2) 4042 forward = 1 4043 else: 4044 srcTransport.SessionNotification(theMessage) 4045 elif theMessage.command in self.__calltypes__: 4046 if macho.mode == 'proxy' and not (theMessage.destination.addressType == const.ADDRESS_TYPE_NODE and theMessage.destination.nodeID == self.nodeID): 4047 forward = 1 4048 else: 4049 with IngressTracer.record_process_call_ms(): 4050 ret = srcTransport.SessionCall(theMessage) 4051 if srcTransport.nodeID in self.transportIDbyProxyNodeID: 4052 IngressTracer.record_ingress_proxy_id(srcTransport.nodeID) 4053 with IngressTracer.record_write_response_ms(): 4054 srcTransport.Write(ret) 4055 elif theMessage.command in self.__responsetypes__ or theMessage.command == cluster.MACHONETMSG_TYPE_PING_RSP and (macho.mode != 'proxy' or theMessage.destination.addressType == const.ADDRESS_TYPE_NODE and theMessage.destination.nodeID == self.nodeID): 4056 if theMessage.destination.callID in self.calls: 4057 if self.calls[theMessage.destination.callID][0].balance < 0: 4058 self.calls[theMessage.destination.callID][0].send(theMessage) 4059 else: 4060 self.LogWarn("Call Response received, but the call wasn't being waited upon in self.calls, so the response will disappear... ", theMessage) 4061 elif theMessage.command == cluster.MACHONETMSG_TYPE_ERRORRESPONSE and theMessage.originalCommand == cluster.MACHONETMSG_TYPE_NOTIFICATION: 4062 self.LogWarn("An error response was received for a notification from us. Notifcations don't have responses, so this message will disappear... ", theMessage) 4063 else: 4064 self.LogError('Call Response received on', macho.mode, ", but the call wasn't in self.calls, so the response will disappear... ", theMessage) 4065 elif theMessage.command == cluster.MACHONETMSG_TYPE_TRANSPORTCLOSED: 4066 srcTransport.SessionClosed(theMessage.clientID, 'Session terminated due to remote transport closed notification', theMessage.isRemote) 4067 elif theMessage.command in self.__pingrequestorresponse__: 4068 if macho.mode == 'proxy' and not (theMessage.destination.addressType == const.ADDRESS_TYPE_NODE and theMessage.destination.nodeID == self.nodeID): 4069 forward = 1 4070 elif theMessage.command == cluster.MACHONETMSG_TYPE_PING_REQ: 4071 theMessage.times.append((blue.os.GetWallclockTime(), blue.os.GetWallclockTimeNow(), macho.mode + '::turnaround')) 4072 theResponse = theMessage.Response(times=theMessage.times[:]) 4073 srcTransport.Write(theResponse) 4074 elif theMessage.command == cluster.MACHONETMSG_TYPE_IDENTIFICATION_REQ: 4075 if self.nodeID is None: 4076 raise RuntimeError('no node ID') 4077 if macho.mode == 'server': 4078 myaddress = self.GetTransport('tcp:packet:machoNet').GetExternalAddress() 4079 others = [ ('%s:%d' % (each.ipAddress, each.port), each.nodeID) for each in self._GetLayTracksTo() ] 4080 else: 4081 myaddress = self.GetTransport('tcp:packet:machoNet').GetInternalAddress() 4082 others = [] 4083 if theMessage.nodeID in self.transportIDbyProxyNodeID or theMessage.nodeID in self.transportIDbySolNodeID: 4084 self.LogWarn('Rejecting connection attempt from ', theMessage.nodeID, ' because I already have such a connection') 4085 response = (False, 'I already have a socket for this node') 4086 elif theMessage.nodeID in self.dontLayTracksTo and theMessage.nodeID > self.nodeID: 4087 self.LogInfo('Rejecting connection attempt from ', theMessage.nodeID, ' because I was already establishing such a connection, and I have priority', self.nodeID, 'vs.', theMessage.nodeID) 4088 response = (False, 'I am already connecting to this node') 4089 else: 4090 self.LogInfo('Accepting connection from ', theMessage.nodeID) 4091 response = (True, 'OK') 4092 if response[0]: 4093 srcTransport.Finalize(theMessage.nodeID) 4094 blue.net.SetTransportNodeID(srcTransport.transportID, theMessage.nodeID) 4095 if theMessage.isProxy: 4096 thedict = self.transportIDbyProxyNodeID 4097 elif theMessage.isApp: 4098 thedict = self.transportIDbyAppNodeID 4099 else: 4100 thedict = self.transportIDbySolNodeID 4101 thedict[theMessage.nodeID] = srcTransport.transportID 4102 try: 4103 isApp = self.nodeID in self.transportIDbyAppNodeID 4104 srcTransport.Write(theMessage.Response(response, self.nodeID, others, macho.mode == 'proxy', isApp, self.serviceMask)) 4105 except: 4106 del thedict[theMessage.nodeID] 4107 raise 4108 4109 self.externalAddressesByNodeID[theMessage.nodeID] = (theMessage.myaddress, theMessage.serviceMask) 4110 if self.clusterStartupPhase: 4111 if macho.mode == 'proxy': 4112 polarisID = self.session.ConnectToSolServerService('machoNet').GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) 4113 else: 4114 polarisID = self.GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) 4115 isPolaris = theMessage.nodeID == polarisID 4116 sm.ScatterEvent('OnNewNode', theMessage.nodeID, theMessage.myaddress, theMessage.isProxy, isPolaris, theMessage.serviceMask) 4117 else: 4118 isPolaris = macho.mode == 'server' and self.GetNodeID() == self.GetNodeFromAddress(cluster.SERVICE_POLARIS, 0) 4119 isApp = self.nodeID in self.transportIDbyAppNodeID 4120 srcTransport.Write(theMessage.Response(response, self.nodeID, others, macho.mode == 'proxy', isApp, self.serviceMask)) 4121 srcTransport.Close(response[1], noSend=True) 4122 others = list(theMessage.others) 4123 random.shuffle(others) 4124 for otherAddress, otherNodeID in others: 4125 self.LayTracksIfNeeded(otherAddress, otherNodeID, 'IdentificationReq.others') 4126 4127 else: 4128 srcTransport.Close('Command %d is not allowed at this node' % theMessage.command) 4129 else: 4130 forward = 1 4131 if forward: 4132 if macho.mode != 'proxy': 4133 self.LogError('TransportReader is rejecting the message violently') 4134 srcTransport.Close("Client and server nodes do not forward packets. Something's wrong, dude.") 4135 return 4136 try: 4137 destTransports = self._GetTransports(theMessage.destination, srcTransport) 4138 if theMessage.destination.addressType == const.ADDRESS_TYPE_BROADCAST and macho.mode == 'proxy': 4139 theMessage.destination.narrowcast = [] 4140 theMessage.Changed() 4141 for each in destTransports: 4142 self.FragileWrite(each, theMessage) 4143 4144 except UnMachoDestination as e: 4145 if theMessage.destination.addressType != const.ADDRESS_TYPE_CLIENT or theMessage.command == cluster.MACHONETMSG_TYPE_CALL_REQ: 4146 srcTransport.Write(theMessage.ErrorResponse(cluster.MACHONETERR_UNMACHODESTINATION, e.payload)) 4147 else: 4148 self.LogWarn('Message abandoned due to unroutable destination address, reason=', e.payload) 4149 self.LogWarn('Packet=', theMessage) 4150 sys.exc_clear() 4151 4152 except GPSTransportClosed as e: 4153 srcTransport.Close('Connected transport closed') 4154 sys.exc_clear() 4155 return 4156 except Exception: 4157 log.LogException('TransportReader standard error') 4158 if srcTransport.transportName == 'tcp:packet:client': 4159 srcTransport.Close('TransportReader blew up', 'UNHANDLEDEXCEPTION') 4160 else: 4161 self.LogError('Playing chicken with a freight train, eh? Sorry dude, but the show must go on...') 4162 sys.exc_clear() 4163 except: 4164 log.LogException('TransportReader non-standard error') 4165 srcTransport.Close('TransportReader blew up - non-standard error', 'UNHANDLEDEXCEPTION') 4166 raise 4167 4168 def ShouldAcceptForwardedMessage(self, transport, theMessage): 4169 if transport.nodeID in self.transportIDbyAppNodeID: 4170 return True 4171 return False 4172 4173 @bluepy.TimedFunction('machoNet::FragileWrite') 4174 def FragileWrite(self, destTransport, message): 4175 try: 4176 if message.command in self.__pingrequestorresponse__: 4177 message.times.append((blue.os.GetWallclockTime(), blue.os.GetWallclockTimeNow(), macho.mode + '::writing')) 4178 message.Changed() 4179 destTransport.Write(message) 4180 except GPSTransportClosed as e: 4181 destTransport.Close('Wrote packet on a closed socket. Close reason was: ' + strx(e.reason)) 4182 sys.exc_clear() 4183 except Exception: 4184 self.LogWarn("Write died of natural or bizzarre causes. Could be natural. Could be bizzarre. Here's a trace:") 4185 log.LogException() 4186 destTransport.Close('Write died of natural or bizzarre causes.') 4187 sys.exc_clear() 4188 except: 4189 destTransport.Close('Write died of natural or bizzarre causes, probably shutdown.') 4190 raise 4191 4192 def GetProxyNodeIDFromClientID(self, clientID): 4193 try: 4194 proxyNodeID = clientID % 10000000000L 4195 except TypeError as e: 4196 raise UnMachoDestination('%s is not a valid clientID' % repr(clientID)) 4197 4198 if macho.mode == 'server' and proxyNodeID not in self.transportIDbyProxyNodeID: 4199 raise UnMachoDestination('The proxy node in question is not reachable') 4200 return proxyNodeID 4201 4202 def GetIDOfAddress(self, address, clientMode = True, customIDData = None): 4203 ipaddr, port = address.split(':') 4204 ipaddr = GetPreferredHostByName(ipaddr) 4205 x = ipaddr.split('.') 4206 if clientMode: 4207 t = self.clientIDOffset 4208 self.clientIDOffset += 1 4209 return t * 10000000000L + self.nodeID 4210 elif customIDData: 4211 basePort, portOffset, idRangeBase = customIDData 4212 hops = (int(port) - basePort - portOffset) / 1000 4213 if hops < 0: 4214 log.LogTraceback('hops (orchestrator agent) fuxx0red') 4215 self.LogError('How is this even related to orchestrator?') 4216 self.LogError('hops: ', hops) 4217 self.LogError('port: ', port) 4218 self.LogError('om: ', offsetMap[macho.mode]['tcp:packet:machoNet']) 4219 return idRangeBase + int(x[3]) + hops * 1000 4220 else: 4221 hops = (int(port) - self.defaultProxyPortOffset - offsetMap[macho.mode]['tcp:packet:client']) / 1000 4222 if hops < 0: 4223 log.LogTraceback('hops fuxx0red') 4224 self.LogError('hops: ', hops) 4225 self.LogError('port: ', port) 4226 self.LogError('dpopo: ', self.defaultProxyPortOffset) 4227 self.LogError('om: ', offsetMap[macho.mode]['tcp:packet:client']) 4228 return cluster.PROXY_MIN_NODE_ID + int(x[3]) + hops * cluster.PROXY_PER_HOPS_INCREASE 4229 4230 def SetStatusKeyValuePair(self, key, value): 4231 self.statusData[key] = value 4232 4233 def SetStatusCallBack(self, key, function): 4234 self.statusCallbacks[key] = function 4235 4236 def StatusSocketLoop(self): 4237 try: 4238 self.LogInfo('Starting status socket') 4239 statusSocket = socket.socket() 4240 statusSocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 4241 port = self.basePortNumber + offsetMap[macho.mode]['tcp:raw:status'] 4242 statusSocket.bind(('', port)) 4243 statusSocket.listen(socket.SOMAXCONN) 4244 while True: 4245 try: 4246 self._SendStatus(statusSocket) 4247 except: 4248 log.LogException('Status socket loop threw an exception') 4249 4250 except: 4251 log.LogException('Status socket threw and exception, and is now stopping') 4252 4253 def _SendStatus(self, statusSocket): 4254 s, address = statusSocket.accept() 4255 try: 4256 self.LogInfo('Sending server status to %s', address) 4257 data = {} 4258 data.update(self.statusData) 4259 stackless.current.block_trap = True 4260 for key, function in self.statusCallbacks.iteritems(): 4261 try: 4262 data[key] = function() 4263 except: 4264 log.LogException('Callback function for key %s failed', key) 4265 4266 stackless.current.block_trap = False 4267 s.send('%s %s\r\n\r\n%s' % ('HTTP/1.0', '200', json.dumps(data))) 4268 finally: 4269 s.close() 4270 4271 def ForwardProxyBroadcast(self, *args): 4272 self.ProxyBroadcast(*args) 4273 4274 def ForwardSinglecastByCharID(self, *args): 4275 self.SinglecastByCharID(*args) 4276 4277 def GetRuntimeDetails(self): 4278 proxy_macho_net = self.session.ConnectToProxyServerService('machoNet') 4279 connectionProps = proxy_macho_net.GetConnectionProperties() 4280 vip_mode = connectionProps['vipMode'] 4281 online_players = self.onlineData.get_logon_count() 4282 build_number = boot.build 4283 return {'players': online_players, 4284 'vip': vip_mode, 4285 'build': build_number, 4286 'start_time': blue.sysinfo.processStartTime} 4287 4288 def GetConnectedClusterName(self): 4289 if monolithconfig.on_client(): 4290 proxy_macho_net = self.session.ConnectToProxyServerService('machoNet') 4291 return proxy_macho_net.GetConnectedClusterName() 4292 return prefs.clusterName 4293 4294 def GetConnectedClusterMode(self): 4295 if monolithconfig.on_client(): 4296 proxy_macho_net = self.session.ConnectToProxyServerService('machoNet') 4297 return proxy_macho_net.GetConnectedClusterMode() 4298 return prefs.clusterMode 4299 4300 def _log_pickle_error(self, message, the_pickle, tags = None): 4301 if tags is None: 4302 tags = {} 4303 tags.update(the_pickle.extract_tags()) 4304 self.logger.error(message, extra=tags) 4305 4306 def _configure_metrics_flag(self, evaluate_flag): 4307 self.metricsEnabled = evaluate_flag 4308 4309 4310 def WhitelistDumper(): 4311 while True: 4312 f = file('wl.txt', 'wt') 4313 vals = blue.marshal.globalsWhitelist.values() 4314 vals.sort() 4315 for v in vals: 4316 print(v[0] + '.' + v[1], file=f) 4317 4318 f.close() 4319 blue.pyos.synchro.SleepWallclock(5000) 4320 4321 4322 def CollectWhitelist(): 4323 blue.marshal.globalsWhitelist = {} 4324 blue.marshal.collectWhitelist = True 4325 uthread.new(WhitelistDumper) 4326 4327 4328 def RegisterPortOffset(mapname, portOffset, modeRoleBootMacho = None): 4329 if modeRoleBootMacho is None: 4330 modeRoleBootMacho = boot.role 4331 offsetMap[modeRoleBootMacho][mapname] = portOffset 4332 4333 4334 def GetPreferredHostByName(name): 4335 addressList = socket.gethostbyname_ex(name)[2] 4336 lookingFor = PREFERRED_SUBNETWORKS.get(macho.mode, DEFAULT_SUBNETWORK) 4337 for address in addressList: 4338 if address.startswith(lookingFor): 4339 return address 4340 4341 if addressList: 4342 return addressList[0] 4343 4344 4345 def CreateInstance(cls, arguments = ()): 4346 instance = cls(*arguments) 4347 for each in getattr(instance, '__persistvars__', []): 4348 if not hasattr(instance, each): 4349 setattr(instance, each, None) 4350 4351 for each in getattr(instance, '__nonpersistvars__', []): 4352 if not hasattr(instance, each): 4353 setattr(instance, each, None) 4354 4355 return instance 4356 4357 4358 mode = boot.role