[Notes] [Git][BuildGrid/buildgrid][mablanch/61-bazel-support] 3 commits: utils.py: New OutputDirectory maker helper



Title: GitLab

Martin Blanchard pushed to branch mablanch/61-bazel-support at BuildGrid / buildgrid

Commits:

3 changed files:

Changes:

  • buildgrid/_app/bots/temp_directory.py
    ... ... @@ -19,71 +19,94 @@ import tempfile
    19 19
     
    
    20 20
     from google.protobuf import any_pb2
    
    21 21
     
    
    22
    -from buildgrid.utils import read_file, create_digest, write_fetch_directory, parse_to_pb2_from_fetch
    
    23
    -from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
    
    22
    +from buildgrid.client.cas import upload
    
    23
    +from buildgrid._protos.build.bazel.remote.execution.v2 import remote_execution_pb2
    
    24 24
     from buildgrid._protos.google.bytestream import bytestream_pb2_grpc
    
    25
    +from buildgrid.utils import write_fetch_directory, parse_to_pb2_from_fetch
    
    26
    +from buildgrid.utils import output_file_maker, output_directory_maker
    
    25 27
     
    
    26 28
     
    
    27 29
     def work_temp_directory(context, lease):
    
    28
    -    """ Bot downloads directories and files into a temp directory,
    
    29
    -    then uploads results back to CAS
    
    30
    +    """Executes a lease for a build action, using host tools.
    
    30 31
         """
    
    31 32
     
    
    32
    -    parent = context.parent
    
    33 33
         stub_bytestream = bytestream_pb2_grpc.ByteStreamStub(context.cas_channel)
    
    34
    +    instance_name = context.parent
    
    35
    +    logger = context.logger
    
    34 36
     
    
    35 37
         action_digest = remote_execution_pb2.Digest()
    
    36 38
         lease.payload.Unpack(action_digest)
    
    37 39
     
    
    38
    -    action = remote_execution_pb2.Action()
    
    40
    +    action = parse_to_pb2_from_fetch(remote_execution_pb2.Action(),
    
    41
    +                                     stub_bytestream, action_digest, instance_name)
    
    39 42
     
    
    40
    -    action = parse_to_pb2_from_fetch(action, stub_bytestream, action_digest, parent)
    
    43
    +    with tempfile.TemporaryDirectory() as temp_directory:
    
    44
    +        command = parse_to_pb2_from_fetch(remote_execution_pb2.Command(),
    
    45
    +                                          stub_bytestream, action.command_digest, instance_name)
    
    41 46
     
    
    42
    -    with tempfile.TemporaryDirectory() as temp_dir:
    
    47
    +        write_fetch_directory(temp_directory, stub_bytestream,
    
    48
    +                              action.input_root_digest, instance_name)
    
    43 49
     
    
    44
    -        command = remote_execution_pb2.Command()
    
    45
    -        command = parse_to_pb2_from_fetch(command, stub_bytestream, action.command_digest, parent)
    
    46
    -
    
    47
    -        arguments = "cd {} &&".format(temp_dir)
    
    50
    +        environment = os.environ.copy()
    
    51
    +        for variable in command.environment_variables:
    
    52
    +            if variable.name not in ['PATH', 'PWD']:
    
    53
    +                environment[variable.name] = variable.value
    
    48 54
     
    
    55
    +        command_line = list()
    
    49 56
             for argument in command.arguments:
    
    50
    -            arguments += " {}".format(argument)
    
    51
    -
    
    52
    -        context.logger.info(arguments)
    
    53
    -
    
    54
    -        write_fetch_directory(temp_dir, stub_bytestream, action.input_root_digest, parent)
    
    55
    -
    
    56
    -        proc = subprocess.Popen(arguments,
    
    57
    -                                shell=True,
    
    58
    -                                stdin=subprocess.PIPE,
    
    59
    -                                stdout=subprocess.PIPE)
    
    60
    -
    
    61
    -        # TODO: Should return the std_out to the user
    
    62
    -        proc.communicate()
    
    63
    -
    
    64
    -        result = remote_execution_pb2.ActionResult()
    
    65
    -        requests = []
    
    66
    -        for output_file in command.output_files:
    
    67
    -            path = os.path.join(temp_dir, output_file)
    
    68
    -            chunk = read_file(path)
    
    69
    -
    
    70
    -            digest = create_digest(chunk)
    
    71
    -
    
    72
    -            result.output_files.extend([remote_execution_pb2.OutputFile(path=output_file,
    
    73
    -                                                                        digest=digest)])
    
    74
    -
    
    75
    -            requests.append(remote_execution_pb2.BatchUpdateBlobsRequest.Request(
    
    76
    -                digest=digest, data=chunk))
    
    77
    -
    
    78
    -        request = remote_execution_pb2.BatchUpdateBlobsRequest(instance_name=parent,
    
    79
    -                                                               requests=requests)
    
    80
    -
    
    81
    -        stub_cas = remote_execution_pb2_grpc.ContentAddressableStorageStub(context.cas_channel)
    
    82
    -        stub_cas.BatchUpdateBlobs(request)
    
    83
    -
    
    84
    -        result_any = any_pb2.Any()
    
    85
    -        result_any.Pack(result)
    
    86
    -
    
    87
    -        lease.result.CopyFrom(result_any)
    
    57
    +            command_line.append(argument.strip())
    
    58
    +
    
    59
    +        working_directory = None
    
    60
    +        if command.working_directory:
    
    61
    +            working_directory = os.path.join(temp_directory,
    
    62
    +                                             command.working_directory)
    
    63
    +            os.makedirs(working_directory, exist_ok=True)
    
    64
    +        else:
    
    65
    +            working_directory = temp_directory
    
    66
    +
    
    67
    +        # Ensure that output files structure exists:
    
    68
    +        for output_path in command.output_files:
    
    69
    +            directory_path = os.path.join(working_directory,
    
    70
    +                                          os.path.dirname(output_path))
    
    71
    +            os.makedirs(directory_path, exist_ok=True)
    
    72
    +
    
    73
    +        logger.debug(' '.join(command_line))
    
    74
    +
    
    75
    +        process = subprocess.Popen(command_line,
    
    76
    +                                   cwd=working_directory,
    
    77
    +                                   universal_newlines=True,
    
    78
    +                                   env=environment,
    
    79
    +                                   stdin=subprocess.PIPE,
    
    80
    +                                   stdout=subprocess.PIPE)
    
    81
    +        # TODO: Should return the stdout and stderr in the ActionResult.
    
    82
    +        process.communicate()
    
    83
    +
    
    84
    +        action_result = remote_execution_pb2.ActionResult()
    
    85
    +
    
    86
    +        with upload(context.cas_channel, instance=instance_name) as cas:
    
    87
    +            for output_path in command.output_files:
    
    88
    +                file_path = os.path.join(working_directory, output_path)
    
    89
    +                # Missing outputs should simply be omitted in ActionResult:
    
    90
    +                if not os.path.isfile(file_path):
    
    91
    +                    continue
    
    92
    +
    
    93
    +                output_file = output_file_maker(file_path, working_directory, cas=cas)
    
    94
    +                action_result.output_files.extend([output_file])
    
    95
    +
    
    96
    +            for output_path in command.output_directories:
    
    97
    +                directory_path = os.path.join(working_directory, output_path)
    
    98
    +                # Missing outputs should simply be omitted in ActionResult:
    
    99
    +                if not os.path.isdir(directory_path):
    
    100
    +                    continue
    
    101
    +
    
    102
    +                # OutputDirectory.path should be relative to the working direcory:
    
    103
    +                output_directory = output_directory_maker(directory_path, working_directory, cas=cas)
    
    104
    +
    
    105
    +                action_result.output_directories.extend([output_directory])
    
    106
    +
    
    107
    +        action_result_any = any_pb2.Any()
    
    108
    +        action_result_any.Pack(action_result)
    
    109
    +
    
    110
    +        lease.result.CopyFrom(action_result_any)
    
    88 111
     
    
    89 112
         return lease

  • buildgrid/server/execution/execution_service.py
    ... ... @@ -86,6 +86,11 @@ class ExecutionService(remote_execution_pb2_grpc.ExecutionServicer):
    86 86
                 yield operations_pb2.Operation()
    
    87 87
     
    
    88 88
         def _get_instance(self, name):
    
    89
    +        # If client does not support multiple instances, it may omit the
    
    90
    +        # instance name request parameter, so better map our default:
    
    91
    +        if not name and len(self._instances) == 1:
    
    92
    +            name = next(iter(self._instances))
    
    93
    +
    
    89 94
             try:
    
    90 95
                 return self._instances[name]
    
    91 96
     
    

  • buildgrid/utils.py
    ... ... @@ -145,6 +145,120 @@ def file_maker(file_path, file_digest):
    145 145
                                              is_executable=os.access(file_path, os.X_OK))
    
    146 146
     
    
    147 147
     
    
    148
    +def directory_maker(directory_path, child_directories=None, cas=None, upload_directories=True):
    
    149
    +    """Creates a :obj:`Directory` from a local directory and possibly upload it.
    
    150
    +
    
    151
    +    Args:
    
    152
    +        directory_path (str): absolute or relative path to a local directory.
    
    153
    +        child_directories (list): output list of of children :obj:`Directory`
    
    154
    +            objects.
    
    155
    +        cas (:obj:`Uploader`): a CAS client uploader.
    
    156
    +        upload_directories (bool): wheter or not to upload the :obj:`Directory`
    
    157
    +            objects along with the files.
    
    158
    +
    
    159
    +    Returns:
    
    160
    +        :obj:`Directory`, :obj:`Digest`: Tuple of a new gRPC :obj:`Directory`
    
    161
    +        for the local directory pointed by `directory_path` and the digest
    
    162
    +        for that object.
    
    163
    +    """
    
    164
    +    if not os.path.isabs(directory_path):
    
    165
    +        directory_path = os.path.abspath(directory_path)
    
    166
    +
    
    167
    +    files, directories, symlinks = list(), list(), list()
    
    168
    +    for directory_entry in os.scandir(directory_path):
    
    169
    +        # Create a FileNode and corresponding BatchUpdateBlobsRequest:
    
    170
    +        if directory_entry.is_file(follow_symlinks=False):
    
    171
    +            if cas is not None:
    
    172
    +                node_digest = cas.upload_file(directory_entry.path)
    
    173
    +            else:
    
    174
    +                node_digest = create_digest(read_file(directory_entry.path))
    
    175
    +
    
    176
    +            node = remote_execution_pb2.FileNode()
    
    177
    +            node.name = directory_entry.name
    
    178
    +            node.digest.CopyFrom(node_digest)
    
    179
    +            node.is_executable = os.access(directory_entry.path, os.X_OK)
    
    180
    +
    
    181
    +            files.append(node)
    
    182
    +
    
    183
    +        # Create a DirectoryNode and corresponding BatchUpdateBlobsRequest:
    
    184
    +        elif directory_entry.is_dir(follow_symlinks=False):
    
    185
    +            _, node_digest = directory_maker(directory_entry.path,
    
    186
    +                                             child_directories=child_directories,
    
    187
    +                                             upload_directories=upload_directories,
    
    188
    +                                             cas=cas)
    
    189
    +
    
    190
    +            node = remote_execution_pb2.DirectoryNode()
    
    191
    +            node.name = directory_entry.name
    
    192
    +            node.digest.CopyFrom(node_digest)
    
    193
    +
    
    194
    +            directories.append(node)
    
    195
    +
    
    196
    +        # Create a SymlinkNode if necessary;
    
    197
    +        elif os.path.islink(directory_entry.path):
    
    198
    +            node_target = os.readlink(directory_entry.path)
    
    199
    +
    
    200
    +            node = remote_execution_pb2.SymlinkNode()
    
    201
    +            node.name = directory_entry.name
    
    202
    +            node.target = node_target
    
    203
    +
    
    204
    +            symlinks.append(node)
    
    205
    +
    
    206
    +    files.sort(key=attrgetter('name'))
    
    207
    +    directories.sort(key=attrgetter('name'))
    
    208
    +    symlinks.sort(key=attrgetter('name'))
    
    209
    +
    
    210
    +    directory = remote_execution_pb2.Directory()
    
    211
    +    directory.files.extend(files)
    
    212
    +    directory.directories.extend(directories)
    
    213
    +    directory.symlinks.extend(symlinks)
    
    214
    +
    
    215
    +    if child_directories is not None:
    
    216
    +        child_directories.append(directory)
    
    217
    +
    
    218
    +    if cas is not None and upload_directories:
    
    219
    +        directory_digest = cas.upload_directory(directory)
    
    220
    +    else:
    
    221
    +        directory_digest = create_digest(directory.SerializeToString())
    
    222
    +
    
    223
    +    return directory, directory_digest
    
    224
    +
    
    225
    +
    
    226
    +def tree_maker(directory_path, cas=None):
    
    227
    +    """Creates a :obj:`Tree` from a local directory and possibly upload it.
    
    228
    +
    
    229
    +    If `cas` is specified, the local directory content will be uploded/stored
    
    230
    +    in remote CAS (the :obj:`Tree` message won't).
    
    231
    +
    
    232
    +    Args:
    
    233
    +        directory_path (str): absolute or relative path to a local directory.
    
    234
    +        cas (:obj:`Uploader`): a CAS client uploader.
    
    235
    +
    
    236
    +    Returns:
    
    237
    +        :obj:`Tree`, :obj:`Digest`: Tuple of a new gRPC :obj:`Tree` for the
    
    238
    +        local directory pointed by `directory_path` and the digest for that
    
    239
    +        object.
    
    240
    +    """
    
    241
    +    if not os.path.isabs(directory_path):
    
    242
    +        directory_path = os.path.abspath(directory_path)
    
    243
    +
    
    244
    +    child_directories = list()
    
    245
    +    directory, _ = directory_maker(directory_path,
    
    246
    +                                   child_directories=child_directories,
    
    247
    +                                   upload_directories=False,
    
    248
    +                                   cas=cas)
    
    249
    +
    
    250
    +    tree = remote_execution_pb2.Tree()
    
    251
    +    tree.children.extend(child_directories)
    
    252
    +    tree.root.CopyFrom(directory)
    
    253
    +
    
    254
    +    if cas is not None:
    
    255
    +        tree_digest = cas.send_message(tree)
    
    256
    +    else:
    
    257
    +        tree_digest = create_digest(tree.SerializeToString())
    
    258
    +
    
    259
    +    return tree, tree_digest
    
    260
    +
    
    261
    +
    
    148 262
     def read_file(file_path):
    
    149 263
         """Loads raw file content in memory.
    
    150 264
     
    
    ... ... @@ -196,3 +310,35 @@ def output_file_maker(file_path, input_path, cas=None):
    196 310
         output_file.is_executable = os.access(file_path, os.X_OK)
    
    197 311
     
    
    198 312
         return output_file
    
    313
    +
    
    314
    +
    
    315
    +def output_directory_maker(directory_path, working_path, cas=None):
    
    316
    +    """Creates an :obj:`OutputDirectory` from a local directory.
    
    317
    +
    
    318
    +    If `cas` is specified, the local directory content will be uploded/stored
    
    319
    +    in remote CAS (the :obj:`OutputDirectory` message won't).
    
    320
    +
    
    321
    +    Note:
    
    322
    +        `directory_path` **must** point inside or be relative to `input_path`.
    
    323
    +
    
    324
    +    Args:
    
    325
    +        directory_path (str): absolute or relative path to a local directory.
    
    326
    +        working_path (str): absolute or relative path to the working directory.
    
    327
    +        cas (:obj:`Uploader`): a CAS client uploader.
    
    328
    +
    
    329
    +    Returns:
    
    330
    +        :obj:`OutputDirectory`: a new gRPC :obj:`OutputDirectory` for the
    
    331
    +        directory pointed by `directory_path`.
    
    332
    +    """
    
    333
    +    if not os.path.isabs(directory_path):
    
    334
    +        directory_path = os.path.abspath(directory_path)
    
    335
    +    if not os.path.isabs(working_path):
    
    336
    +        working_path = os.path.abspath(working_path)
    
    337
    +
    
    338
    +    _, tree_digest = tree_maker(directory_path, cas=cas)
    
    339
    +
    
    340
    +    output_directory = remote_execution_pb2.OutputDirectory()
    
    341
    +    output_directory.tree_digest.CopyFrom(tree_digest)
    
    342
    +    output_directory.path = os.path.relpath(directory_path, start=working_path)
    
    343
    +
    
    344
    +    return output_directory



  • [Date Prev][Date Next]   [Thread Prev][Thread Next]   [Thread Index] [Date Index] [Author Index]