U /e@sddlmZddlZddlZddlZddlZddlmZddlZddl m Z ddl Z ddl m Z mZddlmZddlmZddlmZmZdd lmZGd d d eZdS) ) annotationsN)suppress) HTTPServer)get_address_hostget_tcp_server_addresses)Server)RoutingApplication) DequeHandlerclean_dashboard_address get_versionsc@sZeZdZdZdddZddZddZed d Zd d d ddZ dddZ dddZ dS) ServerNodez? Base class for server nodes in a distributed cluster. NcCs t|dS)N)packagesr )selfrr4/tmp/pip-unpacked-wheel-g426oqom/distributed/node.pyversionsszServerNode.versionsc CsN|dkr d}|jD]0\}}d}t|tr8|\}}nd}t|trP|d}t|ttfrt|dkr|dt|d}}n"t|dkr|d\}}nt |t|tr|\}}ni}z>||fd|j i|}| |dk r|n||f||j |<Wqt k rF}z,tjd|d |d d t|d d W5d}~XYqXqdS)Nz0.0.0.0r:Zio_loopz Could not launch service 'z ' on port z. zGot the following message: ) stacklevel)Z service_specsitems isinstancetuplestrsplitlistlenint ValueErrorZlooplistenservices Exceptionwarningswarn) rZdefault_listen_ipkvZ listen_ipportkwargsserviceerrrstart_services"sB        zServerNode.start_servicescCsPt|dr4|jjD] }t|drt|jr|q|jD] }|q>dS)Nhttp_applicationstop)hasattrr.Z applicationscallabler/r#values)rZ applicationr+rrr stop_servicesKs    zServerNode.stop_servicescCsdd|jDS)NcSsi|]\}}||jqSrr)).0r'r(rrr Usz,ServerNode.service_ports..)r#r)rrrr service_portsSszServerNode.service_portszlogging.LoggerNone)loggerreturncCsPttjdd|_|jttjd||jt ||j |jdS)Nzdistributed.admin.log-length)nzdistributed.admin.log-format) r daskconfigget_deque_handler setFormatterlogging Formatter addHandlerweakreffinalize removeHandler)rr9rrr_setup_loggingWs  zServerNode._setup_loggingrFcCsr|j}g}t|jD]X\}}|r(||ks2|j|kr6qn|rV||j|j||fq||j||fq|S)a! Fetch log entries for this node Parameters ---------- start : float, optional A time (in seconds) to begin filtering log entries from n : int, optional Maximum number of log entries to return from filtered results timestamps : bool, default False Do we want log entries to include the time they were generated? Returns ------- List of tuples containing the log level, message, and (optional) timestamp for each filtered entry )r? enumeratedequecreatedappend levelnameformat)rstartr;Z timestampsZ deque_handlerLcountmsgrrrget_logsaszServerNode.get_logsc Cst||_tjd}tjd}tjd}|rRtj|tjjd}|j ||dt |j|d|_ t |pj|}|D]} | ddkr|j } t| ttfr| d } | rttt| | d<W5QRXd } d } z6| s|j jf| n|j jft| d d iWqrWqtk r.d } | d} | dkr*YqXqqrt|j } | d d|j _|j |jd<tdd|Ddd| DD]6\}}||krx|d krxtd|d|dqxdS)z0This creates an HTTP Server running on this nodez'distributed.scheduler.dashboard.tls.keyz(distributed.scheduler.dashboard.tls.certz+distributed.scheduler.dashboard.tls.ca-file)cafilepurpose)keyfile) ssl_optionsaddressNrFrr)TrZ dashboardcSsg|] }|dqSr4r)r5arrr sz0ServerNode.start_http_server..cSsg|] }|dqS)rr)r5brrrrYszPort z` is already in use. Perhaps you already have a cluster running? Hosting the HTTP server on port z instead)rr.r<r=r>sslcreate_default_contextPurpose CLIENT_AUTHload_cert_chainrZ http_serverr Z_start_addressrrrrr!rr"tlzmerger$rr)r#zipr%r&)rZroutesZdashboard_address default_portrVZtls_keyZtls_certZ tls_ca_fileZhttp_addressesZ http_addressrWZ change_portZ retries_leftZbound_addressesexpectedactualrrrstart_http_server~sV            zServerNode.start_http_server)N)rNF)rN) __name__ __module__ __qualname____doc__rr-r3propertyr7rGrRrfrrrrr s )  r ) __future__rrAr[r%rD contextlibrr`Ztornado.httpserverrr<Zdistributed.commrrZdistributed.corerZdistributed.http.routingrZdistributed.utilsr r Zdistributed.versionsr r rrrrs