You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@buildstream.apache.org by ju...@apache.org on 2024/05/31 14:47:35 UTC
(buildstream) branch juerg/buildbox-asset-remote created (now c1defd0db)
This is an automated email from the ASF dual-hosted git repository.
juergbi pushed a change to branch juerg/buildbox-asset-remote
in repository https://gitbox.apache.org/repos/asf/buildstream.git
at c1defd0db _assetcache.py: Use buildbox-casd as remote asset proxy, if supported
This branch includes the following new commits:
new 6e947605d _cas: Merge `CASDChannel` into `CASDProcessManager`
new e1fbeaebf Move `CASDProcessManager` instance from `CASCache` to `Context`
new 1c58181b5 _remotespec.py: Add `to_localcas_remote()` method
new c1defd0db _assetcache.py: Use buildbox-casd as remote asset proxy, if supported
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(buildstream) 01/04: _cas: Merge `CASDChannel` into `CASDProcessManager`
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
juergbi pushed a commit to branch juerg/buildbox-asset-remote
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 6e947605d6af1e55f3c4d77db5553713755c67e9
Author: Jürg Billeter <j...@bitron.ch>
AuthorDate: Fri May 31 12:32:18 2024 +0200
_cas: Merge `CASDChannel` into `CASDProcessManager`
There is only ever a single instance of `CASDChannel` and it's always
created right after creating `CASDProcessManager`. This is in
preparation of using buildbox-casd outside of `CASCache`.
---
src/buildstream/_cas/cascache.py | 48 ++++++---------
src/buildstream/_cas/casdprocessmanager.py | 83 +++++++-------------------
src/buildstream/_cas/casserver.py | 15 ++---
src/buildstream/_stream.py | 2 +-
src/buildstream/sandbox/_sandboxbuildboxrun.py | 4 +-
tests/artifactcache/pull.py | 3 -
6 files changed, 48 insertions(+), 107 deletions(-)
diff --git a/src/buildstream/_cas/cascache.py b/src/buildstream/_cas/cascache.py
index 42937c1d6..8110337bc 100644
--- a/src/buildstream/_cas/cascache.py
+++ b/src/buildstream/_cas/cascache.py
@@ -80,17 +80,15 @@ class CASCache:
self._remote_cache = bool(remote_cache_spec)
- self._casd_process_manager = None
- self._casd_channel = None
+ self._casd = None
if casd:
assert log_directory is not None, "log_directory is required when casd is True"
log_dir = os.path.join(log_directory, "_casd")
- self._casd_process_manager = CASDProcessManager(
+ self._casd = CASDProcessManager(
path, log_dir, log_level, cache_quota, remote_cache_spec, protect_session_blobs, messenger
)
- self._casd_channel = self._casd_process_manager.create_channel()
- self._cache_usage_monitor = _CASCacheUsageMonitor(self._casd_channel)
+ self._cache_usage_monitor = _CASCacheUsageMonitor(self._casd)
self._cache_usage_monitor.start()
else:
assert not self._remote_cache
@@ -103,16 +101,16 @@ class CASCache:
# Return ContentAddressableStorage stub for buildbox-casd channel.
#
def get_cas(self):
- assert self._casd_channel, "CASCache was created without a channel"
- return self._casd_channel.get_cas()
+ assert self._casd, "CASCache was created without buildbox-casd"
+ return self._casd.get_cas()
# get_local_cas():
#
# Return LocalCAS stub for buildbox-casd channel.
#
def get_local_cas(self):
- assert self._casd_channel, "CASCache was created without a channel"
- return self._casd_channel.get_local_cas()
+ assert self._casd, "CASCache was created without buildbox-casd"
+ return self._casd.get_local_cas()
# preflight():
#
@@ -122,30 +120,18 @@ class CASCache:
if not os.path.join(self.casdir, "objects"):
raise CASCacheError("CAS repository check failed for '{}'".format(self.casdir))
- # close_grpc_channels():
- #
- # Close the casd channel if it exists
- #
- def close_grpc_channels(self):
- if self._casd_channel:
- self._casd_channel.close()
-
# release_resources():
#
# Release resources used by CASCache.
#
def release_resources(self, messenger=None):
- if self._casd_channel:
- self._casd_channel.request_shutdown()
-
if self._cache_usage_monitor:
self._cache_usage_monitor.stop()
self._cache_usage_monitor.join()
- if self._casd_process_manager:
- self.close_grpc_channels()
- self._casd_process_manager.release_resources(messenger)
- self._casd_process_manager = None
+ if self._casd:
+ self._casd.release_resources(messenger)
+ self._casd = None
def get_default_remote(self):
return self._default_remote
@@ -733,16 +719,16 @@ class CASCache:
assert not self._cache_usage_monitor_forbidden
return self._cache_usage_monitor.get_cache_usage()
- # get_casd_process_manager()
+ # get_casd()
#
# Get the underlying buildbox-casd process
#
# Returns:
# (subprocess.Process): The casd process that is used for the current cascache
#
- def get_casd_process_manager(self):
- assert self._casd_process_manager is not None, "Only call this with a running buildbox-casd process"
- return self._casd_process_manager
+ def get_casd(self):
+ assert self._casd is not None, "Only call this with a running buildbox-casd process"
+ return self._casd
# _CASCacheUsage
@@ -783,9 +769,9 @@ class _CASCacheUsage:
# buildbox-casd.
#
class _CASCacheUsageMonitor(threading.Thread):
- def __init__(self, connection):
+ def __init__(self, casd):
super().__init__()
- self._connection = connection
+ self._casd = casd
self._disk_usage = None
self._disk_quota = None
self._should_stop = False
@@ -797,7 +783,7 @@ class _CASCacheUsageMonitor(threading.Thread):
self._should_stop = True
def run(self):
- local_cas = self._connection.get_local_cas()
+ local_cas = self._casd.get_local_cas()
while not self._should_stop:
try:
diff --git a/src/buildstream/_cas/casdprocessmanager.py b/src/buildstream/_cas/casdprocessmanager.py
index 8fa9861a5..b0f75a4a4 100644
--- a/src/buildstream/_cas/casdprocessmanager.py
+++ b/src/buildstream/_cas/casdprocessmanager.py
@@ -109,6 +109,16 @@ class CASDProcessManager:
env=self.__buildbox_casd_env(),
)
+ self._casd_channel = None
+ self._bytestream = None
+ self._casd_cas = None
+ self._local_cas = None
+ self._asset_fetch = None
+ self._asset_push = None
+ self._shutdown_requested = False
+
+ self._lock = threading.Lock()
+
def __buildbox_casd(self):
return utils._get_host_tool_internal("buildbox-casd", search_subprojects_dir="buildbox")
@@ -259,6 +269,17 @@ class CASDProcessManager:
# Terminate the process and release related resources.
#
def release_resources(self, messenger=None):
+ self._shutdown_requested = True
+ with self._lock:
+ if self._casd_channel:
+ self._asset_push = None
+ self._asset_fetch = None
+ self._local_cas = None
+ self._casd_cas = None
+ self._bytestream = None
+ self._casd_channel.close()
+ self._casd_channel = None
+
self._terminate(messenger)
self.process = None
shutil.rmtree(self._socket_tempdir)
@@ -305,31 +326,6 @@ class CASDProcessManager:
"Buildbox-casd didn't exit cleanly. Exit code: {}, Logs: {}".format(return_code, self._logfile)
)
- # create_channel():
- #
- # Return a CASDChannel, note that the actual connection is not necessarily
- # established until it is needed.
- #
- def create_channel(self):
- return CASDChannel(self._socket_path, self._connection_string, self._start_time, self.process.pid)
-
-
-class CASDChannel:
- def __init__(self, socket_path, connection_string, start_time, casd_pid):
- self._socket_path = socket_path
- self._connection_string = connection_string
- self._start_time = start_time
- self._casd_channel = None
- self._bytestream = None
- self._casd_cas = None
- self._local_cas = None
- self._asset_fetch = None
- self._asset_push = None
- self._casd_pid = casd_pid
- self._shutdown_requested = False
-
- self._lock = threading.Lock()
-
def _establish_connection(self):
with self._lock:
if self._casd_channel is not None:
@@ -347,7 +343,7 @@ class CASDChannel:
# check that process is still alive
try:
- proc = psutil.Process(self._casd_pid)
+ proc = psutil.Process(self.process.pid)
if proc.status() == psutil.STATUS_ZOMBIE:
proc.wait()
@@ -409,38 +405,3 @@ class CASDChannel:
if self._casd_channel is None:
self._establish_connection()
return self._asset_push
-
- # is_closed():
- #
- # Return whether this connection is closed or not.
- #
- def is_closed(self):
- return self._casd_channel is None
-
- # request_shutdown():
- #
- # Notify the channel that a shutdown of casd was requested.
- #
- # Thus we know that not being able to establish a connection is expected
- # and no error will be reported in that case.
- def request_shutdown(self) -> None:
- self._shutdown_requested = True
-
- # close():
- #
- # Close the casd channel.
- #
- def close(self):
- assert self._shutdown_requested, "Please request shutdown before closing"
-
- with self._lock:
- if self.is_closed():
- return
-
- self._asset_push = None
- self._asset_fetch = None
- self._local_cas = None
- self._casd_cas = None
- self._bytestream = None
- self._casd_channel.close()
- self._casd_channel = None
diff --git a/src/buildstream/_cas/casserver.py b/src/buildstream/_cas/casserver.py
index 4e46b4fd6..6570df75f 100644
--- a/src/buildstream/_cas/casserver.py
+++ b/src/buildstream/_cas/casserver.py
@@ -90,10 +90,9 @@ def create_server(repo, *, enable_push, quota, index_only, log_level=LogLevel.Le
handler.setFormatter(logging.Formatter(fmt="%(levelname)s: %(funcName)s: %(message)s"))
logger.addHandler(handler)
- casd_manager = CASDProcessManager(
+ casd = CASDProcessManager(
os.path.abspath(repo), os.path.join(os.path.abspath(repo), "logs"), log_level, quota, None, False, None
)
- casd_channel = casd_manager.create_channel()
try:
# Use max_workers default from Python 3.5+
@@ -102,19 +101,19 @@ def create_server(repo, *, enable_push, quota, index_only, log_level=LogLevel.Le
if not index_only:
bytestream_pb2_grpc.add_ByteStreamServicer_to_server(
- _ByteStreamServicer(casd_channel, enable_push=enable_push), server
+ _ByteStreamServicer(casd, enable_push=enable_push), server
)
remote_execution_pb2_grpc.add_ContentAddressableStorageServicer_to_server(
- _ContentAddressableStorageServicer(casd_channel, enable_push=enable_push), server
+ _ContentAddressableStorageServicer(casd, enable_push=enable_push), server
)
remote_execution_pb2_grpc.add_CapabilitiesServicer_to_server(_CapabilitiesServicer(), server)
# Remote Asset API
- remote_asset_pb2_grpc.add_FetchServicer_to_server(_FetchServicer(casd_channel), server)
+ remote_asset_pb2_grpc.add_FetchServicer_to_server(_FetchServicer(casd), server)
if enable_push:
- remote_asset_pb2_grpc.add_PushServicer_to_server(_PushServicer(casd_channel), server)
+ remote_asset_pb2_grpc.add_PushServicer_to_server(_PushServicer(casd), server)
# Ensure we have the signal handler set for SIGTERM
# This allows threads from GRPC to call our methods that do register
@@ -123,9 +122,7 @@ def create_server(repo, *, enable_push, quota, index_only, log_level=LogLevel.Le
yield server
finally:
- casd_channel.request_shutdown()
- casd_channel.close()
- casd_manager.release_resources()
+ casd.release_resources()
class _ByteStreamServicer(bytestream_pb2_grpc.ByteStreamServicer):
diff --git a/src/buildstream/_stream.py b/src/buildstream/_stream.py
index 159436914..075f25bcb 100644
--- a/src/buildstream/_stream.py
+++ b/src/buildstream/_stream.py
@@ -1799,7 +1799,7 @@ class Stream:
self._session_start_callback()
self._running = True
- status = self._scheduler.run(self.queues, self._context.get_cascache().get_casd_process_manager())
+ status = self._scheduler.run(self.queues, self._context.get_cascache().get_casd())
self._running = False
if status == SchedStatus.ERROR:
diff --git a/src/buildstream/sandbox/_sandboxbuildboxrun.py b/src/buildstream/sandbox/_sandboxbuildboxrun.py
index 1987ebac1..42f206459 100644
--- a/src/buildstream/sandbox/_sandboxbuildboxrun.py
+++ b/src/buildstream/sandbox/_sandboxbuildboxrun.py
@@ -87,7 +87,7 @@ class SandboxBuildBoxRun(SandboxREAPI):
context = self._get_context()
cascache = context.get_cascache()
- casd_process_manager = cascache.get_casd_process_manager()
+ casd = cascache.get_casd()
with utils._tempnamedfile() as action_file, utils._tempnamedfile() as result_file:
action_file.write(action.SerializeToString())
@@ -95,7 +95,7 @@ class SandboxBuildBoxRun(SandboxREAPI):
buildbox_command = [
self.__buildbox_run(),
- "--remote={}".format(casd_process_manager._connection_string),
+ "--remote={}".format(casd._connection_string),
"--action={}".format(action_file.name),
"--action-result={}".format(result_file.name),
]
diff --git a/tests/artifactcache/pull.py b/tests/artifactcache/pull.py
index 0c9960c1a..114957b85 100644
--- a/tests/artifactcache/pull.py
+++ b/tests/artifactcache/pull.py
@@ -203,9 +203,6 @@ def test_pull_tree(cli, tmpdir, datafiles):
cli.remove_artifact_from_cache(project_dir, "target.bst")
# Assert that we are not cached locally anymore
- artifactcache.release_resources()
- cas._casd_channel.request_shutdown()
- cas.close_grpc_channels()
assert cli.get_element_state(project_dir, "target.bst") != "cached"
tree_digest = remote_execution_pb2.Digest(hash=tree_hash, size_bytes=tree_size)
(buildstream) 04/04: _assetcache.py: Use buildbox-casd as remote asset proxy, if supported
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
juergbi pushed a commit to branch juerg/buildbox-asset-remote
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit c1defd0dbe8affb00951bf752a19372358802b73
Author: Jürg Billeter <j...@bitron.ch>
AuthorDate: Fri May 31 14:07:52 2024 +0200
_assetcache.py: Use buildbox-casd as remote asset proxy, if supported
buildbox-casd is already used as CAS proxy. Using buildbox-casd for all
remote connections aims to improve robustness (e.g., consistent retry
behavior) and will allow adding support for token-based authentication.
---
src/buildstream/_assetcache.py | 60 ++++++++++++++++++++++++++++--------------
src/buildstream/_remote.py | 3 ---
2 files changed, 40 insertions(+), 23 deletions(-)
diff --git a/src/buildstream/_assetcache.py b/src/buildstream/_assetcache.py
index 6a4806341..6f9dcba7c 100644
--- a/src/buildstream/_assetcache.py
+++ b/src/buildstream/_assetcache.py
@@ -20,17 +20,20 @@ from typing import List, Dict, Tuple, Iterable, Optional
import grpc
from . import utils
-from ._cas import CASRemote, CASCache
+from ._cas import CASRemote, CASCache, CASDProcessManager
from ._exceptions import AssetCacheError, RemoteError
from ._remotespec import RemoteSpec, RemoteType
from ._remote import BaseRemote
from ._protos.build.bazel.remote.asset.v1 import remote_asset_pb2, remote_asset_pb2_grpc
+from ._protos.build.buildgrid import local_cas_pb2
from ._protos.google.rpc import code_pb2
class AssetRemote(BaseRemote):
- def __init__(self, spec):
+ def __init__(self, spec, casd):
super().__init__(spec)
+ self.casd = casd
+ self.instance_name = None
self.fetch_service = None
self.push_service = None
@@ -40,9 +43,24 @@ class AssetRemote(BaseRemote):
super().close()
def _configure_protocols(self):
- # set up remote asset stubs
- self.fetch_service = remote_asset_pb2_grpc.FetchStub(self.channel)
- self.push_service = remote_asset_pb2_grpc.PushStub(self.channel)
+ local_cas = self.casd.get_local_cas()
+ request = local_cas_pb2.GetInstanceNameForRemotesRequest()
+ request.remote_asset.CopyFrom(self.spec.to_localcas_remote())
+ try:
+ response = local_cas.GetInstanceNameForRemotes(request)
+ self.instance_name = response.instance_name
+ self.fetch_service = self.casd.get_asset_fetch()
+ self.push_service = self.casd.get_asset_push()
+ except grpc.RpcError as e:
+ if e.code() == grpc.StatusCode.UNIMPLEMENTED or e.code() == grpc.StatusCode.INVALID_ARGUMENT:
+ # buildbox-casd is too old to support asset-only remotes.
+ # Fall back to direct connection.
+ self.instance_name = self.spec.instance_name
+ self.channel = self.spec.open_channel()
+ self.fetch_service = remote_asset_pb2_grpc.FetchStub(self.channel)
+ self.push_service = remote_asset_pb2_grpc.PushStub(self.channel)
+ else:
+ raise
# _check():
#
@@ -55,8 +73,8 @@ class AssetRemote(BaseRemote):
#
def _check(self):
request = remote_asset_pb2.FetchBlobRequest()
- if self.spec.instance_name:
- request.instance_name = self.spec.instance_name
+ if self.instance_name:
+ request.instance_name = self.instance_name
try:
self.fetch_service.FetchBlob(request)
@@ -74,8 +92,8 @@ class AssetRemote(BaseRemote):
if self.spec.push:
request = remote_asset_pb2.PushBlobRequest()
- if self.spec.instance_name:
- request.instance_name = self.spec.instance_name
+ if self.instance_name:
+ request.instance_name = self.instance_name
try:
self.push_service.PushBlob(request)
@@ -112,8 +130,8 @@ class AssetRemote(BaseRemote):
#
def fetch_blob(self, uris, *, qualifiers=None):
request = remote_asset_pb2.FetchBlobRequest()
- if self.spec.instance_name:
- request.instance_name = self.spec.instance_name
+ if self.instance_name:
+ request.instance_name = self.instance_name
request.uris.extend(uris)
if qualifiers:
request.qualifiers.extend(qualifiers)
@@ -153,8 +171,8 @@ class AssetRemote(BaseRemote):
#
def fetch_directory(self, uris, *, qualifiers=None):
request = remote_asset_pb2.FetchDirectoryRequest()
- if self.spec.instance_name:
- request.instance_name = self.spec.instance_name
+ if self.instance_name:
+ request.instance_name = self.instance_name
request.uris.extend(uris)
if qualifiers:
request.qualifiers.extend(qualifiers)
@@ -194,8 +212,8 @@ class AssetRemote(BaseRemote):
#
def push_blob(self, uris, blob_digest, *, qualifiers=None, references_blobs=None, references_directories=None):
request = remote_asset_pb2.PushBlobRequest()
- if self.spec.instance_name:
- request.instance_name = self.spec.instance_name
+ if self.instance_name:
+ request.instance_name = self.instance_name
request.uris.extend(uris)
request.blob_digest.CopyFrom(blob_digest)
if qualifiers:
@@ -231,8 +249,8 @@ class AssetRemote(BaseRemote):
self, uris, directory_digest, *, qualifiers=None, references_blobs=None, references_directories=None
):
request = remote_asset_pb2.PushDirectoryRequest()
- if self.spec.instance_name:
- request.instance_name = self.spec.instance_name
+ if self.instance_name:
+ request.instance_name = self.instance_name
request.uris.extend(uris)
request.root_directory_digest.CopyFrom(directory_digest)
if qualifiers:
@@ -262,14 +280,14 @@ class AssetRemote(BaseRemote):
# to establish a connection to this remote at initialization time.
#
class RemotePair:
- def __init__(self, cas: CASCache, spec: RemoteSpec):
+ def __init__(self, casd: CASDProcessManager, cas: CASCache, spec: RemoteSpec):
self.index: Optional[AssetRemote] = None
self.storage: Optional[CASRemote] = None
self.error: Optional[str] = None
try:
if spec.remote_type in [RemoteType.INDEX, RemoteType.ALL]:
- index = AssetRemote(spec)
+ index = AssetRemote(spec, casd)
index.check()
self.index = index
if spec.remote_type in [RemoteType.STORAGE, RemoteType.ALL]:
@@ -324,6 +342,8 @@ class AssetCache:
# Hold on to the project specs
self._project_specs = project_specs
+ casd = self.context.get_casd()
+
for spec in specs:
# This can be called multiple times, ensure that we only try
# to instantiate each remote once.
@@ -331,7 +351,7 @@ class AssetCache:
if spec in self._remotes:
continue
- remote = RemotePair(self.cas, spec)
+ remote = RemotePair(casd, self.cas, spec)
if remote.error:
self.context.messenger.warn("Failed to initialize remote {}: {}".format(spec.url, remote.error))
diff --git a/src/buildstream/_remote.py b/src/buildstream/_remote.py
index 453f7f2a0..32584de09 100644
--- a/src/buildstream/_remote.py
+++ b/src/buildstream/_remote.py
@@ -68,9 +68,6 @@ class BaseRemote:
if self._initialized:
return
- if self.spec:
- self.channel = self.spec.open_channel()
-
self._configure_protocols()
self._initialized = True
(buildstream) 02/04: Move `CASDProcessManager` instance from `CASCache` to `Context`
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
juergbi pushed a commit to branch juerg/buildbox-asset-remote
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit e1fbeaebff873c9fc7f29585723ce041fe2dd901
Author: Jürg Billeter <j...@bitron.ch>
AuthorDate: Fri May 31 12:52:16 2024 +0200
Move `CASDProcessManager` instance from `CASCache` to `Context`
---
src/buildstream/_cas/__init__.py | 1 +
src/buildstream/_cas/cascache.py | 37 ++++-----------------
src/buildstream/_cas/casdprocessmanager.py | 2 ++
src/buildstream/_context.py | 32 ++++++++++++------
src/buildstream/_testing/runcli.py | 2 +-
tests/artifactcache/expiry.py | 8 ++---
tests/internals/cascache.py | 27 +++++++--------
tests/internals/storage.py | 8 ++---
tests/internals/storage_vdir_import.py | 38 +++++-----------------
tests/testutils/__init__.py | 1 +
tests/testutils/artifactshare.py | 2 +-
.../_cas/__init__.py => tests/testutils/casd.py | 24 +++++++++++---
12 files changed, 78 insertions(+), 104 deletions(-)
diff --git a/src/buildstream/_cas/__init__.py b/src/buildstream/_cas/__init__.py
index 5f9c4abb1..52f425d56 100644
--- a/src/buildstream/_cas/__init__.py
+++ b/src/buildstream/_cas/__init__.py
@@ -15,4 +15,5 @@
# Tristan Van Berkom <tr...@codethink.co.uk>
from .cascache import CASCache, CASLogLevel
+from .casdprocessmanager import CASDProcessManager
from .casremote import CASRemote
diff --git a/src/buildstream/_cas/cascache.py b/src/buildstream/_cas/cascache.py
index 8110337bc..627778c16 100644
--- a/src/buildstream/_cas/cascache.py
+++ b/src/buildstream/_cas/cascache.py
@@ -32,7 +32,6 @@ from .. import utils
from ..types import FastEnum, SourceRef
from .._exceptions import CASCacheError
-from .casdprocessmanager import CASDProcessManager
from .casremote import CASRemote, _CASBatchRead, _CASBatchUpdate, BlobNotFound
_BUFFER_SIZE = 65536
@@ -52,25 +51,11 @@ class CASLogLevel(FastEnum):
#
# Args:
# path (str): The root directory for the CAS repository
-# casd (bool): True to spawn buildbox-casd (default) or False (testing only)
-# cache_quota (int): User configured cache quota
-# protect_session_blobs (bool): Disable expiry for blobs used in the current session
-# log_level (LogLevel): Log level to give to buildbox-casd for logging
-# log_directory (str): the root of the directory in which to store logs
+# casd (CASDProcessManager): The buildbox-casd manager
+# remote_cache (bool): True if a CAS server is configured as a remote cache (storage-service)
#
class CASCache:
- def __init__(
- self,
- path,
- *,
- casd=True,
- cache_quota=None,
- remote_cache_spec=None,
- protect_session_blobs=True,
- log_level=CASLogLevel.WARNING,
- log_directory=None,
- messenger=None
- ):
+ def __init__(self, path, *, casd, remote_cache=False):
self.casdir = os.path.join(path, "cas")
self.tmpdir = os.path.join(path, "tmp")
os.makedirs(self.tmpdir, exist_ok=True)
@@ -78,16 +63,10 @@ class CASCache:
self._cache_usage_monitor = None
self._cache_usage_monitor_forbidden = False
- self._remote_cache = bool(remote_cache_spec)
+ self._remote_cache = remote_cache
- self._casd = None
+ self._casd = casd
if casd:
- assert log_directory is not None, "log_directory is required when casd is True"
- log_dir = os.path.join(log_directory, "_casd")
- self._casd = CASDProcessManager(
- path, log_dir, log_level, cache_quota, remote_cache_spec, protect_session_blobs, messenger
- )
-
self._cache_usage_monitor = _CASCacheUsageMonitor(self._casd)
self._cache_usage_monitor.start()
else:
@@ -124,15 +103,11 @@ class CASCache:
#
# Release resources used by CASCache.
#
- def release_resources(self, messenger=None):
+ def release_resources(self):
if self._cache_usage_monitor:
self._cache_usage_monitor.stop()
self._cache_usage_monitor.join()
- if self._casd:
- self._casd.release_resources(messenger)
- self._casd = None
-
def get_default_remote(self):
return self._default_remote
diff --git a/src/buildstream/_cas/casdprocessmanager.py b/src/buildstream/_cas/casdprocessmanager.py
index b0f75a4a4..7093880c2 100644
--- a/src/buildstream/_cas/casdprocessmanager.py
+++ b/src/buildstream/_cas/casdprocessmanager.py
@@ -63,6 +63,8 @@ _REQUIRED_CASD_MICRO = 58
#
class CASDProcessManager:
def __init__(self, path, log_dir, log_level, cache_quota, remote_cache_spec, protect_session_blobs, messenger):
+ os.makedirs(path, exist_ok=True)
+
self._log_dir = log_dir
self._socket_path = self._make_socket_path(path)
diff --git a/src/buildstream/_context.py b/src/buildstream/_context.py
index 545e5c37e..31b2df85f 100644
--- a/src/buildstream/_context.py
+++ b/src/buildstream/_context.py
@@ -30,7 +30,7 @@ from ._artifactcache import ArtifactCache
from ._elementsourcescache import ElementSourcesCache
from ._remotespec import RemoteSpec, RemoteExecutionSpec
from ._sourcecache import SourceCache
-from ._cas import CASCache, CASLogLevel
+from ._cas import CASCache, CASDProcessManager, CASLogLevel
from .types import _CacheBuildTrees, _PipelineSelection, _SchedulerErrorAction, _SourceUriPolicy
from ._workspaces import Workspaces, WorkspaceProjectCache
from .node import Node, MappingNode
@@ -223,6 +223,7 @@ class Context:
self._project_overrides: MappingNode = Node.from_dict({})
self._workspaces: Optional[Workspaces] = None
self._workspace_project_cache: WorkspaceProjectCache = WorkspaceProjectCache()
+ self._casd: Optional[CASDProcessManager] = None
self._cascache: Optional[CASCache] = None
# __enter__()
@@ -247,7 +248,11 @@ class Context:
self._sourcecache.release_resources()
if self._cascache:
- self._cascache.release_resources(self.messenger)
+ self._cascache.release_resources()
+
+ if self._casd:
+ self._casd.release_resources(self.messenger)
+ self._casd = None
# load()
#
@@ -677,8 +682,8 @@ class Context:
# value which we cache here too.
return self._strict_build_plan
- def get_cascache(self) -> CASCache:
- if self._cascache is None:
+ def get_casd(self) -> CASDProcessManager:
+ if self._casd is None:
if self.log_debug:
log_level = CASLogLevel.TRACE
elif self.log_verbose:
@@ -686,15 +691,22 @@ class Context:
else:
log_level = CASLogLevel.WARNING
- self._cascache = CASCache(
+ assert self.logdir is not None, "log_directory is required for casd"
+ log_dir = os.path.join(self.logdir, "_casd")
+ self._casd = CASDProcessManager(
self.cachedir,
- casd=self.use_casd,
- cache_quota=self.config_cache_quota,
- remote_cache_spec=self.remote_cache_spec,
- log_level=log_level,
- log_directory=self.logdir,
+ log_dir,
+ log_level,
+ self.config_cache_quota,
+ self.remote_cache_spec,
+ protect_session_blobs=True,
messenger=self.messenger,
)
+ return self._casd
+
+ def get_cascache(self) -> CASCache:
+ if self._cascache is None:
+ self._cascache = CASCache(self.cachedir, casd=self.get_casd(), remote_cache=bool(self.remote_cache_spec))
return self._cascache
######################################################
diff --git a/src/buildstream/_testing/runcli.py b/src/buildstream/_testing/runcli.py
index 8daaf08b6..87682f009 100644
--- a/src/buildstream/_testing/runcli.py
+++ b/src/buildstream/_testing/runcli.py
@@ -725,7 +725,7 @@ class TestArtifact:
def _extract_subdirectory(self, tmpdir, digest):
with tempfile.TemporaryDirectory() as extractdir:
try:
- cas = CASCache(str(tmpdir), casd=False)
+ cas = CASCache(str(tmpdir), casd=None)
cas.checkout(extractdir, digest)
yield extractdir
except FileNotFoundError:
diff --git a/tests/artifactcache/expiry.py b/tests/artifactcache/expiry.py
index 3fd312174..629e34601 100644
--- a/tests/artifactcache/expiry.py
+++ b/tests/artifactcache/expiry.py
@@ -22,20 +22,18 @@ import time
import pytest
-from buildstream._cas import CASCache
from buildstream.exceptions import ErrorDomain, LoadErrorReason
from buildstream._testing import cli # pylint: disable=unused-import
from buildstream._testing._utils.site import have_subsecond_mtime
-from tests.testutils import create_element_size, wait_for_cache_granularity
+from tests.testutils import casd_cache, create_element_size, wait_for_cache_granularity
DATA_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "expiry")
def get_cache_usage(directory):
- cas_cache = CASCache(directory, log_directory=os.path.dirname(directory))
- try:
+ with casd_cache(directory) as cas_cache:
wait = 0.1
for _ in range(0, int(5 / wait)):
used_size = cas_cache.get_cache_usage().used_size
@@ -45,8 +43,6 @@ def get_cache_usage(directory):
assert False, "Unable to retrieve cache usage"
return None
- finally:
- cas_cache.release_resources()
# Ensure that the cache successfully removes an old artifact if we do
diff --git a/tests/internals/cascache.py b/tests/internals/cascache.py
index 23f7575c4..6c51f9045 100644
--- a/tests/internals/cascache.py
+++ b/tests/internals/cascache.py
@@ -15,9 +15,9 @@ import os
import time
from unittest.mock import MagicMock
-from buildstream._cas.cascache import CASCache
from buildstream._cas import casdprocessmanager
from buildstream._messenger import Messenger
+from tests.testutils import casd_cache
#
@@ -42,9 +42,8 @@ def test_report_when_cascache_dies_before_asked_to(tmp_path, monkeypatch):
monkeypatch.setenv("PATH", str(tmp_path), prepend=os.pathsep)
messenger = MagicMock(spec_set=Messenger)
- cache = CASCache(str(tmp_path.joinpath("casd")), casd=True, log_directory=str(tmp_path.joinpath("logs")))
- time.sleep(1)
- cache.release_resources(messenger)
+ with casd_cache(tmp_path.joinpath("casd"), messenger) as cache:
+ time.sleep(1)
assert messenger.bug.call_count == 1
@@ -64,9 +63,8 @@ def test_report_when_cascache_exits_not_cleanly(tmp_path, monkeypatch):
monkeypatch.setattr(casdprocessmanager, "_CASD_TIMEOUT", 0.1)
messenger = MagicMock(spec_set=Messenger)
- cache = CASCache(str(tmp_path.joinpath("casd")), casd=True, log_directory=str(tmp_path.joinpath("logs")))
- time.sleep(1)
- cache.release_resources(messenger)
+ with casd_cache(tmp_path.joinpath("casd"), messenger) as cache:
+ time.sleep(1)
assert messenger.bug.call_count == 1
@@ -86,9 +84,8 @@ def test_report_when_cascache_is_forcefully_killed(tmp_path, monkeypatch):
monkeypatch.setattr(casdprocessmanager, "_CASD_TIMEOUT", 0.1)
messenger = MagicMock(spec_set=Messenger)
- cache = CASCache(str(tmp_path.joinpath("casd")), casd=True, log_directory=str(tmp_path.joinpath("logs")))
- time.sleep(1)
- cache.release_resources(messenger)
+ with casd_cache(tmp_path.joinpath("casd"), messenger) as cache:
+ time.sleep(1)
assert messenger.warn.call_count == 1
@@ -114,9 +111,8 @@ def test_casd_redirects_stderr_to_file_and_rotate(tmp_path, monkeypatch):
# Let's create the first `n_max_log_files` log files
for i in range(1, n_max_log_files + 1):
- cache = CASCache(str(casd_files_path), casd=True, log_directory=str(casd_parent_logs_path))
- time.sleep(0.5)
- cache.release_resources()
+ with casd_cache(casd_files_path) as cache:
+ time.sleep(0.5)
existing_log_files = sorted(casd_logs_path.iterdir())
assert len(existing_log_files) == i
@@ -126,9 +122,8 @@ def test_casd_redirects_stderr_to_file_and_rotate(tmp_path, monkeypatch):
for _ in range(3):
evicted_file = existing_log_files.pop(0)
- cache = CASCache(str(casd_files_path), casd=True, log_directory=str(casd_parent_logs_path))
- time.sleep(0.5)
- cache.release_resources()
+ with casd_cache(casd_files_path) as cache:
+ time.sleep(0.5)
existing_log_files = sorted(casd_logs_path.iterdir())
assert len(existing_log_files) == n_max_log_files
diff --git a/tests/internals/storage.py b/tests/internals/storage.py
index 49bb7043c..f3b4031cb 100644
--- a/tests/internals/storage.py
+++ b/tests/internals/storage.py
@@ -24,10 +24,11 @@ from typing import List, Optional
import pytest
from buildstream import DirectoryError, FileType
-from buildstream._cas import CASCache
from buildstream.storage._casbaseddirectory import CasBasedDirectory
from buildstream.storage._filebaseddirectory import FileBasedDirectory
+from tests.testutils import casd_cache
+
DATA_DIR = os.path.join(os.path.dirname(os.path.realpath(__file__)), "storage")
@@ -38,11 +39,8 @@ def setup_backend(backend_class, tmpdir):
os.mkdir(path)
yield backend_class(path)
else:
- cas_cache = CASCache(os.path.join(tmpdir, "cas"), log_directory=os.path.join(tmpdir, "logs"))
- try:
+ with casd_cache(os.path.join(tmpdir, "cas")) as cas_cache:
yield backend_class(cas_cache)
- finally:
- cas_cache.release_resources()
@pytest.mark.parametrize("backend", [FileBasedDirectory, CasBasedDirectory])
diff --git a/tests/internals/storage_vdir_import.py b/tests/internals/storage_vdir_import.py
index e0ca134b6..637283e69 100644
--- a/tests/internals/storage_vdir_import.py
+++ b/tests/internals/storage_vdir_import.py
@@ -20,10 +20,11 @@ import pytest
from buildstream import DirectoryError
from buildstream.storage._casbaseddirectory import CasBasedDirectory
from buildstream.storage._filebaseddirectory import FileBasedDirectory
-from buildstream._cas import CASCache
from buildstream.utils import _set_file_mtime
from buildstream._testing._utils.site import have_subsecond_mtime
+from tests.testutils import casd_cache
+
# These are comparitive tests that check that FileBasedDirectory and
# CasBasedDirectory act identically.
@@ -189,8 +190,7 @@ def _import_test(tmpdir, original, overlay, generator_function, verify_contents=
if not have_subsecond_mtime(str(tmpdir)):
pytest.skip("Filesystem does not support subsecond mtime precision: {}".format(str(tmpdir)))
- cas_cache = CASCache(tmpdir, log_directory=os.path.join(tmpdir, "logs"))
- try:
+ with casd_cache(os.path.join(tmpdir, "casd")) as cas_cache:
# Create some fake content
generator_function(original, tmpdir)
if original != overlay:
@@ -245,8 +245,6 @@ def _import_test(tmpdir, original, overlay, generator_function, verify_contents=
duplicate_cas._import_files_internal(roundtrip_dir, properties=["mtime"])
assert duplicate_cas._get_digest().hash == d._get_digest().hash
- finally:
- cas_cache.release_resources()
@pytest.mark.parametrize("original", range(1, len(root_filesets) + 1))
@@ -262,8 +260,7 @@ def test_random_cas_import(tmpdir, original, overlay):
def _listing_test(tmpdir, root, generator_function):
- cas_cache = CASCache(tmpdir, log_directory=os.path.join(tmpdir, "logs"))
- try:
+ with casd_cache(os.path.join(tmpdir, "casd")) as cas_cache:
# Create some fake content
generator_function(root, tmpdir)
@@ -274,8 +271,6 @@ def _listing_test(tmpdir, root, generator_function):
filelist2 = list(d2.list_relative_paths())
assert filelist == filelist2
- finally:
- cas_cache.release_resources()
@pytest.mark.parametrize("root", range(1, NUM_RANDOM_TESTS + 1))
@@ -291,8 +286,7 @@ def test_fixed_directory_listing(tmpdir, root):
# Check that the vdir is decending and readable
def test_open_directory(tmpdir):
cas_dir = os.path.join(str(tmpdir), "cas")
- cas_cache = CASCache(cas_dir, log_directory=os.path.join(str(tmpdir), "logs"))
- try:
+ with casd_cache(cas_dir) as cas_cache:
d = CasBasedDirectory(cas_cache)
Content_to_check = "You got me"
@@ -306,8 +300,6 @@ def test_open_directory(tmpdir):
with open(cas_cache.objpath(digest), encoding="utf-8") as fp:
content = fp.read()
assert Content_to_check == content
- finally:
- cas_cache.release_resources()
# Check symlink logic for edgecases
@@ -315,8 +307,7 @@ def test_open_directory(tmpdir):
# to decend in to files or links to files
def test_bad_symlinks(tmpdir):
cas_dir = os.path.join(str(tmpdir), "cas")
- cas_cache = CASCache(cas_dir, log_directory=os.path.join(str(tmpdir), "logs"))
- try:
+ with casd_cache(cas_dir) as cas_cache:
d = CasBasedDirectory(cas_cache)
test_dir = os.path.join(str(tmpdir), "importfrom")
@@ -336,16 +327,13 @@ def test_bad_symlinks(tmpdir):
with pytest.raises(DirectoryError) as error:
d.open_directory("a/f")
assert error.reason == exp_reason
- finally:
- cas_cache.release_resources()
# Check symlink logic for edgecases
# Check decend accross relitive link
def test_relative_symlink(tmpdir):
cas_dir = os.path.join(str(tmpdir), "cas")
- cas_cache = CASCache(cas_dir, log_directory=os.path.join(str(tmpdir), "logs"))
- try:
+ with casd_cache(cas_dir) as cas_cache:
d = CasBasedDirectory(cas_cache)
Content_to_check = "You got me"
@@ -363,16 +351,13 @@ def test_relative_symlink(tmpdir):
with open(cas_cache.objpath(digest), encoding="utf-8") as fp:
content = fp.read()
assert Content_to_check == content
- finally:
- cas_cache.release_resources()
# Check symlink logic for edgecases
# Check deccend accross abs link
def test_abs_symlink(tmpdir):
cas_dir = os.path.join(str(tmpdir), "cas")
- cas_cache = CASCache(cas_dir, log_directory=os.path.join(str(tmpdir), "logs"))
- try:
+ with casd_cache(cas_dir) as cas_cache:
d = CasBasedDirectory(cas_cache)
Content_to_check = "two step file"
@@ -391,16 +376,13 @@ def test_abs_symlink(tmpdir):
with open(cas_cache.objpath(digest), encoding="utf-8") as fp:
content = fp.read()
assert Content_to_check == content
- finally:
- cas_cache.release_resources()
# Check symlink logic for edgecases
# Check symlink can not escape root
def test_bad_sym_escape(tmpdir):
cas_dir = os.path.join(str(tmpdir), "cas")
- cas_cache = CASCache(cas_dir, log_directory=os.path.join(str(tmpdir), "logs"))
- try:
+ with casd_cache(cas_dir) as cas_cache:
d = CasBasedDirectory(cas_cache)
test_dir = os.path.join(str(tmpdir), "importfrom")
@@ -417,5 +399,3 @@ def test_bad_sym_escape(tmpdir):
with pytest.raises(DirectoryError) as error:
d.open_directory("a/l", follow_symlinks=True)
assert error.reason == "directory-not-found"
- finally:
- cas_cache.release_resources()
diff --git a/tests/testutils/__init__.py b/tests/testutils/__init__.py
index 028131e4b..03eb422ea 100644
--- a/tests/testutils/__init__.py
+++ b/tests/testutils/__init__.py
@@ -20,6 +20,7 @@
#
from .artifactshare import create_artifact_share, create_split_share, assert_shared, assert_not_shared, ArtifactShare
+from .casd import casd_cache
from .context import dummy_context
from .element_generators import create_element_size
from .junction import generate_junction
diff --git a/tests/testutils/artifactshare.py b/tests/testutils/artifactshare.py
index 515164aac..fcdbcb597 100644
--- a/tests/testutils/artifactshare.py
+++ b/tests/testutils/artifactshare.py
@@ -142,7 +142,7 @@ class ArtifactShare(BaseArtifactShare):
super().__init__()
# Set after subprocess creation as it's not picklable
- self.cas = CASCache(self.repodir, casd=False)
+ self.cas = CASCache(self.repodir, casd=None)
def _create_server(self):
return create_server(
diff --git a/src/buildstream/_cas/__init__.py b/tests/testutils/casd.py
similarity index 50%
copy from src/buildstream/_cas/__init__.py
copy to tests/testutils/casd.py
index 5f9c4abb1..9a3eb8e71 100644
--- a/src/buildstream/_cas/__init__.py
+++ b/tests/testutils/casd.py
@@ -10,9 +10,23 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-#
-# Authors:
-# Tristan Van Berkom <tr...@codethink.co.uk>
-from .cascache import CASCache, CASLogLevel
-from .casremote import CASRemote
+import os
+from contextlib import contextmanager
+
+from buildstream._cas import CASCache, CASDProcessManager, CASLogLevel
+
+
+@contextmanager
+def casd_cache(path, messenger=None):
+ casd = CASDProcessManager(
+ str(path), os.path.join(str(path), "..", "logs", "_casd"), CASLogLevel.WARNING, None, None, True, None
+ )
+ try:
+ cascache = CASCache(str(path), casd=casd)
+ try:
+ yield cascache
+ finally:
+ cascache.release_resources()
+ finally:
+ casd.release_resources(messenger)
(buildstream) 03/04: _remotespec.py: Add `to_localcas_remote()` method
Posted by ju...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
juergbi pushed a commit to branch juerg/buildbox-asset-remote
in repository https://gitbox.apache.org/repos/asf/buildstream.git
commit 1c58181b5394246af3b40ae0a5a96ef60fd963f0
Author: Jürg Billeter <j...@bitron.ch>
AuthorDate: Fri May 31 14:14:08 2024 +0200
_remotespec.py: Add `to_localcas_remote()` method
---
src/buildstream/_cas/casremote.py | 13 +------------
src/buildstream/_remotespec.py | 20 ++++++++++++++++++++
2 files changed, 21 insertions(+), 12 deletions(-)
diff --git a/src/buildstream/_cas/casremote.py b/src/buildstream/_cas/casremote.py
index f4fbe1223..a6cae7c2f 100644
--- a/src/buildstream/_cas/casremote.py
+++ b/src/buildstream/_cas/casremote.py
@@ -59,18 +59,7 @@ class CASRemote(BaseRemote):
local_cas = self.cascache.get_local_cas()
request = local_cas_pb2.GetInstanceNameForRemotesRequest()
- cas_endpoint = request.content_addressable_storage
- cas_endpoint.url = self.spec.url
- if self.spec.instance_name:
- cas_endpoint.instance_name = self.spec.instance_name
- if self.spec.server_cert:
- cas_endpoint.server_cert = self.spec.server_cert
- if self.spec.client_key:
- cas_endpoint.client_key = self.spec.client_key
- if self.spec.client_cert:
- cas_endpoint.client_cert = self.spec.client_cert
- if self.spec.keepalive_time is not None:
- cas_endpoint.keepalive_time.FromSeconds(self.spec.keepalive_time)
+ request.content_addressable_storage.CopyFrom(self.spec.to_localcas_remote())
try:
response = local_cas.GetInstanceNameForRemotes(request)
except grpc.RpcError as e:
diff --git a/src/buildstream/_remotespec.py b/src/buildstream/_remotespec.py
index 13da04dbb..48a6c0c9f 100644
--- a/src/buildstream/_remotespec.py
+++ b/src/buildstream/_remotespec.py
@@ -20,6 +20,7 @@ from grpc import ChannelCredentials, Channel
from ._exceptions import LoadError, RemoteError
from .exceptions import LoadErrorReason
+from ._protos.build.buildgrid import local_cas_pb2
from .types import FastEnum
from .node import MappingNode
@@ -218,6 +219,25 @@ class RemoteSpec:
return channel
+ # to_localcas_remote()
+ #
+ # Create a `LocalContentAddressableStorage.Remote` proto from the `RemoteSpec` object.
+ #
+ def to_localcas_remote(self) -> local_cas_pb2.Remote:
+ remote = local_cas_pb2.Remote()
+ remote.url = self.url
+ if self.instance_name:
+ remote.instance_name = self.instance_name
+ if self.server_cert:
+ remote.server_cert = self.server_cert
+ if self.client_key:
+ remote.client_key = self.client_key
+ if self.client_cert:
+ remote.client_cert = self.client_cert
+ if self.keepalive_time is not None:
+ remote.keepalive_time.FromSeconds(self.keepalive_time)
+ return remote
+
# new_from_node():
#
# Creates a RemoteSpec() from a YAML loaded node.