signature:
def general_params ( cls ) :
    """Define all the possible config params"""

implementation:
params = { }
# GENERAL CONFIG
params_general = { "general" : { "min_update_delay" : { "optional" : True , "default" : 60 , "type" : int , "description" : "Short delay between tasks (collect, enrich ...)" } , "update" : { "optional" : False , "default" : False , "type" : bool , "description" : "Execute the tasks in loop" } , "short_name" : { "optional" : False , "default" : "Short name" , "type" : str , "description" : "Short name of the project" } , "debug" : { "optional" : False , "default" : True , "type" : bool , "description" : "Debug mode (logging mainly)" } , "logs_dir" : { "optional" : False , "default" : "logs" , "type" : str , "description" : "Directory with the logs of sirmordred" } , "log_handler" : { "optional" : True , "default" : "file" , "type" : str , "description" : "use rotate for rotating the logs automatically" } , "log_max_bytes" : { "optional" : True , "default" : 104857600 , # 100MB
"type" : int , "description" : "Max number of bytes per log file" } , "log_backup_count" : { "optional" : True , "default" : 5 , "type" : int , "description" : "Number of rotate logs files to preserve" } , "bulk_size" : { "optional" : True , "default" : 1000 , "type" : int , "description" : "Number of items to write in Elasticsearch using bulk operations" } , "scroll_size" : { "optional" : True , "default" : 100 , "type" : int , "description" : "Number of items to read from Elasticsearch when scrolling" } , "aliases_file" : { "optional" : True , "default" : ALIASES_JSON , "type" : str , "description" : "JSON file to define aliases for raw and enriched indexes" } , "menu_file" : { "optional" : True , "default" : MENU_YAML , "type" : str , "description" : "YAML file to define the menus to be shown in Kibiter" } , "retention_time" : { "optional" : True , "default" : None , "type" : int , "description" : "The maximum number of minutes wrt the current date to retain the data" } } }
params_projects = { "projects" : { "projects_file" : { "optional" : True , "default" : PROJECTS_JSON , "type" : str , "description" : "Projects file path with repositories to be collected group by projects" } , "projects_url" : { "optional" : True , "default" : None , "type" : str , "description" : "Projects file URL" } , "load_eclipse" : { "optional" : True , "default" : False , "type" : bool , "description" : "Load the projects from Eclipse" } } }
params_phases = { "phases" : { "collection" : { "optional" : False , "default" : True , "type" : bool , "description" : "Activate collection of items" } , "enrichment" : { "optional" : False , "default" : True , "type" : bool , "description" : "Activate enrichment of items" } , "identities" : { "optional" : False , "default" : True , "type" : bool , "description" : "Do the identities tasks" } , "panels" : { "optional" : False , "default" : True , "type" : bool , "description" : "Load panels, create alias and other tasks related" } , "track_items" : { "optional" : True , "default" : False , "type" : bool , "description" : "Track specific items from a gerrit repository" } , "report" : { "optional" : True , "default" : False , "type" : bool , "description" : "Generate the PDF report for a project (alpha)" } } }
general_config_params = [ params_general , params_projects , params_phases ]
for section_params in general_config_params :
    params . update ( section_params )
# Config provided by tasks
params_collection = { "es_collection" : { "password" : { "optional" : True , "default" : None , "type" : str , "description" : "Password for connection to Elasticsearch" } , "user" : { "optional" : True , "default" : None , "type" : str , "description" : "User for connection to Elasticsearch" } , "url" : { "optional" : False , "default" : "http://172.17.0.1:9200" , "type" : str , "description" : "Elasticsearch URL" } , "arthur" : { "optional" : True , "default" : False , "type" : bool , "description" : "Use arthur for collecting items from perceval" } , "arthur_url" : { "optional" : True , "default" : None , "type" : str , "description" : "URL for the arthur service" } , "redis_url" : { "optional" : True , "default" : None , "type" : str , "description" : "URL for the redis service" } } }
params_enrichment = { "es_enrichment" : { "url" : { "optional" : False , "default" : "http://172.17.0.1:9200" , "type" : str , "description" : "Elasticsearch URL" } , "autorefresh" : { "optional" : True , "default" : True , "type" : bool , "description" : "Execute the autorefresh of identities" } , "autorefresh_interval" : { "optional" : True , "default" : 2 , "type" : int , "description" : "Set time interval (days) for autorefresh identities" } , "user" : { "optional" : True , "default" : None , "type" : str , "description" : "User for connection to Elasticsearch" } , "password" : { "optional" : True , "default" : None , "type" : str , "description" : "Password for connection to Elasticsearch" } } }
params_panels = { "panels" : { "strict" : { "optional" : True , "default" : True , "type" : bool , "description" : "Enable strict panels loading" } , "kibiter_time_from" : { "optional" : True , "default" : "now-90d" , "type" : str , "description" : "Default time interval for Kibiter" } , "kibiter_default_index" : { "optional" : True , "default" : "git" , "type" : str , "description" : "Default index pattern for Kibiter" } , "kibiter_url" : { "optional" : False , "default" : None , "type" : str , "description" : "Kibiter URL" } , "kibiter_version" : { "optional" : True , "default" : None , "type" : str , "description" : "Kibiter version" } , "community" : { "optional" : True , "default" : True , "type" : bool , "description" : "Enable community structure menu" } , "kafka" : { "optional" : True , "default" : False , "type" : bool , "description" : "Enable kafka menu" } , "github-repos" : { "optional" : True , "default" : False , "type" : bool , "description" : "Enable GitHub repo stats menu" } , "gitlab-issues" : { "optional" : True , "default" : False , "type" : bool , "description" : "Enable GitLab issues menu" } , "gitlab-merges" : { "optional" : True , "default" : False , "type" : bool , "description" : "Enable GitLab merge requests menu" } , "mattermost" : { "optional" : True , "default" : False , "type" : bool , "description" : "Enable Mattermost menu" } } }
params_report = { "report" : { "start_date" : { "optional" : False , "default" : "1970-01-01" , "type" : str , "description" : "Start date for the report" } , "end_date" : { "optional" : False , "default" : "2100-01-01" , "type" : str , "description" : "End date for the report" } , "interval" : { "optional" : False , "default" : "quarter" , "type" : str , "description" : "Interval for the report" } , "config_file" : { "optional" : False , "default" : "report.cfg" , "type" : str , "description" : "Config file for the report" } , "data_dir" : { "optional" : False , "default" : "report_data" , "type" : str , "description" : "Directory in which to store the report data" } , "filters" : { "optional" : True , "default" : [ ] , "type" : list , "description" : "General filters to be applied to all queries" } , "offset" : { "optional" : True , "default" : None , "type" : str , "description" : "Date offset to be applied to start and end" } } }
params_sortinghat = { "sortinghat" : { "affiliate" : { "optional" : False , "default" : "True" , "type" : bool , "description" : "Affiliate identities to organizations" } , "unaffiliated_group" : { "optional" : False , "default" : "Unknown" , "type" : str , "description" : "Name of the organization for unaffiliated identities" } , "matching" : { "optional" : False , "default" : [ "email" ] , "type" : list , "description" : "Algorithm for matching identities in Sortinghat" } , "sleep_for" : { "optional" : False , "default" : 3600 , "type" : int , "description" : "Delay between task identities executions" } , "database" : { "optional" : False , "default" : "sortinghat_db" , "type" : str , "description" : "Name of the Sortinghat database" } , "host" : { "optional" : False , "default" : "mariadb" , "type" : str , "description" : "Host with the Sortinghat database" } , "user" : { "optional" : False , "default" : "root" , "type" : str , "description" : "User to access the Sortinghat database" } , "password" : { "optional" : False , "default" : "" , "type" : str , "description" : "Password to access the Sortinghat database" } , "autoprofile" : { "optional" : False , "default" : [ "customer" , "git" , "github" ] , "type" : list , "description" : "Order in which to get the identities information for filling the profile" } , "load_orgs" : { "optional" : True , "default" : False , "type" : bool , "deprecated" : "Load organizations in Sortinghat database" , "description" : "" } , "identities_format" : { "optional" : True , "default" : "sortinghat" , "type" : str , "description" : "Format of the identities data to be loaded" } , "strict_mapping" : { "optional" : True , "default" : True , "type" : bool , "description" : "rigorous check of values in identities matching " "(i.e, well formed email addresses)" } , "reset_on_load" : { "optional" : True , "default" : False , "type" : bool , "description" : "Unmerge and remove affiliations for all identities on load" } , "orgs_file" : { "optional" : True , "default" : None , "type" : str , "description" : "File path with the organizations to be loaded in Sortinghat" } , "identities_file" : { "optional" : True , "default" : [ ] , "type" : list , "description" : "File path with the identities to be loaded in Sortinghat" } , "identities_export_url" : { "optional" : True , "default" : None , "type" : str , "description" : "URL in which to export the identities in Sortinghat" } , "identities_api_token" : { "optional" : True , "default" : None , "type" : str , "description" : "API token for remote operation with GitHub and Gitlab" } , "bots_names" : { "optional" : True , "default" : [ ] , "type" : list , "description" : "Name of the identities to be marked as bots" } , "no_bots_names" : { "optional" : True , "default" : [ ] , "type" : list , "description" : "Name of the identities to be unmarked as bots" } , "autogender" : { "optional" : True , "default" : False , "type" : bool , "description" : "Add gender to the profiles (executes autogender)" } } }
params_track_items = { "track_items" : { "project" : { "optional" : False , "default" : "TrackProject" , "type" : str , "description" : "Gerrit project to track" } , "upstream_raw_es_url" : { "optional" : False , "default" : "" , "type" : str , "description" : "URL with the file with the gerrit reviews to track" } , "raw_index_gerrit" : { "optional" : False , "default" : "" , "type" : str , "description" : "Name of the gerrit raw index" } , "raw_index_git" : { "optional" : False , "default" : "" , "type" : str , "description" : "Name of the git raw index" } } }
tasks_config_params = [ params_collection , params_enrichment , params_panels , params_report , params_sortinghat , params_track_items ]
for section_params in tasks_config_params :
    params . update ( section_params )
return params


signature:
def set_param ( self , section , param , value ) :
    """Change a param in the config"""

implementation:
if section not in self . conf or param not in self . conf [ section ] :
    logger . error ( 'Config section %s and param %s not exists' , section , param )
else :
    self . conf [ section ] [ param ] = value


signature:
def _add_to_conf ( self , new_conf ) :
    """Add new configuration to self . conf .
 Adds configuration parameters in new _ con to self . conf .
 If they already existed in conf , overwrite them .
 : param new _ conf : new configuration , to add"""

implementation:
for section in new_conf :
    if section not in self . conf :
        self . conf [ section ] = new_conf [ section ]
    else :
        for param in new_conf [ section ] :
            self . conf [ section ] [ param ] = new_conf [ section ] [ param ]


signature:
def es_version ( self , url ) :
    """Get Elasticsearch version .
 Get the version of Elasticsearch . This is useful because
 Elasticsearch and Kibiter are paired ( same major version for 5 , 6 ) .
 : param url : Elasticseearch url hosting Kibiter indices
 : returns : major version , as string"""

implementation:
try :
    res = self . grimoire_con . get ( url )
    res . raise_for_status ( )
    major = res . json ( ) [ 'version' ] [ 'number' ] . split ( "." ) [ 0 ]
except Exception :
    logger . error ( "Error retrieving Elasticsearch version: " + url )
    raise
return major


signature:
def execute_nonstop_tasks ( self , tasks_cls ) :
    """Just a wrapper to the execute _ batch _ tasks method"""

implementation: self . execute_batch_tasks ( tasks_cls , self . conf [ 'sortinghat' ] [ 'sleep_for' ] , self . conf [ 'general' ] [ 'min_update_delay' ] , False )


signature:
def __execute_initial_load ( self ) :
    """Tasks that should be done just one time"""

implementation:
if self . conf [ 'phases' ] [ 'panels' ] :
    tasks_cls = [ TaskPanels , TaskPanelsMenu ]
    self . execute_tasks ( tasks_cls )
if self . conf [ 'phases' ] [ 'identities' ] :
    tasks_cls = [ TaskInitSortingHat ]
    self . execute_tasks ( tasks_cls )
logger . info ( "Loading projects" )
tasks_cls = [ TaskProjects ]
self . execute_tasks ( tasks_cls )
logger . info ( "Done" )
return


signature:
def start ( self ) :
    """This method defines the workflow of SirMordred . So it calls to :
 - initialize the databases
 - execute the different phases for the first iteration
 ( collection , identities , enrichment )
 - start the collection and enrichment in parallel by data source
 - start also the Sorting Hat merge"""

implementation:
# logger . debug ( " Starting SirMordred engine . . . " )
logger . info ( "" )
logger . info ( "----------------------------" )
logger . info ( "Starting SirMordred engine ..." )
logger . info ( "- - - - - - - - - - - - - - " )
# check we have access to the needed ES
if not self . check_es_access ( ) :
    print ( 'Can not access Elasticsearch service. Exiting sirmordred ...' )
    sys . exit ( 1 )
# If arthur is configured check that it is working
if self . conf [ 'es_collection' ] [ 'arthur' ] :
    if not self . check_redis_access ( ) :
        print ( 'Can not access redis service. Exiting sirmordred ...' )
        sys . exit ( 1 )
    if not self . check_arthur_access ( ) :
        print ( 'Can not access arthur service. Exiting sirmordred ...' )
        sys . exit ( 1 )
# If bestiary is configured check that it is working
if self . conf [ 'projects' ] [ 'projects_url' ] :
    if not self . check_bestiary_access ( ) :
        print ( 'Can not access bestiary service. Exiting sirmordred ...' )
        sys . exit ( 1 )
# Initial round : panels and projects loading
self . __execute_initial_load ( )
# Tasks to be executed during updating process
all_tasks_cls = [ ]
all_tasks_cls . append ( TaskProjects )
# projects update is always needed
if self . conf [ 'phases' ] [ 'collection' ] :
    if not self . conf [ 'es_collection' ] [ 'arthur' ] :
        all_tasks_cls . append ( TaskRawDataCollection )
    else :
        all_tasks_cls . append ( TaskRawDataArthurCollection )
if self . conf [ 'phases' ] [ 'identities' ] : # load identities and orgs periodically for updates
    all_tasks_cls . append ( TaskIdentitiesLoad )
    all_tasks_cls . append ( TaskIdentitiesMerge )
    all_tasks_cls . append ( TaskIdentitiesExport )
    # This is done in enrichement before doing the enrich
    # if self . conf [ ' phases ' ] [ ' collection ' ] :
    # all _ tasks _ cls . append ( TaskIdentitiesCollection )
if self . conf [ 'phases' ] [ 'enrichment' ] :
    all_tasks_cls . append ( TaskEnrich )
if self . conf [ 'phases' ] [ 'track_items' ] :
    all_tasks_cls . append ( TaskTrackItems )
if self . conf [ 'phases' ] [ 'report' ] :
    all_tasks_cls . append ( TaskReport )
# this is the main loop , where the execution should spend
# most of its time
while True :
    if not all_tasks_cls :
        logger . warning ( "No tasks to execute." )
        break
    try :
        if not self . conf [ 'general' ] [ 'update' ] :
            self . execute_batch_tasks ( all_tasks_cls , self . conf [ 'sortinghat' ] [ 'sleep_for' ] , self . conf [ 'general' ] [ 'min_update_delay' ] )
            self . execute_batch_tasks ( all_tasks_cls , self . conf [ 'sortinghat' ] [ 'sleep_for' ] , self . conf [ 'general' ] [ 'min_update_delay' ] )
            break
        else :
            self . execute_nonstop_tasks ( all_tasks_cls )
        # FIXME this point is never reached so despite the exception is
        # handled and the error is shown , the traceback is not printed
    except DataCollectionError as e :
        logger . error ( str ( e ) )
        var = traceback . format_exc ( )
        logger . error ( var )
    except DataEnrichmentError as e :
        logger . error ( str ( e ) )
        var = traceback . format_exc ( )
        logger . error ( var )
logger . info ( "Finished SirMordred engine ..." )


signature:
def run ( self , halt_on_nonzero = True , quiet = False , q = False , streaming = False ) :
    """After building your commands , call ` run ( ) ` to have your code executed ."""

implementation:
commands = str ( self )
if not ( quiet or q ) :
    self . _echo . cmd ( commands )
env = self . _context [ 0 ] . get ( 'env' , { } ) if len ( self . _context ) > 0 else os . environ
executable = self . current_context . get ( 'executable' )
try :
    process = subprocess . Popen ( commands , bufsize = 1 , shell = True , env = env , stdin = subprocess . PIPE , stdout = subprocess . PIPE , stderr = subprocess . PIPE , executable = executable , universal_newlines = True )
    result = Result ( process , commands , self . _context , streaming , halt_on_nonzero = halt_on_nonzero )
except Exception as e :
    result = Result ( None , commands , self . _context , exception = e )
    result . dump_exception ( )
    if halt_on_nonzero :
        raise e
finally :
    self . clear ( )
return result


signature:
def validate_config ( self ) :
    '''Validates the provided config to make sure all the required fields are
 there .'''

implementation:
# first ensure that all the required fields are there
for key , key_config in self . params_map . items ( ) :
    if key_config [ 'required' ] :
        if key not in self . config :
            raise ValueError ( "Invalid Configuration! Required parameter '%s' was not provided to Sultan." )
# second ensure that the fields that were pased were actually fields that
# can be used
for key in self . config . keys ( ) :
    if key not in self . params_map :
        raise ValueError ( "Invalid Configuration! The parameter '%s' provided is not used by Sultan!" % key )


signature:
def stdout ( self ) :
    """Converts stdout string to a list ."""

implementation:
if self . _streaming :
    stdout = [ ]
    while not self . __stdout . empty ( ) :
        try :
            line = self . __stdout . get_nowait ( )
            stdout . append ( line )
        except :
            pass
else :
    stdout = self . __stdout
return stdout


signature:
def stderr ( self ) :
    """Converts stderr string to a list ."""

implementation:
if self . _streaming :
    stderr = [ ]
    while not self . __stderr . empty ( ) :
        try :
            line = self . __stderr . get_nowait ( )
            stderr . append ( line )
        except :
            pass
else :
    stderr = self . __stderr
return stderr


signature:
def print_stdout ( self , always_print = False ) :
    """Prints the stdout to console - if there is any stdout , otherwise does nothing .
 : param always _ print : print the stdout , even if there is nothing in the buffer ( default : false )"""

implementation:
if self . __stdout or always_print :
    self . __echo . info ( "--{ STDOUT }---" + "-" * 100 )
    self . __format_lines_info ( self . stdout )
    self . __echo . info ( "---------------" + "-" * 100 )


signature:
def print_stderr ( self , always_print = False ) :
    """Prints the stderr to console - if there is any stdout , otherwise does nothing .
 : param always _ print : print the stderr , even if there is nothing in the buffer ( default : false )"""

implementation:
if self . __stderr or always_print :
    self . __echo . critical ( "--{ STDERR }---" + "-" * 100 )
    self . __format_lines_error ( self . stderr )
    self . __echo . critical ( "---------------" + "-" * 100 )


signature:
def print_traceback ( self , always_print = False ) :
    """Prints the traceback to console - if there is any traceback , otherwise does nothing .
 : param always _ print : print the traceback , even if there is nothing in the buffer ( default : false )"""

implementation:
if self . _exception or always_print :
    self . __echo . critical ( "--{ TRACEBACK }" + "-" * 100 )
    self . __format_lines_error ( self . traceback )
    self . __echo . critical ( "---------------" + "-" * 100 )


signature:
def format ( self , record ) :
    """Customize the message format based on the log level ."""

implementation:
if isinstance ( self . fmt , dict ) :
    self . _fmt = self . fmt [ record . levelname ]
    if sys . version_info > ( 3 , 2 ) : # Update self . _ style because we ' ve changed self . _ fmt
    # ( code based on stdlib ' s logging . Formatter . _ _ init _ _ ( ) )
        if self . style not in logging . _STYLES :
            raise ValueError ( 'Style must be one of: %s' % ',' . join ( list ( logging . _STYLES . keys ( ) ) ) )
        self . _style = logging . _STYLES [ self . style ] [ 0 ] ( self . _fmt )
if sys . version_info > ( 2 , 7 ) :
    message = super ( LevelFormatter , self ) . format ( record )
else :
    message = ColoredFormatter . format ( self , record )
return message


signature:
def replace_print ( fileobj = sys . stderr ) :
    """Sys . out replacer , by default with stderr .
 Use it like this :
 with replace _ print _ with ( fileobj ) :
 print " hello " # writes to the file
 print " done " # prints to stdout
 Args :
 fileobj : a file object to replace stdout .
 Yields :
 The printer ."""

implementation:
printer = _Printer ( fileobj )
previous_stdout = sys . stdout
sys . stdout = printer
try :
    yield printer
finally :
    sys . stdout = previous_stdout


signature:
def compact_interval_string ( value_list ) :
    """Compact a list of integers into a comma - separated string of intervals .
 Args :
 value _ list : A list of sortable integers such as a list of numbers
 Returns :
 A compact string representation , such as " 1-5,8,12-15" """

implementation:
if not value_list :
    return ''
value_list . sort ( )
# Start by simply building up a list of separate contiguous intervals
interval_list = [ ]
curr = [ ]
for val in value_list :
    if curr and ( val > curr [ - 1 ] + 1 ) :
        interval_list . append ( ( curr [ 0 ] , curr [ - 1 ] ) )
        curr = [ val ]
    else :
        curr . append ( val )
if curr :
    interval_list . append ( ( curr [ 0 ] , curr [ - 1 ] ) )
# For each interval collapse it down to " first , last " or just " first " if
# if first = = last .
return ',' . join ( [ '{}-{}' . format ( pair [ 0 ] , pair [ 1 ] ) if pair [ 0 ] != pair [ 1 ] else str ( pair [ 0 ] ) for pair in interval_list ] )


signature:
def _get_storage_service ( credentials ) :
    """Get a storage client using the provided credentials or defaults ."""

implementation:
if credentials is None :
    credentials = oauth2client . client . GoogleCredentials . get_application_default ( )
return discovery . build ( 'storage' , 'v1' , credentials = credentials )


signature:
def _retry_storage_check ( exception ) :
    """Return True if we should retry , False otherwise ."""

implementation:
now = datetime . now ( ) . strftime ( '%Y-%m-%d %H:%M:%S.%f' )
print_error ( '%s: Exception %s: %s' % ( now , type ( exception ) . __name__ , str ( exception ) ) )
return isinstance ( exception , oauth2client . client . AccessTokenRefreshError )


signature:
def _load_file_from_gcs ( gcs_file_path , credentials = None ) :
    """Load context from a text file in gcs .
 Args :
 gcs _ file _ path : The target file path ; should have the ' gs : / / ' prefix .
 credentials : Optional credential to be used to load the file from gcs .
 Returns :
 The content of the text file as a string ."""

implementation:
gcs_service = _get_storage_service ( credentials )
bucket_name , object_name = gcs_file_path [ len ( 'gs://' ) : ] . split ( '/' , 1 )
request = gcs_service . objects ( ) . get_media ( bucket = bucket_name , object = object_name )
file_handle = io . BytesIO ( )
downloader = MediaIoBaseDownload ( file_handle , request , chunksize = 1024 * 1024 )
done = False
while not done :
    _ , done = _downloader_next_chunk ( downloader )
filevalue = file_handle . getvalue ( )
if not isinstance ( filevalue , six . string_types ) :
    filevalue = filevalue . decode ( )
return six . StringIO ( filevalue )


signature:
def load_file ( file_path , credentials = None ) :
    """Load a file from either local or gcs .
 Args :
 file _ path : The target file path , which should have the prefix ' gs : / / ' if
 to be loaded from gcs .
 credentials : Optional credential to be used to load the file from gcs .
 Returns :
 A python File object if loading file from local or a StringIO object if
 loading from gcs ."""

implementation:
if file_path . startswith ( 'gs://' ) :
    return _load_file_from_gcs ( file_path , credentials )
else :
    return open ( file_path , 'r' )


signature:
def _file_exists_in_gcs ( gcs_file_path , credentials = None ) :
    """Check whether the file exists , in GCS .
 Args :
 gcs _ file _ path : The target file path ; should have the ' gs : / / ' prefix .
 credentials : Optional credential to be used to load the file from gcs .
 Returns :
 True if the file ' s there ."""

implementation:
gcs_service = _get_storage_service ( credentials )
bucket_name , object_name = gcs_file_path [ len ( 'gs://' ) : ] . split ( '/' , 1 )
request = gcs_service . objects ( ) . get ( bucket = bucket_name , object = object_name , projection = 'noAcl' )
try :
    request . execute ( )
    return True
except errors . HttpError :
    return False


signature:
def file_exists ( file_path , credentials = None ) :
    """Check whether the file exists , on local disk or GCS .
 Args :
 file _ path : The target file path ; should have the ' gs : / / ' prefix if in gcs .
 credentials : Optional credential to be used to load the file from gcs .
 Returns :
 True if the file ' s there ."""

implementation:
if file_path . startswith ( 'gs://' ) :
    return _file_exists_in_gcs ( file_path , credentials )
else :
    return os . path . isfile ( file_path )


signature:
def _prefix_exists_in_gcs ( gcs_prefix , credentials = None ) :
    """Check whether there is a GCS object whose name starts with the prefix .
 Since GCS doesn ' t actually have folders , this is how we check instead .
 Args :
 gcs _ prefix : The path ; should start with ' gs : / / ' .
 credentials : Optional credential to be used to load the file from gcs .
 Returns :
 True if the prefix matches at least one object in GCS .
 Raises :
 errors . HttpError : if it can ' t talk to the server"""

implementation:
gcs_service = _get_storage_service ( credentials )
bucket_name , prefix = gcs_prefix [ len ( 'gs://' ) : ] . split ( '/' , 1 )
# documentation in
# https : / / cloud . google . com / storage / docs / json _ api / v1 / objects / list
request = gcs_service . objects ( ) . list ( bucket = bucket_name , prefix = prefix , maxResults = 1 )
response = request . execute ( )
return response . get ( 'items' , None )


signature:
def simple_pattern_exists_in_gcs ( file_pattern , credentials = None ) :
    """True iff an object exists matching the input GCS pattern .
 The GCS pattern must be a full object reference or a " simple pattern " that
 conforms to the dsub input and output parameter restrictions :
 * No support for * * , ? wildcards or [ ] character ranges
 * Wildcards may only appear in the file name
 Args :
 file _ pattern : eg . ' gs : / / foo / ba * '
 credentials : Optional credential to be used to load the file from gcs .
 Raises :
 ValueError : if file _ pattern breaks the rules .
 Returns :
 True iff a file exists that matches that pattern ."""

implementation:
if '*' not in file_pattern :
    return _file_exists_in_gcs ( file_pattern , credentials )
if not file_pattern . startswith ( 'gs://' ) :
    raise ValueError ( 'file name must start with gs://' )
gcs_service = _get_storage_service ( credentials )
bucket_name , prefix = file_pattern [ len ( 'gs://' ) : ] . split ( '/' , 1 )
if '*' in bucket_name :
    raise ValueError ( 'Wildcards may not appear in the bucket name' )
# There is a ' * ' in prefix because we checked there ' s one in file _ pattern
# and there isn ' t one in bucket _ name . Hence it must be in prefix .
assert '*' in prefix
prefix_no_wildcard = prefix [ : prefix . index ( '*' ) ]
request = gcs_service . objects ( ) . list ( bucket = bucket_name , prefix = prefix_no_wildcard )
response = request . execute ( )
if 'items' not in response :
    return False
items_list = [ i [ 'name' ] for i in response [ 'items' ] ]
return any ( fnmatch . fnmatch ( i , prefix ) for i in items_list )


signature:
def outputs_are_present ( outputs ) :
    """True if each output contains at least one file or no output specified ."""

implementation:
# outputs are OutputFileParam ( see param _ util . py )
# If outputs contain a pattern , then there is no way for ` dsub ` to verify
# that * all * output is present . The best that ` dsub ` can do is to verify
# that * some * output was created for each such parameter .
for o in outputs :
    if not o . value :
        continue
    if o . recursive :
        if not folder_exists ( o . value ) :
            return False
    else :
        if not simple_pattern_exists_in_gcs ( o . value ) :
            return False
return True


signature:
def _build_pipeline_input_file_param ( cls , var_name , docker_path ) :
    """Return a dict object representing a pipeline input argument ."""

implementation:
# If the filename contains a wildcard , then the target Docker path must
# be a directory in order to ensure consistency whether the source pattern
# contains 1 or multiple files .
# In that case , we set the docker _ path to explicitly have a trailing slash
# ( for the Pipelines API " gsutil cp " handling , and then override the
# associated var _ name environment variable in the generated Docker command .
path , filename = os . path . split ( docker_path )
if '*' in filename :
    return cls . _build_pipeline_file_param ( var_name , path + '/' )
else :
    return cls . _build_pipeline_file_param ( var_name , docker_path )


signature:
def _build_pipeline_docker_command ( cls , script_name , inputs , outputs , envs ) :
    """Return a multi - line string of the full pipeline docker command ."""

implementation:
# We upload the user script as an environment argument
# and write it to SCRIPT _ DIR ( preserving its local file name ) .
# The docker _ command :
# * writes the script body to a file
# * installs gcloud if there are recursive copies to do
# * sets environment variables for inputs with wildcards
# * sets environment variables for recursive input directories
# * recursively copies input directories
# * creates output directories
# * sets environment variables for recursive output directories
# * sets the DATA _ ROOT environment variable to / mnt / data
# * sets the working directory to $ { DATA _ ROOT }
# * executes the user script
# * recursively copies output directories
recursive_input_dirs = [ var for var in inputs if var . recursive and var . value ]
recursive_output_dirs = [ var for var in outputs if var . recursive and var . value ]
install_cloud_sdk = ''
if recursive_input_dirs or recursive_output_dirs :
    install_cloud_sdk = INSTALL_CLOUD_SDK
export_input_dirs = ''
copy_input_dirs = ''
if recursive_input_dirs :
    export_input_dirs = providers_util . build_recursive_localize_env ( providers_util . DATA_MOUNT_POINT , inputs )
    copy_input_dirs = providers_util . build_recursive_localize_command ( providers_util . DATA_MOUNT_POINT , inputs , job_model . P_GCS )
export_output_dirs = ''
copy_output_dirs = ''
if recursive_output_dirs :
    export_output_dirs = providers_util . build_recursive_gcs_delocalize_env ( providers_util . DATA_MOUNT_POINT , outputs )
    copy_output_dirs = providers_util . build_recursive_delocalize_command ( providers_util . DATA_MOUNT_POINT , outputs , job_model . P_GCS )
docker_paths = [ var . docker_path if var . recursive else os . path . dirname ( var . docker_path ) for var in outputs if var . value ]
mkdirs = '\n' . join ( [ 'mkdir -p {0}/{1}' . format ( providers_util . DATA_MOUNT_POINT , path ) for path in docker_paths ] )
inputs_with_wildcards = [ var for var in inputs if not var . recursive and var . docker_path and '*' in os . path . basename ( var . docker_path ) ]
export_inputs_with_wildcards = '\n' . join ( [ 'export {0}="{1}/{2}"' . format ( var . name , providers_util . DATA_MOUNT_POINT , var . docker_path ) for var in inputs_with_wildcards ] )
export_empty_envs = '\n' . join ( [ 'export {0}=""' . format ( var . name ) for var in envs | inputs | outputs if not var . value ] )
return DOCKER_COMMAND . format ( mk_runtime_dirs = MK_RUNTIME_DIRS_COMMAND , script_path = '%s/%s' % ( providers_util . SCRIPT_DIR , script_name ) , install_cloud_sdk = install_cloud_sdk , export_inputs_with_wildcards = export_inputs_with_wildcards , export_input_dirs = export_input_dirs , copy_input_dirs = copy_input_dirs , mk_output_dirs = mkdirs , export_output_dirs = export_output_dirs , export_empty_envs = export_empty_envs , tmpdir = providers_util . TMP_DIR , working_dir = providers_util . WORKING_DIR , copy_output_dirs = copy_output_dirs )


signature:
def build_pipeline ( cls , project , zones , min_cores , min_ram , disk_size , boot_disk_size , preemptible , accelerator_type , accelerator_count , image , script_name , envs , inputs , outputs , pipeline_name ) :
    """Builds a pipeline configuration for execution .
 Args :
 project : string name of project .
 zones : list of zone names for jobs to be run at .
 min _ cores : int number of CPU cores required per job .
 min _ ram : int GB of RAM required per job .
 disk _ size : int GB of disk to attach under / mnt / data .
 boot _ disk _ size : int GB of disk for boot .
 preemptible : use a preemptible VM for the job
 accelerator _ type : string GCE defined accelerator type .
 accelerator _ count : int number of accelerators of the specified type to
 attach .
 image : string Docker image name in which to run .
 script _ name : file name of the script to run .
 envs : list of EnvParam objects specifying environment variables to set
 within each job .
 inputs : list of FileParam objects specifying input variables to set
 within each job .
 outputs : list of FileParam objects specifying output variables to set
 within each job .
 pipeline _ name : string name of pipeline .
 Returns :
 A nested dictionary with one entry under the key ephemeralPipeline
 containing the pipeline configuration ."""

implementation:
if min_cores is None :
    min_cores = job_model . DEFAULT_MIN_CORES
if min_ram is None :
    min_ram = job_model . DEFAULT_MIN_RAM
if disk_size is None :
    disk_size = job_model . DEFAULT_DISK_SIZE
if boot_disk_size is None :
    boot_disk_size = job_model . DEFAULT_BOOT_DISK_SIZE
if preemptible is None :
    preemptible = job_model . DEFAULT_PREEMPTIBLE
# Format the docker command
docker_command = cls . _build_pipeline_docker_command ( script_name , inputs , outputs , envs )
# Pipelines inputParameters can be both simple name / value pairs which get
# set as environment variables , as well as input file paths which the
# Pipelines controller will automatically localize to the Pipeline VM .
# In the ephemeralPipeline object , the inputParameters are only defined ;
# the values are passed in the pipelineArgs .
# Pipelines outputParameters are only output file paths , which the
# Pipelines controller can automatically de - localize after the docker
# command completes .
# The Pipelines API does not support recursive copy of file parameters ,
# so it is implemented within the dsub - generated pipeline .
# Any inputs or outputs marked as " recursive " are completely omitted here ;
# their environment variables will be set in the docker command , and
# recursive copy code will be generated there as well .
# The Pipelines API does not accept empty environment variables . Set them to
# empty in DOCKER _ COMMAND instead .
input_envs = [ { 'name' : SCRIPT_VARNAME } ] + [ { 'name' : env . name } for env in envs if env . value ]
input_files = [ cls . _build_pipeline_input_file_param ( var . name , var . docker_path ) for var in inputs if not var . recursive and var . value ]
# Outputs are an array of file parameters
output_files = [ cls . _build_pipeline_file_param ( var . name , var . docker_path ) for var in outputs if not var . recursive and var . value ]
# The ephemeralPipeline provides the template for the pipeline .
# pyformat : disable
return { 'ephemeralPipeline' : { 'projectId' : project , 'name' : pipeline_name , # Define the resources needed for this pipeline .
'resources' : { 'minimumCpuCores' : min_cores , 'minimumRamGb' : min_ram , 'bootDiskSizeGb' : boot_disk_size , 'preemptible' : preemptible , 'zones' : google_base . get_zones ( zones ) , 'acceleratorType' : accelerator_type , 'acceleratorCount' : accelerator_count , # Create a data disk that is attached to the VM and destroyed
# when the pipeline terminates .
'disks' : [ { 'name' : 'datadisk' , 'autoDelete' : True , 'sizeGb' : disk_size , 'mountPoint' : providers_util . DATA_MOUNT_POINT , } ] , } , 'inputParameters' : input_envs + input_files , 'outputParameters' : output_files , 'docker' : { 'imageName' : image , 'cmd' : docker_command , } } }


signature:
def build_pipeline_args ( cls , project , script , job_params , task_params , reserved_labels , preemptible , logging_uri , scopes , keep_alive ) :
    """Builds pipeline args for execution .
 Args :
 project : string name of project .
 script : Body of the script to execute .
 job _ params : dictionary of values for labels , envs , inputs , and outputs
 for this job .
 task _ params : dictionary of values for labels , envs , inputs , and outputs
 for this task .
 reserved _ labels : dictionary of reserved labels ( e . g . task - id ,
 task - attempt )
 preemptible : use a preemptible VM for the job
 logging _ uri : path for job logging output .
 scopes : list of scope .
 keep _ alive : Seconds to keep VM alive on failure
 Returns :
 A nested dictionary with one entry under the key pipelineArgs containing
 the pipeline arguments ."""

implementation:
# For the Pipelines API , envs and file inputs are all " inputs " .
inputs = { }
inputs . update ( { SCRIPT_VARNAME : script } )
inputs . update ( { var . name : var . value for var in job_params [ 'envs' ] | task_params [ 'envs' ] if var . value } )
inputs . update ( { var . name : var . uri for var in job_params [ 'inputs' ] | task_params [ 'inputs' ] if not var . recursive and var . value } )
# Remove wildcard references for non - recursive output . When the pipelines
# controller generates a delocalize call , it must point to a bare directory
# for patterns . The output param OUTFILE = gs : / / bucket / path / * . bam should
# delocalize with a call similar to :
# gsutil cp / mnt / data / output / gs / bucket / path / * . bam gs : / / bucket / path /
outputs = { }
for var in job_params [ 'outputs' ] | task_params [ 'outputs' ] :
    if var . recursive or not var . value :
        continue
    if '*' in var . uri . basename :
        outputs [ var . name ] = var . uri . path
    else :
        outputs [ var . name ] = var . uri
labels = { }
labels . update ( { label . name : label . value if label . value else '' for label in ( reserved_labels | job_params [ 'labels' ] | task_params [ 'labels' ] ) } )
# pyformat : disable
args = { 'pipelineArgs' : { 'projectId' : project , 'resources' : { 'preemptible' : preemptible , } , 'inputs' : inputs , 'outputs' : outputs , 'labels' : labels , 'serviceAccount' : { 'email' : 'default' , 'scopes' : scopes , } , # Pass the user - specified GCS destination for pipeline logging .
'logging' : { 'gcsPath' : logging_uri } , } }
# pyformat : enable
if keep_alive :
    args [ 'pipelineArgs' ] [ 'keep_vm_alive_on_failure_duration' ] = '%ss' % keep_alive
return args


signature:
def _datetime_to_utc_int ( date ) :
    """Convert the integer UTC time value into a local datetime ."""

implementation:
if date is None :
    return None
# Convert localized datetime to a UTC integer
epoch = dsub_util . replace_timezone ( datetime . utcfromtimestamp ( 0 ) , pytz . utc )
return ( date - epoch ) . total_seconds ( )


signature:
def get_filter ( project , status = None , user_id = None , job_id = None , job_name = None , labels = None , task_id = None , task_attempt = None , create_time_min = None , create_time_max = None ) :
    """Return a filter string for operations . list ( ) ."""

implementation:
ops_filter = [ 'projectId = %s' % project ]
if status and status != '*' :
    ops_filter . append ( 'status = %s' % status )
if user_id != '*' :
    ops_filter . append ( 'labels.user-id = %s' % user_id )
if job_id != '*' :
    ops_filter . append ( 'labels.job-id = %s' % job_id )
if job_name != '*' :
    ops_filter . append ( 'labels.job-name = %s' % job_name )
if task_id != '*' :
    ops_filter . append ( 'labels.task-id = %s' % task_id )
if task_attempt != '*' :
    ops_filter . append ( 'labels.task-attempt = %s' % task_attempt )
# Even though labels are nominally ' arbitrary strings ' they are trusted
# since param _ util restricts the character set .
if labels :
    for l in labels :
        ops_filter . append ( 'labels.%s = %s' % ( l . name , l . value ) )
epoch = dsub_util . replace_timezone ( datetime . utcfromtimestamp ( 0 ) , pytz . utc )
if create_time_min :
    create_time_min_utc_int = ( create_time_min - epoch ) . total_seconds ( )
    ops_filter . append ( 'createTime >= %d' % create_time_min_utc_int )
if create_time_max :
    create_time_max_utc_int = ( create_time_max - epoch ) . total_seconds ( )
    ops_filter . append ( 'createTime <= %d' % create_time_max_utc_int )
return ' AND ' . join ( ops_filter )


signature:
def is_dsub_operation ( cls , op ) :
    """Determine if a pipelines operation is a dsub request .
 We don ' t have a rigorous way to identify an operation as being submitted
 by dsub . Our best option is to check for certain fields that have always
 been part of dsub operations .
 - labels : job - id , job - name , and user - id have always existed
 - envs : _ SCRIPT has always existed .
 In order to keep a simple heuristic this test only uses labels .
 Args :
 op : a pipelines operation .
 Returns :
 Boolean , true if the pipeline run was generated by dsub ."""

implementation:
if not cls . is_pipelines_operation ( op ) :
    return False
for name in [ 'job-id' , 'job-name' , 'user-id' ] :
    if not cls . get_operation_label ( op , name ) :
        return False
return True


signature:
def list ( cls , service , ops_filter , page_size = 0 ) :
    """Gets the list of operations for the specified filter .
 Args :
 service : Google Genomics API service object
 ops _ filter : string filter of operations to return
 page _ size : the number of operations to requested on each list operation to
 the pipelines API ( if 0 or None , the API default is used )
 Yields :
 Operations matching the filter criteria ."""

implementation:
page_token = None
more_operations = True
documented_default_page_size = 256
documented_max_page_size = 2048
if not page_size :
    page_size = documented_default_page_size
page_size = min ( page_size , documented_max_page_size )
while more_operations :
    api = service . operations ( ) . list ( name = 'operations' , filter = ops_filter , pageToken = page_token , pageSize = page_size )
    response = google_base . Api . execute ( api )
    ops = response . get ( 'operations' , [ ] )
    for op in ops :
        if cls . is_dsub_operation ( op ) :
            yield GoogleOperation ( op )
    page_token = response . get ( 'nextPageToken' )
    more_operations = bool ( page_token )


signature:
def prepare_job_metadata ( self , script , job_name , user_id , create_time ) :
    """Returns a dictionary of metadata fields for the job ."""

implementation: return google_base . prepare_job_metadata ( script , job_name , user_id , create_time )


signature:
def _build_pipeline_request ( self , task_view ) :
    """Returns a Pipeline objects for the job ."""

implementation:
job_metadata = task_view . job_metadata
job_params = task_view . job_params
job_resources = task_view . job_resources
task_metadata = task_view . task_descriptors [ 0 ] . task_metadata
task_params = task_view . task_descriptors [ 0 ] . task_params
task_resources = task_view . task_descriptors [ 0 ] . task_resources
script = task_view . job_metadata [ 'script' ]
reserved_labels = google_base . build_pipeline_labels ( job_metadata , task_metadata , task_id_pattern = 'task-%d' )
# Build the ephemeralPipeline for this job .
# The ephemeralPipeline definition changes for each job because file
# parameters localCopy . path changes based on the remote _ uri .
pipeline = _Pipelines . build_pipeline ( project = self . _project , zones = job_resources . zones , min_cores = job_resources . min_cores , min_ram = job_resources . min_ram , disk_size = job_resources . disk_size , boot_disk_size = job_resources . boot_disk_size , preemptible = job_resources . preemptible , accelerator_type = job_resources . accelerator_type , accelerator_count = job_resources . accelerator_count , image = job_resources . image , script_name = script . name , envs = job_params [ 'envs' ] | task_params [ 'envs' ] , inputs = job_params [ 'inputs' ] | task_params [ 'inputs' ] , outputs = job_params [ 'outputs' ] | task_params [ 'outputs' ] , pipeline_name = job_metadata [ 'pipeline-name' ] )
# Build the pipelineArgs for this job .
logging_uri = task_resources . logging_path . uri
scopes = job_resources . scopes or google_base . DEFAULT_SCOPES
pipeline . update ( _Pipelines . build_pipeline_args ( self . _project , script . value , job_params , task_params , reserved_labels , job_resources . preemptible , logging_uri , scopes , job_resources . keep_alive ) )
return pipeline


signature:
def lookup_job_tasks ( self , statuses , user_ids = None , job_ids = None , job_names = None , task_ids = None , task_attempts = None , labels = None , create_time_min = None , create_time_max = None , max_tasks = 0 , page_size = 0 ) :
    """Yields operations based on the input criteria .
 If any of the filters are empty or { ' * ' } , then no filtering is performed on
 that field . Filtering by both a job id list and job name list is
 unsupported .
 Args :
 statuses : { ' * ' } , or a list of job status strings to return . Valid
 status strings are ' RUNNING ' , ' SUCCESS ' , ' FAILURE ' , or ' CANCELED ' .
 user _ ids : a list of ids for the user ( s ) who launched the job .
 job _ ids : a list of job ids to return .
 job _ names : a list of job names to return .
 task _ ids : a list of specific tasks within the specified job ( s ) to return .
 task _ attempts : a list of specific attempts within the specified tasks ( s )
 to return .
 labels : a list of LabelParam with user - added labels . All labels must
 match the task being fetched .
 create _ time _ min : a timezone - aware datetime value for the earliest create
 time of a task , inclusive .
 create _ time _ max : a timezone - aware datetime value for the most recent
 create time of a task , inclusive .
 max _ tasks : the maximum number of job tasks to return or 0 for no limit .
 page _ size : the page size to use for each query to the pipelins API .
 Raises :
 ValueError : if both a job id list and a job name list are provided
 Yeilds :
 Genomics API Operations objects ."""

implementation:
# Server - side , we can filter on status , job _ id , user _ id , task _ id , but there
# is no OR filter ( only AND ) , so we can ' t handle lists server side .
# Therefore we construct a set of queries for each possible combination of
# these criteria .
statuses = statuses if statuses else { '*' }
user_ids = user_ids if user_ids else { '*' }
job_ids = job_ids if job_ids else { '*' }
job_names = job_names if job_names else { '*' }
task_ids = task_ids if task_ids else { '*' }
task_attempts = task_attempts if task_attempts else { '*' }
# The task - id label value of " task - n " instead of just " n " is a hold - over
# from early label value character restrictions .
# Accept both forms , " task - n " and " n " , for lookups by task - id .
task_ids = { 'task-{}' . format ( t ) if t . isdigit ( ) else t for t in task_ids }
if job_ids != { '*' } and job_names != { '*' } :
    raise ValueError ( 'Filtering by both job IDs and job names is not supported' )
# AND filter rule arguments .
labels = labels if labels else set ( )
# The results of all these queries need to be sorted by create - time
# ( descending ) . To accomplish this , each query stream ( already sorted by
# create - time ) is added to a SortedGeneratorIterator which is a wrapper
# around a PriorityQueue of iterators ( sorted by each stream ' s newest task ' s
# create - time ) . A sorted list can then be built by stepping through this
# iterator and adding tasks until none are left or we hit max _ tasks .
now = datetime . now ( )
def _desc_date_sort_key ( t ) :
    return now - dsub_util . replace_timezone ( t . get_field ( 'create-time' ) , None )
query_queue = sorting_util . SortedGeneratorIterator ( key = _desc_date_sort_key )
for status , job_id , job_name , user_id , task_id , task_attempt in ( itertools . product ( statuses , job_ids , job_names , user_ids , task_ids , task_attempts ) ) :
    ops_filter = _Operations . get_filter ( self . _project , status = status , user_id = user_id , job_id = job_id , job_name = job_name , labels = labels , task_id = task_id , task_attempt = task_attempt , create_time_min = create_time_min , create_time_max = create_time_max )
    # The pipelines API returns operations sorted by create - time date . We can
    # use this sorting guarantee to merge - sort the streams together and only
    # retrieve more tasks as needed .
    stream = _Operations . list ( self . _service , ops_filter , page_size = page_size )
    query_queue . add_generator ( stream )
tasks_yielded = 0
for task in query_queue :
    yield task
    tasks_yielded += 1
    if 0 < max_tasks <= tasks_yielded :
        break


signature:
def delete_jobs ( self , user_ids , job_ids , task_ids , labels , create_time_min = None , create_time_max = None ) :
    """Kills the operations associated with the specified job or job . task .
 Args :
 user _ ids : List of user ids who " own " the job ( s ) to cancel .
 job _ ids : List of job _ ids to cancel .
 task _ ids : List of task - ids to cancel .
 labels : List of LabelParam , each must match the job ( s ) to be canceled .
 create _ time _ min : a timezone - aware datetime value for the earliest create
 time of a task , inclusive .
 create _ time _ max : a timezone - aware datetime value for the most recent
 create time of a task , inclusive .
 Returns :
 A list of tasks canceled and a list of error messages ."""

implementation:
# Look up the job ( s )
tasks = list ( self . lookup_job_tasks ( { 'RUNNING' } , user_ids = user_ids , job_ids = job_ids , task_ids = task_ids , labels = labels , create_time_min = create_time_min , create_time_max = create_time_max ) )
print ( 'Found %d tasks to delete.' % len ( tasks ) )
return google_base . cancel ( self . _service . new_batch_http_request , self . _service . operations ( ) . cancel , tasks )


signature:
def get_field ( self , field , default = None ) :
    """Returns a value from the operation for a specific set of field names .
 Args :
 field : a dsub - specific job metadata key
 default : default value to return if field does not exist or is empty .
 Returns :
 A text string for the field or a list for ' inputs ' .
 Raises :
 ValueError : if the field label is not supported by the operation"""

implementation:
metadata = self . _op . get ( 'metadata' )
value = None
if field == 'internal-id' :
    value = self . _op [ 'name' ]
elif field == 'job-id' :
    value = metadata [ 'labels' ] . get ( 'job-id' )
elif field == 'job-name' :
    value = metadata [ 'labels' ] . get ( 'job-name' )
elif field == 'task-id' :
    value = metadata [ 'labels' ] . get ( 'task-id' )
elif field == 'task-attempt' :
    value = metadata [ 'labels' ] . get ( 'task-attempt' )
elif field == 'user-id' :
    value = metadata [ 'labels' ] . get ( 'user-id' )
elif field == 'dsub-version' :
    value = metadata [ 'labels' ] . get ( 'dsub-version' )
elif field == 'task-status' :
    value = self . _operation_status ( )
elif field == 'logging' :
    value = metadata [ 'request' ] [ 'pipelineArgs' ] [ 'logging' ] [ 'gcsPath' ]
elif field == 'envs' :
    value = self . _get_operation_input_field_values ( metadata , False )
elif field == 'labels' : # Reserved labels are filtered from dsub task output .
    value = { k : v for k , v in metadata [ 'labels' ] . items ( ) if k not in job_model . RESERVED_LABELS }
elif field == 'inputs' :
    value = self . _get_operation_input_field_values ( metadata , True )
elif field == 'outputs' :
    value = self . _get_operation_output_field_values ( metadata )
elif field == 'mounts' :
    value = None
elif field == 'create-time' :
    value = google_base . parse_rfc3339_utc_string ( metadata [ 'createTime' ] )
elif field == 'start-time' : # Look through the events list for all " start " events ( only one expected ) .
    start_events = [ e for e in metadata . get ( 'events' , [ ] ) if e [ 'description' ] == 'start' ]
    # Get the startTime from the last " start " event .
    if start_events :
        value = google_base . parse_rfc3339_utc_string ( start_events [ - 1 ] [ 'startTime' ] )
elif field == 'end-time' :
    if 'endTime' in metadata :
        value = google_base . parse_rfc3339_utc_string ( metadata [ 'endTime' ] )
elif field == 'status' :
    value = self . _operation_status ( )
elif field in [ 'status-message' , 'status-detail' ] :
    status , last_update = self . _operation_status_message ( )
    value = status
elif field == 'last-update' :
    status , last_update = self . _operation_status_message ( )
    value = last_update
elif field == 'provider' :
    return _PROVIDER_NAME
elif field == 'provider-attributes' : # Use soft getting of keys to address a race condition and to
# pull the null values found in jobs that fail prior to scheduling .
    gce_data = metadata . get ( 'runtimeMetadata' , { } ) . get ( 'computeEngine' , { } )
    if 'machineType' in gce_data :
        machine_type = gce_data . get ( 'machineType' ) . rpartition ( '/' ) [ 2 ]
    else :
        machine_type = None
    instance_name = gce_data . get ( 'instanceName' )
    instance_zone = gce_data . get ( 'zone' )
    value = { 'machine-type' : machine_type , 'instance-name' : instance_name , 'zone' : instance_zone , }
elif field == 'events' :
    events = metadata . get ( 'events' , [ ] )
    value = [ ]
    for event in events :
        event_value = { 'name' : event . get ( 'description' , '' ) , 'start-time' : google_base . parse_rfc3339_utc_string ( event [ 'startTime' ] ) }
        if 'endTime' in event :
            event_value [ 'end-time' ] = google_base . parse_rfc3339_utc_string ( event [ 'endTime' ] )
        value . append ( event_value )
elif field in [ 'user-project' , 'script-name' , 'script' , 'input-recursives' , 'output-recursives' ] : # Supported in local and google - v2 providers .
    value = None
else :
    raise ValueError ( 'Unsupported field: "%s"' % field )
return value if value else default


signature:
def _operation_status_message ( self ) :
    """Returns the most relevant status string and last updated date string .
 This string is meant for display only .
 Returns :
 A printable status string and date string ."""

implementation:
metadata = self . _op [ 'metadata' ]
if not self . _op [ 'done' ] :
    if 'events' in metadata and metadata [ 'events' ] : # Get the last event
        last_event = metadata [ 'events' ] [ - 1 ]
        msg = last_event [ 'description' ]
        ds = last_event [ 'startTime' ]
    else :
        msg = 'Pending'
        ds = metadata [ 'createTime' ]
else :
    ds = metadata [ 'endTime' ]
    if 'error' in self . _op :
        msg = self . _op [ 'error' ] [ 'message' ]
    else :
        msg = 'Success'
return ( msg , google_base . parse_rfc3339_utc_string ( ds ) )


signature:
def _get_operation_input_field_values ( self , metadata , file_input ) :
    """Returns a dictionary of envs or file inputs for an operation .
 Args :
 metadata : operation metadata field
 file _ input : True to return a dict of file inputs , False to return envs .
 Returns :
 A dictionary of input field name value pairs"""

implementation:
# To determine input parameter type , we iterate through the
# pipeline inputParameters .
# The values come from the pipelineArgs inputs .
input_args = metadata [ 'request' ] [ 'ephemeralPipeline' ] [ 'inputParameters' ]
vals_dict = metadata [ 'request' ] [ 'pipelineArgs' ] [ 'inputs' ]
# Get the names for files or envs
names = [ arg [ 'name' ] for arg in input_args if ( 'localCopy' in arg ) == file_input ]
# Build the return dict
return { name : vals_dict [ name ] for name in names if name in vals_dict }


signature:
def error_message ( self ) :
    """Returns an error message if the operation failed for any reason .
 Failure as defined here means ; ended for any reason other than ' success ' .
 This means that a successful cancelation will also create an error message
 here .
 Returns :
 string , string will be empty if job did not error ."""

implementation:
if 'error' in self . _op :
    if 'task-id' in self . _op [ 'metadata' ] [ 'labels' ] :
        job_id = self . _op [ 'metadata' ] [ 'labels' ] [ 'task-id' ]
    else :
        job_id = self . _op [ 'metadata' ] [ 'labels' ] [ 'job-id' ]
    return 'Error in job %s - code %s: %s' % ( job_id , self . _op [ 'error' ] [ 'code' ] , self . _op [ 'error' ] [ 'message' ] )
else :
    return ''


signature:
def _format_task_name ( job_id , task_id , task_attempt ) :
    """Create a task name from a job - id , task - id , and task - attempt .
 Task names are used internally by dsub as well as by the docker task runner .
 The name is formatted as " < job - id > . < task - id > [ . task - attempt ] " . Task names
 follow formatting conventions allowing them to be safely used as a docker
 name .
 Args :
 job _ id : ( str ) the job ID .
 task _ id : ( str ) the task ID .
 task _ attempt : ( int ) the task attempt .
 Returns :
 a task name string ."""

implementation:
docker_name = '%s.%s' % ( job_id , 'task' if task_id is None else task_id )
if task_attempt is not None :
    docker_name += '.' + str ( task_attempt )
# Docker container names must match : [ a - zA - Z0-9 ] [ a - zA - Z0-9 _ . - ]
# So 1 ) prefix it with " dsub - " and 2 ) change all invalid characters to " - " .
return 'dsub-{}' . format ( _convert_suffix_to_docker_chars ( docker_name ) )


signature:
def _convert_suffix_to_docker_chars ( suffix ) :
    """Rewrite string so that all characters are valid in a docker name suffix ."""

implementation:
# Docker container names must match : [ a - zA - Z0-9 ] [ a - zA - Z0-9 _ . - ]
accepted_characters = string . ascii_letters + string . digits + '_.-'
def label_char_transform ( char ) :
    if char in accepted_characters :
        return char
    return '-'
return '' . join ( label_char_transform ( c ) for c in suffix )


signature:
def _task_sort_function ( task ) :
    """Return a tuple for sorting ' most recent first ' ."""

implementation: return ( task . get_field ( 'create-time' ) , int ( task . get_field ( 'task-id' , 0 ) ) , int ( task . get_field ( 'task-attempt' , 0 ) ) )


signature:
def _datetime_in_range ( self , dt , dt_min = None , dt_max = None ) :
    """Determine if the provided time is within the range , inclusive ."""

implementation:
# The pipelines API stores operation create - time with second granularity .
# We mimic this behavior in the local provider by truncating to seconds .
dt = dt . replace ( microsecond = 0 )
if dt_min :
    dt_min = dt_min . replace ( microsecond = 0 )
else :
    dt_min = dsub_util . replace_timezone ( datetime . datetime . min , pytz . utc )
if dt_max :
    dt_max = dt_max . replace ( microsecond = 0 )
else :
    dt_max = dsub_util . replace_timezone ( datetime . datetime . max , pytz . utc )
return dt_min <= dt <= dt_max


signature:
def _get_task_from_task_dir ( self , job_id , user_id , task_id , task_attempt ) :
    """Return a Task object with this task ' s info ."""

implementation:
# We need to be very careful about how we read and interpret the contents
# of the task directory . The directory could be changing because a new
# task is being created . The directory could be changing because a task
# is ending .
# If the meta . yaml does not exist , the task does not yet exist .
# If the meta . yaml exists , it means the task is scheduled . It does not mean
# it is yet running .
# If the task . pid file exists , it means that the runner . sh was started .
task_dir = self . _task_directory ( job_id , task_id , task_attempt )
job_descriptor = self . _read_task_metadata ( task_dir )
if not job_descriptor :
    return None
# If we read up an old task , the user - id will not be in the job _ descriptor .
if not job_descriptor . job_metadata . get ( 'user-id' ) :
    job_descriptor . job_metadata [ 'user-id' ] = user_id
# Get the pid of the runner
pid = - 1
try :
    with open ( os . path . join ( task_dir , 'task.pid' ) , 'r' ) as f :
        pid = int ( f . readline ( ) . strip ( ) )
except ( IOError , OSError ) :
    pass
# Get the script contents
script = None
script_name = job_descriptor . job_metadata . get ( 'script-name' )
if script_name :
    script = self . _read_script ( task_dir , script_name )
# Read the files written by the runner . sh .
# For new tasks , these may not have been written yet .
end_time = self . _get_end_time_from_task_dir ( task_dir )
last_update = self . _get_last_update_time_from_task_dir ( task_dir )
events = self . _get_events_from_task_dir ( task_dir )
status = self . _get_status_from_task_dir ( task_dir )
log_detail = self . _get_log_detail_from_task_dir ( task_dir )
# If the status file is not yet written , then mark the task as pending
if not status :
    status = 'RUNNING'
    log_detail = [ 'Pending' ]
return LocalTask ( task_status = status , events = events , log_detail = log_detail , job_descriptor = job_descriptor , end_time = end_time , last_update = last_update , pid = pid , script = script )


signature:
def _delocalize_logging_command ( self , logging_path , user_project ) :
    """Returns a command to delocalize logs .
 Args :
 logging _ path : location of log files .
 user _ project : name of the project to be billed for the request .
 Returns :
 eg . ' gs : / / bucket / path / myfile ' or ' gs : / / bucket / script - foobar - 12'"""

implementation:
# Get the logging prefix ( everything up to " . log " )
logging_prefix = os . path . splitext ( logging_path . uri ) [ 0 ]
# Set the provider - specific mkdir and file copy commands
if logging_path . file_provider == job_model . P_LOCAL :
    mkdir_cmd = 'mkdir -p "%s"\n' % os . path . dirname ( logging_prefix )
    cp_cmd = 'cp'
elif logging_path . file_provider == job_model . P_GCS :
    mkdir_cmd = ''
    if user_project :
        cp_cmd = 'gsutil -u {} -mq cp' . format ( user_project )
    else :
        cp_cmd = 'gsutil -mq cp'
else :
    assert False
# Construct the copy command
copy_logs_cmd = textwrap . dedent ( """\
      local cp_cmd="{cp_cmd}"
      local prefix="{prefix}"
    """ ) . format ( cp_cmd = cp_cmd , prefix = logging_prefix )
# Build up the command
body = textwrap . dedent ( """\
      {mkdir_cmd}
      {copy_logs_cmd}
    """ ) . format ( mkdir_cmd = mkdir_cmd , copy_logs_cmd = copy_logs_cmd )
return body


signature:
def _task_directory ( self , job_id , task_id , task_attempt ) :
    """The local dir for staging files for that particular task ."""

implementation:
dir_name = 'task' if task_id is None else str ( task_id )
if task_attempt :
    dir_name = '%s.%s' % ( dir_name , task_attempt )
return self . _provider_root ( ) + '/' + job_id + '/' + dir_name


signature:
def _make_environment ( self , inputs , outputs , mounts ) :
    """Return a dictionary of environment variables for the container ."""

implementation:
env = { }
env . update ( providers_util . get_file_environment_variables ( inputs ) )
env . update ( providers_util . get_file_environment_variables ( outputs ) )
env . update ( providers_util . get_file_environment_variables ( mounts ) )
return env


signature:
def _localize_inputs_recursive_command ( self , task_dir , inputs ) :
    """Returns a command that will stage recursive inputs ."""

implementation:
data_dir = os . path . join ( task_dir , _DATA_SUBDIR )
provider_commands = [ providers_util . build_recursive_localize_command ( data_dir , inputs , file_provider ) for file_provider in _SUPPORTED_INPUT_PROVIDERS ]
return '\n' . join ( provider_commands )


signature:
def _get_input_target_path ( self , local_file_path ) :
    """Returns a directory or file path to be the target for " gsutil cp " .
 If the filename contains a wildcard , then the target path must
 be a directory in order to ensure consistency whether the source pattern
 contains one or multiple files .
 Args :
 local _ file _ path : A full path terminating in a file or a file wildcard .
 Returns :
 The path to use as the " gsutil cp " target ."""

implementation:
path , filename = os . path . split ( local_file_path )
if '*' in filename :
    return path + '/'
else :
    return local_file_path


signature:
def _localize_inputs_command ( self , task_dir , inputs , user_project ) :
    """Returns a command that will stage inputs ."""

implementation:
commands = [ ]
for i in inputs :
    if i . recursive or not i . value :
        continue
    source_file_path = i . uri
    local_file_path = task_dir + '/' + _DATA_SUBDIR + '/' + i . docker_path
    dest_file_path = self . _get_input_target_path ( local_file_path )
    commands . append ( 'mkdir -p "%s"' % os . path . dirname ( local_file_path ) )
    if i . file_provider in [ job_model . P_LOCAL , job_model . P_GCS ] : # The semantics that we expect here are implemented consistently in
    # " gsutil cp " , and are a bit different than " cp " when it comes to
    # wildcard handling , so use it for both local and GCS :
    # - ` cp path / * dest / ` will error if " path " has subdirectories .
    # - ` cp " path / * " " dest / " ` will fail ( it expects wildcard expansion
    # to come from shell ) .
        if user_project :
            command = 'gsutil -u %s -mq cp "%s" "%s"' % ( user_project , source_file_path , dest_file_path )
        else :
            command = 'gsutil -mq cp "%s" "%s"' % ( source_file_path , dest_file_path )
        commands . append ( command )
return '\n' . join ( commands )


signature:
def _delocalize_outputs_commands ( self , task_dir , outputs , user_project ) :
    """Copy outputs from local disk to GCS ."""

implementation:
commands = [ ]
for o in outputs :
    if o . recursive or not o . value :
        continue
    # The destination path is o . uri . path , which is the target directory
    # ( rather than o . uri , which includes the filename or wildcard ) .
    dest_path = o . uri . path
    local_path = task_dir + '/' + _DATA_SUBDIR + '/' + o . docker_path
    if o . file_provider == job_model . P_LOCAL :
        commands . append ( 'mkdir -p "%s"' % dest_path )
    # Use gsutil even for local files ( explained in _ localize _ inputs _ command ) .
    if o . file_provider in [ job_model . P_LOCAL , job_model . P_GCS ] :
        if user_project :
            command = 'gsutil -u %s -mq cp "%s" "%s"' % ( user_project , local_path , dest_path )
        else :
            command = 'gsutil -mq cp "%s" "%s"' % ( local_path , dest_path )
        commands . append ( command )
return '\n' . join ( commands )


signature:
def get_dsub_version ( ) :
    """Get the dsub version out of the _ dsub _ version . py source file .
 Setup . py should not import dsub version from dsub directly since ambiguity in
 import order could lead to an old version of dsub setting the version number .
 Parsing the file directly is simpler than using import tools ( whose interface
 varies between python 2.7 , 3.4 , and 3.5 ) .
 Returns :
 string of dsub version .
 Raises :
 ValueError : if the version is not found ."""

implementation:
filename = os . path . join ( os . path . dirname ( __file__ ) , 'dsub/_dsub_version.py' )
with open ( filename , 'r' ) as versionfile :
    for line in versionfile :
        if line . startswith ( 'DSUB_VERSION =' ) : # Get the version then strip whitespace and quote characters .
            version = line . partition ( '=' ) [ 2 ]
            return version . strip ( ) . strip ( '\'"' )
raise ValueError ( 'Could not find version.' )


signature:
def get_filtered_normalized_events ( self ) :
    """Filter the granular v2 events down to events of interest .
 Filter through the large number of granular events returned by the
 pipelines API , and extract only those that are interesting to a user . This
 is implemented by filtering out events which are known to be uninteresting
 ( i . e . the default actions run for every job ) and by explicitly matching
 specific events which are interesting and mapping those to v1 style naming .
 Events which are not whitelisted or blacklisted will still be output ,
 meaning any events which are added in the future won ' t be masked .
 We don ' t want to suppress display of events that we don ' t recognize .
 They may be important .
 Returns :
 A list of maps containing the normalized , filtered events ."""

implementation:
# Need the user - image to look for the right " pulling image " event
user_image = google_v2_operations . get_action_image ( self . _op , _ACTION_USER_COMMAND )
# Only create an " ok " event for operations with SUCCESS status .
need_ok = google_v2_operations . is_success ( self . _op )
# Events are keyed by name for easier deletion .
events = { }
# Events are assumed to be ordered by timestamp ( newest to oldest ) .
for event in google_v2_operations . get_events ( self . _op ) :
    if self . _filter ( event ) :
        continue
    mapped , match = self . _map ( event )
    name = mapped [ 'name' ]
    if name == 'ok' : # If we want the " ok " event , we grab the first ( most recent ) .
        if not need_ok or 'ok' in events :
            continue
    if name == 'pulling-image' :
        if match . group ( 1 ) != user_image :
            continue
    events [ name ] = mapped
return sorted ( events . values ( ) , key = operator . itemgetter ( 'start-time' ) )


signature:
def _map ( self , event ) :
    """Extract elements from an operation event and map to a named event ."""

implementation:
description = event . get ( 'description' , '' )
start_time = google_base . parse_rfc3339_utc_string ( event . get ( 'timestamp' , '' ) )
for name , regex in _EVENT_REGEX_MAP . items ( ) :
    match = regex . match ( description )
    if match :
        return { 'name' : name , 'start-time' : start_time } , match
return { 'name' : description , 'start-time' : start_time } , None


signature:
def _get_logging_env ( self , logging_uri , user_project ) :
    """Returns the environment for actions that copy logging files ."""

implementation:
if not logging_uri . endswith ( '.log' ) :
    raise ValueError ( 'Logging URI must end in ".log": {}' . format ( logging_uri ) )
logging_prefix = logging_uri [ : - len ( '.log' ) ]
return { 'LOGGING_PATH' : '{}.log' . format ( logging_prefix ) , 'STDOUT_PATH' : '{}-stdout.log' . format ( logging_prefix ) , 'STDERR_PATH' : '{}-stderr.log' . format ( logging_prefix ) , 'USER_PROJECT' : user_project , }


signature:
def _get_prepare_env ( self , script , job_descriptor , inputs , outputs , mounts ) :
    """Return a dict with variables for the ' prepare ' action ."""

implementation:
# Add the _ SCRIPT _ REPR with the repr ( script ) contents
# Add the _ META _ YAML _ REPR with the repr ( meta ) contents
# Add variables for directories that need to be created , for example :
# DIR _ COUNT : 2
# DIR _ 0 : / mnt / data / input / gs / bucket / path1/
# DIR _ 1 : / mnt / data / output / gs / bucket / path2
# List the directories in sorted order so that they are created in that
# order . This is primarily to ensure that permissions are set as we create
# each directory .
# For example :
# mkdir - m 777 - p / root / first / second
# mkdir - m 777 - p / root / first
# * may * not actually set 777 on / root / first
docker_paths = sorted ( [ var . docker_path if var . recursive else os . path . dirname ( var . docker_path ) for var in inputs | outputs | mounts if var . value ] )
env = { _SCRIPT_VARNAME : repr ( script . value ) , _META_YAML_VARNAME : repr ( job_descriptor . to_yaml ( ) ) , 'DIR_COUNT' : str ( len ( docker_paths ) ) }
for idx , path in enumerate ( docker_paths ) :
    env [ 'DIR_{}' . format ( idx ) ] = os . path . join ( providers_util . DATA_MOUNT_POINT , path )
return env


signature:
def _get_localization_env ( self , inputs , user_project ) :
    """Return a dict with variables for the ' localization ' action ."""

implementation:
# Add variables for paths that need to be localized , for example :
# INPUT _ COUNT : 1
# INPUT _ 0 : MY _ INPUT _ FILE
# INPUT _ RECURSIVE _ 0 : 0
# INPUT _ SRC _ 0 : gs : / / mybucket / mypath / myfile
# INPUT _ DST _ 0 : / mnt / data / inputs / mybucket / mypath / myfile
non_empty_inputs = [ var for var in inputs if var . value ]
env = { 'INPUT_COUNT' : str ( len ( non_empty_inputs ) ) }
for idx , var in enumerate ( non_empty_inputs ) :
    env [ 'INPUT_{}' . format ( idx ) ] = var . name
    env [ 'INPUT_RECURSIVE_{}' . format ( idx ) ] = str ( int ( var . recursive ) )
    env [ 'INPUT_SRC_{}' . format ( idx ) ] = var . value
    # For wildcard paths , the destination must be a directory
    dst = os . path . join ( providers_util . DATA_MOUNT_POINT , var . docker_path )
    path , filename = os . path . split ( dst )
    if '*' in filename :
        dst = '{}/' . format ( path )
    env [ 'INPUT_DST_{}' . format ( idx ) ] = dst
env [ 'USER_PROJECT' ] = user_project
return env


signature:
def _get_delocalization_env ( self , outputs , user_project ) :
    """Return a dict with variables for the ' delocalization ' action ."""

implementation:
# Add variables for paths that need to be delocalized , for example :
# OUTPUT _ COUNT : 1
# OUTPUT _ 0 : MY _ OUTPUT _ FILE
# OUTPUT _ RECURSIVE _ 0 : 0
# OUTPUT _ SRC _ 0 : gs : / / mybucket / mypath / myfile
# OUTPUT _ DST _ 0 : / mnt / data / outputs / mybucket / mypath / myfile
non_empty_outputs = [ var for var in outputs if var . value ]
env = { 'OUTPUT_COUNT' : str ( len ( non_empty_outputs ) ) }
for idx , var in enumerate ( non_empty_outputs ) :
    env [ 'OUTPUT_{}' . format ( idx ) ] = var . name
    env [ 'OUTPUT_RECURSIVE_{}' . format ( idx ) ] = str ( int ( var . recursive ) )
    env [ 'OUTPUT_SRC_{}' . format ( idx ) ] = os . path . join ( providers_util . DATA_MOUNT_POINT , var . docker_path )
    # For wildcard paths , the destination must be a directory
    if '*' in var . uri . basename :
        dst = var . uri . path
    else :
        dst = var . uri
    env [ 'OUTPUT_DST_{}' . format ( idx ) ] = dst
env [ 'USER_PROJECT' ] = user_project
return env


signature:
def _build_user_environment ( self , envs , inputs , outputs , mounts ) :
    """Returns a dictionary of for the user container environment ."""

implementation:
envs = { env . name : env . value for env in envs }
envs . update ( providers_util . get_file_environment_variables ( inputs ) )
envs . update ( providers_util . get_file_environment_variables ( outputs ) )
envs . update ( providers_util . get_file_environment_variables ( mounts ) )
return envs


signature:
def _get_mount_actions ( self , mounts , mnt_datadisk ) :
    """Returns a list of two actions per gcs bucket to mount ."""

implementation:
actions_to_add = [ ]
for mount in mounts :
    bucket = mount . value [ len ( 'gs://' ) : ]
    mount_path = mount . docker_path
    actions_to_add . extend ( [ google_v2_pipelines . build_action ( name = 'mount-{}' . format ( bucket ) , flags = [ 'ENABLE_FUSE' , 'RUN_IN_BACKGROUND' ] , image_uri = _GCSFUSE_IMAGE , mounts = [ mnt_datadisk ] , commands = [ '--implicit-dirs' , '--foreground' , '-o ro' , bucket , os . path . join ( providers_util . DATA_MOUNT_POINT , mount_path ) ] ) , google_v2_pipelines . build_action ( name = 'mount-wait-{}' . format ( bucket ) , flags = [ 'ENABLE_FUSE' ] , image_uri = _GCSFUSE_IMAGE , mounts = [ mnt_datadisk ] , commands = [ 'wait' , os . path . join ( providers_util . DATA_MOUNT_POINT , mount_path ) ] ) ] )
return actions_to_add


signature:
def _build_pipeline_request ( self , task_view ) :
    """Returns a Pipeline objects for the task ."""

implementation:
job_metadata = task_view . job_metadata
job_params = task_view . job_params
job_resources = task_view . job_resources
task_metadata = task_view . task_descriptors [ 0 ] . task_metadata
task_params = task_view . task_descriptors [ 0 ] . task_params
task_resources = task_view . task_descriptors [ 0 ] . task_resources
# Set up VM - specific variables
mnt_datadisk = google_v2_pipelines . build_mount ( disk = _DATA_DISK_NAME , path = providers_util . DATA_MOUNT_POINT , read_only = False )
scopes = job_resources . scopes or google_base . DEFAULT_SCOPES
# Set up the task labels
labels = { label . name : label . value if label . value else '' for label in google_base . build_pipeline_labels ( job_metadata , task_metadata ) | job_params [ 'labels' ] | task_params [ 'labels' ] }
# Set local variables for the core pipeline values
script = task_view . job_metadata [ 'script' ]
user_project = task_view . job_metadata [ 'user-project' ] or ''
envs = job_params [ 'envs' ] | task_params [ 'envs' ]
inputs = job_params [ 'inputs' ] | task_params [ 'inputs' ]
outputs = job_params [ 'outputs' ] | task_params [ 'outputs' ]
mounts = job_params [ 'mounts' ]
gcs_mounts = param_util . get_gcs_mounts ( mounts )
persistent_disk_mount_params = param_util . get_persistent_disk_mounts ( mounts )
persistent_disks = [ google_v2_pipelines . build_disk ( name = disk . name . replace ( '_' , '-' ) , # Underscores not allowed
size_gb = disk . disk_size or job_model . DEFAULT_MOUNTED_DISK_SIZE , source_image = disk . value , disk_type = disk . disk_type or job_model . DEFAULT_DISK_TYPE ) for disk in persistent_disk_mount_params ]
persistent_disk_mounts = [ google_v2_pipelines . build_mount ( disk = persistent_disk . get ( 'name' ) , path = os . path . join ( providers_util . DATA_MOUNT_POINT , persistent_disk_mount_param . docker_path ) , read_only = True ) for persistent_disk , persistent_disk_mount_param in zip ( persistent_disks , persistent_disk_mount_params ) ]
# The list of " actions " ( 1 - based ) will be :
# 1 - continuous copy of log files off to Cloud Storage
# 2 - prepare the shared mount point ( write the user script )
# 3 - localize objects from Cloud Storage to block storage
# 4 - execute user command
# 5 - delocalize objects from block storage to Cloud Storage
# 6 - final copy of log files off to Cloud Storage
# If the user has requested an SSH server be started , it will be inserted
# after logging is started , and all subsequent action numbers above will be
# incremented by 1.
# If the user has requested to mount one or more buckets , two actions per
# bucket will be inserted after the prepare step , and all subsequent action
# numbers will be incremented by the number of actions added .
# We need to track the action numbers specifically for the user action and
# the final logging action .
optional_actions = 0
if job_resources . ssh :
    optional_actions += 1
mount_actions = self . _get_mount_actions ( gcs_mounts , mnt_datadisk )
optional_actions += len ( mount_actions )
user_action = 4 + optional_actions
final_logging_action = 6 + optional_actions
# Set up the commands and environment for the logging actions
logging_cmd = _LOGGING_CMD . format ( log_cp_fn = _GSUTIL_CP_FN , log_cp_cmd = _LOG_CP_CMD . format ( user_action = user_action , logging_action = 'logging_action' ) )
continuous_logging_cmd = _CONTINUOUS_LOGGING_CMD . format ( log_msg_fn = _LOG_MSG_FN , log_cp_fn = _GSUTIL_CP_FN , log_cp_cmd = _LOG_CP_CMD . format ( user_action = user_action , logging_action = 'continuous_logging_action' ) , final_logging_action = final_logging_action , log_interval = job_resources . log_interval or '60s' )
logging_env = self . _get_logging_env ( task_resources . logging_path . uri , user_project )
# Set up command and environments for the prepare , localization , user ,
# and de - localization actions
script_path = os . path . join ( providers_util . SCRIPT_DIR , script . name )
prepare_command = _PREPARE_CMD . format ( log_msg_fn = _LOG_MSG_FN , mk_runtime_dirs = _MK_RUNTIME_DIRS_CMD , script_var = _SCRIPT_VARNAME , python_decode_script = _PYTHON_DECODE_SCRIPT , script_path = script_path , mk_io_dirs = _MK_IO_DIRS )
prepare_env = self . _get_prepare_env ( script , task_view , inputs , outputs , mounts )
localization_env = self . _get_localization_env ( inputs , user_project )
user_environment = self . _build_user_environment ( envs , inputs , outputs , mounts )
delocalization_env = self . _get_delocalization_env ( outputs , user_project )
# Build the list of actions
actions = [ ]
actions . append ( google_v2_pipelines . build_action ( name = 'logging' , flags = 'RUN_IN_BACKGROUND' , image_uri = _CLOUD_SDK_IMAGE , environment = logging_env , entrypoint = '/bin/bash' , commands = [ '-c' , continuous_logging_cmd ] ) )
if job_resources . ssh :
    actions . append ( google_v2_pipelines . build_action ( name = 'ssh' , image_uri = _SSH_IMAGE , mounts = [ mnt_datadisk ] , entrypoint = 'ssh-server' , port_mappings = { _DEFAULT_SSH_PORT : _DEFAULT_SSH_PORT } , flags = 'RUN_IN_BACKGROUND' ) )
actions . append ( google_v2_pipelines . build_action ( name = 'prepare' , image_uri = _PYTHON_IMAGE , mounts = [ mnt_datadisk ] , environment = prepare_env , entrypoint = '/bin/bash' , commands = [ '-c' , prepare_command ] ) , )
actions . extend ( mount_actions )
actions . extend ( [ google_v2_pipelines . build_action ( name = 'localization' , image_uri = _CLOUD_SDK_IMAGE , mounts = [ mnt_datadisk ] , environment = localization_env , entrypoint = '/bin/bash' , commands = [ '-c' , _LOCALIZATION_CMD . format ( log_msg_fn = _LOG_MSG_FN , recursive_cp_fn = _GSUTIL_RSYNC_FN , cp_fn = _GSUTIL_CP_FN , cp_loop = _LOCALIZATION_LOOP ) ] ) , google_v2_pipelines . build_action ( name = 'user-command' , image_uri = job_resources . image , mounts = [ mnt_datadisk ] + persistent_disk_mounts , environment = user_environment , entrypoint = '/usr/bin/env' , commands = [ 'bash' , '-c' , _USER_CMD . format ( tmp_dir = providers_util . TMP_DIR , working_dir = providers_util . WORKING_DIR , user_script = script_path ) ] ) , google_v2_pipelines . build_action ( name = 'delocalization' , image_uri = _CLOUD_SDK_IMAGE , mounts = [ mnt_datadisk ] , environment = delocalization_env , entrypoint = '/bin/bash' , commands = [ '-c' , _LOCALIZATION_CMD . format ( log_msg_fn = _LOG_MSG_FN , recursive_cp_fn = _GSUTIL_RSYNC_FN , cp_fn = _GSUTIL_CP_FN , cp_loop = _DELOCALIZATION_LOOP ) ] ) , google_v2_pipelines . build_action ( name = 'final_logging' , flags = 'ALWAYS_RUN' , image_uri = _CLOUD_SDK_IMAGE , environment = logging_env , entrypoint = '/bin/bash' , commands = [ '-c' , logging_cmd ] ) , ] )
assert len ( actions ) - 2 == user_action
assert len ( actions ) == final_logging_action
# Prepare the VM ( resources ) configuration
disks = [ google_v2_pipelines . build_disk ( _DATA_DISK_NAME , job_resources . disk_size , source_image = None , disk_type = job_resources . disk_type or job_model . DEFAULT_DISK_TYPE ) ]
disks . extend ( persistent_disks )
network = google_v2_pipelines . build_network ( job_resources . network , job_resources . subnetwork , job_resources . use_private_address )
if job_resources . machine_type :
    machine_type = job_resources . machine_type
elif job_resources . min_cores or job_resources . min_ram :
    machine_type = GoogleV2CustomMachine . build_machine_type ( job_resources . min_cores , job_resources . min_ram )
else :
    machine_type = job_model . DEFAULT_MACHINE_TYPE
accelerators = None
if job_resources . accelerator_type :
    accelerators = [ google_v2_pipelines . build_accelerator ( job_resources . accelerator_type , job_resources . accelerator_count ) ]
service_account = google_v2_pipelines . build_service_account ( job_resources . service_account or 'default' , scopes )
resources = google_v2_pipelines . build_resources ( self . _project , job_resources . regions , google_base . get_zones ( job_resources . zones ) , google_v2_pipelines . build_machine ( network = network , machine_type = machine_type , preemptible = job_resources . preemptible , service_account = service_account , boot_disk_size_gb = job_resources . boot_disk_size , disks = disks , accelerators = accelerators , nvidia_driver_version = job_resources . nvidia_driver_version , labels = labels , cpu_platform = job_resources . cpu_platform ) , )
# Build the pipeline request
pipeline = google_v2_pipelines . build_pipeline ( actions , resources , None , job_resources . timeout )
return { 'pipeline' : pipeline , 'labels' : labels }


signature:
def submit_job ( self , job_descriptor , skip_if_output_present ) :
    """Submit the job ( or tasks ) to be executed .
 Args :
 job _ descriptor : all parameters needed to launch all job tasks
 skip _ if _ output _ present : ( boolean ) if true , skip tasks whose output
 is present ( see - - skip flag for more explanation ) .
 Returns :
 A dictionary containing the ' user - id ' , ' job - id ' , and ' task - id ' list .
 For jobs that are not task array jobs , the task - id list should be empty .
 Raises :
 ValueError : if job resources or task data contain illegal values ."""

implementation:
# Validate task data and resources .
param_util . validate_submit_args_or_fail ( job_descriptor , provider_name = _PROVIDER_NAME , input_providers = _SUPPORTED_INPUT_PROVIDERS , output_providers = _SUPPORTED_OUTPUT_PROVIDERS , logging_providers = _SUPPORTED_LOGGING_PROVIDERS )
# Prepare and submit jobs .
launched_tasks = [ ]
requests = [ ]
for task_view in job_model . task_view_generator ( job_descriptor ) :
    job_params = task_view . job_params
    task_params = task_view . task_descriptors [ 0 ] . task_params
    outputs = job_params [ 'outputs' ] | task_params [ 'outputs' ]
    if skip_if_output_present : # check whether the output ' s already there
        if dsub_util . outputs_are_present ( outputs ) :
            print ( 'Skipping task because its outputs are present' )
            continue
    request = self . _build_pipeline_request ( task_view )
    if self . _dry_run :
        requests . append ( request )
    else :
        task_id = self . _submit_pipeline ( request )
        launched_tasks . append ( task_id )
# If this is a dry - run , emit all the pipeline request objects
if self . _dry_run :
    print ( json . dumps ( requests , indent = 2 , sort_keys = True , separators = ( ',' , ': ' ) ) )
if not requests and not launched_tasks :
    return { 'job-id' : dsub_util . NO_JOB }
return { 'job-id' : job_descriptor . job_metadata [ 'job-id' ] , 'user-id' : job_descriptor . job_metadata [ 'user-id' ] , 'task-id' : [ task_id for task_id in launched_tasks if task_id ] , }


signature:
def _operations_list ( self , ops_filter , max_tasks , page_size , page_token ) :
    """Gets the list of operations for the specified filter .
 Args :
 ops _ filter : string filter of operations to return
 max _ tasks : the maximum number of job tasks to return or 0 for no limit .
 page _ size : the number of operations to requested on each list operation to
 the pipelines API ( if 0 or None , the API default is used )
 page _ token : page token returned by a previous _ operations _ list call .
 Returns :
 Operations matching the filter criteria ."""

implementation:
# We are not using the documented default page size of 256,
# nor allowing for the maximum page size of 2048 as larger page sizes
# currently cause the operations . list ( ) API to return an error :
# HttpError 429 . . . Resource has been exhausted ( e . g . check quota ) .
max_page_size = 128
# Set the page size to the smallest ( non - zero ) size we can
page_size = min ( sz for sz in [ page_size , max_page_size , max_tasks ] if sz )
# Execute operations . list ( ) and return all of the dsub operations
api = self . _service . projects ( ) . operations ( ) . list ( name = 'projects/{}/operations' . format ( self . _project ) , filter = ops_filter , pageToken = page_token , pageSize = page_size )
response = google_base . Api . execute ( api )
return [ GoogleOperation ( op ) for op in response . get ( 'operations' , [ ] ) if google_v2_operations . is_dsub_operation ( op ) ] , response . get ( 'nextPageToken' )


signature:
def lookup_job_tasks ( self , statuses , user_ids = None , job_ids = None , job_names = None , task_ids = None , task_attempts = None , labels = None , create_time_min = None , create_time_max = None , max_tasks = 0 , page_size = 0 ) :
    """Yields operations based on the input criteria .
 If any of the filters are empty or { ' * ' } , then no filtering is performed on
 that field . Filtering by both a job id list and job name list is
 unsupported .
 Args :
 statuses : { ' * ' } , or a list of job status strings to return . Valid
 status strings are ' RUNNING ' , ' SUCCESS ' , ' FAILURE ' , or ' CANCELED ' .
 user _ ids : a list of ids for the user ( s ) who launched the job .
 job _ ids : a list of job ids to return .
 job _ names : a list of job names to return .
 task _ ids : a list of specific tasks within the specified job ( s ) to return .
 task _ attempts : a list of specific attempts within the specified tasks ( s )
 to return .
 labels : a list of LabelParam with user - added labels . All labels must
 match the task being fetched .
 create _ time _ min : a timezone - aware datetime value for the earliest create
 time of a task , inclusive .
 create _ time _ max : a timezone - aware datetime value for the most recent
 create time of a task , inclusive .
 max _ tasks : the maximum number of job tasks to return or 0 for no limit .
 page _ size : the page size to use for each query to the pipelins API .
 Raises :
 ValueError : if both a job id list and a job name list are provided
 Yeilds :
 Genomics API Operations objects ."""

implementation:
# Build a filter for operations to return
ops_filter = self . _build_query_filter ( statuses , user_ids , job_ids , job_names , task_ids , task_attempts , labels , create_time_min , create_time_max )
# Execute the operations . list ( ) API to get batches of operations to yield
page_token = None
tasks_yielded = 0
while True : # If max _ tasks is set , let operations . list ( ) know not to send more than
# we need .
    max_to_fetch = None
    if max_tasks :
        max_to_fetch = max_tasks - tasks_yielded
    ops , page_token = self . _operations_list ( ops_filter , max_to_fetch , page_size , page_token )
    for op in ops :
        yield op
        tasks_yielded += 1
    assert ( max_tasks >= tasks_yielded or not max_tasks )
    if not page_token or 0 < max_tasks <= tasks_yielded :
        break


signature:
def _operation_status ( self ) :
    """Returns the status of this operation .
 Raises :
 ValueError : if the operation status cannot be determined .
 Returns :
 A printable status string ( RUNNING , SUCCESS , CANCELED or FAILURE ) ."""

implementation:
if not google_v2_operations . is_done ( self . _op ) :
    return 'RUNNING'
if google_v2_operations . is_success ( self . _op ) :
    return 'SUCCESS'
if google_v2_operations . is_canceled ( self . _op ) :
    return 'CANCELED'
if google_v2_operations . is_failed ( self . _op ) :
    return 'FAILURE'
raise ValueError ( 'Status for operation {} could not be determined' . format ( self . _op [ 'name' ] ) )


signature:
def _operation_status_message ( self ) :
    """Returns the most relevant status string and failed action .
 This string is meant for display only .
 Returns :
 A printable status string and name of failed action ( if any ) ."""

implementation:
msg = None
action = None
if not google_v2_operations . is_done ( self . _op ) :
    last_event = google_v2_operations . get_last_event ( self . _op )
    if last_event :
        msg = last_event [ 'description' ]
        action_id = last_event . get ( 'details' , { } ) . get ( 'actionId' )
        if action_id :
            action = google_v2_operations . get_action_by_id ( self . _op , action_id )
    else :
        msg = 'Pending'
else :
    failed_events = google_v2_operations . get_failed_events ( self . _op )
    if failed_events :
        failed_event = failed_events [ - 1 ]
        msg = failed_event . get ( 'details' , { } ) . get ( 'stderr' )
        action_id = failed_event . get ( 'details' , { } ) . get ( 'actionId' )
        if action_id :
            action = google_v2_operations . get_action_by_id ( self . _op , action_id )
    if not msg :
        error = google_v2_operations . get_error ( self . _op )
        if error :
            msg = error [ 'message' ]
        else :
            msg = 'Success'
return msg , action


signature:
def error_message ( self ) :
    """Returns an error message if the operation failed for any reason .
 Failure as defined here means ended for any reason other than ' success ' .
 This means that a successful cancelation will also return an error message .
 Returns :
 string , string will be empty if job did not error ."""

implementation:
error = google_v2_operations . get_error ( self . _op )
if error :
    job_id = self . get_field ( 'job-id' )
    task_id = self . get_field ( 'task-id' )
    task_str = job_id if task_id is None else '{} (task: {})' . format ( job_id , task_id )
    return 'Error in {} - code {}: {}' . format ( task_str , error [ 'code' ] , error [ 'message' ] )
return ''