[Notes] [Git][BuildStream/buildstream][jmac/remote_execution_client] 14 commits: cascache.py: Modifications necessary for remote execution

[Martin Blanchard <gitlab mg gitlab com>]

Title: GitLab

Martin Blanchard pushed to branch jmac/remote_execution_client at BuildStream / buildstream

Commits:

• 10032dc8
  by Jim MacArthur at 2018-09-04T11:08:26Z

  cascache.py: Modifications necessary for remote execution
  
  Add helper method for object pushing and pulling: push_directory() and
  pull_tree() and a verifier verify_key_pushed().
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• a788daf5
  by Martin Blanchard at 2018-09-04T11:08:28Z

  _casbaseddirectory.py: Add a method for hash recalculation
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• 29d79445
  by Jim MacArthur at 2018-09-04T11:08:28Z

  sandbox.py: Allow setting the virtual directory
  
  This is for use after remote execution has finished, since remote
  execution produces a new output directory rather than modifying
  the initial directory.
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• 1e543f12
  by Jim MacArthur at 2018-09-04T11:08:28Z

  Add "remote-execution" project configuration option.
  
  This just adds one option, "remote-execution/url". Affects multiple files.
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• f286e267
  by Jim MacArthur at 2018-09-04T11:08:28Z

  format_project.rst: Document remote-execution option
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• 97d821f3
  by Jim MacArthur at 2018-09-04T11:08:28Z

  _sandboxremote.py: Implement the REAPI client
  
  The remote execution client is implemented as a remote sandbox that
  sends sources and build commands to a REAPI server and fetches results
  once remotely executed. New file.
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• f1791579
  by Jim MacArthur at 2018-09-04T11:08:28Z

  sandbox/__init__.py: Add SandboxRemote
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• e7fdb96e
  by Jim MacArthur at 2018-09-04T11:08:28Z

  element.py: Get the updated virtual directory after running
  
  Executing run() on a sandbox can now replace the virtual directory,
  since remote execution returns a potentially different directory rather
  than an update to the existing one. Call get_virtual_directory() again
  after running to accout for this.
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• 5787aeff
  by Jim MacArthur at 2018-09-04T11:08:28Z

  element.py: Switch to SandboxRemote if config option is set
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• ef92a11f
  by Jim MacArthur at 2018-09-04T11:08:28Z

  autotools.py: Mark this as a BST_VIRTUAL_DIRECTORY plugin
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• 40ad5608
  by Martin Blanchard at 2018-09-04T11:08:28Z

  cmake.py: Mark this as a BST_VIRTUAL_DIRECTORY plugin
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• b6f4cb20
  by Martin Blanchard at 2018-09-04T11:08:28Z

  make.py:‧Mark‧this‧as‧a‧BST_VIRTUAL_DIRECTORY‧plugin
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• 111214d4
  by Martin Blanchard at 2018-09-04T11:08:28Z

  meson.py:‧Mark‧this‧as‧a‧BST_VIRTUAL_DIRECTORY‧plugin
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

• c571b9da
  by Martin Blanchard at 2018-09-04T11:08:28Z

  qmake.py:‧Mark‧this‧as‧a‧BST_VIRTUAL_DIRECTORY‧plugin
  
  https://gitlab.com/BuildStream/buildstream/issues/454
  

15 changed files:

• buildstream/_artifactcache/cascache.py
• buildstream/_project.py
• buildstream/buildelement.py
• buildstream/data/projectconfig.yaml
• buildstream/element.py
• buildstream/plugins/elements/autotools.py
• buildstream/plugins/elements/cmake.py
• buildstream/plugins/elements/make.py
• buildstream/plugins/elements/meson.py
• buildstream/plugins/elements/qmake.py
• buildstream/sandbox/__init__.py
• + buildstream/sandbox/_sandboxremote.py
• buildstream/sandbox/sandbox.py
• buildstream/storage/_casbaseddirectory.py
• doc/source/format_project.rst

Changes:

buildstream/_artifactcache/cascache.py

---

...	...	@@ -19,6 +19,7 @@
19	19
20	20	import hashlib
21	21	import itertools
	22	+import io
22	23	import multiprocessing
23	24	import os
24	25	import signal
...	...	@@ -66,6 +67,10 @@ class CASCache(ArtifactCache):
66	67	self._calculate_cache_quota()
67	68
68	69	self._enable_push = enable_push
	70	+ if self._enable_push:
	71	+ self._uuid = str(uuid.uuid4())
	72	+ else:
	73	+ self._uuid = None
69	74
70	75	# Per-project list of _CASRemote instances.
71	76	self._remotes = {}
...	...	@@ -76,6 +81,7 @@ class CASCache(ArtifactCache):
76	81	################################################
77	82	# Implementation of abstract methods #
78	83	################################################
	84	+
79	85	def contains(self, element, key):
80	86	refpath = self._refpath(self.get_artifact_fullname(element, key))
81	87
...	...	@@ -259,6 +265,25 @@ class CASCache(ArtifactCache):
259	265
260	266	return False
261	267
	268	+ def pull_tree(self, project, digest):
	269	+ """ Pull a single Tree rather than an artifact.
	270	+ Does not update local refs. """
	271	+
	272	+ for remote in self._remotes[project]:
	273	+ try:
	274	+ remote.init()
	275	+
	276	+ digest = self._fetch_tree(remote, digest)
	277	+
	278	+ # no need to pull from additional remotes
	279	+ return digest
	280	+
	281	+ except grpc.RpcError as e:
	282	+ if e.code() != grpc.StatusCode.NOT_FOUND:
	283	+ raise
	284	+
	285	+ return None
	286	+
262	287	def link_key(self, element, oldkey, newkey):
263	288	oldref = self.get_artifact_fullname(element, oldkey)
264	289	newref = self.get_artifact_fullname(element, newkey)
...	...	@@ -267,8 +292,46 @@ class CASCache(ArtifactCache):
267	292
268	293	self.set_ref(newref, tree)
269	294
	295	+ def _push_refs_to_remote(self, refs, remote):
	296	+ skipped_remote = True
	297	+ try:
	298	+ for ref in refs:
	299	+ tree = self.resolve_ref(ref)
	300	+
	301	+ # Check whether ref is already on the server in which case
	302	+ # there is no need to push the artifact
	303	+ try:
	304	+ request = buildstream_pb2.GetReferenceRequest()
	305	+ request.key = ref
	306	+ response = remote.ref_storage.GetReference(request)
	307	+
	308	+ if response.digest.hash == tree.hash and response.digest.size_bytes == tree.size_bytes:
	309	+ # ref is already on the server with the same tree
	310	+ continue
	311	+
	312	+ except grpc.RpcError as e:
	313	+ if e.code() != grpc.StatusCode.NOT_FOUND:
	314	+ # Intentionally re-raise RpcError for outer except block.
	315	+ raise
	316	+
	317	+ self._send_directory(remote, tree)
	318	+
	319	+ request = buildstream_pb2.UpdateReferenceRequest()
	320	+ request.keys.append(ref)
	321	+ request.digest.hash = tree.hash
	322	+ request.digest.size_bytes = tree.size_bytes
	323	+ remote.ref_storage.UpdateReference(request)
	324	+
	325	+ skipped_remote = False
	326	+ except grpc.RpcError as e:
	327	+ if e.code() != grpc.StatusCode.RESOURCE_EXHAUSTED:
	328	+ raise ArtifactError("Failed to push artifact {}: {}".format(refs, e), temporary=True) from e
	329	+
	330	+ return not skipped_remote
	331	+
270	332	def push(self, element, keys):
271		- refs = [self.get_artifact_fullname(element, key) for key in keys]
	333	+
	334	+ refs = [self.get_artifact_fullname(element, key) for key in list(keys)]
272	335
273	336	project = element._get_project()
274	337
...	...	@@ -278,95 +341,77 @@ class CASCache(ArtifactCache):
278	341
279	342	for remote in push_remotes:
280	343	remote.init()
281		- skipped_remote = True
282		- element.info("Pushing {} -> {}".format(element._get_brief_display_key(), remote.spec.url))
283		-
284		- try:
285		- for ref in refs:
286		- tree = self.resolve_ref(ref)
287		-
288		- # Check whether ref is already on the server in which case
289		- # there is no need to push the artifact
290		- try:
291		- request = buildstream_pb2.GetReferenceRequest()
292		- request.key = ref
293		- response = remote.ref_storage.GetReference(request)
294		-
295		- if response.digest.hash == tree.hash and response.digest.size_bytes == tree.size_bytes:
296		- # ref is already on the server with the same tree
297		- continue
298		-
299		- except grpc.RpcError as e:
300		- if e.code() != grpc.StatusCode.NOT_FOUND:
301		- # Intentionally re-raise RpcError for outer except block.
302		- raise
303		-
304		- missing_blobs = {}
305		- required_blobs = self._required_blobs(tree)
306		-
307		- # Limit size of FindMissingBlobs request
308		- for required_blobs_group in _grouper(required_blobs, 512):
309		- request = remote_execution_pb2.FindMissingBlobsRequest()
310		-
311		- for required_digest in required_blobs_group:
312		- d = request.blob_digests.add()
313		- d.hash = required_digest.hash
314		- d.size_bytes = required_digest.size_bytes
315		-
316		- response = remote.cas.FindMissingBlobs(request)
317		- for digest in response.missing_blob_digests:
318		- d = remote_execution_pb2.Digest()
319		- d.hash = digest.hash
320		- d.size_bytes = digest.size_bytes
321		- missing_blobs[d.hash] = d
322		-
323		- # Upload any blobs missing on the server
324		- skipped_remote = False
325		- for digest in missing_blobs.values():
326		- uuid_ = uuid.uuid4()
327		- resource_name = '/'.join(['uploads', str(uuid_), 'blobs',
328		- digest.hash, str(digest.size_bytes)])
329		-
330		- def request_stream(resname):
331		- with open(self.objpath(digest), 'rb') as f:
332		- assert os.fstat(f.fileno()).st_size == digest.size_bytes
333		- offset = 0
334		- finished = False
335		- remaining = digest.size_bytes
336		- while not finished:
337		- chunk_size = min(remaining, 64 * 1024)
338		- remaining -= chunk_size
339		-
340		- request = bytestream_pb2.WriteRequest()
341		- request.write_offset = offset
342		- # max. 64 kB chunks
343		- request.data = f.read(chunk_size)
344		- request.resource_name = resname
345		- request.finish_write = remaining <= 0
346		- yield request
347		- offset += chunk_size
348		- finished = request.finish_write
349		- response = remote.bytestream.Write(request_stream(resource_name))
350		-
351		- request = buildstream_pb2.UpdateReferenceRequest()
352		- request.keys.append(ref)
353		- request.digest.hash = tree.hash
354		- request.digest.size_bytes = tree.size_bytes
355		- remote.ref_storage.UpdateReference(request)
356		-
357		- pushed = True
358	344
359		- except grpc.RpcError as e:
360		- if e.code() != grpc.StatusCode.RESOURCE_EXHAUSTED:
361		- raise ArtifactError("Failed to push artifact {}: {}".format(refs, e), temporary=True) from e
	345	+ element.info("Pushing {} -> {}".format(element._get_brief_display_key(), remote.spec.url))
362	346
363		- if skipped_remote:
	347	+ if self._push_refs_to_remote(refs, remote):
	348	+ pushed = True
	349	+ else:
364	350	self.context.message(Message(
365	351	None,
366	352	MessageType.SKIPPED,
367	353	"Remote ({}) already has {} cached".format(
368	354	remote.spec.url, element._get_brief_display_key())
369	355	))
	356	+
	357	+ return pushed
	358	+
	359	+ def push_directory(self, project, directory):
	360	+
	361	+ push_remotes = [r for r in self._remotes[project] if r.spec.push]
	362	+
	363	+ if directory.ref is None:
	364	+ return None
	365	+
	366	+ for remote in push_remotes:
	367	+ remote.init()
	368	+
	369	+ self._send_directory(remote, directory.ref)
	370	+
	371	+ return directory.ref
	372	+
	373	+ def push_message(self, project, message):
	374	+
	375	+ push_remotes = [r for r in self._remotes[project] if r.spec.push]
	376	+
	377	+ message_buffer = message.SerializeToString()
	378	+ message_sha = hashlib.sha256(message_buffer)
	379	+ message_digest = remote_execution_pb2.Digest()
	380	+ message_digest.hash = message_sha.hexdigest()
	381	+ message_digest.size_bytes = len(message_buffer)
	382	+
	383	+ for remote in push_remotes:
	384	+ remote.init()
	385	+
	386	+ with io.BytesIO(message_buffer) as b:
	387	+ self._send_blob(remote, message_digest, b)
	388	+
	389	+ return message_digest
	390	+
	391	+ def _verify_digest_on_remote(self, remote, digest):
	392	+ # Check whether ref is already on the server in which case
	393	+ # there is no need to push the artifact
	394	+ request = remote_execution_pb2.FindMissingBlobsRequest()
	395	+ request.blob_digests.extend([digest])
	396	+
	397	+ response = remote.cas.FindMissingBlobs(request)
	398	+ if digest in response.missing_blob_digests:
	399	+ return False
	400	+
	401	+ return True
	402	+
	403	+ def verify_key_pushed(self, project, digest):
	404	+
	405	+ push_remotes = [r for r in self._remotes[project] if r.spec.push]
	406	+
	407	+ pushed = False
	408	+
	409	+ for remote in push_remotes:
	410	+ remote.init()
	411	+
	412	+ if self._verify_digest_on_remote(remote, digest):
	413	+ pushed = True
	414	+
370	415	return pushed
371	416
372	417	################################################
...	...	@@ -599,6 +644,7 @@ class CASCache(ArtifactCache):
599	644	################################################
600	645	# Local Private Methods #
601	646	################################################
	647	+
602	648	def _checkout(self, dest, tree):
603	649	os.makedirs(dest, exist_ok=True)
604	650
...	...	@@ -782,16 +828,16 @@ class CASCache(ArtifactCache):
782	828	for dirnode in directory.directories:
783	829	yield from self._required_blobs(dirnode.digest)
784	830
785		- def _fetch_blob(self, remote, digest, out):
	831	+ def _fetch_blob(self, remote, digest, stream):
786	832	resource_name = '/'.join(['blobs', digest.hash, str(digest.size_bytes)])
787	833	request = bytestream_pb2.ReadRequest()
788	834	request.resource_name = resource_name
789	835	request.read_offset = 0
790	836	for response in remote.bytestream.Read(request):
791		- out.write(response.data)
	837	+ stream.write(response.data)
	838	+ stream.flush()
792	839
793		- out.flush()
794		- assert digest.size_bytes == os.fstat(out.fileno()).st_size
	840	+ assert digest.size_bytes == os.fstat(stream.fileno()).st_size
795	841
796	842	def _fetch_directory(self, remote, tree):
797	843	objpath = self.objpath(tree)
...	...	@@ -827,6 +873,92 @@ class CASCache(ArtifactCache):
827	873	digest = self.add_object(path=out.name)
828	874	assert digest.hash == tree.hash
829	875
	876	+ def _fetch_tree(self, remote, digest):
	877	+ # download but do not store the Tree object
	878	+ with tempfile.NamedTemporaryFile(dir=self.tmpdir) as out:
	879	+ self._fetch_blob(remote, digest, out)
	880	+
	881	+ tree = remote_execution_pb2.Tree()
	882	+
	883	+ with open(out.name, 'rb') as f:
	884	+ tree.ParseFromString(f.read())
	885	+
	886	+ tree.children.extend([tree.root])
	887	+ for directory in tree.children:
	888	+ for filenode in directory.files:
	889	+ fileobjpath = self.objpath(filenode.digest)
	890	+ if os.path.exists(fileobjpath):
	891	+ # already in local cache
	892	+ continue
	893	+
	894	+ with tempfile.NamedTemporaryFile(dir=self.tmpdir) as f:
	895	+ self._fetch_blob(remote, filenode.digest, f)
	896	+
	897	+ added_digest = self.add_object(path=f.name)
	898	+ assert added_digest.hash == filenode.digest.hash
	899	+
	900	+ # place directory blob only in final location when we've downloaded
	901	+ # all referenced blobs to avoid dangling references in the repository
	902	+ dirbuffer = directory.SerializeToString()
	903	+ dirdigest = self.add_object(buffer=dirbuffer)
	904	+ assert dirdigest.size_bytes == len(dirbuffer)
	905	+
	906	+ return dirdigest
	907	+
	908	+ def _send_blob(self, remote, digest, stream):
	909	+ resource_name = '/'.join(['uploads', self._uuid, 'blobs',
	910	+ digest.hash, str(digest.size_bytes)])
	911	+
	912	+ def request_stream(resname, instream):
	913	+ offset = 0
	914	+ finished = False
	915	+ remaining = digest.size_bytes
	916	+ while not finished:
	917	+ chunk_size = min(remaining, 64 * 1024)
	918	+ remaining -= chunk_size
	919	+
	920	+ request = bytestream_pb2.WriteRequest()
	921	+ request.write_offset = offset
	922	+ # max. 64 kB chunks
	923	+ request.data = instream.read(chunk_size)
	924	+ request.resource_name = resname
	925	+ request.finish_write = remaining <= 0
	926	+
	927	+ yield request
	928	+
	929	+ offset += chunk_size
	930	+ finished = request.finish_write
	931	+
	932	+ response = remote.bytestream.Write(request_stream(resource_name, stream))
	933	+
	934	+ assert response.committed_size == digest.size_bytes
	935	+
	936	+ def _send_directory(self, remote, digest):
	937	+ required_blobs = self._required_blobs(digest)
	938	+
	939	+ missing_blobs = dict()
	940	+ # Limit size of FindMissingBlobs request
	941	+ for required_blobs_group in _grouper(required_blobs, 512):
	942	+ request = remote_execution_pb2.FindMissingBlobsRequest()
	943	+
	944	+ for required_digest in required_blobs_group:
	945	+ d = request.blob_digests.add()
	946	+ d.hash = required_digest.hash
	947	+ d.size_bytes = required_digest.size_bytes
	948	+
	949	+ response = remote.cas.FindMissingBlobs(request)
	950	+ for missing_digest in response.missing_blob_digests:
	951	+ d = remote_execution_pb2.Digest()
	952	+ d.hash = missing_digest.hash
	953	+ d.size_bytes = missing_digest.size_bytes
	954	+ missing_blobs[d.hash] = d
	955	+
	956	+ # Upload any blobs missing on the server
	957	+ for blob_digest in missing_blobs.values():
	958	+ with open(self.objpath(blob_digest), 'rb') as f:
	959	+ assert os.fstat(f.fileno()).st_size == blob_digest.size_bytes
	960	+ self._send_blob(remote, blob_digest, f)
	961	+
830	962
831	963	# Represents a single remote CAS cache.
832	964	#

buildstream/_project.py

---

...	...	@@ -129,6 +129,7 @@ class Project():
129	129
130	130	self.artifact_cache_specs = None
131	131	self._sandbox = None
	132	+ self._remote_execution = None
132	133	self._splits = None
133	134
134	135	self._context.add_project(self)
...	...	@@ -471,7 +472,7 @@ class Project():
471	472	'aliases', 'name',
472	473	'artifacts', 'options',
473	474	'fail-on-overlap', 'shell', 'fatal-warnings',
474		- 'ref-storage', 'sandbox', 'mirrors'
	475	+ 'ref-storage', 'sandbox', 'mirrors', 'remote-execution'
475	476	])
476	477
477	478	#
...	...	@@ -489,6 +490,9 @@ class Project():
489	490	# Load sandbox configuration
490	491	self._sandbox = _yaml.node_get(config, Mapping, 'sandbox')
491	492
	493	+ # Load remote execution configuration
	494	+ self._remote_execution = _yaml.node_get(config, Mapping, 'remote-execution')
	495	+
492	496	# Load project split rules
493	497	self._splits = _yaml.node_get(config, Mapping, 'split-rules')
494	498

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/data/projectconfig.yaml

---

...	...	@@ -204,3 +204,6 @@ shell:
204	204	# Command to run when `bst shell` does not provide a command
205	205	#
206	206	command: [ 'sh', '-i' ]
	207	+
	208	+remote-execution:
	209	+ url: ""
		\ No newline at end of file

buildstream/element.py

---

...	...	@@ -95,6 +95,7 @@ from . import _site
95	95	from ._platform import Platform
96	96	from .plugin import CoreWarnings
97	97	from .sandbox._config import SandboxConfig
	98	+from .sandbox._sandboxremote import SandboxRemote
98	99
99	100	from .storage.directory import Directory
100	101	from .storage._filebaseddirectory import FileBasedDirectory
...	...	@@ -250,6 +251,9 @@ class Element(Plugin):
250	251	# Extract Sandbox config
251	252	self.__sandbox_config = self.__extract_sandbox_config(meta)
252	253
	254	+ # Extract remote execution URL
	255	+ self.__remote_execution_url = self.__extract_remote_execution_config(meta)
	256	+
253	257	def __lt__(self, other):
254	258	return self.name < other.name
255	259
...	...	@@ -1570,6 +1574,8 @@ class Element(Plugin):
1570	1574	finally:
1571	1575	if collect is not None:
1572	1576	try:
	1577	+ # Sandbox will probably have replaced its virtual directory, so get it again
	1578	+ sandbox_vroot = sandbox.get_virtual_directory()
1573	1579	collectvdir = sandbox_vroot.descend(collect.lstrip(os.sep).split(os.sep))
1574	1580	except VirtualDirectoryError:
1575	1581	# No collect directory existed
...	...	@@ -2146,7 +2152,32 @@ class Element(Plugin):
2146	2152	project = self._get_project()
2147	2153	platform = Platform.get_platform()
2148	2154
2149		- if directory is not None and os.path.exists(directory):
	2155	+ if self.__remote_execution_url and self.BST_VIRTUAL_DIRECTORY:
	2156	+ if not self.__artifacts.has_push_remotes(element=self):
	2157	+ # Give an early warning if remote execution will not work
	2158	+ raise ElementError("Artifact {} is configured to use remote execution but has no push remotes. "
	2159	+ .format(self.name) +
	2160	+ "The remote artifact server(s) may not be correctly configured or contactable.")
	2161	+
	2162	+ self.info("Using a remote sandbox for artifact {}".format(self.name))
	2163	+
	2164	+ sandbox = SandboxRemote(context, project,
	2165	+ directory,
	2166	+ stdout=stdout,
	2167	+ stderr=stderr,
	2168	+ config=config,
	2169	+ server_url=self.__remote_execution_url,
	2170	+ allow_real_directory=False)
	2171	+ yield sandbox
	2172	+
	2173	+ elif directory is not None and os.path.exists(directory):
	2174	+ if self.__remote_execution_url:
	2175	+ self.warn("Artifact {} is configured to use remote execution but element plugin does not support it."
	2176	+ .format(self.name), detail="Element plugin '{kind}' does not support virtual directories."
	2177	+ .format(kind=self.get_kind()), warning_token="remote-failure")
	2178	+
	2179	+ self.info("Falling back to local sandbox for artifact {}".format(self.name))
	2180	+
2150	2181	sandbox = platform.create_sandbox(context, project,
2151	2182	directory,
2152	2183	stdout=stdout,
...	...	@@ -2318,6 +2349,20 @@ class Element(Plugin):
2318	2349	return SandboxConfig(self.node_get_member(sandbox_config, int, 'build-uid'),
2319	2350	self.node_get_member(sandbox_config, int, 'build-gid'))
2320	2351
	2352	+ # Remote execution configuration data (server URL), to be used by the remote sandbox.
	2353	+ #
	2354	+ def __extract_remote_execution_config(self, meta):
	2355	+ if self.__is_junction:
	2356	+ return None
	2357	+ else:
	2358	+ project = self._get_project()
	2359	+ project.ensure_fully_loaded()
	2360	+ if project._remote_execution:
	2361	+ rexec_config = _yaml.node_chain_copy(project._remote_execution)
	2362	+ return self.node_get_member(rexec_config, str, 'url')
	2363	+ else:
	2364	+ return None
	2365	+
2321	2366	# This makes a special exception for the split rules, which
2322	2367	# elements may extend but whos defaults are defined in the project.
2323	2368	#

buildstream/plugins/elements/autotools.py

---

...	...	@@ -57,7 +57,8 @@ from buildstream import BuildElement
57	57
58	58	# Element implementation for the 'autotools' kind.
59	59	class AutotoolsElement(BuildElement):
60		- pass
	60	+ # Supports virtual directories (required for remote execution)
	61	+ BST_VIRTUAL_DIRECTORY = True
61	62
62	63
63	64	# Plugin entry point

buildstream/plugins/elements/cmake.py

---

...	...	@@ -56,7 +56,8 @@ from buildstream import BuildElement
56	56
57	57	# Element implementation for the 'cmake' kind.
58	58	class CMakeElement(BuildElement):
59		- pass
	59	+ # Supports virtual directories (required for remote execution)
	60	+ BST_VIRTUAL_DIRECTORY = True
60	61
61	62
62	63	# Plugin entry point

buildstream/plugins/elements/make.py

---

...	...	@@ -38,7 +38,8 @@ from buildstream import BuildElement
38	38
39	39	# Element implementation for the 'make' kind.
40	40	class MakeElement(BuildElement):
41		- pass
	41	+ # Supports virtual directories (required for remote execution)
	42	+ BST_VIRTUAL_DIRECTORY = True
42	43
43	44
44	45	# Plugin entry point

buildstream/plugins/elements/meson.py

---

...	...	@@ -53,7 +53,8 @@ from buildstream import BuildElement
53	53
54	54	# Element implementation for the 'meson' kind.
55	55	class MesonElement(BuildElement):
56		- pass
	56	+ # Supports virtual directories (required for remote execution)
	57	+ BST_VIRTUAL_DIRECTORY = True
57	58
58	59
59	60	# Plugin entry point

buildstream/plugins/elements/qmake.py

---

...	...	@@ -33,7 +33,8 @@ from buildstream import BuildElement
33	33
34	34	# Element implementation for the 'qmake' kind.
35	35	class QMakeElement(BuildElement):
36		- pass
	36	+ # Supports virtual directories (required for remote execution)
	37	+ BST_VIRTUAL_DIRECTORY = True
37	38
38	39
39	40	# Plugin entry point

buildstream/sandbox/__init__.py

---

...	...	@@ -20,3 +20,4 @@
20	20	from .sandbox import Sandbox, SandboxFlags
21	21	from ._sandboxchroot import SandboxChroot
22	22	from ._sandboxbwrap import SandboxBwrap
	23	+from ._sandboxremote import SandboxRemote

buildstream/sandbox/_sandboxremote.py

---

	1	+#!/usr/bin/env python3
	2	+#
	3	+# Copyright (C) 2018 Bloomberg LP
	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 re
	23	+from urllib.parse import urlparse
	24	+
	25	+import grpc
	26	+
	27	+from . import Sandbox
	28	+from ..storage._filebaseddirectory import FileBasedDirectory
	29	+from ..storage._casbaseddirectory import CasBasedDirectory
	30	+from .._protos.build.bazel.remote.execution.v2 import remote_execution_pb2, remote_execution_pb2_grpc
	31	+from .._artifactcache.cascache import CASCache
	32	+
	33	+
	34	+class SandboxError(Exception):
	35	+ pass
	36	+
	37	+
	38	+# SandboxRemote()
	39	+#
	40	+# This isn't really a sandbox, it's a stub which sends all the sources and build
	41	+# commands to a remote server and retrieves the results from it.
	42	+#
	43	+class SandboxRemote(Sandbox):
	44	+
	45	+ def __init__(self, *args, **kwargs):
	46	+ super().__init__(*args, **kwargs)
	47	+ self.cascache = None
	48	+
	49	+ url = urlparse(kwargs['server_url'])
	50	+ if not url.scheme or not url.hostname or not url.port:
	51	+ raise SandboxError("Configured remote URL '{}' does not match the expected layout. "
	52	+ .format(self.server_url) +
	53	+ "It should be of the form <protocol>://<domain name>:<port>.")
	54	+ elif url.scheme != 'http':
	55	+ raise SandboxError("Configured remote '{}' uses an unsupported protocol. "
	56	+ "Only plain HTTP is currenlty supported (no HTTPS).")
	57	+
	58	+ self.server_url = '{}:{}'.format(url.hostname, url.port)
	59	+
	60	+ def _get_cascache(self):
	61	+ if self.cascache is None:
	62	+ self.cascache = CASCache(self._get_context())
	63	+ self.cascache.setup_remotes(use_config=True)
	64	+ return self.cascache
	65	+
	66	+ def run_remote_command(self, command, input_root_digest, working_directory, environment):
	67	+ # Sends an execution request to the remote execution server.
	68	+ #
	69	+ # This function blocks until it gets a response from the server.
	70	+ #
	71	+ environment_variables = [remote_execution_pb2.Command.
	72	+ EnvironmentVariable(name=k, value=v)
	73	+ for (k, v) in environment.items()]
	74	+
	75	+ # Create and send the Command object.
	76	+ remote_command = remote_execution_pb2.Command(arguments=command,
	77	+ working_directory=working_directory,
	78	+ environment_variables=environment_variables,
	79	+ output_files=[],
	80	+ output_directories=[self._output_directory],
	81	+ platform=None)
	82	+
	83	+ cascache = self._get_cascache()
	84	+ # Upload the Command message to the remote CAS server
	85	+ command_digest = cascache.push_message(self._get_project(), remote_command)
	86	+ if not command_digest or not cascache.verify_key_pushed(self._get_project(), command_digest):
	87	+ # Command push failed
	88	+ return None
	89	+
	90	+ # Create and send the action.
	91	+ action = remote_execution_pb2.Action(command_digest=command_digest,
	92	+ input_root_digest=input_root_digest,
	93	+ timeout=None,
	94	+ do_not_cache=True)
	95	+
	96	+ # Upload the Action message to the remote CAS server
	97	+ action_digest = cascache.push_message(self._get_project(), action)
	98	+ if not action_digest or not cascache.verify_key_pushed(self._get_project(), action_digest):
	99	+ # Action push failed
	100	+ return None
	101	+
	102	+ # Next, try to create a communication channel to the BuildGrid server.
	103	+ channel = grpc.insecure_channel(self.server_url)
	104	+ stub = remote_execution_pb2_grpc.ExecutionStub(channel)
	105	+ request = remote_execution_pb2.ExecuteRequest(action_digest=action_digest,
	106	+ skip_cache_lookup=True)
	107	+ try:
	108	+ operation_iterator = stub.Execute(request)
	109	+ except grpc.RpcError:
	110	+ return None
	111	+
	112	+ operation = None
	113	+ with self._get_context().timed_activity("Waiting for the remote build to complete"):
	114	+ # It is advantageous to check operation_iterator.code() is grpc.StatusCode.OK here,
	115	+ # which will check the server is actually contactable. However, calling it when the
	116	+ # server is available seems to cause .code() to hang forever.
	117	+ for operation in operation_iterator:
	118	+ if operation.done:
	119	+ break
	120	+
	121	+ return operation
	122	+
	123	+ def process_job_output(self, output_directories, output_files):
	124	+ # Reads the remote execution server response to an execution request.
	125	+ #
	126	+ # output_directories is an array of OutputDirectory objects.
	127	+ # output_files is an array of OutputFile objects.
	128	+ #
	129	+ # We only specify one output_directory, so it's an error
	130	+ # for there to be any output files or more than one directory at the moment.
	131	+ #
	132	+ if output_files:
	133	+ raise SandboxError("Output files were returned when we didn't request any.")
	134	+ elif not output_directories:
	135	+ error_text = "No output directory was returned from the build server."
	136	+ raise SandboxError(error_text)
	137	+ elif len(output_directories) > 1:
	138	+ error_text = "More than one output directory was returned from the build server: {}."
	139	+ raise SandboxError(error_text.format(output_directories))
	140	+
	141	+ tree_digest = output_directories[0].tree_digest
	142	+ if tree_digest is None or not tree_digest.hash:
	143	+ raise SandboxError("Output directory structure had no digest attached.")
	144	+
	145	+ cascache = self._get_cascache()
	146	+ # Now do a pull to ensure we have the necessary parts.
	147	+ dir_digest = cascache.pull_tree(self._get_project(), tree_digest)
	148	+ if dir_digest is None or not dir_digest.hash or not dir_digest.size_bytes:
	149	+ raise SandboxError("Output directory structure pulling from remote failed.")
	150	+
	151	+ path_components = os.path.split(self._output_directory)
	152	+
	153	+ # Now what we have is a digest for the output. Once we return, the calling process will
	154	+ # attempt to descend into our directory and find that directory, so we need to overwrite
	155	+ # that.
	156	+
	157	+ if not path_components:
	158	+ # The artifact wants the whole directory; we could just return the returned hash in its
	159	+ # place, but we don't have a means to do that yet.
	160	+ raise SandboxError("Unimplemented: Output directory is empty or equal to the sandbox root.")
	161	+
	162	+ # At the moment, we will get the whole directory back in the first directory argument and we need
	163	+ # to replace the sandbox's virtual directory with that. Creating a new virtual directory object
	164	+ # from another hash will be interesting, though...
	165	+
	166	+ new_dir = CasBasedDirectory(self._get_context(), ref=dir_digest)
	167	+ self._set_virtual_directory(new_dir)
	168	+
	169	+ def run(self, command, flags, *, cwd=None, env=None):
	170	+ # Upload sources
	171	+ upload_vdir = self.get_virtual_directory()
	172	+
	173	+ if isinstance(upload_vdir, FileBasedDirectory):
	174	+ # Make a new temporary directory to put source in
	175	+ upload_vdir = CasBasedDirectory(self._get_context(), ref=None)
	176	+ upload_vdir.import_files(self.get_virtual_directory()._get_underlying_directory())
	177	+
	178	+ upload_vdir.recalculate_hash()
	179	+
	180	+ cascache = self._get_cascache()
	181	+ # Now, push that key (without necessarily needing a ref) to the remote.
	182	+ vdir_digest = cascache.push_directory(self._get_project(), upload_vdir)
	183	+ if not vdir_digest or not cascache.verify_key_pushed(self._get_project(), vdir_digest):
	184	+ raise SandboxError("Failed to verify that source has been pushed to the remote artifact cache.")
	185	+
	186	+ # Set up environment and working directory
	187	+ if cwd is None:
	188	+ cwd = self._get_work_directory()
	189	+
	190	+ if cwd is None:
	191	+ cwd = '/'
	192	+
	193	+ if env is None:
	194	+ env = self._get_environment()
	195	+
	196	+ # We want command args as a list of strings
	197	+ if isinstance(command, str):
	198	+ command = [command]
	199	+
	200	+ # Now transmit the command to execute
	201	+ operation = self.run_remote_command(command, upload_vdir.ref, cwd, env)
	202	+
	203	+ if operation is None:
	204	+ # Failure of remote execution, usually due to an error in BuildStream
	205	+ # NB This error could be raised in __run_remote_command
	206	+ raise SandboxError("No response returned from server")
	207	+
	208	+ assert(not operation.HasField('error') and operation.HasField('response'))
	209	+
	210	+ execution_response = remote_execution_pb2.ExecuteResponse()
	211	+ # The response is expected to be an ExecutionResponse message
	212	+ assert(operation.response.Is(execution_response.DESCRIPTOR))
	213	+
	214	+ operation.response.Unpack(execution_response)
	215	+
	216	+ if execution_response.status.code != 0:
	217	+ # A normal error during the build: the remote execution system
	218	+ # has worked correctly but the command failed.
	219	+ # execution_response.error also contains 'message' (str) and
	220	+ # 'details' (iterator of Any) which we ignore at the moment.
	221	+ return execution_response.status.code
	222	+
	223	+ action_result = execution_response.result
	224	+
	225	+ self.process_job_output(action_result.output_directories, action_result.output_files)
	226	+
	227	+ 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]:
...	...	@@ -144,11 +146,17 @@ class Sandbox():
144	146	self._vdir = FileBasedDirectory(self._root)
145	147	return self._vdir
146	148
	149	+ def _set_virtual_directory(self, virtual_directory):
	150	+ """ Sets virtual directory. Useful after remote execution
	151	+ has rewritten the working directory.
	152	+ """
	153	+ self._vdir = virtual_directory
	154	+
147	155	def set_environment(self, environment):
148	156	"""Sets the environment variables for the sandbox
149	157
150	158	Args:
151		- directory (dict): The environment variables to use in the sandbox
	159	+ environment (dict): The environment variables to use in the sandbox
152	160	"""
153	161	self.__env = environment
154	162
...	...	@@ -160,6 +168,15 @@ class Sandbox():
160	168	"""
161	169	self.__cwd = directory
162	170
	171	+ def set_output_directory(self, directory):
	172	+ """Sets the output directory - the directory which is preserved
	173	+ as an artifact after assembly.
	174	+
	175	+ Args:
	176	+ directory (str): An absolute path within the sandbox
	177	+ """
	178	+ self._output_directory = directory
	179	+
163	180	def mark_directory(self, directory, *, artifact=False):
164	181	"""Marks a sandbox directory and ensures it will exist
165	182

buildstream/storage/_casbaseddirectory.py

---

...	...	@@ -543,6 +543,15 @@ class CasBasedDirectory(Directory):
543	543	filelist.append(k)
544	544	return filelist
545	545
	546	+ def recalculate_hash(self):
	547	+ """ Recalcuates the hash for this directory and store the results in
	548	+ the cache. If this directory has a parent, tell it to
	549	+ recalculate (since changing this directory changes an entry in
	550	+ the parent). Hashes for subdirectories also get recalculated.
	551	+ """
	552	+ self._recalculate_recursing_up()
	553	+ self._recalculate_recursing_down()
	554	+
546	555	def _get_identifier(self):
547	556	path = ""
548	557	if self.parent:

doc/source/format_project.rst

---

...	...	@@ -204,6 +204,24 @@ with an artifact share.
204	204	You can also specify a list of caches here; earlier entries in the list
205	205	will have higher priority than later ones.
206	206
	207	+Remote execution
	208	+~~~~~~~~~~~~~~~~
	209	+BuildStream supports remote execution using the Google Remote Execution API
	210	+(REAPI). A description of how remote execution works is beyond the scope
	211	+of this document, but you can specify a remote server complying with the REAPI
	212	+using the `remote-execution` option:
	213	+
	214	+.. code:: yaml
	215	+
	216	+ remote-execution:
	217	+
	218	+ # A url defining a remote execution server
	219	+ url: http://buildserver.example.com:50051
	220	+
	221	+The url should contain a hostname and port separated by ':'. Only plain HTTP is
	222	+currently suported (no HTTPS).
	223	+
	224	+The Remote Execution API can be found via https://github.com/bazelbuild/remote-apis.
207	225
208	226	.. _project_essentials_mirrors:
209	227

—
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.