GNOME.org
 

[Notes] [Git][BuildStream/buildstream][jmac/remote_execution_rebase] 11 commits: cascache: pep8

Title: GitLab

Jim MacArthur pushed to branch jmac/remote_execution_rebase at BuildStream / buildstream

Commits:

5 changed files:

Changes:

  • buildstream/_artifactcache/cascache.py
    ... ... @@ -213,7 +213,6 @@ class CASCache(ArtifactCache):
    213 213 
                 remotes_for_project = self._remotes[element._get_project()]
    214 214 
                 return any(remote.spec.push for remote in remotes_for_project)
    215 215
    216 
    -
    217 216 
         def pull_key(self, key, size_bytes, project):
    218 217 
             """ Pull a single key rather than an artifact.
    219 218 
             Does not update local refs. """
    ... ... @@ -362,9 +361,9 @@ class CASCache(ArtifactCache):
    362 361 
             refs = [self.get_artifact_fullname(element, key) for key in keys]
    363 362
    364 363 
             project = element._get_project()
    365 
    -        return self.push_refs(element, refs, project)
    364 
    +        return self.push_refs(refs, project, element=element)
    366 365
    367 
    -    def push_refs(self, element, refs, project, may_have_dependencies=True):
    366 
    +    def push_refs(self, refs, project, may_have_dependencies=True, element=None):
    368 367
    369 368 
             push_remotes = [r for r in self._remotes[project] if r.spec.push]
    370 369
    ... ... @@ -374,7 +373,7 @@ class CASCache(ArtifactCache):
    374 373 
                 remote.init()
    375 374 
                 if self._push_refs_to_remote(refs, remote, may_have_dependencies):
    376 375 
                     pushed = True
    377 
    -            else:
    376 
    +            elif element:
    378 377 
                     self.context.message(Message(
    379 378 
                         None,
    380 379 
                         MessageType.SKIPPED,
    ... ... @@ -399,7 +398,6 @@ class CASCache(ArtifactCache):
    399 398 
             return pushed
    400 399
    401 400 
         def _verify_ref_on_remote(self, ref, remote):
    402 
    -        pushed = False
    403 401 
             tree = self.resolve_ref(ref)
    404 402
    405 403 
             # Check whether ref is already on the server in which case

  • buildstream/buildelement.py
    ... ... @@ -155,6 +155,9 @@ class BuildElement(Element):
    155 155 
                 command_dir = build_root
    156 156 
             sandbox.set_work_directory(command_dir)
    157 157
    158 
    +        # Tell sandbox which directory is preserved in the finished artifact
    159 
    +        sandbox.set_output_directory(install_root)
    160 
    +
    158 161 
             # Setup environment
    159 162 
             sandbox.set_environment(self.get_environment())
    160 163

  • buildstream/element.py
    ... ... @@ -1537,6 +1537,8 @@ class Element(Plugin):
    1537 1537 
                     finally:
    1538 1538 
                         if collect is not None:
    1539 1539 
                             try:
    1540 
    +                            # Sandbox will probably have replaced its virtual directory, so get it again
    1541 
    +                            sandbox_vroot = sandbox.get_virtual_directory()
    1540 1542 
                                 collectvdir = sandbox_vroot.descend(collect.lstrip(os.sep).split(os.sep))
    1541 1543 
                             except VirtualDirectoryError:
    1542 1544 
                                 # No collect directory existed

  • buildstream/sandbox/_sandboxremote.py
    1 
    +#!/usr/bin/env python3
    2 
    +#
    3 
    +#  Copyright (C) 2018 Codethink Limited
    4 
    +#
    5 
    +#  This program is free software; you can redistribute it and/or
    6 
    +#  modify it under the terms of the GNU Lesser General Public
    7 
    +#  License as published by the Free Software Foundation; either
    8 
    +#  version 2 of the License, or (at your option) any later version.
    9 
    +#
    10 
    +#  This library is distributed in the hope that it will be useful,
    11 
    +#  but WITHOUT ANY WARRANTY; without even the implied warranty of
    12 
    +#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.	 See the GNU
    13 
    +#  Lesser General Public License for more details.
    14 
    +#
    15 
    +#  You should have received a copy of the GNU Lesser General Public
    16 
    +#  License along with this library. If not, see <http://www.gnu.org/licenses/>.
    17 
    +#
    18 
    +#  Authors:
    19 
    +#        Jim MacArthur <jim macarthur codethink co uk>
    20 
    +
    21 
    +import os
    22 
    +import time
    23 
    +
    24 
    +import grpc
    25 
    +
    26 
    +from . import Sandbox
    27 
    +from ..storage._filebaseddirectory import FileBasedDirectory
    28 
    +from ..storage._casbaseddirectory import CasBasedDirectory
    29 
    +from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
    30 
    +from .._protos.google.longrunning import operations_pb2, operations_pb2_grpc
    31 
    +
    32 
    +from .._artifactcache.cascache import CASCache
    33 
    +
    34 
    +
    35 
    +class SandboxError(Exception):
    36 
    +    pass
    37 
    +
    38 
    +
    39 
    +# SandboxRemote()
    40 
    +#
    41 
    +# This isn't really a sandbox, it's a stub which sends all the source
    42 
    +# to a remote server and retrieves the results from it.
    43 
    +#
    44 
    +class SandboxRemote(Sandbox):
    45 
    +
    46 
    +    def __init__(self, *args, **kwargs):
    47 
    +        super().__init__(*args, **kwargs)
    48 
    +        self.user_ns_available = kwargs['user_ns_available']
    49 
    +        self.die_with_parent_available = kwargs['die_with_parent_available']
    50 
    +        self.cascache = None
    51 
    +
    52 
    +    def _get_cascache(self):
    53 
    +        if self.cascache is None:
    54 
    +            self.cascache = CASCache(self._get_context())
    55 
    +            self.cascache.setup_remotes(use_config=True)
    56 
    +        return self.cascache
    57 
    +
    58 
    +    def __run_remote_command(self, cascache, command, input_root_digest, environment):
    59 
    +        environment_variables = []
    60 
    +        for(k, v) in environment.items():
    61 
    +            environment_variables.append(remote_execution_pb2.Command.EnvironmentVariable(name=k, value=v))
    62 
    +        remote_command = remote_execution_pb2.Command(arguments=command, environment_variables=environment_variables)
    63 
    +
    64 
    +        # Serialise this into the cascache...
    65 
    +        command_digest = cascache.add_object(buffer=remote_command.SerializeToString())
    66 
    +
    67 
    +        command_ref = 'worker-command/{}'.format(command_digest.hash)
    68 
    +        cascache.set_ref(command_ref, command_digest)
    69 
    +
    70 
    +        command_push_successful = cascache.push_refs([command_ref], self._get_project(), may_have_dependencies=False)
    71 
    +        if command_push_successful or cascache.verify_key_pushed(command_ref, self._get_project()):
    72 
    +            # Next, try to create a communication channel
    73 
    +            port = 50051
    74 
    +            channel = grpc.insecure_channel('dekatron.office.codethink.co.uk:{}'.format(port))
    75 
    +            stub = remote_execution_pb2_grpc.ExecutionStub(channel)
    76 
    +            ops_stub = operations_pb2_grpc.OperationsStub(channel)
    77 
    +
    78 
    +            # Having done that, create and send the action.
    79 
    +
    80 
    +            action = remote_execution_pb2.Action(command_digest=command_digest,
    81 
    +                                                 input_root_digest=input_root_digest,
    82 
    +                                                 output_files=[],
    83 
    +                                                 output_directories=[self._output_directory],
    84 
    +                                                 platform=None,
    85 
    +                                                 timeout=None,
    86 
    +                                                 do_not_cache=True)
    87 
    +
    88 
    +            request = remote_execution_pb2.ExecuteRequest(instance_name='default',
    89 
    +                                                          action=action,
    90 
    +                                                          skip_cache_lookup=True)
    91 
    +
    92 
    +            operation = stub.Execute(request)  # Returns Operation
    93 
    +            job_name = operation.name
    94 
    +        else:
    95 
    +            # Source push failed
    96 
    +            return None
    97 
    +        while True:
    98 
    +            # TODO: Timeout
    99 
    +            # Refresh the operation data periodically using the name
    100 
    +            request = operations_pb2.GetOperationRequest(name=job_name)
    101 
    +            operation = ops_stub.GetOperation(request)
    102 
    +            time.sleep(1)
    103 
    +            if operation.done:
    104 
    +                break
    105 
    +        return operation
    106 
    +
    107 
    +    def process_job_output(self, output_directories, output_files):
    108 
    +        # output_directories is an array of OutputDirectory objects.
    109 
    +        # output_files is an array of OutputFile objects.
    110 
    +        #
    111 
    +        # We only specify one output_directory, so it's an error
    112 
    +        # for there to be any output files or more than one directory at the moment.
    113 
    +
    114 
    +        if output_files:
    115 
    +            raise SandboxError("Output files were returned when we didn't request any.")
    116 
    +        elif len(output_directories) > 1:
    117 
    +            error_text = "More than one output directory was returned from the build server: {}"
    118 
    +            raise SandboxError(error_text.format(output_directories))
    119 
    +
    120 
    +        digest = output_directories[0].tree_digest
    121 
    +        if digest is None or digest.hash is None or digest.hash == "":
    122 
    +            raise SandboxError("Output directory structure had no digest attached.")
    123 
    +
    124 
    +        # Now do a pull to ensure we have the necessary parts.
    125 
    +        cascache = self._get_cascache()
    126 
    +        cascache.pull_key(digest.hash, digest.size_bytes, self._get_project())
    127 
    +        path_components = os.path.split(self._output_directory)
    128 
    +
    129 
    +        # Now what we have is a digest for the output. Once we return, the calling process will
    130 
    +        # attempt to descend into our directory and find that directory, so we need to overwrite
    131 
    +        # that.
    132 
    +
    133 
    +        if not path_components:
    134 
    +            # The artifact wants the whole directory; we could just return the returned hash in its
    135 
    +            # place, but we don't have a means to do that yet.
    136 
    +            raise SandboxError("Unimplemented: Output directory is empty or equal to the sandbox root.")
    137 
    +
    138 
    +        # At the moment, we will get the whole directory back in the first directory argument and we need
    139 
    +        # to replace the sandbox's virtual directory with that. Creating a new virtual directory object
    140 
    +        # from another hash will be interesting, though...
    141 
    +
    142 
    +        new_dir = CasBasedDirectory(self._get_context(), ref=digest)
    143 
    +        self.set_virtual_directory(new_dir)
    144 
    +
    145 
    +    def run(self, command, flags, *, cwd=None, env=None):
    146 
    +        # Upload sources
    147 
    +        upload_vdir = self.get_virtual_directory()
    148 
    +        if isinstance(upload_vdir, FileBasedDirectory):
    149 
    +            # Make a new temporary directory to put source in
    150 
    +            upload_vdir = CasBasedDirectory(self._get_context(), ref=None)
    151 
    +            upload_vdir.import_files(self.get_virtual_directory().get_underlying_directory())
    152 
    +
    153 
    +        # Now, push that key (without necessarily needing a ref) to the remote.
    154 
    +        cascache = self._get_cascache()
    155 
    +
    156 
    +        ref = 'worker-source/{}'.format(upload_vdir.ref.hash)
    157 
    +        upload_vdir._save(ref)
    158 
    +        source_push_successful = cascache.push_refs([ref], self._get_project())
    159 
    +        # Fallback to the sandbox default settings for
    160 
    +        # the cwd and environment.
    161 
    +
    162 
    +        if env is None:
    163 
    +            env = self._get_environment()
    164 
    +
    165 
    +        # We want command args as a list of strings
    166 
    +        if isinstance(command, str):
    167 
    +            command = [command]
    168 
    +
    169 
    +        # Now transmit the command to execute
    170 
    +        if source_push_successful or cascache.verify_key_pushed(ref, self._get_project()):
    171 
    +            response = self.__run_remote_command(cascache, command, upload_vdir.ref, env)
    172 
    +
    173 
    +            if response is None or response.HasField("error"):
    174 
    +                # Build failed, so return a failure code
    175 
    +                return 1
    176 
    +            else:
    177 
    +
    178 
    +                # At the moment, response can either be an
    179 
    +                # ExecutionResponse containing an ActionResult, or an
    180 
    +                # ActionResult directly.
    181 
    +                executeResponse = remote_execution_pb2.ExecuteResponse()
    182 
    +                if response.response.Is(executeResponse.DESCRIPTOR):
    183 
    +                    # Unpack ExecuteResponse and set response to its response
    184 
    +                    response.response.Unpack(executeResponse)
    185 
    +                    response = executeResponse
    186 
    +
    187 
    +                actionResult = remote_execution_pb2.ActionResult()
    188 
    +                if response.response.Is(actionResult.DESCRIPTOR):
    189 
    +                    response.response.Unpack(actionResult)
    190 
    +                    self.process_job_output(actionResult.output_directories, actionResult.output_files)
    191 
    +                else:
    192 
    +                    raise SandboxError("Received unknown message from server (expected ExecutionResponse).")
    193 
    +        else:
    194 
    +            raise SandboxError("Failed to verify that source has been pushed to the remote artifact cache.")
    195 
    +        return 0

  • buildstream/sandbox/sandbox.py
    ... ... @@ -99,9 +99,11 @@ class Sandbox():
    99 99 
             self.__stdout = kwargs['stdout']
    100 100 
             self.__stderr = kwargs['stderr']
    101 101
    102 
    -        # Setup the directories. Root should be available to subclasses, hence
    103 
    -        # being single-underscore. The others are private to this class.
    102 
    +        # Setup the directories. Root and output_directory should be
    103 
    +        # available to subclasses, hence being single-underscore. The
    104 
    +        # others are private to this class.
    104 105 
             self._root = os.path.join(directory, 'root')
    106 
    +        self._output_directory = None
    105 107 
             self.__directory = directory
    106 108 
             self.__scratch = os.path.join(self.__directory, 'scratch')
    107 109 
             for directory_ in [self._root, self.__scratch]:
    ... ... @@ -142,11 +144,29 @@ class Sandbox():
    142 144 
                     self._vdir = FileBasedDirectory(self._root)
    143 145 
             return self._vdir
    144 146
    147 
    +    def set_virtual_directory(self, vdir):
    148 
    +        """ Sets virtual directory. Useful after remote execution
    149 
    +        has rewritten the working directory. """
    150 
    +        self._vdir = vdir
    151 
    +
    152 
    +    def get_virtual_toplevel_directory(self):
    153 
    +        """Fetches the sandbox's toplevel directory
    154 
    +
    155 
    +        The toplevel directory contains 'root', 'scratch' and later
    156 
    +        'artifact' where output is copied to.
    157 
    +
    158 
    +        Returns:
    159 
    +           (str): The sandbox toplevel directory
    160 
    +
    161 
    +        """
    162 
    +        # For now, just create a new Directory every time we're asked
    163 
    +        return FileBasedDirectory(self.__directory)
    164 
    +
    145 165 
         def set_environment(self, environment):
    146 166 
             """Sets the environment variables for the sandbox
    147 167
    148 168 
             Args:
    149 
    -           directory (dict): The environment variables to use in the sandbox
    169 
    +           environment (dict): The environment variables to use in the sandbox
    150 170 
             """
    151 171 
             self.__env = environment
    152 172
    ... ... @@ -158,6 +178,15 @@ class Sandbox():
    158 178 
             """
    159 179 
             self.__cwd = directory
    160 180
    181 
    +    def set_output_directory(self, directory):
    182 
    +        """Sets the output directory - the directory which is preserved
    183 
    +        as an artifact after assembly.
    184 
    +
    185 
    +        Args:
    186 
    +           directory (str): An absolute path within the sandbox
    187 
    +        """
    188 
    +        self._output_directory = directory
    189 
    +
    161 190 
         def mark_directory(self, directory, *, artifact=False):
    162 191 
             """Marks a sandbox directory and ensures it will exist
    163 192

    • —
    View it on GitLab.
    You're receiving this email because of your account on gitlab.com. If you'd like to receive fewer emails, you can adjust your notification settings.

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