<s> """<STR_LIT>""" <EOL> import docker <EOL> import functools <EOL> import io <EOL> import json <EOL> import logging <EOL> import mesos . interface <EOL> import os <EOL> import pesos . executor <EOL> import re <EOL> import signal <EOL> import subprocess <EOL> import threading <EOL> import time <EOL> import traceback <EOL> import tempfile <EOL> import tarfile <EOL> from pesos . vendor . mesos import mesos_pb2 <EOL> from portainer . app import subcommand <EOL> from portainer . proto import portainer_pb2 <EOL> from portainer . util . squash import get_squash_layers , download_layers_for_image , extract_layer_tar , apply_layer , generate_tarball , rewrite_image_parent <EOL> logger = logging . getLogger ( "<STR_LIT>" ) <EOL> @ subcommand ( "<STR_LIT>" ) <EOL> def main ( args ) : <EOL> driver = pesos . executor . PesosExecutorDriver ( Executor ( ) ) <EOL> thread = threading . Thread ( target = driver . run ) <EOL> thread . setDaemon ( True ) <EOL> thread . start ( ) <EOL> while thread . isAlive ( ) : <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> class Executor ( mesos . interface . Executor ) : <EOL> TASK_STARTING = mesos_pb2 . TASK_STARTING <EOL> TASK_RUNNING = mesos_pb2 . TASK_RUNNING <EOL> TASK_FINISHED = mesos_pb2 . TASK_FINISHED <EOL> TASK_FAILED = mesos_pb2 . TASK_FAILED <EOL> def __init__ ( self ) : <EOL> self . build_task = None <EOL> self . docker = None <EOL> self . docker_daemon_up = False <EOL> def registered ( self , driver , executorInfo , frameworkInfo , slaveInfo ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> try : <EOL> build_task = portainer_pb2 . BuildTask ( ) <EOL> build_task . ParseFromString ( executorInfo . data ) <EOL> except Exception : <EOL> logger . error ( "<STR_LIT>" ) <EOL> raise <EOL> self . build_task = build_task <EOL> def launch_docker_daemon ( ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> env = dict ( os . environ ) <EOL> env [ "<STR_LIT>" ] = "<STR_LIT>" % ( <EOL> os . path . join ( os . environ . get ( "<STR_LIT>" , os . environ [ "<STR_LIT>" ] ) , "<STR_LIT>" ) <EOL> ) <EOL> for reg in build_task . daemon . insecure_registries : <EOL> env [ "<STR_LIT>" ] += "<STR_LIT>" % reg <EOL> proc = subprocess . Popen ( [ "<STR_LIT>" ] , env = env ) <EOL> self . docker = docker . Client ( ) <EOL> while True : <EOL> try : <EOL> self . docker . ping ( ) <EOL> except : <EOL> logger . info ( "<STR_LIT>" ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> else : <EOL> self . docker_daemon_up = True <EOL> break <EOL> proc . wait ( ) <EOL> if not build_task . daemon . HasField ( "<STR_LIT>" ) : <EOL> daemon_thread = threading . Thread ( target = launch_docker_daemon ) <EOL> daemon_thread . setDaemon ( True ) <EOL> daemon_thread . start ( ) <EOL> else : <EOL> self . docker = docker . Client ( build_task . daemon . docker_host ) <EOL> self . docker_daemon_up = True <EOL> def disconnected ( self , driver ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> def reregistered ( self , driver , slaveInfo ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> def launchTask ( self , driver , taskInfo ) : <EOL> logger . info ( "<STR_LIT>" , taskInfo . task_id . value ) <EOL> thread = threading . Thread ( target = functools . partial ( <EOL> self . _build_image , <EOL> driver , <EOL> taskInfo , <EOL> self . build_task <EOL> ) ) <EOL> thread . setDaemon ( True ) <EOL> thread . start ( ) <EOL> def shutdown ( self , driver ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> if os . path . exists ( "<STR_LIT>" ) : <EOL> try : <EOL> docker_pid = int ( open ( "<STR_LIT>" , "<STR_LIT:r>" ) . read ( ) ) <EOL> os . kill ( docker_pid , signal . SIGTERM ) <EOL> except Exception , e : <EOL> logger . error ( "<STR_LIT>" ) <EOL> logger . error ( e ) <EOL> else : <EOL> logger . warning ( "<STR_LIT>" ) <EOL> def _wrap_docker_stream ( self , stream ) : <EOL> """<STR_LIT>""" <EOL> for msg in stream : <EOL> logger . info ( "<STR_LIT>" , msg . rstrip ( ) ) <EOL> try : <EOL> update = json . loads ( msg . rstrip ( ) ) <EOL> except Exception , e : <EOL> logger . error ( "<STR_LIT>" , msg , e ) <EOL> else : <EOL> if "<STR_LIT:error>" in update : <EOL> logger . error ( "<STR_LIT>" , update [ "<STR_LIT:error>" ] ) <EOL> yield update [ "<STR_LIT:error>" ] , False <EOL> raise Exception ( "<STR_LIT>" ) <EOL> friendly_message = None <EOL> is_stream = False <EOL> if "<STR_LIT>" in update : <EOL> is_stream = True <EOL> friendly_message = re . sub ( r'<STR_LIT>' , '<STR_LIT>' , <EOL> update [ "<STR_LIT>" ] . rstrip ( ) ) <EOL> if "<STR_LIT:status>" in update : <EOL> friendly_message = update [ "<STR_LIT:status>" ] . rstrip ( ) <EOL> if "<STR_LIT:id>" in update : <EOL> friendly_message = "<STR_LIT>" % ( update [ "<STR_LIT:id>" ] , friendly_message ) <EOL> if "<STR_LIT>" in update : <EOL> friendly_message += "<STR_LIT>" % update [ "<STR_LIT>" ] <EOL> if friendly_message is not None : <EOL> yield friendly_message , is_stream <EOL> def _build_image ( self , driver , taskInfo , buildTask ) : <EOL> """<STR_LIT>""" <EOL> driver . sendStatusUpdate ( mesos_pb2 . TaskStatus ( <EOL> task_id = taskInfo . task_id , <EOL> state = self . TASK_STARTING <EOL> ) ) <EOL> logger . info ( "<STR_LIT>" ) <EOL> timeout = <NUM_LIT:30> <EOL> while timeout > <NUM_LIT:1> and not self . docker_daemon_up : <EOL> timeout -= <NUM_LIT:1> <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> try : <EOL> if not self . docker_daemon_up : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> driver . sendStatusUpdate ( mesos_pb2 . TaskStatus ( <EOL> task_id = taskInfo . task_id , <EOL> state = self . TASK_RUNNING <EOL> ) ) <EOL> if not buildTask : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> if not buildTask . context and not buildTask . dockerfile : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> registry_url = None <EOL> if buildTask . image . HasField ( "<STR_LIT>" ) : <EOL> registry_url = buildTask . image . registry . hostname <EOL> if buildTask . image . registry . HasField ( "<STR_LIT:port>" ) : <EOL> registry_url += "<STR_LIT>" % buildTask . image . registry . port <EOL> registry_url += "<STR_LIT:/>" <EOL> if not registry_url : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> image_name = registry_url + buildTask . image . repository <EOL> logger . info ( "<STR_LIT>" , image_name ) <EOL> base_image_id = None <EOL> sandbox_dir = os . environ . get ( "<STR_LIT>" , os . environ [ "<STR_LIT>" ] ) <EOL> if buildTask . dockerfile : <EOL> build_request = self . docker . build ( <EOL> fileobj = io . StringIO ( buildTask . dockerfile ) , <EOL> stream = True <EOL> ) <EOL> for message , is_stream in self . _wrap_docker_stream ( build_request ) : <EOL> if message . startswith ( u"<STR_LIT>" ) and not base_image_id : <EOL> base_image_id = message [ <NUM_LIT:6> : ] <EOL> if not is_stream or ( is_stream and buildTask . stream ) : <EOL> driver . sendFrameworkMessage ( <EOL> ( "<STR_LIT>" % ( image_name , message ) ) . encode ( '<STR_LIT>' ) <EOL> ) <EOL> else : <EOL> context_path = os . path . join ( sandbox_dir , buildTask . context ) <EOL> if not os . path . exists ( context_path ) : <EOL> raise Exception ( "<STR_LIT>" % ( context_path ) ) <EOL> with open ( context_path , "<STR_LIT:r>" ) as context : <EOL> build_request = self . docker . build ( <EOL> fileobj = context , <EOL> custom_context = True , <EOL> encoding = "<STR_LIT>" , <EOL> stream = True <EOL> ) <EOL> for message , is_stream in self . _wrap_docker_stream ( build_request ) : <EOL> if message . startswith ( u"<STR_LIT>" ) and not base_image_id : <EOL> base_image_id = message [ <NUM_LIT:6> : ] <EOL> if not is_stream or ( is_stream and buildTask . stream ) : <EOL> driver . sendFrameworkMessage ( <EOL> ( "<STR_LIT>" % ( image_name , message ) ) . encode ( '<STR_LIT>' ) <EOL> ) <EOL> match = re . search ( r'<STR_LIT>' , message ) <EOL> if not match : <EOL> raise Exception ( "<STR_LIT>" % message ) <EOL> image_id = match . group ( <NUM_LIT:1> ) <EOL> if buildTask . HasField ( "<STR_LIT>" ) and buildTask . squash : <EOL> if not base_image_id : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> image_id = self . _squash_image ( <EOL> driver , <EOL> sandbox_dir , <EOL> image_name , <EOL> base_image_id , <EOL> image_id <EOL> ) <EOL> tags = buildTask . image . tag or [ "<STR_LIT>" ] <EOL> driver . sendFrameworkMessage ( str ( "<STR_LIT>" % ( image_name , image_id ) ) ) <EOL> for tag in tags : <EOL> try : <EOL> self . docker . tag ( <EOL> image = image_id , <EOL> repository = image_name , <EOL> tag = tag , <EOL> force = True <EOL> ) <EOL> driver . sendFrameworkMessage ( str ( "<STR_LIT>" % ( image_name , tag ) ) ) <EOL> except Exception , e : <EOL> raise e <EOL> driver . sendFrameworkMessage ( str ( "<STR_LIT>" % image_name ) ) <EOL> push_request = self . docker . push ( image_name , stream = True ) <EOL> for message , is_stream in self . _wrap_docker_stream ( push_request ) : <EOL> if not is_stream or ( is_stream and buildTask . stream ) : <EOL> driver . sendFrameworkMessage ( <EOL> str ( "<STR_LIT>" % ( image_name , message ) ) <EOL> ) <EOL> driver . sendStatusUpdate ( mesos_pb2 . TaskStatus ( <EOL> task_id = taskInfo . task_id , <EOL> state = self . TASK_FINISHED <EOL> ) ) <EOL> except Exception , e : <EOL> logger . error ( "<STR_LIT>" , e ) <EOL> driver . sendStatusUpdate ( mesos_pb2 . TaskStatus ( <EOL> task_id = taskInfo . task_id , <EOL> state = self . TASK_FAILED , <EOL> message = str ( e ) , <EOL> data = traceback . format_exc ( ) <EOL> ) ) <EOL> raise <EOL> def _squash_image ( self , driver , sandbox_dir , image_name , <EOL> base_image_id , head_image_id ) : <EOL> """<STR_LIT>""" <EOL> logger . info ( "<STR_LIT>" , base_image_id [ : <NUM_LIT:12> ] , head_image_id [ : <NUM_LIT:12> ] ) <EOL> driver . sendFrameworkMessage ( str ( "<STR_LIT>" % image_name ) ) <EOL> base_image_id , head_image_id , total_bytes , new_layers = get_squash_layers ( <EOL> self . docker , base_image_id , head_image_id <EOL> ) <EOL> total_mb = total_bytes / <NUM_LIT> / <NUM_LIT> <EOL> driver . sendFrameworkMessage ( str ( "<STR_LIT>" % ( image_name , len ( new_layers ) , total_mb ) ) ) <EOL> driver . sendFrameworkMessage ( str ( "<STR_LIT>" % image_name ) ) <EOL> layers_tar_fh = download_layers_for_image ( self . docker , sandbox_dir , head_image_id ) <EOL> layers_tar = tarfile . open ( fileobj = layers_tar_fh ) <EOL> working_dir = tempfile . mkdtemp ( dir = sandbox_dir ) <EOL> seen_paths = set ( ) <EOL> for layer_id in new_layers : <EOL> short_layer_id = layer_id [ : <NUM_LIT:12> ] <EOL> driver . sendFrameworkMessage ( str ( "<STR_LIT>" % ( image_name , short_layer_id ) ) ) <EOL> layer_tar_fh = extract_layer_tar ( sandbox_dir , layers_tar , layer_id ) <EOL> layer_tar = tarfile . open ( fileobj = layer_tar_fh ) <EOL> driver . sendFrameworkMessage ( str ( "<STR_LIT>" % ( image_name , short_layer_id ) ) ) <EOL> seen_paths = apply_layer ( working_dir , layer_tar , seen_paths ) <EOL> driver . sendFrameworkMessage ( str ( "<STR_LIT>" % image_name ) ) <EOL> new_layer_tarball_path = generate_tarball ( sandbox_dir , working_dir ) <EOL> driver . sendFrameworkMessage ( str ( "<STR_LIT>" % ( image_name , head_image_id [ : <NUM_LIT:12> ] , base_image_id [ : <NUM_LIT:12> ] ) ) ) <EOL> _ , new_image_tarball_path = tempfile . mkstemp ( dir = sandbox_dir , suffix = "<STR_LIT>" ) <EOL> new_image_tarball = tarfile . open ( new_image_tarball_path , "<STR_LIT>" ) <EOL> new_image_tarball . add ( new_layer_tarball_path , arcname = os . path . join ( head_image_id , "<STR_LIT>" ) ) <EOL> layers_tar . extract ( os . path . join ( head_image_id , "<STR_LIT>" ) , path = sandbox_dir ) <EOL> layers_tar . extract ( os . path . join ( head_image_id , "<STR_LIT>" ) , path = sandbox_dir ) <EOL> with open ( os . path . join ( sandbox_dir , head_image_id , "<STR_LIT>" ) , "<STR_LIT>" ) as json_info : <EOL> image_info = rewrite_image_parent ( json . load ( json_info ) , base_image_id ) <EOL> json_info . seek ( <NUM_LIT:0> ) <EOL> json . dump ( image_info , json_info ) <EOL> json_info . truncate ( ) <EOL> new_image_tarball . add ( os . path . join ( sandbox_dir , head_image_id , "<STR_LIT>" ) , arcname = os . path . join ( head_image_id , "<STR_LIT>" ) ) <EOL> new_image_tarball . add ( os . path . join ( sandbox_dir , head_image_id , "<STR_LIT>" ) , arcname = os . path . join ( head_image_id , "<STR_LIT>" ) ) <EOL> new_image_tarball . close ( ) <EOL> driver . sendFrameworkMessage ( str ( "<STR_LIT>" % image_name ) ) <EOL> with open ( new_image_tarball_path , "<STR_LIT:r>" ) as fh : <EOL> self . docker . remove_image ( head_image_id , force = True ) <EOL> self . docker . load_image ( fh ) <EOL> return head_image_id </s>
